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

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.79 by root, Thu Oct 26 16:28:33 2006 UTC vs.
Revision 1.87 by root, Mon Oct 30 23:42:27 2006 UTC

…		…
43	# else	43	# else
44	# error sendfile support requested but not available	44	# error sendfile support requested but not available
45	# endif	45	# endif
46	#endif	46	#endif
47		47
		48	/* number of seconds after which idle threads exit */
		49	#define IDLE_TIMEOUT 10
		50
48	/* used for struct dirent, AIX doesn't provide it */	51	/* used for struct dirent, AIX doesn't provide it */
49	#ifndef NAME_MAX	52	#ifndef NAME_MAX
50	# define NAME_MAX 4096	53	# define NAME_MAX 4096
51	#endif	54	#endif
52		55
…		…
65		68
66	/* wether word reads are potentially non-atomic.	69	/* wether word reads are potentially non-atomic.
67	* this is conservatice, likely most arches this runs	70	* this is conservatice, likely most arches this runs
68	* on have atomic word read/writes.	71	* on have atomic word read/writes.
69	*/	72	*/
70	#ifndef WORDREAD_UNSAFE	73	#ifndef WORDACCESS_UNSAFE
71	# if __i386 \|\| __x86_64	74	# if __i386 \|\| __x86_64
72	# define WORDREAD_UNSAFE 0	75	# define WORDACCESS_UNSAFE 0
73	# else	76	# else
74	# define WORDREAD_UNSAFE 1	77	# define WORDACCESS_UNSAFE 1
75	# endif	78	# endif
76	#endif	79	#endif
77		80
78	/* buffer size for various temporary buffers */	81	/* buffer size for various temporary buffers */
79	#define AIO_BUFSIZE 65536	82	#define AIO_BUFSIZE 65536
…		…
92	REQ_READ, REQ_WRITE, REQ_READAHEAD,	95	REQ_READ, REQ_WRITE, REQ_READAHEAD,
93	REQ_SENDFILE,	96	REQ_SENDFILE,
94	REQ_STAT, REQ_LSTAT, REQ_FSTAT,	97	REQ_STAT, REQ_LSTAT, REQ_FSTAT,
95	REQ_FSYNC, REQ_FDATASYNC,	98	REQ_FSYNC, REQ_FDATASYNC,
96	REQ_UNLINK, REQ_RMDIR, REQ_RENAME,	99	REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
97	REQ_READDIR,	100	REQ_MKNOD, REQ_READDIR,
98	REQ_LINK, REQ_SYMLINK,	101	REQ_LINK, REQ_SYMLINK, REQ_READLINK,
99	REQ_GROUP, REQ_NOP,	102	REQ_GROUP, REQ_NOP,
100	REQ_BUSY,	103	REQ_BUSY,
101	};	104	};
102		105
103	#define AIO_REQ_KLASS "IO::AIO::REQ"	106	#define AIO_REQ_KLASS "IO::AIO::REQ"
…		…
105		108
106	typedef struct aio_cb	109	typedef struct aio_cb
107	{	110	{
108	struct aio_cb *volatile next;	111	struct aio_cb *volatile next;
109		112
110	SV data, callback;	113	SV callback, fh;
111	SV fh, fh2;	114	SV sv1, sv2;
112	void dataptr, data2ptr;	115	void ptr1, ptr2;
113	Stat_t *statdata;
114	off_t offset;	116	off_t offs;
115	size_t length;	117	size_t size;
116	ssize_t result;	118	ssize_t result;
117		119
118	STRLEN dataoffset;	120	STRLEN stroffset;
119	int type;	121	int type;
120	int fd, fd2;	122	int int1, int2;
121	int errorno;	123	int errorno;
122	mode_t mode; /* open */	124	mode_t mode; /* open */
123		125
124	unsigned char flags;	126	unsigned char flags;
125	unsigned char pri;	127	unsigned char pri;
…		…
127	SV self; / the perl counterpart of this request, if any */	129	SV self; / the perl counterpart of this request, if any */
128	struct aio_cb grp, grp_prev, grp_next, grp_first;	130	struct aio_cb grp, grp_prev, grp_next, grp_first;
129	} aio_cb;	131	} aio_cb;
130		132
131	enum {	133	enum {
132	FLAG_CANCELLED = 0x01,	134	FLAG_CANCELLED = 0x01,
		135	FLAG_SV1_RO_OFF = 0x40, /* data was set readonly */
		136	FLAG_PTR2_FREE = 0x80, /* need free(ptr2) */
133	};	137	};
134		138
135	typedef aio_cb *aio_req;	139	typedef aio_cb *aio_req;
136	typedef aio_cb *aio_req_ornot;	140	typedef aio_cb *aio_req_ornot;
137		141
…		…
142	DEFAULT_PRI = 0,	146	DEFAULT_PRI = 0,
143	PRI_BIAS = -PRI_MIN,	147	PRI_BIAS = -PRI_MIN,
144	NUM_PRI = PRI_MAX + PRI_BIAS + 1,	148	NUM_PRI = PRI_MAX + PRI_BIAS + 1,
145	};	149	};
146		150
		151	#define AIO_TICKS ((1000000 + 1023) >> 10)
		152
		153	static unsigned int max_poll_time = 0;
		154	static unsigned int max_poll_reqs = 0;
		155
		156	/* calculcate time difference in ~1/AIO_TICKS of a second */
		157	static int tvdiff (struct timeval tv1, struct timeval tv2)
		158	{
		159	return (tv2->tv_sec - tv1->tv_sec ) * AIO_TICKS
		160	+ ((tv2->tv_usec - tv1->tv_usec) >> 10);
		161	}
		162
147	static int next_pri = DEFAULT_PRI + PRI_BIAS;	163	static int next_pri = DEFAULT_PRI + PRI_BIAS;
148		164
149	static unsigned int started, wanted;	165	static unsigned int started, idle, wanted;
150	static volatile unsigned int nreqs, nready, npending;
151	static volatile unsigned int max_outstanding = 0xffffffff;
152	static int respipe [2];
153		166
154	#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)	167	#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
155	# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP	168	# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
156	#else	169	#else
157	# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER	170	# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
…		…
198	wrk->prev->next = wrk->next;	211	wrk->prev->next = wrk->next;
199		212
200	free (wrk);	213	free (wrk);
201	}	214	}
202		215
		216	static volatile unsigned int nreqs, nready, npending;
		217	static volatile unsigned int max_idle = 4;
		218	static volatile unsigned int max_outstanding = 0xffffffff;
		219	static int respipe [2];
		220
