[ViewVC] Diff of: cvs/IO-AIO/AIO.xs

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.9 by root, Tue Jul 12 11:02:54 2005 UTC vs.
Revision 1.19 by root, Sun Jul 31 19:04:45 2005 UTC

…		…
		1	#ifndef _REENTRANT
		2	# define _REENTRANT 1
		3	#endif
		4	#include <errno.h>
		5
1	#include "EXTERN.h"	6	#include "EXTERN.h"
2	#include "perl.h"	7	#include "perl.h"
3	#include "XSUB.h"	8	#include "XSUB.h"
4		9
		10	#include "autoconf/config.h"
		11
5	#include <sys/types.h>	12	#include <sys/types.h>
6	#include <sys/stat.h>	13	#include <sys/stat.h>
		14
7	#include <unistd.h>	15	#include <unistd.h>
8	#include <fcntl.h>	16	#include <fcntl.h>
9	#include <signal.h>	17	#include <signal.h>
10	#include <sched.h>	18	#include <sched.h>
11		19
12	#include <pthread.h>	20	#include <pthread.h>
13	#include <sys/syscall.h>
14		21
15	typedef void InputStream; / hack, but 5.6.1 is simply toooo old ;) */	22	typedef void InputStream; / hack, but 5.6.1 is simply toooo old ;) */
16	typedef void OutputStream; / hack, but 5.6.1 is simply toooo old ;) */	23	typedef void OutputStream; / hack, but 5.6.1 is simply toooo old ;) */
17	typedef void InOutStream; / hack, but 5.6.1 is simply toooo old ;) */	24	typedef void InOutStream; / hack, but 5.6.1 is simply toooo old ;) */
18		25
39	off_t offset;	46	off_t offset;
40	size_t length;	47	size_t length;
41	ssize_t result;	48	ssize_t result;
42	mode_t mode; /* open */	49	mode_t mode; /* open */
43	int errorno;	50	int errorno;
44	SV data, callback;	51	SV data, callback, *fh;
45	void *dataptr;	52	void *dataptr;
46	STRLEN dataoffset;	53	STRLEN dataoffset;
47		54
48	Stat_t *statdata;	55	Stat_t *statdata;
49	} aio_cb;	56	} aio_cb;
50		57
51	typedef aio_cb *aio_req;	58	typedef aio_cb *aio_req;
52		59
53	static int started;	60	static int started;
54	static int nreqs;	61	static volatile int nreqs;
55	static int max_outstanding = 1<<30;	62	static int max_outstanding = 1<<30;
56	static int respipe [2];	63	static int respipe [2];
57		64
58	static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER;	65	static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER;
59	static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER;	66	static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER;
…		…
63	static volatile aio_req ress, rese; /* queue start, queue end */	70	static volatile aio_req ress, rese; /* queue start, queue end */
64		71
65	static void	72	static void
66	poll_wait ()	73	poll_wait ()
67	{	74	{
68	if (!nreqs)	75	if (nreqs && !ress)
69	return;	76	{
70
71	fd_set rfd;	77	fd_set rfd;
72	FD_ZERO(&rfd);	78	FD_ZERO(&rfd);
73	FD_SET(respipe [0], &rfd);	79	FD_SET(respipe [0], &rfd);
74		80
75	select (respipe [0] + 1, &rfd, 0, 0, 0);	81	select (respipe [0] + 1, &rfd, 0, 0, 0);
		82	}
76	}	83	}
77		84
78	static int	85	static int
79	poll_cb ()	86	poll_cb ()
80	{	87	{
81	dSP;	88	dSP;
82	int count = 0;	89	int count = 0;
83	aio_req req;	90	aio_req req, prv;
84		91
		92	pthread_mutex_lock (&reslock);
		93
85	{	94	{
86	/* read and signals sent by the worker threads */	95	/* read any signals sent by the worker threads */
87	char buf [32];	96	char buf [32];
88	while (read (respipe [0], buf, 32) > 0)	97	while (read (respipe [0], buf, 32) > 0)
89	;	98	;
90	}	99	}
91		100
92	for (;;)	101	req = ress;
		102	ress = rese = 0;
		103
		104	pthread_mutex_unlock (&reslock);
		105
		106	while (req)
93	{	107	{
94	pthread_mutex_lock (&reslock);
95
96	req = ress;
97
98	if (ress)
99	{
100	ress = ress->next;
101	if (!ress) rese = 0;
102	}
103
104	pthread_mutex_unlock (&reslock);
105
106	if (!req)
107	break;
108
109	nreqs--;	108	nreqs--;
110		109
111	if (req->type == REQ_QUIT)	110	if (req->type == REQ_QUIT)
112	started--;	111	started--;
113	else	112	else
…		…
120	+ req->result > 0 ? req->result : 0);	119	+ req->result > 0 ? req->result : 0);
121		120
122	if (req->data)	121	if (req->data)
123	SvREFCNT_dec (req->data);	122	SvREFCNT_dec (req->data);
124		123
		124	if (req->fh)
		125	SvREFCNT_dec (req->fh);
		126
125	if (req->type == REQ_STAT \|\| req->type == REQ_LSTAT \|\| req->type == REQ_FSTAT)	127	if (req->type == REQ_STAT \|\| req->type == REQ_LSTAT \|\| req->type == REQ_FSTAT)
126	{	128	{
127	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;	129	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
128	PL_laststatval = req->result;	130	PL_laststatval = req->result;
129	PL_statcache = *(req->statdata);	131	PL_statcache = *(req->statdata);
130		132
131	Safefree (req->statdata);	133	Safefree (req->statdata);
132	}	134	}
133		135
		136	ENTER;
134	PUSHMARK (SP);	137	PUSHMARK (SP);
135	XPUSHs (sv_2mortal (newSViv (req->result)));	138	XPUSHs (sv_2mortal (newSViv (req->result)));
136		139
137	if (req->type == REQ_OPEN)	140	if (req->type == REQ_OPEN)
138	{	141	{
…		…
141		144
142	PUTBACK;	145	PUTBACK;
143	call_pv ("IO::AIO::_fd2fh", G_SCALAR \| G_EVAL);	146	call_pv ("IO::AIO::_fd2fh", G_SCALAR \| G_EVAL);
144	SPAGAIN;	147	SPAGAIN;
145		148
146	fh = POPs;	149	fh = SvREFCNT_inc (POPs);
147		150
148	PUSHMARK (SP);	151	PUSHMARK (SP);
149	XPUSHs (fh);	152	XPUSHs (sv_2mortal (fh));
150	}	153	}
151		154
152	if (SvOK (req->callback))	155	if (SvOK (req->callback))
153	{	156	{
154	PUTBACK;	157	PUTBACK;
155	call_sv (req->callback, G_VOID \| G_EVAL);	158	call_sv (req->callback, G_VOID \| G_EVAL);
156	SPAGAIN;	159	SPAGAIN;
157	}	160	}
		161
		162	LEAVE;
158		163
159	if (req->callback)	164	if (req->callback)
160	SvREFCNT_dec (req->callback);	165	SvREFCNT_dec (req->callback);
161		166
162	errno = errorno;	167	errno = errorno;
163	count++;	168	count++;
164	}	169	}
165		170
		171	prv = req;
		172	req = req->next;
