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

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.44 by root, Sat Oct 21 23:20:29 2006 UTC vs.
Revision 1.52 by root, Sun Oct 22 22:14:54 2006 UTC

…		…
9		9
10	#include <pthread.h>	10	#include <pthread.h>
11		11
12	#include <stddef.h>	12	#include <stddef.h>
13	#include <errno.h>	13	#include <errno.h>
		14	#include <sys/time.h>
		15	#include <sys/select.h>
14	#include <sys/types.h>	16	#include <sys/types.h>
15	#include <sys/stat.h>	17	#include <sys/stat.h>
16	#include <limits.h>	18	#include <limits.h>
17	#include <unistd.h>	19	#include <unistd.h>
18	#include <fcntl.h>	20	#include <fcntl.h>
…		…
53	REQ_STAT, REQ_LSTAT, REQ_FSTAT,	55	REQ_STAT, REQ_LSTAT, REQ_FSTAT,
54	REQ_FSYNC, REQ_FDATASYNC,	56	REQ_FSYNC, REQ_FDATASYNC,
55	REQ_UNLINK, REQ_RMDIR, REQ_RENAME,	57	REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
56	REQ_READDIR,	58	REQ_READDIR,
57	REQ_LINK, REQ_SYMLINK,	59	REQ_LINK, REQ_SYMLINK,
		60	REQ_SLEEP,
58	REQ_GROUP,	61	REQ_GROUP,
59	};	62	};
60		63
61	#define AIO_REQ_KLASS "IO::AIO::REQ"	64	#define AIO_REQ_KLASS "IO::AIO::REQ"
62	#define AIO_GRP_KLASS "IO::AIO::GRP"	65	#define AIO_GRP_KLASS "IO::AIO::GRP"
63		66
64	typedef struct aio_cb	67	typedef struct aio_cb
65	{	68	{
66	struct aio_cb grp, grp_prev, *grp_next;
67
68	struct aio_cb *volatile next;	69	struct aio_cb *volatile next;
		70
		71	struct aio_cb grp, grp_prev, grp_next, grp_first;
69		72
70	SV self; / the perl counterpart of this request, if any */	73	SV self; / the perl counterpart of this request, if any */
71		74
72	SV data, callback;	75	SV data, callback;
73	SV fh, fh2;	76	SV fh, fh2;
…		…
85	unsigned char cancelled;	88	unsigned char cancelled;
86	} aio_cb;	89	} aio_cb;
87		90
88	typedef aio_cb *aio_req;	91	typedef aio_cb *aio_req;
89	typedef aio_cb *aio_req_ornot;	92	typedef aio_cb *aio_req_ornot;
90	typedef aio_cb *aio_group;
91		93
92	static int started, wanted;	94	static int started, wanted;
93	static volatile int nreqs;	95	static volatile int nreqs;
94	static int max_outstanding = 1<<30;	96	static int max_outstanding = 1<<30;
95	static int respipe [2];	97	static int respipe [2];
…		…
99	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;	101	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
100		102
101	static volatile aio_req reqs, reqe; /* queue start, queue end */	103	static volatile aio_req reqs, reqe; /* queue start, queue end */
102	static volatile aio_req ress, rese; /* queue start, queue end */	104	static volatile aio_req ress, rese; /* queue start, queue end */
103		105
		106	static void req_invoke (aio_req req);
		107	static void req_free (aio_req req);
		108
104	/* must be called at most once */	109	/* must be called at most once */
105	static SV req_sv (aio_req req, const char klass)	110	static SV req_sv (aio_req req, const char klass)
106	{	111	{
		112	if (!req->self)
		113	{
107	req->self = (SV *)newHV ();	114	req->self = (SV *)newHV ();
108	sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0);	115	sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0);
		116	}
