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

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.2 by root, Sun Jul 10 18:16:49 2005 UTC vs.
Revision 1.17 by root, Sun Jul 31 18:45:48 2005 UTC

…		…
1	#define PERL_NO_GET_CONTEXT
2
3	#include "EXTERN.h"	1	#include "EXTERN.h"
4	#include "perl.h"	2	#include "perl.h"
5	#include "XSUB.h"	3	#include "XSUB.h"
6		4
		5	#include "autoconf/config.h"
		6
7	#include <sys/types.h>	7	#include <sys/types.h>
8	#include <sys/stat.h>	8	#include <sys/stat.h>
		9
9	#include <unistd.h>	10	#include <unistd.h>
10	#include <fcntl.h>	11	#include <fcntl.h>
11	#include <signal.h>	12	#include <signal.h>
12	#include <sched.h>	13	#include <sched.h>
13	#include <endian.h>
14		14
15	#include <pthread.h>	15	#include <pthread.h>
16	#include <sys/syscall.h>
17		16
18	typedef void InputStream; / hack, but 5.6.1 is simply toooo old ;) */	17	typedef void InputStream; / hack, but 5.6.1 is simply toooo old ;) */
19	typedef void OutputStream; / hack, but 5.6.1 is simply toooo old ;) */	18	typedef void OutputStream; / hack, but 5.6.1 is simply toooo old ;) */
20	typedef void InOutStream; / hack, but 5.6.1 is simply toooo old ;) */	19	typedef void InOutStream; / hack, but 5.6.1 is simply toooo old ;) */
21		20
22	#if __i386 \|\| __amd64
23	# define STACKSIZE ( 256 * sizeof (long))
24	#elif __ia64	21	#if __ia64
25	# define STACKSIZE (8192 * sizeof (long))	22	# define STACKSIZE 65536
26	#else	23	#else
27	# define STACKSIZE ( 512 * sizeof (long))	24	# define STACKSIZE 4096
28	#endif	25	#endif
29		26
30	enum {	27	enum {
31	REQ_QUIT,	28	REQ_QUIT,
32	REQ_OPEN, REQ_CLOSE,	29	REQ_OPEN, REQ_CLOSE,
34	REQ_STAT, REQ_LSTAT, REQ_FSTAT, REQ_UNLINK,	31	REQ_STAT, REQ_LSTAT, REQ_FSTAT, REQ_UNLINK,
35	REQ_FSYNC, REQ_FDATASYNC,	32	REQ_FSYNC, REQ_FDATASYNC,
36	};	33	};
37		34
38	typedef struct aio_cb {	35	typedef struct aio_cb {
39	struct aio_cb *next;	36	struct aio_cb *volatile next;
40		37
41	int type;	38	int type;
42		39
43	int fd;	40	int fd;
44	off_t offset;	41	off_t offset;
45	size_t length;	42	size_t length;
46	ssize_t result;	43	ssize_t result;
47	mode_t mode; /* open */	44	mode_t mode; /* open */
48	int errorno;	45	int errorno;
49	SV data, callback;	46	SV data, callback, *fh;
50	void *dataptr;	47	void *dataptr;
51	STRLEN dataoffset;	48	STRLEN dataoffset;
52		49
53	Stat_t *statdata;	50	Stat_t *statdata;
54	} aio_cb;	51	} aio_cb;
55		52
56	typedef aio_cb *aio_req;	53	typedef aio_cb *aio_req;
57		54
58	static int started;	55	static int started;
59	static int nreqs;	56	static volatile int nreqs;
		57	static int max_outstanding = 1<<30;
60	static int reqpipe[2], respipe[2];	58	static int respipe [2];
61		59
		60	static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER;
		61	static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER;
		62	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
		63