166	Safefree (req);	173	Safefree (prv);
		174
		175	/* TODO: croak on errors? */
167	}	176	}
168		177
169	return count;	178	return count;
170	}	179	}
171		180
…		…
226	req->type = REQ_QUIT;	235	req->type = REQ_QUIT;
227		236
228	send_req (req);	237	send_req (req);
229	}	238	}
230		239
231	static void	240	/* work around various missing functions */
232	read_write (int dowrite, int fd, off_t offset, size_t length,
233	SV data, STRLEN dataoffset, SV callback)
234	{
235	aio_req req;
236	STRLEN svlen;
237	char *svptr = SvPV (data, svlen);
238		241
239	SvUPGRADE (data, SVt_PV);	242	#if !HAVE_PREADWRITE
240	SvPOK_on (data);	243	# define pread aio_pread
		244	# define pwrite aio_pwrite
241		245
242	if (dataoffset < 0)	246	/*
243	dataoffset += svlen;	247	* make our pread/pwrite safe against themselves, but not against
		248	* normal read/write by using a mutex. slows down execution a lot,
		249	* but that's your problem, not mine.
		250	*/
		251	static pthread_mutex_t iolock = PTHREAD_MUTEX_INITIALIZER;
244		252
245	if (dataoffset < 0 \|\| dataoffset > svlen)	253	static ssize_t
246	croak ("data offset outside of string");	254	pread (int fd, void *buf, size_t count, off_t offset)
		255	{
		256	ssize_t res;
		257	off_t ooffset;
247		258
248	if (dowrite)	259	pthread_mutex_lock (&iolock);
		260	ooffset = lseek (fd, 0, SEEK_CUR);
		261	lseek (fd, offset, SEEK_SET);
		262	res = read (fd, buf, count);
		263	lseek (fd, ooffset, SEEK_SET);
		264	pthread_mutex_unlock (&iolock);
		265
		266	return res;
		267	}
		268
		269	static ssize_t
		270	pwrite (int fd, void *buf, size_t count, off_t offset)
		271	{
		272	ssize_t res;
		273	off_t ooffset;
		274
		275	pthread_mutex_lock (&iolock);
		276	ooffset = lseek (fd, 0, SEEK_CUR);
		277	lseek (fd, offset, SEEK_SET);
		278	res = write (fd, buf, count);
		279	lseek (fd, offset, SEEK_SET);
		280	pthread_mutex_unlock (&iolock);
		281
		282	return res;
		283	}
		284	#endif
		285
		286	#if !HAVE_FDATASYNC
		287	# define fdatasync fsync
		288	#endif
		289
		290	#if !HAVE_READAHEAD
		291	# define readahead aio_readahead
		292
		293	static char readahead_buf[4096];
		294
		295	static ssize_t
		296	readahead (int fd, off_t offset, size_t count)
		297	{
		298	while (count > 0)
249	{	299	{
250	/* write: check length and adjust. */	300	size_t len = count < sizeof (readahead_buf) ? count : sizeof (readahead_buf);
251	if (length < 0 \|\| length + dataoffset > svlen)	301
252	length = svlen - dataoffset;	302	pread (fd, readahead_buf, len, offset);
		303	offset += len;
		304	count -= len;
253	}	305	}
254	else
255	{
256	/* read: grow scalar as necessary */
257	svptr = SvGROW (data, length + dataoffset);
258	}
259		306
260	if (length < 0)	307	errno = 0;
261	croak ("length must not be negative");
262
263	Newz (0, req, 1, aio_cb);
264
265	if (!req)
266	croak ("out of memory during aio_req allocation");
267
268	req->type = dowrite ? REQ_WRITE : REQ_READ;
269	req->fd = fd;
270	req->offset = offset;
271	req->length = length;
272	req->data = SvREFCNT_inc (data);
273	req->dataptr = (char *)svptr + dataoffset;
274	req->callback = SvREFCNT_inc (callback);
275
276	send_req (req);
277	}	308	}
		309	#endif