109		117
110	return sv_2mortal (sv_bless (newRV_noinc (req->self), gv_stashpv (klass, 1)));	118	return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1)));
111	}	119	}
112		120
113	static aio_req SvAIO_REQ (SV sv, const char klass)	121	static aio_req SvAIO_REQ (SV *sv)
114	{	122	{
115	if (!sv_derived_from (sv, klass) \|\| !SvROK (sv))	123	if (!sv_derived_from (sv, AIO_REQ_KLASS) \|\| !SvROK (sv))
116	croak ("object of class %s expected", klass);	124	croak ("object of class " AIO_REQ_KLASS " expected");
117		125
118	MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext);	126	MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext);
119		127
120	return mg ? (aio_req)mg->mg_ptr : 0;	128	return mg ? (aio_req)mg->mg_ptr : 0;
121	}	129	}
122		130
123	static void req_free (aio_req req)	131	static void aio_grp_feed (aio_req grp)
124	{	132	{
125	if (req->self)	133	while (grp->length < grp->fd2)
126	{
127	sv_unmagic (req->self, PERL_MAGIC_ext);
128	SvREFCNT_dec (req->self);
129	}	134	{
		135	int old_len = grp->length;
130		136
131	if (req->data)	137	if (grp->fh2 && SvOK (grp->fh2))
132	SvREFCNT_dec (req->data);	138	{
		139	dSP;
133		140
134	if (req->fh)	141	ENTER;
135	SvREFCNT_dec (req->fh);	142	SAVETMPS;
		143	PUSHMARK (SP);
		144	XPUSHs (req_sv (grp, AIO_GRP_KLASS));
		145	PUTBACK;
		146	call_sv (grp->fh2, G_VOID \| G_EVAL);
		147	SPAGAIN;
		148	FREETMPS;
		149	LEAVE;
		150	}
136		151
137	if (req->fh2)	152	/* stop if no progress has been made */
		153	if (old_len == grp->length)
		154	{
138	SvREFCNT_dec (req->fh2);	155	SvREFCNT_dec (grp->fh2);
139		156	grp->fh2 = 0;
140	if (req->statdata)	157	break;
141	Safefree (req->statdata);	158	}
142		159	}
143	if (req->callback)
144	SvREFCNT_dec (req->callback);
145
146	if (req->type == REQ_READDIR && req->result >= 0)
147	free (req->data2ptr);
148
149	Safefree (req);
150	}	160	}
151		161
152	static void	162	static void aio_grp_dec (aio_req grp)
153	poll_wait ()	163	{
		164	--grp->length;
		165
		166	/* call feeder, if applicable */
		167	aio_grp_feed (grp);
		168
		169	/* finish, if done */
		170	if (!grp->length && grp->fd)
		171	{
		172	req_invoke (grp);
		173	req_free (grp);
		174	}
		175	}
		176
		177	static void poll_wait ()