62	static aio_req qs, qe; /* queue start, queue end */	64	static volatile aio_req reqs, reqe; /* queue start, queue end */
63		65	static volatile aio_req ress, rese; /* queue start, queue end */
64	static void aio_proc(void arg);
65
66	static void
67	start_thread (void)
68	{
69	sigset_t fullsigset, oldsigset;
70	pthread_t tid;
71	pthread_attr_t attr;
72
73	pthread_attr_init (&attr);
74	pthread_attr_setstacksize (&attr, STACKSIZE);
75	pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
76
77	sigfillset (&fullsigset);
78	sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
79
80	if (pthread_create (&tid, &attr, aio_proc, 0) == 0)
81	started++;
82
83	sigprocmask (SIG_SETMASK, &oldsigset, 0);
84	}
85
86	static void
87	send_reqs (void)
88	{
89	/* this write is atomic */
90	while (qs && write (reqpipe[1], &qs, sizeof qs) == sizeof qs)
91	{
92	qs = qs->next;
93	if (!qs) qe = 0;
94	}
95	}
96
97	static void
98	send_req (aio_req req)
99	{
100	nreqs++;
101	req->next = 0;
102
103	if (qe)
104	{
105	qe->next = req;
106	qe = req;
107	}
108	else
109	qe = qs = req;
110
111	send_reqs ();
112	}
113
114	static void
115	end_thread (void)
116	{
117	aio_req req;
118	New (0, req, 1, aio_cb);
119	req->type = REQ_QUIT;
120
121	send_req (req);
122	}
123
124	static void
125	read_write (pTHX_
126	int dowrite, int fd, off_t offset, size_t length,
127	SV data, STRLEN dataoffset, SV callback)
128	{
129	aio_req req;
130	STRLEN svlen;
131	char *svptr = SvPV (data, svlen);
132
133	SvUPGRADE (data, SVt_PV);
134	SvPOK_on (data);
135
136	if (dataoffset < 0)
137	dataoffset += svlen;
138
139	if (dataoffset < 0 \|\| dataoffset > svlen)
140	croak ("data offset outside of string");
141
142	if (dowrite)
143	{
144	/* write: check length and adjust. */
145	if (length < 0 \|\| length + dataoffset > svlen)
146	length = svlen - dataoffset;
147	}
148	else
149	{
150	/* read: grow scalar as necessary */
151	svptr = SvGROW (data, length + dataoffset);
152	}
153
154	if (length < 0)
155	croak ("length must not be negative");
156
157	Newz (0, req, 1, aio_cb);
158
159	if (!req)
160	croak ("out of memory during aio_req allocation");
161
162	req->type = dowrite ? REQ_WRITE : REQ_READ;
163	req->fd = fd;
164	req->offset = offset;
165	req->length = length;
166	req->data = SvREFCNT_inc (data);
167	req->dataptr = (char *)svptr + dataoffset;
168	req->callback = SvREFCNT_inc (callback);
169
170	send_req (req);
171	}
172		66
173	static void	67	static void
174	poll_wait ()	68	poll_wait ()
175	{	69	{
		70	if (nreqs && !ress)
		71	{
176	fd_set rfd;	72	fd_set rfd;
177	FD_ZERO(&rfd);	73	FD_ZERO(&rfd);
178	FD_SET(respipe[0], &rfd);	74	FD_SET(respipe [0], &rfd);
179		75
180	select (respipe[0] + 1, &rfd, 0, 0, 0);	76	select (respipe [0] + 1, &rfd, 0, 0, 0);
		77	}
181	}	78	}
182		79
183	static int	80	static int
184	poll_cb (pTHX)	81	poll_cb ()
185	{	82	{
186	dSP;	83	dSP;
187	int count = 0;	84	int count = 0;
188	aio_req req;	85	aio_req req, prv;
189		86
190	while (read (respipe[0], (void *)&req, sizeof (req)) == sizeof (req))	87	pthread_mutex_lock (&reslock);
		88
		89	{
		90	/* read any signals sent by the worker threads */
		91	char buf [32];
		92	while (read (respipe [0], buf, 32) > 0)
		93	;
		94	}
		95
		96	req = ress;
		97	ress = rese = 0;
		98
		99	pthread_mutex_unlock (&reslock);
		100
		101	while (req)
191	{	102	{
192	nreqs--;	103	nreqs--;
193		104
194	if (req->type == REQ_QUIT)	105	if (req->type == REQ_QUIT)
195	started--;	106	started--;
…		…
203	+ req->result > 0 ? req->result : 0);	114	+ req->result > 0 ? req->result : 0);
204		115
205	if (req->data)	116	if (req->data)
206	SvREFCNT_dec (req->data);	117	SvREFCNT_dec (req->data);
207		118
		119	if (req->fh)
		120	SvREFCNT_dec (req->fh);
		121
208	if (req->type == REQ_STAT \|\| req->type == REQ_LSTAT \|\| req->type == REQ_FSTAT)	122	if (req->type == REQ_STAT \|\| req->type == REQ_LSTAT \|\| req->type == REQ_FSTAT)
209	{	123	{
210	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;	124	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
211	PL_laststatval = req->result;	125	PL_laststatval = req->result;
212	PL_statcache = *(req->statdata);	126	PL_statcache = *(req->statdata);
213		127
214	Safefree (req->statdata);	128	Safefree (req->statdata);
215	}	129	}
216		130
		131	ENTER;
217	PUSHMARK (SP);	132	PUSHMARK (SP);
218	XPUSHs (sv_2mortal (newSViv (req->result)));	133	XPUSHs (sv_2mortal (newSViv (req->result)));
219		134
220	if (req->type == REQ_OPEN)	135	if (req->type == REQ_OPEN)
221	{	136	{
…		…
224		139
225	PUTBACK;	140	PUTBACK;
226	call_pv ("IO::AIO::_fd2fh", G_SCALAR \| G_EVAL);	141	call_pv ("IO::AIO::_fd2fh", G_SCALAR \| G_EVAL);
227	SPAGAIN;	142	SPAGAIN;
228		143
229	fh = POPs;	144	fh = SvREFCNT_inc (POPs);
230		145
231	PUSHMARK (SP);	146	PUSHMARK (SP);
232	XPUSHs (fh);	147	XPUSHs (sv_2mortal (fh));
233	}	148	}
234		149
		150	if (SvOK (req->callback))
		151	{
235	PUTBACK;	152	PUTBACK;
236	call_sv (req->callback, G_VOID \| G_EVAL);	153	call_sv (req->callback, G_VOID \| G_EVAL);
237	SPAGAIN;	154	SPAGAIN;
		155	}
		156
		157	LEAVE;
238		158
239	if (req->callback)	159	if (req->callback)
240	SvREFCNT_dec (req->callback);	160	SvREFCNT_dec (req->callback);
241		161
242	errno = errorno;	162	errno = errorno;
243	count++;	163	count++;
244	}	164	}
245		165
		166	prv = req;
		167	req = req->next;