203	static pthread_mutex_t reslock = AIO_MUTEX_INIT;	221	static pthread_mutex_t reslock = AIO_MUTEX_INIT;
204	static pthread_mutex_t reqlock = AIO_MUTEX_INIT;	222	static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
205	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;	223	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
		224
		225	#if WORDACCESS_UNSAFE
		226
		227	static unsigned int get_nready ()
		228	{
		229	unsigned int retval;
		230
		231	LOCK (reqlock);
		232	retval = nready;
		233	UNLOCK (reqlock);
		234
		235	return retval;
		236	}
		237
		238	static unsigned int get_npending ()
		239	{
		240	unsigned int retval;
		241
		242	LOCK (reslock);
		243	retval = npending;
		244	UNLOCK (reslock);
		245
		246	return retval;
		247	}
		248
		249	static unsigned int get_nthreads ()
		250	{
		251	unsigned int retval;
		252
		253	LOCK (wrklock);
		254	retval = started;
		255	UNLOCK (wrklock);
		256
		257	return retval;
		258	}
		259
		260	#else
		261
		262	# define get_nready() nready
		263	# define get_npending() npending
		264	# define get_nthreads() started
		265
		266	#endif
206		267
207	/*	268	/*
208	* a somewhat faster data structure might be nice, but	269	* a somewhat faster data structure might be nice, but
209	* with 8 priorities this actually needs <20 insns	270	* with 8 priorities this actually needs <20 insns
210	* per shift, the most expensive operation.	271	* per shift, the most expensive operation.
…		…
256	}	317	}
257		318
258	abort ();	319	abort ();
259	}	320	}
260		321
261	static int poll_cb (int max);	322	static int poll_cb ();
262	static void req_invoke (aio_req req);	323	static void req_invoke (aio_req req);
263	static void req_free (aio_req req);	324	static void req_free (aio_req req);
264	static void req_cancel (aio_req req);	325	static void req_cancel (aio_req req);
265		326
266	/* must be called at most once */	327	/* must be called at most once */
…		…
287	return mg ? (aio_req)mg->mg_ptr : 0;	348	return mg ? (aio_req)mg->mg_ptr : 0;
288	}	349	}
289		350
290	static void aio_grp_feed (aio_req grp)	351	static void aio_grp_feed (aio_req grp)
291	{	352	{
292	while (grp->length < grp->fd2 && !(grp->flags & FLAG_CANCELLED))	353	while (grp->size < grp->int2 && !(grp->flags & FLAG_CANCELLED))
293	{	354	{
294	int old_len = grp->length;	355	int old_len = grp->size;
295		356
296	if (grp->fh2 && SvOK (grp->fh2))	357	if (grp->sv2 && SvOK (grp->sv2))
297	{	358	{
298	dSP;	359	dSP;
299		360
300	ENTER;	361	ENTER;
301	SAVETMPS;	362	SAVETMPS;
302	PUSHMARK (SP);	363	PUSHMARK (SP);
303	XPUSHs (req_sv (grp, AIO_GRP_KLASS));	364	XPUSHs (req_sv (grp, AIO_GRP_KLASS));
304	PUTBACK;	365	PUTBACK;
305	call_sv (grp->fh2, G_VOID \| G_EVAL \| G_KEEPERR);	366	call_sv (grp->sv2, G_VOID \| G_EVAL \| G_KEEPERR);
306	SPAGAIN;	367	SPAGAIN;
307	FREETMPS;	368	FREETMPS;
308	LEAVE;	369	LEAVE;
309	}	370	}
310		371
311	/* stop if no progress has been made */	372	/* stop if no progress has been made */
312	if (old_len == grp->length)	373	if (old_len == grp->size)
313	{	374	{
314	SvREFCNT_dec (grp->fh2);	375	SvREFCNT_dec (grp->sv2);
315	grp->fh2 = 0;	376	grp->sv2 = 0;
316	break;	377	break;
317	}	378	}
318	}	379	}
319	}	380	}
320		381
321	static void aio_grp_dec (aio_req grp)	382	static void aio_grp_dec (aio_req grp)
322	{	383	{
323	--grp->length;	384	--grp->size;
324		385
325	/* call feeder, if applicable */	386	/* call feeder, if applicable */
326	aio_grp_feed (grp);	387	aio_grp_feed (grp);
327		388
328	/* finish, if done */	389	/* finish, if done */
329	if (!grp->length && grp->fd)	390	if (!grp->size && grp->int1)
330	{	391	{
331	req_invoke (grp);	392	req_invoke (grp);
332	req_free (grp);	393	req_free (grp);
333	}	394	}
334	}	395	}
335		396
336	static void poll_wait ()
337	{
338	fd_set rfd;
339
340	while (nreqs)
341	{
342	int size;
343	if (WORDREAD_UNSAFE) LOCK (reslock);
344	size = res_queue.size;
345	if (WORDREAD_UNSAFE) UNLOCK (reslock);
346
347	if (size)
348	return;
349
350	FD_ZERO(&rfd);
351	FD_SET(respipe [0], &rfd);
352
353	select (respipe [0] + 1, &rfd, 0, 0, 0);
354	}
355	}
356
357	static void req_invoke (aio_req req)	397	static void req_invoke (aio_req req)
358	{	398	{
359	dSP;	399	dSP;
		400
		401	if (req->flags & FLAG_SV1_RO_OFF)
		402	SvREADONLY_off (req->sv1);
360		403
361	if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))	404	if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))
362	{	405	{
363	ENTER;	406	ENTER;
364	SAVETMPS;	407	SAVETMPS;
…		…
372	SV *rv = &PL_sv_undef;	415	SV *rv = &PL_sv_undef;
373		416
374	if (req->result >= 0)	417	if (req->result >= 0)
375	{	418	{
376	int i;	419	int i;
377	char *buf = req->data2ptr;	420	char *buf = req->ptr2;
378	AV *av = newAV ();	421	AV *av = newAV ();
379		422
380	av_extend (av, req->result - 1);	423	av_extend (av, req->result - 1);
381		424
382	for (i = 0; i < req->result; ++i)	425	for (i = 0; i < req->result; ++i)
…		…
410	XPUSHs (sv_2mortal (fh));	453	XPUSHs (sv_2mortal (fh));
411	}	454	}
412	break;	455	break;
413		456
414	case REQ_GROUP:	457	case REQ_GROUP:
415	req->fd = 2; /* mark group as finished */	458	req->int1 = 2; /* mark group as finished */
416		459
417	if (req->data)	460	if (req->sv1)
418	{	461	{
419	int i;	462	int i;
420	AV av = (AV )req->data;	463	AV av = (AV )req->sv1;
421		464
422	EXTEND (SP, AvFILL (av) + 1);	465	EXTEND (SP, AvFILL (av) + 1);
423	for (i = 0; i <= AvFILL (av); ++i)	466	for (i = 0; i <= AvFILL (av); ++i)
424	PUSHs (*av_fetch (av, i, 0));	467	PUSHs (*av_fetch (av, i, 0));
425	}	468	}
…		…
427		470
428	case REQ_NOP:	471	case REQ_NOP:
429	case REQ_BUSY:	472	case REQ_BUSY:
430	break;	473	break;
431		474
		475	case REQ_READLINK:
		476	if (req->result > 0)
		477	{
		478	SvCUR_set (req->sv1, req->result);
		479	*SvEND (req->sv1) = 0;
		480	PUSHs (req->sv1);
		481	}
		482	break;
		483
		484	case REQ_STAT:
		485	case REQ_LSTAT:
		486	case REQ_FSTAT:
		487	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
		488	PL_laststatval = req->result;
		489	PL_statcache = (Stat_t )(req->ptr2);
		490	PUSHs (sv_2mortal (newSViv (req->result)));
		491	break;
		492
		493	case REQ_READ:
		494	SvCUR_set (req->sv1, req->stroffset + (req->result > 0 ? req->result : 0));
		495	*SvEND (req->sv1) = 0;
		496	PUSHs (sv_2mortal (newSViv (req->result)));
		497	break;
		498
432	default:	499	default:
433	PUSHs (sv_2mortal (newSViv (req->result)));	500	PUSHs (sv_2mortal (newSViv (req->result)));
434	break;	501	break;
435	}	502	}
436		503
…		…
471	{	538	{
472	sv_unmagic (req->self, PERL_MAGIC_ext);	539	sv_unmagic (req->self, PERL_MAGIC_ext);
473	SvREFCNT_dec (req->self);	540	SvREFCNT_dec (req->self);
474	}	541	}
475		542
476	SvREFCNT_dec (req->data);
477	SvREFCNT_dec (req->fh);	543	SvREFCNT_dec (req->fh);
		544	SvREFCNT_dec (req->sv1);
