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

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

Diff Legend

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

Comparing IO-AIO/AIO.xs (file contents): Revision 1.10 by root, Wed Jul 13 00:13:09 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.10 by root, Wed Jul 13 00:13:09 2005 UTC vs.
Revision 1.19 by root, Sun Jul 31 19:04:45 2005 UTC