246	Safefree (req);	168	Safefree (prv);
		169
		170	/* TODO: croak on errors? */
247	}	171	}
248		172
249	if (qs)
250	send_reqs ();
251
252	return count;	173	return count;
253	}	174	}
		175
		176	static void aio_proc(void arg);
		177
		178	static void
		179	start_thread (void)
		180	{
		181	sigset_t fullsigset, oldsigset;
		182	pthread_t tid;
		183	pthread_attr_t attr;
		184
		185	pthread_attr_init (&attr);
		186	pthread_attr_setstacksize (&attr, STACKSIZE);
		187	pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
		188
		189	sigfillset (&fullsigset);
		190	sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
		191
		192	if (pthread_create (&tid, &attr, aio_proc, 0) == 0)
		193	started++;
		194
		195	sigprocmask (SIG_SETMASK, &oldsigset, 0);
		196	}
		197
		198	static void
		199	send_req (aio_req req)
		200	{
		201	nreqs++;
		202
		203	pthread_mutex_lock (&reqlock);
		204
		205	req->next = 0;
		206
		207	if (reqe)
		208	{
		209	reqe->next = req;
		210	reqe = req;
		211	}
		212	else
		213	reqe = reqs = req;
		214
		215	pthread_cond_signal (&reqwait);
		216	pthread_mutex_unlock (&reqlock);
		217
		218	while (nreqs > max_outstanding)
		219	{
		220	poll_wait ();
		221	poll_cb ();
		222	}
		223	}
		224
		225	static void
		226	end_thread (void)
		227	{
		228	aio_req req;
		229	New (0, req, 1, aio_cb);
		230	req->type = REQ_QUIT;
		231
		232	send_req (req);
		233	}
		234
		235	/* work around various missing functions */
		236
		237	#if !HAVE_PREADWRITE
		238	# define pread aio_pread
		239	# define pwrite aio_pwrite
		240
		241	/*
		242	* make our pread/pwrite safe against themselves, but not against
		243	* normal read/write by using a mutex. slows down execution a lot,
		244	* but that's your problem, not mine.
		245	*/
		246	static pthread_mutex_t iolock = PTHREAD_MUTEX_INITIALIZER;
		247
		248	static ssize_t
		249	pread (int fd, void *buf, size_t count, off_t offset)
		250	{
		251	ssize_t res;
		252	off_t ooffset;
		253
		254	pthread_mutex_lock (&iolock);
		255	ooffset = lseek (fd, 0, SEEK_CUR);
		256	lseek (fd, offset, SEEK_SET);
		257	res = read (fd, buf, count);
		258	lseek (fd, ooffset, SEEK_SET);
		259	pthread_mutex_unlock (&iolock);
		260
		261	return res;
		262	}
		263
		264	static ssize_t
		265	pwrite (int fd, void *buf, size_t count, off_t offset)
		266	{
		267	ssize_t res;
		268	off_t ooffset;
		269
		270	pthread_mutex_lock (&iolock);
		271	ooffset = lseek (fd, 0, SEEK_CUR);
		272	lseek (fd, offset, SEEK_SET);
		273	res = write (fd, buf, count);
		274	lseek (fd, offset, SEEK_SET);
		275	pthread_mutex_unlock (&iolock);
		276
		277	return res;
		278	}
		279	#endif
		280
		281	#if !HAVE_FDATASYNC
		282	# define fdatasync fsync
		283	#endif
		284
		285	#if !HAVE_READAHEAD
		286	# define readahead aio_readahead
		287
		288	static char readahead_buf[4096];
		289
		290	static ssize_t
		291	readahead (int fd, off_t offset, size_t count)
		292	{
		293	while (count > 0)
		294	{
		295	size_t len = count < sizeof (readahead_buf) ? count : sizeof (readahead_buf);
		296
		297	pread (fd, readahead_buf, len, offset);
		298	offset += len;
		299	count -= len;
		300	}
		301
		302	errno = 0;
		303	}
		304	#endif