478	SvREFCNT_dec (req->fh2);	545	SvREFCNT_dec (req->sv2);
479	SvREFCNT_dec (req->callback);	546	SvREFCNT_dec (req->callback);
480	Safefree (req->statdata);
481		547
482	if (req->type == REQ_READDIR)	548	if (req->flags & FLAG_PTR2_FREE)
483	free (req->data2ptr);	549	free (req->ptr2);
484		550
485	Safefree (req);	551	Safefree (req);
486	}	552	}
487		553
488	static void req_cancel_subs (aio_req grp)	554	static void req_cancel_subs (aio_req grp)
…		…
490	aio_req sub;	556	aio_req sub;
491		557
492	if (grp->type != REQ_GROUP)	558	if (grp->type != REQ_GROUP)
493	return;	559	return;
494		560
495	SvREFCNT_dec (grp->fh2);	561	SvREFCNT_dec (grp->sv2);
496	grp->fh2 = 0;	562	grp->sv2 = 0;
497		563
498	for (sub = grp->grp_first; sub; sub = sub->grp_next)	564	for (sub = grp->grp_first; sub; sub = sub->grp_next)
499	req_cancel (sub);	565	req_cancel (sub);
500	}	566	}
501		567
502	static void req_cancel (aio_req req)	568	static void req_cancel (aio_req req)
503	{	569	{
504	req->flags \|= FLAG_CANCELLED;	570	req->flags \|= FLAG_CANCELLED;
505		571
506	req_cancel_subs (req);	572	req_cancel_subs (req);
507	}
508
509	static int poll_cb (int max)
510	{
511	dSP;
512	int count = 0;
513	int do_croak = 0;
514	aio_req req;
515
516	for (;;)
517	{
518	while (max <= 0 \|\| count < max)
519	{
520	LOCK (reslock);
521	req = reqq_shift (&res_queue);
522
523	if (req)
524	{
525	--npending;
526
527	if (!res_queue.size)
528	{
529	/* read any signals sent by the worker threads */
530	char buf [32];
531	while (read (respipe [0], buf, 32) == 32)
532	;
533	}
534	}
535
536	UNLOCK (reslock);
537
538	if (!req)
539	break;
540
541	--nreqs;
542
543	if (req->type == REQ_QUIT)
544	--started;
545	else if (req->type == REQ_GROUP && req->length)
546	{
547	req->fd = 1; /* mark request as delayed */
548	continue;
549	}
550	else
551	{
552	if (req->type == REQ_READ)
553	SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
554
555	if (req->data2ptr && (req->type == REQ_READ \|\| req->type == REQ_WRITE))
556	SvREADONLY_off (req->data);
557
558	if (req->statdata)
559	{
560	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
561	PL_laststatval = req->result;
562	PL_statcache = *(req->statdata);
563	}
564
565	req_invoke (req);
566
567	count++;
568	}
569
570	req_free (req);
571	}
572
573	if (nreqs <= max_outstanding)
574	break;
575
576	poll_wait ();
577
578	max = 0;
579	}
580
581	return count;
582	}	573	}
583		574
584	static void aio_proc(void arg);	575	static void aio_proc(void arg);
585		576
586	static void start_thread (void)	577	static void start_thread (void)
…		…
594	croak ("unable to allocate worker thread data");	585	croak ("unable to allocate worker thread data");
595		586
596	pthread_attr_init (&attr);	587	pthread_attr_init (&attr);
597	pthread_attr_setstacksize (&attr, STACKSIZE);	588	pthread_attr_setstacksize (&attr, STACKSIZE);
598	pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);	589	pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
		590	#ifdef PTHREAD_SCOPE_PROCESS
		591	pthread_attr_setscope (&attr, PTHREAD_SCOPE_PROCESS);
		592	#endif
599		593
600	sigfillset (&fullsigset);	594	sigfillset (&fullsigset);
601		595
602	LOCK (wrklock);	596	LOCK (wrklock);
603	sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);	597	sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
…		…
615		609
616	sigprocmask (SIG_SETMASK, &oldsigset, 0);	610	sigprocmask (SIG_SETMASK, &oldsigset, 0);
617	UNLOCK (wrklock);	611	UNLOCK (wrklock);
618	}	612	}
619		613
		614	static void maybe_start_thread ()
		615	{
		616	if (get_nthreads () >= wanted)
		617	return;
		618
		619	/* todo: maybe use idle here, but might be less exact */
		620	if (0 <= (int)get_nthreads () + (int)get_npending () - (int)nreqs)
		621	return;
		622
		623	start_thread ();
		624	}
		625