154	{	178	{
155	if (nreqs && !ress)	179	if (nreqs && !ress)
156	{	180	{
157	fd_set rfd;	181	fd_set rfd;
158	FD_ZERO(&rfd);	182	FD_ZERO(&rfd);
…		…
160		184
161	select (respipe [0] + 1, &rfd, 0, 0, 0);	185	select (respipe [0] + 1, &rfd, 0, 0, 0);
162	}	186	}
163	}	187	}
164		188
165	static int	189	static void req_invoke (aio_req req)
166	poll_cb ()	190	{
		191	dSP;
		192	int errorno = errno;
		193
		194	if (req->cancelled \|\| !SvOK (req->callback))
		195	return;
		196
		197	errno = req->errorno;
		198
		199	ENTER;
		200	SAVETMPS;
		201	PUSHMARK (SP);
		202	EXTEND (SP, 1);
		203
		204	switch (req->type)
		205	{
		206	case REQ_READDIR:
		207	{
		208	SV *rv = &PL_sv_undef;
		209
		210	if (req->result >= 0)
		211	{
		212	char *buf = req->data2ptr;
		213	AV *av = newAV ();
		214
		215	while (req->result)
		216	{
		217	SV *sv = newSVpv (buf, 0);
		218
		219	av_push (av, sv);
		220	buf += SvCUR (sv) + 1;
		221	req->result--;
		222	}
		223
		224	rv = sv_2mortal (newRV_noinc ((SV *)av));
		225	}
		226
		227	PUSHs (rv);
		228	}
		229	break;
		230
		231	case REQ_OPEN:
		232	{
		233	/* convert fd to fh */
		234	SV *fh;
		235
		236	PUSHs (sv_2mortal (newSViv (req->result)));
		237	PUTBACK;
		238	call_pv ("IO::AIO::_fd2fh", G_SCALAR \| G_EVAL);
		239	SPAGAIN;
		240
		241	fh = SvREFCNT_inc (POPs);
		242
		243	PUSHMARK (SP);
		244	XPUSHs (sv_2mortal (fh));
		245	}
		246	break;
		247
		248	case REQ_GROUP:
		249	req->fd = 2; /* mark group as finished */
		250
		251	if (req->data)
		252	{
		253	int i;
		254	AV av = (AV )req->data;
		255
		256	EXTEND (SP, AvFILL (av) + 1);
		257	for (i = 0; i <= AvFILL (av); ++i)
		258	PUSHs (*av_fetch (av, i, 0));
		259	}
		260	break;
		261
		262	case REQ_SLEEP:
		263	break;
		264
		265	default:
		266	PUSHs (sv_2mortal (newSViv (req->result)));
		267	break;
		268	}
		269
		270
		271	PUTBACK;
		272	call_sv (req->callback, G_VOID \| G_EVAL);
		273	SPAGAIN;
		274
		275	FREETMPS;
		276	LEAVE;
		277
		278	errno = errorno;
		279
		280	if (SvTRUE (ERRSV))
		281	{
		282	req_free (req);
		283	croak (0);
		284	}
		285	}
		286
		287	static void req_free (aio_req req)
		288	{
		289	if (req->grp)
		290	{
		291	aio_req grp = req->grp;
		292
		293	/* unlink request */
		294	if (req->grp_next) req->grp_next->grp_prev = req->grp_prev;
		295	if (req->grp_prev) req->grp_prev->grp_next = req->grp_next;
		296
		297	if (grp->grp_first == req)
		298	grp->grp_first = req->grp_next;
		299
		300	aio_grp_dec (grp);
		301	}
		302
		303	if (req->self)
		304	{
		305	sv_unmagic (req->self, PERL_MAGIC_ext);
		306	SvREFCNT_dec (req->self);
		307	}
		308
		309	SvREFCNT_dec (req->data);
		310	SvREFCNT_dec (req->fh);
		311	SvREFCNT_dec (req->fh2);
		312	SvREFCNT_dec (req->callback);
		313	Safefree (req->statdata);
		314
		315	if (req->type == REQ_READDIR && req->result >= 0)
		316	free (req->data2ptr);
		317
		318	Safefree (req);
		319	}
		320
		321	static void req_cancel (aio_req req)
		322	{
		323	req->cancelled = 1;
		324
		325	if (req->type == REQ_GROUP)
		326	{
		327	aio_req sub;
		328
		329	for (sub = req->grp_first; sub; sub = sub->grp_next)
		330	req_cancel (sub);
		331	}
		332	}
		333
		334	static int poll_cb ()