254		305
255	static void *	306	static void *
256	aio_proc (void *thr_arg)	307	aio_proc (void *thr_arg)
257	{	308	{
258	aio_req req;	309	aio_req req;
		310	int type;
259		311
260	/* then loop */	312	do
261	while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
262	{	313	{
		314	pthread_mutex_lock (&reqlock);
		315
		316	for (;;)
		317	{
		318	req = reqs;
		319
		320	if (reqs)
		321	{
		322	reqs = reqs->next;
		323	if (!reqs) reqe = 0;
		324	}
		325
		326	if (req)
		327	break;
		328
		329	pthread_cond_wait (&reqwait, &reqlock);
		330	}
		331
		332	pthread_mutex_unlock (&reqlock);
		333
263	errno = 0; /* strictly unnecessary */	334	errno = 0; /* strictly unnecessary */
264		335
		336	type = req->type;
		337
265	switch (req->type)	338	switch (type)
266	{	339	{
267	case REQ_READ: req->result = pread64 (req->fd, req->dataptr, req->length, req->offset); break;	340	case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
268	case REQ_WRITE: req->result = pwrite64 (req->fd, req->dataptr, req->length, req->offset); break;	341	case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
269	#if SYS_readahead	342
270	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;	343	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
271	#else
272	case REQ_READAHEAD: req->result = -1; errno = ENOSYS; break;
273	#endif
274		344
275	case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break;	345	case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break;
276	case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break;	346	case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break;
277	case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break;	347	case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break;
278		348
279	case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break;	349	case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break;
280	case REQ_CLOSE: req->result = close (req->fd); break;	350	case REQ_CLOSE: req->result = close (req->fd); break;
281	case REQ_UNLINK: req->result = unlink (req->dataptr); break;	351	case REQ_UNLINK: req->result = unlink (req->dataptr); break;
282		352
		353	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
283	case REQ_FSYNC: req->result = fsync (req->fd); break;	354	case REQ_FSYNC: req->result = fsync (req->fd); break;
284	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
285		355
286	case REQ_QUIT:	356	case REQ_QUIT:
287	write (respipe[1], (void *)&req, sizeof (req));	357	break;
288	return 0;
289		358
290	default:	359	default:
291	req->result = ENOSYS;	360	req->result = ENOSYS;
292	break;	361	break;
293	}	362	}
294		363
295	req->errorno = errno;	364	req->errorno = errno;
296	write (respipe[1], (void *)&req, sizeof (req));	365
		366	pthread_mutex_lock (&reslock);
		367
		368	req->next = 0;
		369
		370	if (rese)
		371	{
		372	rese->next = req;
		373	rese = req;
		374	}
		375	else
		376	{
		377	rese = ress = req;
		378
		379	/* write a dummy byte to the pipe so fh becomes ready */
		380	write (respipe [1], &respipe, 1);
		381	}
		382
		383	pthread_mutex_unlock (&reslock);
297	}	384	}
		385	while (type != REQ_QUIT);
298		386
299	return 0;	387	return 0;
300	}	388	}
301		389
302	MODULE = IO::AIO PACKAGE = IO::AIO	390	MODULE = IO::AIO PACKAGE = IO::AIO
303		391
		392	PROTOTYPES: ENABLE
		393