620	static void req_send (aio_req req)	626	static void req_send (aio_req req)
621	{	627	{
622	while (started < wanted && nreqs >= started)
623	start_thread ();
624
625	++nreqs;	628	++nreqs;
626		629
627	LOCK (reqlock);	630	LOCK (reqlock);
628	++nready;	631	++nready;
629	reqq_push (&req_queue, req);	632	reqq_push (&req_queue, req);
630	pthread_cond_signal (&reqwait);	633	pthread_cond_signal (&reqwait);
631	UNLOCK (reqlock);	634	UNLOCK (reqlock);
		635
		636	maybe_start_thread ();
632	}	637	}
633		638
634	static void end_thread (void)	639	static void end_thread (void)
635	{	640	{
636	aio_req req;	641	aio_req req;
…		…
638	Newz (0, req, 1, aio_cb);	643	Newz (0, req, 1, aio_cb);
639		644
640	req->type = REQ_QUIT;	645	req->type = REQ_QUIT;
641	req->pri = PRI_MAX + PRI_BIAS;	646	req->pri = PRI_MAX + PRI_BIAS;
642		647
643	req_send (req);	648	LOCK (reqlock);
		649	reqq_push (&req_queue, req);
		650	pthread_cond_signal (&reqwait);
		651	UNLOCK (reqlock);
		652
		653	LOCK (wrklock);
		654	--started;
		655	UNLOCK (wrklock);
		656	}
		657
		658	static void set_max_idle (int nthreads)
		659	{
		660	if (WORDACCESS_UNSAFE) LOCK (reqlock);
		661	max_idle = nthreads <= 0 ? 1 : nthreads;
		662	if (WORDACCESS_UNSAFE) UNLOCK (reqlock);
644	}	663	}
645		664
646	static void min_parallel (int nthreads)	665	static void min_parallel (int nthreads)
647	{	666	{
648	if (wanted < nthreads)	667	if (wanted < nthreads)
649	wanted = nthreads;	668	wanted = nthreads;
650	}	669	}
651		670
652	static void max_parallel (int nthreads)	671	static void max_parallel (int nthreads)
653	{	672	{
654	int cur = started;
655
656	if (wanted > nthreads)	673	if (wanted > nthreads)
657	wanted = nthreads;	674	wanted = nthreads;
658		675
659	while (cur > wanted)	676	while (started > wanted)
		677	end_thread ();
		678	}
		679
		680	static void poll_wait ()
		681	{
		682	fd_set rfd;
		683
		684	while (nreqs)
660	{	685	{
		686	int size;
		687	if (WORDACCESS_UNSAFE) LOCK (reslock);
		688	size = res_queue.size;
		689	if (WORDACCESS_UNSAFE) UNLOCK (reslock);
		690
		691	if (size)
		692	return;
		693
661	end_thread ();	694	maybe_start_thread ();
662	cur--;	695
		696	FD_ZERO(&rfd);
		697	FD_SET(respipe [0], &rfd);
		698
		699	select (respipe [0] + 1, &rfd, 0, 0, 0);
663	}	700	}
		701	}
664		702
665	while (started > wanted)	703	static int poll_cb ()
		704	{
		705	dSP;
		706	int count = 0;
		707	int maxreqs = max_poll_reqs;
		708	int do_croak = 0;
		709	struct timeval tv_start, tv_now;
		710	aio_req req;
		711
		712	if (max_poll_time)
		713	gettimeofday (&tv_start, 0);
		714
		715	for (;;)
666	{	716	{
		717	for (;;)
		718	{
		719	maybe_start_thread ();
		720
		721	LOCK (reslock);
		722	req = reqq_shift (&res_queue);
		723
		724	if (req)
		725	{
		726	--npending;
		727
		728	if (!res_queue.size)
		729	{
		730	/* read any signals sent by the worker threads */
		731	char buf [32];
		732	while (read (respipe [0], buf, 32) == 32)
		733	;
		734	}
		735	}
		736
		737	UNLOCK (reslock);
		738
		739	if (!req)
		740	break;
		741
		742	--nreqs;
		743
		744	if (req->type == REQ_GROUP && req->size)
		745	{
		746	req->int1 = 1; /* mark request as delayed */
		747	continue;
		748	}
		749	else
		750	{
		751	req_invoke (req);
		752
		753	count++;
		754	}
		755
		756	req_free (req);
		757
		758	if (maxreqs && !--maxreqs)
		759	break;
		760
		761	if (max_poll_time)
		762	{
		763	gettimeofday (&tv_now, 0);
		764
		765	if (tvdiff (&tv_start, &tv_now) >= max_poll_time)
		766	break;
		767	}
		768	}
		769
		770	if (nreqs <= max_outstanding)
		771	break;
		772
667	poll_wait ();	773	poll_wait ();
668	poll_cb (0);	774
		775	++maxreqs;
669	}	776	}
		777
		778	return count;
670	}	779	}
671		780
672	static void create_pipe ()	781	static void create_pipe ()
673	{	782	{
674	if (pipe (respipe))	783	if (pipe (respipe))
…		…
889	int memofs = 0;	998	int memofs = 0;
890	int res = 0;	999	int res = 0;
891	int errorno;	1000	int errorno;
892		1001
893	LOCK (wrklock);	1002	LOCK (wrklock);
894	self->dirp = dirp = opendir (req->dataptr);	1003	self->dirp = dirp = opendir (req->ptr1);
895	self->dbuf = u = malloc (sizeof (*u));	1004	self->dbuf = u = malloc (sizeof (*u));
		1005	req->flags \|= FLAG_PTR2_FREE;
896	req->data2ptr = names = malloc (memlen);	1006	req->ptr2 = names = malloc (memlen);
897	UNLOCK (wrklock);	1007	UNLOCK (wrklock);
898		1008
899	if (dirp && u && names)	1009	if (dirp && u && names)
900	for (;;)	1010	for (;;)
901	{	1011	{
…		…
915		1025
916	while (memofs + len > memlen)	1026	while (memofs + len > memlen)
917	{	1027	{
918	memlen *= 2;	1028	memlen *= 2;
919	LOCK (wrklock);	1029	LOCK (wrklock);
920	req->data2ptr = names = realloc (names, memlen);	1030	req->ptr2 = names = realloc (names, memlen);
921	UNLOCK (wrklock);	1031	UNLOCK (wrklock);
922		1032
923	if (!names)	1033	if (!names)
924	break;	1034	break;
925	}	1035	}
…		…
938	/*****************************************************************************/	1048	/*****************************************************************************/
939		1049
940	static void aio_proc (void thr_arg)	1050	static void aio_proc (void thr_arg)
941	{	1051	{
942	aio_req req;	1052	aio_req req;
943	int type;	1053	struct timespec ts;
944	worker self = (worker )thr_arg;	1054	worker self = (worker )thr_arg;
945		1055
946	do	1056	/* try to distribute timeouts somewhat evenly */
		1057	ts.tv_nsec = (((unsigned long)self + (unsigned long)ts.tv_sec) & 1023UL)
		1058	* (1000000000UL / 1024UL);
		1059
		1060	for (;;)
947	{	1061	{
		1062	ts.tv_sec = time (0) + IDLE_TIMEOUT;
		1063
948	LOCK (reqlock);	1064	LOCK (reqlock);
949		1065
950	for (;;)	1066	for (;;)
951	{	1067	{
952	self->req = req = reqq_shift (&req_queue);	1068	self->req = req = reqq_shift (&req_queue);
953		1069
954	if (req)	1070	if (req)
955	break;	1071	break;
956		1072
		1073	++idle;
		1074
		1075	if (pthread_cond_timedwait (&reqwait, &reqlock, &ts)
		1076	== ETIMEDOUT)
		1077	{
		1078	if (idle > max_idle)
		1079	{
		1080	--idle;
		1081	UNLOCK (reqlock);
		1082	LOCK (wrklock);
		1083	--started;
		1084	UNLOCK (wrklock);
		1085	goto quit;
		1086	}
		1087
		1088	/* we are allowed to idle, so do so without any timeout */
957	pthread_cond_wait (&reqwait, &reqlock);	1089	pthread_cond_wait (&reqwait, &reqlock);
		1090	ts.tv_sec = time (0) + IDLE_TIMEOUT;
		1091	}
		1092
		1093	--idle;
958	}	1094	}
959		1095
960	--nready;	1096	--nready;
961		1097
962	UNLOCK (reqlock);	1098	UNLOCK (reqlock);
963		1099
964	errno = 0; /* strictly unnecessary */	1100	errno = 0; /* strictly unnecessary */
965	type = req->type; /* remember type for QUIT check */
966		1101
967	if (!(req->flags & FLAG_CANCELLED))	1102	if (!(req->flags & FLAG_CANCELLED))
968	switch (type)	1103	switch (req->type)
969	{	1104	{
970	case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;	1105	case REQ_READ: req->result = pread (req->int1, req->ptr1, req->size, req->offs); break;
971	case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;	1106	case REQ_WRITE: req->result = pwrite (req->int1, req->ptr1, req->size, req->offs); break;
972		1107
973	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;	1108	case REQ_READAHEAD: req->result = readahead (req->int1, req->offs, req->size); break;
974	case REQ_SENDFILE: req->result = sendfile_ (req->fd, req->fd2, req->offset, req->length, self); break;	1109	case REQ_SENDFILE: req->result = sendfile_ (req->int1, req->int2, req->offs, req->size, self); break;
975		1110
976	case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break;	1111	case REQ_STAT: req->result = stat (req->ptr1, (Stat_t *)req->ptr2); break;
977	case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break;	1112	case REQ_LSTAT: req->result = lstat (req->ptr1, (Stat_t *)req->ptr2); break;
978	case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break;	1113	case REQ_FSTAT: req->result = fstat (req->int1, (Stat_t *)req->ptr2); break;
979		1114
980	case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break;	1115	case REQ_OPEN: req->result = open (req->ptr1, req->int1, req->mode); break;
981	case REQ_CLOSE: req->result = close (req->fd); break;	1116	case REQ_CLOSE: req->result = close (req->int1); break;
982	case REQ_UNLINK: req->result = unlink (req->dataptr); break;	1117	case REQ_UNLINK: req->result = unlink (req->ptr1); break;
983	case REQ_RMDIR: req->result = rmdir (req->dataptr); break;	1118	case REQ_RMDIR: req->result = rmdir (req->ptr1); break;
984	case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break;	1119	case REQ_RENAME: req->result = rename (req->ptr2, req->ptr1); break;
985	case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break;	1120	case REQ_LINK: req->result = link (req->ptr2, req->ptr1); break;
986	case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;	1121	case REQ_SYMLINK: req->result = symlink (req->ptr2, req->ptr1); break;
		1122	case REQ_MKNOD: req->result = mknod (req->ptr2, req->mode, (dev_t)req->offs); break;
		1123	case REQ_READLINK: req->result = readlink (req->ptr2, req->ptr1, NAME_MAX); break;
987		1124
988	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;	1125	case REQ_FDATASYNC: req->result = fdatasync (req->int1); break;
989	case REQ_FSYNC: req->result = fsync (req->fd); break;	1126	case REQ_FSYNC: req->result = fsync (req->int1); break;
990	case REQ_READDIR: scandir_ (req, self); break;	1127	case REQ_READDIR: scandir_ (req, self); break;
991		1128
992	case REQ_BUSY:	1129	case REQ_BUSY:
993	{	1130	{
994	struct timeval tv;	1131	struct timeval tv;
995		1132
996	tv.tv_sec = req->fd;	1133	tv.tv_sec = req->int1;
997	tv.tv_usec = req->fd2;	1134	tv.tv_usec = req->int2;
998		1135
999	req->result = select (0, 0, 0, 0, &tv);	1136	req->result = select (0, 0, 0, 0, &tv);
1000	}	1137	}
1001		1138
1002	case REQ_GROUP:	1139	case REQ_GROUP:
1003	case REQ_NOP:	1140	case REQ_NOP:
		1141	break;
		1142
1004	case REQ_QUIT:	1143	case REQ_QUIT:
1005	break;	1144	goto quit;
1006		1145
1007	default:	1146	default:
1008	req->result = ENOSYS;	1147	req->result = ENOSYS;
1009	break;	1148	break;
1010	}	1149	}
…		…
1022	self->req = 0;	1161	self->req = 0;
1023	worker_clear (self);	1162	worker_clear (self);
1024		1163
1025	UNLOCK (reslock);	1164	UNLOCK (reslock);
1026	}	1165	}
1027	while (type != REQ_QUIT);
1028		1166
		1167	quit:
1029	LOCK (wrklock);	1168	LOCK (wrklock);
1030	worker_free (self);	1169	worker_free (self);
1031	UNLOCK (wrklock);	1170	UNLOCK (wrklock);
1032		1171
1033	return 0;	1172	return 0;
…		…
1080		1219
1081	worker_clear (wrk);	1220	worker_clear (wrk);
1082	worker_free (wrk);	1221	worker_free (wrk);
1083	}	1222	}
1084		1223
1085	started = 0;	1224	started = 0;
		1225	idle = 0;
1086	nreqs = 0;	1226	nreqs = 0;
		1227	nready = 0;
		1228	npending = 0;
1087		1229
1088	close (respipe [0]);	1230	close (respipe [0]);
1089	close (respipe [1]);	1231	close (respipe [1]);
1090	create_pipe ();	1232	create_pipe ();
1091		1233
…		…
1118	PROTOTYPES: ENABLE	1260	PROTOTYPES: ENABLE
1119		1261
1120	BOOT:	1262	BOOT:
1121	{	1263	{
1122	HV *stash = gv_stashpv ("IO::AIO", 1);	1264	HV *stash = gv_stashpv ("IO::AIO", 1);
		1265
1123	newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));	1266	newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));
1124	newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));	1267	newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));
1125	newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));	1268	newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));
		1269	newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT));
		1270	newCONSTSUB (stash, "O_TRUNC", newSViv (O_TRUNC));
		1271	newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO));