167	{	335	{
168	dSP;	336	dSP;
169	int count = 0;	337	int count = 0;
170	int do_croak = 0;	338	int do_croak = 0;
171	aio_req req;	339	aio_req req;
…		…
193	pthread_mutex_unlock (&reslock);	361	pthread_mutex_unlock (&reslock);
194		362
195	if (!req)	363	if (!req)
196	break;	364	break;
197		365
198	nreqs--;	366	--nreqs;
199		367
200	if (req->type == REQ_QUIT)	368	if (req->type == REQ_QUIT)
201	started--;	369	started--;
		370	else if (req->type == REQ_GROUP && req->length)
		371	{
		372	req->fd = 1; /* mark request as delayed */
		373	continue;
		374	}
202	else	375	else
203	{	376	{
204	int errorno = errno;
205	errno = req->errorno;
206
207	if (req->type == REQ_READ)	377	if (req->type == REQ_READ)
208	SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));	378	SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
209		379
210	if (req->data2ptr && (req->type == REQ_READ \|\| req->type == REQ_WRITE))	380	if (req->data2ptr && (req->type == REQ_READ \|\| req->type == REQ_WRITE))
211	SvREADONLY_off (req->data);	381	SvREADONLY_off (req->data);
…		…
215	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;	385	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
216	PL_laststatval = req->result;	386	PL_laststatval = req->result;
217	PL_statcache = *(req->statdata);	387	PL_statcache = *(req->statdata);
218	}	388	}
219		389
220	ENTER;	390	req_invoke (req);
221	PUSHMARK (SP);
222		391
223	if (req->type == REQ_READDIR)
224	{
225	SV *rv = &PL_sv_undef;
226
227	if (req->result >= 0)
228	{
229	char *buf = req->data2ptr;
230	AV *av = newAV ();
231
232	while (req->result)
233	{
234	SV *sv = newSVpv (buf, 0);
235
236	av_push (av, sv);
237	buf += SvCUR (sv) + 1;
238	req->result--;
239	}
240
241	rv = sv_2mortal (newRV_noinc ((SV *)av));
242	}
243
244	XPUSHs (rv);
245	}
246	else
247	{
248	XPUSHs (sv_2mortal (newSViv (req->result)));
249
250	if (req->type == REQ_OPEN)
251	{
252	/* convert fd to fh */
253	SV *fh;
254
255	PUTBACK;
256	call_pv ("IO::AIO::_fd2fh", G_SCALAR \| G_EVAL);
257	SPAGAIN;
258
259	fh = SvREFCNT_inc (POPs);
260
261	PUSHMARK (SP);
262	XPUSHs (sv_2mortal (fh));
263	}
264	}
265
266	if (SvOK (req->callback) && !req->cancelled)
267	{
268	PUTBACK;
269	call_sv (req->callback, G_VOID \| G_EVAL);
270	SPAGAIN;
271
272	if (SvTRUE (ERRSV))
273	{
274	req_free (req);
275	croak (0);
276	}
277	}
278
279	LEAVE;
280
281	errno = errorno;
282	count++;	392	count++;
283	}	393	}
284		394
285	req_free (req);	395	req_free (req);
286	}	396	}
…		…
288	return count;	398	return count;
289	}	399	}
290		400
291	static void aio_proc(void arg);	401	static void aio_proc(void arg);
292		402
293	static void
294	start_thread (void)	403	static void start_thread (void)
295	{	404	{
296	sigset_t fullsigset, oldsigset;	405	sigset_t fullsigset, oldsigset;
297	pthread_t tid;	406	pthread_t tid;
298	pthread_attr_t attr;	407	pthread_attr_t attr;
299		408
…		…
308	started++;	417	started++;
309		418
310	sigprocmask (SIG_SETMASK, &oldsigset, 0);	419	sigprocmask (SIG_SETMASK, &oldsigset, 0);
311	}	420	}
312		421
313	static void
314	req_send (aio_req req)	422	static void req_send (aio_req req)
315	{	423	{
316	while (started < wanted && nreqs >= started)	424	while (started < wanted && nreqs >= started)
317	start_thread ();	425	start_thread ();
318		426
319	nreqs++;	427	++nreqs;
320		428
321	pthread_mutex_lock (&reqlock);	429	pthread_mutex_lock (&reqlock);
322		430
323	req->next = 0;	431	req->next = 0;
324		432
…		…
343		451
344	poll_wait ();	452	poll_wait ();
345	}	453	}
346	}	454	}
347		455
348	static void	456	static void end_thread (void)
349	end_thread (void)
350	{	457	{
351	aio_req req;	458	aio_req req;
352	Newz (0, req, 1, aio_cb);	459	Newz (0, req, 1, aio_cb);
353	req->type = REQ_QUIT;	460	req->type = REQ_QUIT;
354		461
…		…
405	* normal read/write by using a mutex. slows down execution a lot,	512	* normal read/write by using a mutex. slows down execution a lot,
406	* but that's your problem, not mine.	513	* but that's your problem, not mine.
407	*/	514	*/
408	static pthread_mutex_t preadwritelock = PTHREAD_MUTEX_INITIALIZER;	515	static pthread_mutex_t preadwritelock = PTHREAD_MUTEX_INITIALIZER;
409		516
410	static ssize_t	517	static ssize_t pread (int fd, void *buf, size_t count, off_t offset)
411	pread (int fd, void *buf, size_t count, off_t offset)
412	{	518	{
413	ssize_t res;	519	ssize_t res;
414	off_t ooffset;	520	off_t ooffset;
415		521
416	pthread_mutex_lock (&preadwritelock);	522	pthread_mutex_lock (&preadwritelock);
…		…
421	pthread_mutex_unlock (&preadwritelock);	527	pthread_mutex_unlock (&preadwritelock);
422		528
423	return res;	529	return res;
424	}	530	}
425		531
426	static ssize_t
427	pwrite (int fd, void *buf, size_t count, off_t offset)	532	static ssize_t pwrite (int fd, void *buf, size_t count, off_t offset)
428	{	533	{
429	ssize_t res;	534	ssize_t res;
430	off_t ooffset;	535	off_t ooffset;
431		536
432	pthread_mutex_lock (&preadwritelock);	537	pthread_mutex_lock (&preadwritelock);
…		…
445	#endif	550	#endif
446		551
447	#if !HAVE_READAHEAD	552	#if !HAVE_READAHEAD
448	# define readahead aio_readahead	553	# define readahead aio_readahead
449		554
450	static ssize_t
451	readahead (int fd, off_t offset, size_t count)	555	static ssize_t readahead (int fd, off_t offset, size_t count)
452	{	556	{
453	char readahead_buf[4096];	557	char readahead_buf[4096];
454		558
455	while (count > 0)	559	while (count > 0)
456	{	560	{
…		…
468	#if !HAVE_READDIR_R	572	#if !HAVE_READDIR_R
469	# define readdir_r aio_readdir_r	573	# define readdir_r aio_readdir_r
470		574
471	static pthread_mutex_t readdirlock = PTHREAD_MUTEX_INITIALIZER;	575	static pthread_mutex_t readdirlock = PTHREAD_MUTEX_INITIALIZER;
472		576
473	static int
474	readdir_r (DIR dirp, struct dirent ent, struct dirent **res)	577	static int readdir_r (DIR dirp, struct dirent ent, struct dirent **res)
475	{	578	{
476	struct dirent *e;	579	struct dirent *e;
477	int errorno;	580	int errorno;
478		581
479	pthread_mutex_lock (&readdirlock);	582	pthread_mutex_lock (&readdirlock);
…		…
495	return e ? 0 : -1;	598	return e ? 0 : -1;
496	}	599	}
497	#endif	600	#endif
498		601
499	/* sendfile always needs emulation */	602	/* sendfile always needs emulation */
500	static ssize_t
501	sendfile_ (int ofd, int ifd, off_t offset, size_t count)	603	static ssize_t sendfile_ (int ofd, int ifd, off_t offset, size_t count)
502	{	604	{
503	ssize_t res;	605	ssize_t res;
504		606
505	if (!count)	607	if (!count)
506	return 0;	608	return 0;
…		…
588		690
589	return res;	691	return res;
590	}	692	}
591		693
592	/* read a full directory */	694	/* read a full directory */
593	static int
594	scandir_ (const char path, void *namesp)	695	static int scandir_ (const char path, void *namesp)
595	{	696	{
596	DIR *dirp = opendir (path);	697	DIR *dirp = opendir (path);
597	union	698	union
598	{	699	{
599	struct dirent d;	700	struct dirent d;
…		…
653	return res;	754	return res;
654	}	755	}
655		756
656	/*****************************************************************************/	757	/*****************************************************************************/
657		758
658	static void *
659	aio_proc (void *thr_arg)	759	static void aio_proc (void thr_arg)
660	{	760	{
661	aio_req req;	761	aio_req req;
662	int type;	762	int type;
663		763
664	do	764	do
…		…
684	pthread_mutex_unlock (&reqlock);	784	pthread_mutex_unlock (&reqlock);
685		785
686	errno = 0; /* strictly unnecessary */	786	errno = 0; /* strictly unnecessary */
687		787
688	if (!req->cancelled)	788	if (!req->cancelled)
689	switch (req->type)	789	switch (type = req->type) /* remember type for QUIT check */
690	{	790	{
691	case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;	791	case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
692	case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;	792	case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
693		793
694	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;	794	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
…		…
707	case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;	807	case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;
708		808
709	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;	809	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
710	case REQ_FSYNC: req->result = fsync (req->fd); break;	810	case REQ_FSYNC: req->result = fsync (req->fd); break;
711	case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break;	811	case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break;
		812
		813	case REQ_SLEEP:
		814	{
		815	struct timeval tv;
		816
		817	tv.tv_sec = req->fd;
		818	tv.tv_usec = req->fd2;
		819
		820	req->result = select (0, 0, 0, 0, &tv);
		821	}
712		822
713	case REQ_QUIT:	823	case REQ_QUIT:
714	break;	824	break;
715		825
716	default:	826	default:
…		…
1073		1183
1074	REQ_SEND;	1184	REQ_SEND;
1075	}	1185	}
1076		1186
1077	void	1187	void
		1188	aio_sleep (delay,callback=&PL_sv_undef)
		1189	double delay
		1190	SV * callback
		1191	PPCODE:
		1192	{
		1193	dREQ;
		1194
		1195	req->type = REQ_SLEEP;
		1196	req->fd = delay < 0. ? 0 : delay;
		1197	req->fd2 = delay < 0. ? 0 : 1000. * (delay - req->fd);
		1198
		1199	REQ_SEND;
		1200	}
		1201
		1202	void
1078	aio_group (callback=&PL_sv_undef)	1203	aio_group (callback=&PL_sv_undef)
1079	SV * callback	1204	SV * callback
1080	PROTOTYPE: ;&	1205	PROTOTYPE: ;$
1081	PPCODE:	1206	PPCODE:
1082	{	1207	{
1083	dREQ;	1208	dREQ;
1084	req->type = REQ_GROUP;	1209	req->type = REQ_GROUP;
		1210	req_send (req);
1085	XPUSHs (req_sv (req, AIO_GRP_KLASS));	1211	XPUSHs (req_sv (req, AIO_GRP_KLASS));
1086	}	1212	}
1087		1213
1088	void	1214	void
1089	flush ()	1215	flush ()
…		…
1134	CODE:	1260	CODE:
1135	RETVAL = nreqs;	1261	RETVAL = nreqs;
1136	OUTPUT:	1262	OUTPUT:
1137	RETVAL	1263	RETVAL
1138		1264
		1265	PROTOTYPES: DISABLE
		1266
1139	MODULE = IO::AIO PACKAGE = IO::AIO::REQ	1267	MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1140		1268
1141	void	1269	void
1142	cancel (aio_req_ornot req)	1270	cancel (aio_req_ornot req)
1143	PROTOTYPE:	1271	PROTOTYPE:
1144	CODE:	1272	CODE:
1145	req->cancelled = 1;	1273	req_cancel (req);
1146		1274
		1275	MODULE = IO::AIO PACKAGE = IO::AIO::GRP
		1276
		1277	void
		1278	add (aio_req grp, ...)
		1279	PPCODE:
		1280	{
		1281	int i;
		1282
		1283	if (grp->fd == 2)
		1284	croak ("cannot add requests to IO::AIO::GRP after the group finished");
		1285
		1286	for (i = 1; i < items; ++i )
		1287	{
		1288	if (GIMME_V != G_VOID)
		1289	XPUSHs (sv_2mortal (newSVsv (ST (i))));
		1290
		1291	aio_req req = SvAIO_REQ (ST (i));
		1292
		1293	if (req)
		1294	{
		1295	++grp->length;
		1296	req->grp = grp;
		1297
		1298	req->grp_prev = 0;
		1299	req->grp_next = grp->grp_first;
		1300
		1301	if (grp->grp_first)
		1302	grp->grp_first->grp_prev = req;
		1303
		1304	grp->grp_first = req;
		1305	}
		1306	}
		1307	}
		1308
		1309	void
		1310	result (aio_req grp, ...)
		1311	CODE:
		1312	{
		1313	int i;
		1314	AV *av = newAV ();
		1315
		1316	for (i = 1; i < items; ++i )
		1317	av_push (av, newSVsv (ST (i)));
		1318
		1319	SvREFCNT_dec (grp->data);
		1320	grp->data = (SV *)av;
		1321	}
		1322
		1323	void
		1324	lock (aio_req grp)
		1325	CODE:
		1326	++grp->length;
		1327
		1328	void
		1329	unlock (aio_req grp)
		1330	CODE:
		1331	aio_grp_dec (grp);
		1332
		1333	void
		1334	feeder_limit (aio_req grp, int limit)
		1335	CODE:
		1336	grp->fd2 = limit;
		1337	aio_grp_feed (grp);
		1338
		1339	void
		1340	set_feeder (aio_req grp, SV *callback=&PL_sv_undef)
		1341	CODE:
		1342	{
		1343	SvREFCNT_dec (grp->fh2);
		1344	grp->fh2 = newSVsv (callback);
		1345
		1346	if (grp->fd2 <= 0)
		1347	grp->fd2 = 2;
		1348
		1349	aio_grp_feed (grp);
		1350	}
		1351

Diff Legend

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

Comparing IO-AIO/AIO.xs (file contents): Revision 1.44 by root, Sat Oct 21 23:20:29 2006 UTC vs. Revision 1.52 by root, Sun Oct 22 22:14:54 2006 UTC

Diff Legend

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.44 by root, Sat Oct 21 23:20:29 2006 UTC vs.
Revision 1.52 by root, Sun Oct 22 22:14:54 2006 UTC