304	BOOT:	394	BOOT:
305	{	395	{
306	if (pipe (reqpipe) \|\| pipe (respipe))	396	if (pipe (respipe))
307	croak ("unable to initialize request or result pipe");	397	croak ("unable to initialize result pipe");
308		398
309	if (fcntl (reqpipe[1], F_SETFL, O_NONBLOCK))	399	if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
310	croak ("cannot set result pipe to nonblocking mode");	400	croak ("cannot set result pipe to nonblocking mode");
311		401
312	if (fcntl (respipe[0], F_SETFL, O_NONBLOCK))	402	if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
313	croak ("cannot set result pipe to nonblocking mode");	403	croak ("cannot set result pipe to nonblocking mode");
314	}	404	}
315		405
316	void	406	void
317	min_parallel(nthreads)	407	min_parallel(nthreads)
…		…
335	}	425	}
336		426
337	while (started > nthreads)	427	while (started > nthreads)
338	{	428	{
339	poll_wait ();	429	poll_wait ();
340	poll_cb (aTHX);	430	poll_cb ();
341	}	431	}
342	}	432	}
343		433
		434	int
		435	max_outstanding(nreqs)
		436	int nreqs
		437	PROTOTYPE: $
		438	CODE:
		439	RETVAL = max_outstanding;
		440	max_outstanding = nreqs;
		441
344	void	442	void
345	aio_open(pathname,flags,mode,callback)	443	aio_open(pathname,flags,mode,callback=&PL_sv_undef)
346	SV * pathname	444	SV * pathname
347	int flags	445	int flags
348	int mode	446	int mode
349	SV * callback	447	SV * callback
350	PROTOTYPE: $$$$	448	PROTOTYPE: $$$;$
351	CODE:	449	CODE:
352	{	450	{
353	aio_req req;	451	aio_req req;
354		452
355	Newz (0, req, 1, aio_cb);	453	Newz (0, req, 1, aio_cb);
…		…
366		464
367	send_req (req);	465	send_req (req);
368	}	466	}
369		467
370	void	468	void
371	aio_close(fh,callback)	469	aio_close(fh,callback=&PL_sv_undef)
372	InputStream fh	470	SV * fh
373	SV * callback	471	SV * callback
374	PROTOTYPE: $$	472	PROTOTYPE: $;$
375	ALIAS:	473	ALIAS:
376	aio_close = REQ_CLOSE	474	aio_close = REQ_CLOSE
377	aio_fsync = REQ_FSYNC	475	aio_fsync = REQ_FSYNC
378	aio_fdatasync = REQ_FDATASYNC	476	aio_fdatasync = REQ_FDATASYNC
379	CODE:	477	CODE:
…		…
384		482
385	if (!req)	483	if (!req)
386	croak ("out of memory during aio_req allocation");	484	croak ("out of memory during aio_req allocation");
387		485
388	req->type = ix;	486	req->type = ix;
		487	req->fh = newSVsv (fh);