1126		1272
1127	create_pipe ();	1273	create_pipe ();
1128	pthread_atfork (atfork_prepare, atfork_parent, atfork_child);	1274	pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
1129	}	1275	}
1130		1276
1131	void	1277	void
		1278	max_poll_reqs (int nreqs)
		1279	PROTOTYPE: $
		1280	CODE:
		1281	max_poll_reqs = nreqs;
		1282
		1283	void
		1284	max_poll_time (double nseconds)
		1285	PROTOTYPE: $
		1286	CODE:
		1287	max_poll_time = nseconds * AIO_TICKS;
		1288
		1289	void
1132	min_parallel (int nthreads)	1290	min_parallel (int nthreads)
1133	PROTOTYPE: $	1291	PROTOTYPE: $
1134		1292
1135	void	1293	void
1136	max_parallel (int nthreads)	1294	max_parallel (int nthreads)
1137	PROTOTYPE: $	1295	PROTOTYPE: $
		1296
		1297	void
		1298	max_idle (int nthreads)
		1299	PROTOTYPE: $
		1300	CODE:
		1301	set_max_idle (nthreads);
1138		1302
1139	int	1303	int
1140	max_outstanding (int maxreqs)	1304	max_outstanding (int maxreqs)
1141	PROTOTYPE: $	1305	PROTOTYPE: $
1142	CODE:	1306	CODE:
…		…
1155	PPCODE:	1319	PPCODE:
1156	{	1320	{
1157	dREQ;	1321	dREQ;
1158		1322
1159	req->type = REQ_OPEN;	1323	req->type = REQ_OPEN;
1160	req->data = newSVsv (pathname);	1324	req->sv1 = newSVsv (pathname);
1161	req->dataptr = SvPVbyte_nolen (req->data);	1325	req->ptr1 = SvPVbyte_nolen (pathname);
1162	req->fd = flags;	1326	req->int1 = flags;
1163	req->mode = mode;	1327	req->mode = mode;
1164		1328
1165	REQ_SEND;	1329	REQ_SEND;
1166	}	1330	}
1167		1331
…		…
1177	PPCODE:	1341	PPCODE:
1178	{	1342	{
1179	dREQ;	1343	dREQ;
1180		1344
1181	req->type = ix;	1345	req->type = ix;
1182	req->fh = newSVsv (fh);	1346	req->fh = newSVsv (fh);
1183	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));	1347	req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh)));
1184		1348
1185	REQ_SEND (req);	1349	REQ_SEND (req);
1186	}	1350	}
1187		1351
1188	void	1352	void
…		…
1197	aio_read = REQ_READ	1361	aio_read = REQ_READ
1198	aio_write = REQ_WRITE	1362	aio_write = REQ_WRITE
1199	PROTOTYPE: $$$$$;$	1363	PROTOTYPE: $$$$$;$
1200	PPCODE:	1364	PPCODE:
1201	{	1365	{
1202	aio_req req;
1203	STRLEN svlen;	1366	STRLEN svlen;
1204	char *svptr = SvPVbyte (data, svlen);	1367	char *svptr = SvPVbyte (data, svlen);
1205		1368
1206	SvUPGRADE (data, SVt_PV);	1369	SvUPGRADE (data, SVt_PV);
1207	SvPOK_on (data);	1370	SvPOK_on (data);
…		…
1229		1392
1230	{	1393	{
1231	dREQ;	1394	dREQ;
1232		1395
1233	req->type = ix;	1396	req->type = ix;
1234	req->fh = newSVsv (fh);	1397	req->fh = newSVsv (fh);
1235	req->fd = PerlIO_fileno (ix == REQ_READ ? IoIFP (sv_2io (fh))	1398	req->int1 = PerlIO_fileno (ix == REQ_READ ? IoIFP (sv_2io (fh))
1236	: IoOFP (sv_2io (fh)));	1399	: IoOFP (sv_2io (fh)));
1237	req->offset = offset;	1400	req->offs = offset;
1238	req->length = length;	1401	req->size = length;
1239	req->data = SvREFCNT_inc (data);	1402	req->sv1 = SvREFCNT_inc (data);
1240	req->dataptr = (char *)svptr + dataoffset;	1403	req->ptr1 = (char *)svptr + dataoffset;
		1404	req->stroffset = dataoffset;
1241		1405
1242	if (!SvREADONLY (data))	1406	if (!SvREADONLY (data))
1243	{	1407	{
1244	SvREADONLY_on (data);	1408	SvREADONLY_on (data);
1245	req->data2ptr = (void *)data;	1409	req->flags \|= FLAG_SV1_RO_OFF;
1246	}	1410	}
1247		1411
1248	REQ_SEND;	1412	REQ_SEND;
1249	}	1413	}
		1414	}
		1415
		1416	void
		1417	aio_readlink (path,callback=&PL_sv_undef)
		1418	SV * path
		1419	SV * callback
		1420	PROTOTYPE: $$;$
		1421	PPCODE:
		1422	{
		1423	SV *data;
		1424	dREQ;
		1425
		1426	data = newSV (NAME_MAX);
		1427	SvPOK_on (data);
		1428
		1429	req->type = REQ_READLINK;
		1430	req->fh = newSVsv (path);
		1431	req->ptr2 = SvPVbyte_nolen (path);
		1432	req->sv1 = data;
		1433	req->ptr1 = SvPVbyte_nolen (data);
		1434
		1435	REQ_SEND;
1250	}	1436	}
1251		1437
1252	void	1438	void
1253	aio_sendfile (out_fh,in_fh,in_offset,length,callback=&PL_sv_undef)	1439	aio_sendfile (out_fh,in_fh,in_offset,length,callback=&PL_sv_undef)
1254	SV * out_fh	1440	SV * out_fh
…		…
1260	PPCODE:	1446	PPCODE:
1261	{	1447	{
1262	dREQ;	1448	dREQ;
1263		1449
1264	req->type = REQ_SENDFILE;	1450	req->type = REQ_SENDFILE;
1265	req->fh = newSVsv (out_fh);	1451	req->fh = newSVsv (out_fh);
1266	req->fd = PerlIO_fileno (IoIFP (sv_2io (out_fh)));	1452	req->int1 = PerlIO_fileno (IoIFP (sv_2io (out_fh)));
1267	req->fh2 = newSVsv (in_fh);	1453	req->sv2 = newSVsv (in_fh);
1268	req->fd2 = PerlIO_fileno (IoIFP (sv_2io (in_fh)));	1454	req->int2 = PerlIO_fileno (IoIFP (sv_2io (in_fh)));
1269	req->offset = in_offset;	1455	req->offs = in_offset;
1270	req->length = length;	1456	req->size = length;
1271		1457
1272	REQ_SEND;	1458	REQ_SEND;
1273	}	1459	}
1274		1460
1275	void	1461	void
…		…
1282	PPCODE:	1468	PPCODE:
1283	{	1469	{
1284	dREQ;	1470	dREQ;
1285		1471
1286	req->type = REQ_READAHEAD;	1472	req->type = REQ_READAHEAD;
1287	req->fh = newSVsv (fh);	1473	req->fh = newSVsv (fh);
1288	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));	1474	req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh)));
1289	req->offset = offset;	1475	req->offs = offset;
1290	req->length = length;	1476	req->size = length;
1291		1477
1292	REQ_SEND;	1478	REQ_SEND;
1293	}	1479	}
1294		1480
1295	void	1481	void
…		…
1301	aio_lstat = REQ_LSTAT	1487	aio_lstat = REQ_LSTAT
1302	PPCODE:	1488	PPCODE:
1303	{	1489	{
1304	dREQ;	1490	dREQ;
1305		1491
1306	New (0, req->statdata, 1, Stat_t);	1492	req->ptr2 = malloc (sizeof (Stat_t));
1307	if (!req->statdata)	1493	if (!req->ptr2)
1308	{	1494	{
1309	req_free (req);	1495	req_free (req);
1310	croak ("out of memory during aio_req->statdata allocation");	1496	croak ("out of memory during aio_req statdata allocation");
1311	}	1497	}
		1498
		1499	req->flags \|= FLAG_PTR2_FREE;
1312		1500
1313	if (SvPOK (fh_or_path))	1501	if (SvPOK (fh_or_path))
1314	{	1502	{
1315	req->type = ix;	1503	req->type = ix;
1316	req->data = newSVsv (fh_or_path);	1504	req->sv1 = newSVsv (fh_or_path);
1317	req->dataptr = SvPVbyte_nolen (req->data);	1505	req->ptr1 = SvPVbyte_nolen (fh_or_path);
1318	}	1506	}
1319	else	1507	else
1320	{	1508	{
1321	req->type = REQ_FSTAT;	1509	req->type = REQ_FSTAT;
1322	req->fh = newSVsv (fh_or_path);	1510	req->fh = newSVsv (fh_or_path);
1323	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));	1511	req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
1324	}	1512	}
1325		1513
1326	REQ_SEND;	1514	REQ_SEND;
1327	}	1515	}
1328		1516
1329	void	1517	void
1330	aio_unlink (pathname,callback=&PL_sv_undef)	1518	aio_unlink (pathname,callback=&PL_sv_undef)
1331	SV * pathname	1519	SV * pathname
1332	SV * callback	1520	SV * callback
1333	ALIAS:	1521	ALIAS:
1334	aio_unlink = REQ_UNLINK	1522	aio_unlink = REQ_UNLINK
1335	aio_rmdir = REQ_RMDIR	1523	aio_rmdir = REQ_RMDIR
1336	aio_readdir = REQ_READDIR	1524	aio_readdir = REQ_READDIR
1337	PPCODE:	1525	PPCODE:
1338	{	1526	{
1339	dREQ;	1527	dREQ;
1340		1528
1341	req->type = ix;	1529	req->type = ix;
1342	req->data = newSVsv (pathname);	1530	req->sv1 = newSVsv (pathname);
1343	req->dataptr = SvPVbyte_nolen (req->data);	1531	req->ptr1 = SvPVbyte_nolen (pathname);
1344		1532
1345	REQ_SEND;	1533	REQ_SEND;
1346	}	1534	}
1347		1535
1348	void	1536	void
1349	aio_link (oldpath,newpath,callback=&PL_sv_undef)	1537	aio_link (oldpath,newpath,callback=&PL_sv_undef)
1350	SV * oldpath	1538	SV * oldpath
1351	SV * newpath	1539	SV * newpath
1352	SV * callback	1540	SV * callback
1353	ALIAS:	1541	ALIAS:
1354	aio_link = REQ_LINK	1542	aio_link = REQ_LINK
1355	aio_symlink = REQ_SYMLINK	1543	aio_symlink = REQ_SYMLINK
1356	aio_rename = REQ_RENAME	1544	aio_rename = REQ_RENAME
1357	PPCODE:	1545	PPCODE:
1358	{	1546	{
1359	dREQ;	1547	dREQ;
1360		1548
1361	req->type = ix;	1549	req->type = ix;
1362	req->fh = newSVsv (oldpath);	1550	req->fh = newSVsv (oldpath);
1363	req->data2ptr = SvPVbyte_nolen (req->fh);	1551	req->ptr2 = SvPVbyte_nolen (req->fh);
1364	req->data = newSVsv (newpath);	1552	req->sv1 = newSVsv (newpath);
1365	req->dataptr = SvPVbyte_nolen (req->data);	1553	req->ptr1 = SvPVbyte_nolen (newpath);
1366		1554
1367	REQ_SEND;	1555	REQ_SEND;
1368	}	1556	}
1369		1557
1370	void	1558	void
		1559	aio_mknod (pathname,mode,dev,callback=&PL_sv_undef)
		1560	SV * pathname
		1561	SV * callback
		1562	UV mode
		1563	UV dev
		1564	PPCODE:
		1565	{
		1566	dREQ;
		1567
		1568	req->type = REQ_MKNOD;
		1569	req->sv1 = newSVsv (pathname);
		1570	req->ptr1 = SvPVbyte_nolen (pathname);
		1571	req->mode = (mode_t)mode;
		1572	req->offs = dev;
		1573
		1574	REQ_SEND;
		1575	}
		1576
		1577	void