278		310
279	static void *	311	static void *
280	aio_proc (void *thr_arg)	312	aio_proc (void *thr_arg)
281	{	313	{
282	aio_req req;	314	aio_req req;
…		…
308		340
309	type = req->type;	341	type = req->type;
310		342
311	switch (type)	343	switch (type)
312	{	344	{
313	case REQ_READ: req->result = pread64 (req->fd, req->dataptr, req->length, req->offset); break;	345	case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
314	case REQ_WRITE: req->result = pwrite64 (req->fd, req->dataptr, req->length, req->offset); break;	346	case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
315	#if SYS_readahead	347
316	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;	348	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
317	#else
318	case REQ_READAHEAD: req->result = -1; errno = ENOSYS; break;
319	#endif
320		349
321	case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break;	350	case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break;
322	case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break;	351	case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break;
323	case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break;	352	case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break;
324		353
325	case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break;	354	case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break;
326	case REQ_CLOSE: req->result = close (req->fd); break;	355	case REQ_CLOSE: req->result = close (req->fd); break;
327	case REQ_UNLINK: req->result = unlink (req->dataptr); break;	356	case REQ_UNLINK: req->result = unlink (req->dataptr); break;
328		357
		358	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
329	case REQ_FSYNC: req->result = fsync (req->fd); break;	359	case REQ_FSYNC: req->result = fsync (req->fd); break;
330	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
331		360
332	case REQ_QUIT:	361	case REQ_QUIT:
333	break;	362	break;
334		363
335	default:	364	default:
…		…
441	send_req (req);	470	send_req (req);
442	}	471	}
443		472
444	void	473	void
445	aio_close(fh,callback=&PL_sv_undef)	474	aio_close(fh,callback=&PL_sv_undef)
446	InputStream fh	475	SV * fh
447	SV * callback	476	SV * callback
448	PROTOTYPE: $;$	477	PROTOTYPE: $;$
449	ALIAS:	478	ALIAS:
450	aio_close = REQ_CLOSE	479	aio_close = REQ_CLOSE
451	aio_fsync = REQ_FSYNC	480	aio_fsync = REQ_FSYNC
452	aio_fdatasync = REQ_FDATASYNC	481	aio_fdatasync = REQ_FDATASYNC
…		…
458		487
459	if (!req)	488	if (!req)
460	croak ("out of memory during aio_req allocation");	489	croak ("out of memory during aio_req allocation");
461		490
462	req->type = ix;	491	req->type = ix;
		492	req->fh = newSVsv (fh);
463	req->fd = PerlIO_fileno (fh);	493	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));
464	req->callback = SvREFCNT_inc (callback);	494	req->callback = SvREFCNT_inc (callback);
465		495
466	send_req (req);	496	send_req (req);
467	}	497	}
468		498
469	void	499	void
470	aio_read(fh,offset,length,data,dataoffset,callback=&PL_sv_undef)	500	aio_read(fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
471	InputStream fh	501	SV * fh
472	UV offset	502	UV offset
473	IV length	503	IV length
474	SV * data	504	SV * data
475	IV dataoffset	505	IV dataoffset
476	SV * callback	506	SV * callback
		507	ALIAS:
		508	aio_read = REQ_READ
		509	aio_write = REQ_WRITE
477	PROTOTYPE: $$$$$;$	510	PROTOTYPE: $$$$$;$
478	CODE:	511	CODE:
479	read_write (0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);	512	{
		513	aio_req req;
		514	STRLEN svlen;
		515	char *svptr = SvPV (data, svlen);
480		516
481	void	517	SvUPGRADE (data, SVt_PV);
482	aio_write(fh,offset,length,data,dataoffset,callback=&PL_sv_undef)	518	SvPOK_on (data);
483	OutputStream fh	519
484	UV offset	520	if (dataoffset < 0)
485	IV length	521	dataoffset += svlen;
486	SV * data	522
487	IV dataoffset	523	if (dataoffset < 0 \|\| dataoffset > svlen)
488	SV * callback	524	croak ("data offset outside of string");
489	PROTOTYPE: $$$$$;$	525
490	CODE:	526	if (ix == REQ_WRITE)
491	read_write (1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);	527	{
		528	/* write: check length and adjust. */
		529	if (length < 0 \|\| length + dataoffset > svlen)
		530	length = svlen - dataoffset;
		531	}
		532	else
		533	{
		534	/* read: grow scalar as necessary */
		535	svptr = SvGROW (data, length + dataoffset);
		536	}
		537
		538	if (length < 0)
		539	croak ("length must not be negative");
		540
		541	Newz (0, req, 1, aio_cb);
		542
		543	if (!req)
		544	croak ("out of memory during aio_req allocation");
		545
		546	req->type = ix;
		547	req->fh = newSVsv (fh);
		548	req->fd = PerlIO_fileno (ix == REQ_READ ? IoIFP (sv_2io (fh))
		549	: IoOFP (sv_2io (fh)));
		550	req->offset = offset;
		551	req->length = length;
		552	req->data = SvREFCNT_inc (data);
		553	req->dataptr = (char *)svptr + dataoffset;
		554	req->callback = SvREFCNT_inc (callback);
		555
		556	send_req (req);
		557	}
492		558
493	void	559	void
494	aio_readahead(fh,offset,length,callback=&PL_sv_undef)	560	aio_readahead(fh,offset,length,callback=&PL_sv_undef)
495	InputStream fh	561	SV * fh
496	UV offset	562	UV offset
497	IV length	563	IV length
498	SV * callback	564	SV * callback
499	PROTOTYPE: $$$;$	565	PROTOTYPE: $$$;$
500	CODE:	566	CODE:
501	{	567	{
502	aio_req req;	568	aio_req req;
503		569
…		…
508		574
509	if (!req)	575	if (!req)
510	croak ("out of memory during aio_req allocation");	576	croak ("out of memory during aio_req allocation");
511		577
512	req->type = REQ_READAHEAD;	578	req->type = REQ_READAHEAD;
		579	req->fh = newSVsv (fh);
513	req->fd = PerlIO_fileno (fh);	580	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));
514	req->offset = offset;	581	req->offset = offset;
515	req->length = length;	582	req->length = length;
516	req->callback = SvREFCNT_inc (callback);	583	req->callback = SvREFCNT_inc (callback);
517		584
518	send_req (req);	585	send_req (req);
…		…
546	req->dataptr = SvPV_nolen (req->data);	613	req->dataptr = SvPV_nolen (req->data);
547	}	614	}
548	else	615	else
549	{	616	{
550	req->type = REQ_FSTAT;	617	req->type = REQ_FSTAT;
		618	req->fh = newSVsv (fh_or_path);
551	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));	619	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
552	}	620	}
553		621
554	req->callback = SvREFCNT_inc (callback);	622	req->callback = SvREFCNT_inc (callback);
555		623

Diff Legend

-–
+Removed lines
-+
+Added lines
-<
+Changed lines
->
+Changed lines

Comparing IO-AIO/AIO.xs (file contents): Revision 1.9 by root, Tue Jul 12 11:02:54 2005 UTC vs. Revision 1.19 by root, Sun Jul 31 19:04:45 2005 UTC

Diff Legend

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.9 by root, Tue Jul 12 11:02:54 2005 UTC vs.
Revision 1.19 by root, Sun Jul 31 19:04:45 2005 UTC