389	req->fd = PerlIO_fileno (fh);	488	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));
390	req->callback = SvREFCNT_inc (callback);	489	req->callback = SvREFCNT_inc (callback);
391		490
392	send_req (req);	491	send_req (req);
393	}	492	}
394		493
395	void	494	void
396	aio_read(fh,offset,length,data,dataoffset,callback)	495	aio_read(fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
397	InputStream fh	496	SV * fh
398	UV offset	497	UV offset
399	IV length	498	IV length
400	SV * data	499	SV * data
401	IV dataoffset	500	IV dataoffset
402	SV * callback	501	SV * callback
		502	ALIAS:
		503	aio_read = REQ_READ
		504	aio_write = REQ_WRITE
403	PROTOTYPE: $$$$$$	505	PROTOTYPE: $$$$$;$
404	CODE:	506	CODE:
405	read_write (aTHX_ 0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
406
407	void
408	aio_write(fh,offset,length,data,dataoffset,callback)
409	OutputStream fh
410	UV offset
411	IV length
412	SV * data
413	IV dataoffset
414	SV * callback
415	PROTOTYPE: $$$$$$
416	CODE:
417	read_write (aTHX_ 1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
418
419	void
420	aio_readahead(fh,offset,length,callback)
421	InputStream fh
422	UV offset
423	IV length
424	SV * callback
425	PROTOTYPE: $$$$
426	CODE:
427	{	507	{
428	aio_req req;	508	aio_req req;
		509	STRLEN svlen;
		510	char *svptr = SvPV (data, svlen);
		511
		512	SvUPGRADE (data, SVt_PV);
		513	SvPOK_on (data);
		514
		515	if (dataoffset < 0)
		516	dataoffset += svlen;
		517
		518	if (dataoffset < 0 \|\| dataoffset > svlen)
		519	croak ("data offset outside of string");
		520
		521	if (ix == REQ_WRITE)
		522	{
		523	/* write: check length and adjust. */
		524	if (length < 0 \|\| length + dataoffset > svlen)
		525	length = svlen - dataoffset;
		526	}
		527	else
		528	{
		529	/* read: grow scalar as necessary */
		530	svptr = SvGROW (data, length + dataoffset);
		531	}
429		532
430	if (length < 0)	533	if (length < 0)
431	croak ("length must not be negative");	534	croak ("length must not be negative");
432		535
433	Newz (0, req, 1, aio_cb);	536	Newz (0, req, 1, aio_cb);
434		537
435	if (!req)	538	if (!req)
436	croak ("out of memory during aio_req allocation");	539	croak ("out of memory during aio_req allocation");
437		540
		541	req->type = ix;
		542	req->fh = newSVsv (fh);
		543	req->fd = PerlIO_fileno (ix == REQ_READ ? IoIFP (sv_2io (fh))
		544	: IoOFP (sv_2io (fh)));
		545	req->offset = offset;
		546	req->length = length;
		547	req->data = SvREFCNT_inc (data);
		548	req->dataptr = (char *)svptr + dataoffset;
		549	req->callback = SvREFCNT_inc (callback);
		550
		551	send_req (req);
		552	}
		553
		554	void
		555	aio_readahead(fh,offset,length,callback=&PL_sv_undef)
		556	SV * fh
		557	UV offset
		558	IV length
		559	SV * callback
		560	PROTOTYPE: $$$;$
		561	CODE:
		562	{
		563	aio_req req;
		564
		565	if (length < 0)
		566	croak ("length must not be negative");
		567
		568	Newz (0, req, 1, aio_cb);
		569
		570	if (!req)
		571	croak ("out of memory during aio_req allocation");
		572
438	req->type = REQ_READAHEAD;	573	req->type = REQ_READAHEAD;
		574	req->fh = newSVsv (fh);
439	req->fd = PerlIO_fileno (fh);	575	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));
440	req->offset = offset;	576	req->offset = offset;
441	req->length = length;	577	req->length = length;
442	req->callback = SvREFCNT_inc (callback);	578	req->callback = SvREFCNT_inc (callback);
443		579
444	send_req (req);	580	send_req (req);
445	}	581	}
446		582
447	void	583	void
448	aio_stat(fh_or_path,callback)	584	aio_stat(fh_or_path,callback=&PL_sv_undef)
449	SV * fh_or_path	585	SV * fh_or_path
450	SV * callback	586	SV * callback
451	PROTOTYPE: $$
452	ALIAS:	587	ALIAS:
		588	aio_stat = REQ_STAT
453	aio_lstat = 1	589	aio_lstat = REQ_LSTAT
454	CODE:	590	CODE:
455	{	591	{
456	aio_req req;	592	aio_req req;
457		593
458	Newz (0, req, 1, aio_cb);	594	Newz (0, req, 1, aio_cb);
…		…
465	if (!req->statdata)	601	if (!req->statdata)
466	croak ("out of memory during aio_req->statdata allocation");	602	croak ("out of memory during aio_req->statdata allocation");
467		603
468	if (SvPOK (fh_or_path))	604	if (SvPOK (fh_or_path))
469	{	605	{
470	req->type = ix ? REQ_LSTAT : REQ_STAT;	606	req->type = ix;
471	req->data = newSVsv (fh_or_path);	607	req->data = newSVsv (fh_or_path);
472	req->dataptr = SvPV_nolen (req->data);	608	req->dataptr = SvPV_nolen (req->data);
473	}	609	}
474	else	610	else
475	{	611	{
476	req->type = REQ_FSTAT;	612	req->type = REQ_FSTAT;
		613	req->fh = newSVsv (fh_or_path);
477	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));	614	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
478	}	615	}
479		616
480	req->callback = SvREFCNT_inc (callback);	617	req->callback = SvREFCNT_inc (callback);
481		618
482	send_req (req);	619	send_req (req);
483	}	620	}
484		621
485	void	622	void
486	aio_unlink(pathname,callback)	623	aio_unlink(pathname,callback=&PL_sv_undef)
487	SV * pathname	624	SV * pathname
488	SV * callback	625	SV * callback
489	PROTOTYPE: $$
490	CODE:	626	CODE:
491	{	627	{
492	aio_req req;	628	aio_req req;
493		629
494	Newz (0, req, 1, aio_cb);	630	Newz (0, req, 1, aio_cb);
…		…
502	req->callback = SvREFCNT_inc (callback);	638	req->callback = SvREFCNT_inc (callback);
503		639
504	send_req (req);	640	send_req (req);
505	}	641	}
506		642
		643	void
		644	flush()
		645	PROTOTYPE:
		646	CODE:
		647	while (nreqs)
		648	{
		649	poll_wait ();
		650	poll_cb ();
		651	}
		652
		653	void
		654	poll()
		655	PROTOTYPE:
		656	CODE:
		657	if (nreqs)
		658	{
		659	poll_wait ();
		660	poll_cb ();
		661	}
		662