1371	aio_busy (delay,callback=&PL_sv_undef)	1578	aio_busy (delay,callback=&PL_sv_undef)
1372	double delay	1579	double delay
1373	SV * callback	1580	SV * callback
1374	PPCODE:	1581	PPCODE:
1375	{	1582	{
1376	dREQ;	1583	dREQ;
1377		1584
1378	req->type = REQ_BUSY;	1585	req->type = REQ_BUSY;
1379	req->fd = delay < 0. ? 0 : delay;	1586	req->int1 = delay < 0. ? 0 : delay;
1380	req->fd2 = delay < 0. ? 0 : 1000. * (delay - req->fd);	1587	req->int2 = delay < 0. ? 0 : 1000. * (delay - req->int1);
1381		1588
1382	REQ_SEND;	1589	REQ_SEND;
1383	}	1590	}
1384		1591
1385	void	1592	void
…		…
1389	PPCODE:	1596	PPCODE:
1390	{	1597	{
1391	dREQ;	1598	dREQ;
1392		1599
1393	req->type = REQ_GROUP;	1600	req->type = REQ_GROUP;
		1601
1394	req_send (req);	1602	req_send (req);
1395
1396	XPUSHs (req_sv (req, AIO_GRP_KLASS));	1603	XPUSHs (req_sv (req, AIO_GRP_KLASS));
1397	}	1604	}
1398		1605
1399	void	1606	void
1400	aio_nop (callback=&PL_sv_undef)	1607	aio_nop (callback=&PL_sv_undef)
1401	SV * callback	1608	SV * callback
1402	PPCODE:	1609	PPCODE:
1403	{	1610	{
1404	dREQ;	1611	dREQ;
1405		1612
1406	req->type = REQ_NOP;	1613	req->type = REQ_NOP;
…		…
1460		1667
1461	int	1668	int
1462	poll_cb(...)	1669	poll_cb(...)
1463	PROTOTYPE:	1670	PROTOTYPE:
1464	CODE:	1671	CODE:
1465	RETVAL = poll_cb (0);	1672	RETVAL = poll_cb ();
1466	OUTPUT:
1467	RETVAL
1468
1469	int
1470	poll_some(int max = 0)
1471	PROTOTYPE: $
1472	CODE:
1473	RETVAL = poll_cb (max);
1474	OUTPUT:	1673	OUTPUT:
1475	RETVAL	1674	RETVAL
1476		1675
1477	void	1676	void
1478	poll_wait()	1677	poll_wait()
…		…
1491		1690
1492	int	1691	int
1493	nready()	1692	nready()
1494	PROTOTYPE:	1693	PROTOTYPE:
1495	CODE:	1694	CODE:
1496	if (WORDREAD_UNSAFE) LOCK (reqlock);
1497	RETVAL = nready;	1695	RETVAL = get_nready ();
1498	if (WORDREAD_UNSAFE) UNLOCK (reqlock);
1499	OUTPUT:	1696	OUTPUT:
1500	RETVAL	1697	RETVAL
1501		1698
1502	int	1699	int
1503	npending()	1700	npending()
1504	PROTOTYPE:	1701	PROTOTYPE:
1505	CODE:	1702	CODE:
1506	if (WORDREAD_UNSAFE) LOCK (reslock);
1507	RETVAL = npending;	1703	RETVAL = get_npending ();
		1704	OUTPUT:
		1705	RETVAL
		1706
		1707	int
		1708	nthreads()
		1709	PROTOTYPE:
		1710	CODE:
		1711	if (WORDACCESS_UNSAFE) LOCK (wrklock);
		1712	RETVAL = started;
1508	if (WORDREAD_UNSAFE) UNLOCK (reslock);	1713	if (WORDACCESS_UNSAFE) UNLOCK (wrklock);
1509	OUTPUT:	1714	OUTPUT:
1510	RETVAL	1715	RETVAL
1511		1716
1512	PROTOTYPES: DISABLE	1717	PROTOTYPES: DISABLE
1513		1718
…		…
1531	PPCODE:	1736	PPCODE:
1532	{	1737	{
1533	int i;	1738	int i;
1534	aio_req req;	1739	aio_req req;
1535		1740
1536	if (grp->fd == 2)	1741	if (grp->int1 == 2)
1537	croak ("cannot add requests to IO::AIO::GRP after the group finished");	1742	croak ("cannot add requests to IO::AIO::GRP after the group finished");
1538		1743
1539	for (i = 1; i < items; ++i )	1744	for (i = 1; i < items; ++i )
1540	{	1745	{
1541	if (GIMME_V != G_VOID)	1746	if (GIMME_V != G_VOID)
…		…
1543		1748
1544	req = SvAIO_REQ (ST (i));	1749	req = SvAIO_REQ (ST (i));
1545		1750
1546	if (req)	1751	if (req)
1547	{	1752	{
1548	++grp->length;	1753	++grp->size;
1549	req->grp = grp;	1754	req->grp = grp;
1550		1755
1551	req->grp_prev = 0;	1756	req->grp_prev = 0;
1552	req->grp_next = grp->grp_first;	1757	req->grp_next = grp->grp_first;
1553		1758
…		…
1576	av = newAV ();	1781	av = newAV ();
1577		1782
1578	for (i = 1; i < items; ++i )	1783	for (i = 1; i < items; ++i )
1579	av_push (av, newSVsv (ST (i)));	1784	av_push (av, newSVsv (ST (i)));
1580		1785
1581	SvREFCNT_dec (grp->data);	1786	SvREFCNT_dec (grp->sv1);
1582	grp->data = (SV *)av;	1787	grp->sv1 = (SV *)av;
1583	}	1788	}
1584		1789
1585	void	1790	void
1586	errno (aio_req grp, int errorno = errno)	1791	errno (aio_req grp, int errorno = errno)
1587	CODE:	1792	CODE:
1588	grp->errorno = errorno;	1793	grp->errorno = errorno;
1589		1794
1590	void	1795	void
1591	limit (aio_req grp, int limit)	1796	limit (aio_req grp, int limit)
1592	CODE:	1797	CODE:
1593	grp->fd2 = limit;	1798	grp->int2 = limit;
1594	aio_grp_feed (grp);	1799	aio_grp_feed (grp);
1595		1800
1596	void	1801	void
1597	feed (aio_req grp, SV *callback=&PL_sv_undef)	1802	feed (aio_req grp, SV *callback=&PL_sv_undef)
1598	CODE:	1803	CODE:
1599	{	1804	{
1600	SvREFCNT_dec (grp->fh2);	1805	SvREFCNT_dec (grp->sv2);
1601	grp->fh2 = newSVsv (callback);	1806	grp->sv2 = newSVsv (callback);
1602		1807
1603	if (grp->fd2 <= 0)	1808	if (grp->int2 <= 0)
1604	grp->fd2 = 2;	1809	grp->int2 = 2;
1605		1810
1606	aio_grp_feed (grp);	1811	aio_grp_feed (grp);
1607	}	1812	}
1608		1813

Diff Legend

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

Comparing IO-AIO/AIO.xs (file contents): Revision 1.79 by root, Thu Oct 26 16:28:33 2006 UTC vs. Revision 1.87 by root, Mon Oct 30 23:42:27 2006 UTC

Diff Legend

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.79 by root, Thu Oct 26 16:28:33 2006 UTC vs.
Revision 1.87 by root, Mon Oct 30 23:42:27 2006 UTC