507	int	663	int
508	poll_fileno()	664	poll_fileno()
509	PROTOTYPE:	665	PROTOTYPE:
510	CODE:	666	CODE:
511	RETVAL = respipe[0];	667	RETVAL = respipe [0];
512	OUTPUT:	668	OUTPUT:
513	RETVAL	669	RETVAL
514		670
515	int	671	int
516	poll_cb(...)	672	poll_cb(...)
517	PROTOTYPE:	673	PROTOTYPE:
518	CODE:	674	CODE:
519	RETVAL = poll_cb (aTHX);	675	RETVAL = poll_cb ();
520	OUTPUT:	676	OUTPUT:
521	RETVAL	677	RETVAL
522		678
523	void	679	void
524	poll_wait()	680	poll_wait()
525	PROTOTYPE:	681	PROTOTYPE:
526	CODE:	682	CODE:
		683	if (nreqs)
527	poll_wait ();	684	poll_wait ();
528		685
529	int	686	int
530	nreqs()	687	nreqs()
531	PROTOTYPE:	688	PROTOTYPE:
532	CODE:	689	CODE:

Diff Legend

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

Comparing IO-AIO/AIO.xs (file contents): Revision 1.2 by root, Sun Jul 10 18:16:49 2005 UTC vs. Revision 1.17 by root, Sun Jul 31 18:45:48 2005 UTC

Diff Legend

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.2 by root, Sun Jul 10 18:16:49 2005 UTC vs.
Revision 1.17 by root, Sun Jul 31 18:45:48 2005 UTC