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

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.71 by root, Tue Oct 24 16:35:04 2006 UTC vs.
Revision 1.84 by root, Sat Oct 28 01:24:19 2006 UTC

…		…
1	/* solaris */	1	/* solaris */
2	#define _POSIX_PTHREAD_SEMANTICS 1	2	#define _POSIX_PTHREAD_SEMANTICS 1
3		3
4	#if __linux	4	#if __linux && !defined(_GNU_SOURCE)
5	# define _GNU_SOURCE	5	# define _GNU_SOURCE
6	#endif	6	#endif
7		7
		8	/* just in case */
8	#define _REENTRANT 1	9	#define _REENTRANT 1
9		10
10	#include <errno.h>	11	#include <errno.h>
11		12
12	#include "EXTERN.h"	13	#include "EXTERN.h"
42	# else	43	# else
43	# error sendfile support requested but not available	44	# error sendfile support requested but not available
44	# endif	45	# endif
45	#endif	46	#endif
46		47
		48	/* number of seconds after which idle threads exit */
		49	#define IDLE_TIMEOUT 10
		50
47	/* used for struct dirent, AIX doesn't provide it */	51	/* used for struct dirent, AIX doesn't provide it */
48	#ifndef NAME_MAX	52	#ifndef NAME_MAX
49	# define NAME_MAX 4096	53	# define NAME_MAX 4096
		54	#endif
		55
		56	#ifndef PTHREAD_STACK_MIN
		57	/* care for broken platforms, e.g. windows */
		58	# define PTHREAD_STACK_MIN 16384
50	#endif	59	#endif
51		60
52	#if __ia64	61	#if __ia64
53	# define STACKSIZE 65536	62	# define STACKSIZE 65536
54	#elif __i386 \|\| __x86_64 /* 16k is unreasonably high :( */	63	#elif __i386 \|\| __x86_64 /* 16k is unreasonably high :( */
55	# define STACKSIZE PTHREAD_STACK_MIN	64	# define STACKSIZE PTHREAD_STACK_MIN
56	#else	65	#else
57	# define STACKSIZE 16384	66	# define STACKSIZE 16384
		67	#endif
		68
		69	/* wether word reads are potentially non-atomic.
		70	* this is conservatice, likely most arches this runs
		71	* on have atomic word read/writes.
		72	*/
		73	#ifndef WORDREAD_UNSAFE
		74	# if __i386 \|\| __x86_64
		75	# define WORDREAD_UNSAFE 0
		76	# else
		77	# define WORDREAD_UNSAFE 1
		78	# endif
58	#endif	79	#endif
59		80
60	/* buffer size for various temporary buffers */	81	/* buffer size for various temporary buffers */
61	#define AIO_BUFSIZE 65536	82	#define AIO_BUFSIZE 65536
62		83
…		…
74	REQ_READ, REQ_WRITE, REQ_READAHEAD,	95	REQ_READ, REQ_WRITE, REQ_READAHEAD,
75	REQ_SENDFILE,	96	REQ_SENDFILE,
76	REQ_STAT, REQ_LSTAT, REQ_FSTAT,	97	REQ_STAT, REQ_LSTAT, REQ_FSTAT,
77	REQ_FSYNC, REQ_FDATASYNC,	98	REQ_FSYNC, REQ_FDATASYNC,
78	REQ_UNLINK, REQ_RMDIR, REQ_RENAME,	99	REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
79	REQ_READDIR,	100	REQ_MKNOD, REQ_READDIR,
80	REQ_LINK, REQ_SYMLINK,	101	REQ_LINK, REQ_SYMLINK,
81	REQ_GROUP, REQ_NOP,	102	REQ_GROUP, REQ_NOP,
82	REQ_BUSY,	103	REQ_BUSY,
83	};	104	};
84		105
…		…
126	NUM_PRI = PRI_MAX + PRI_BIAS + 1,	147	NUM_PRI = PRI_MAX + PRI_BIAS + 1,
127	};	148	};
128		149
129	static int next_pri = DEFAULT_PRI + PRI_BIAS;	150	static int next_pri = DEFAULT_PRI + PRI_BIAS;
130		151
131	static int started, wanted;	152	static unsigned int started, idle, wanted;
132	static volatile int nreqs;
133	static int max_outstanding = 1<<30;
134	static int respipe [2];
135		153
136	#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)	154	#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
137	# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP	155	# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
138	#else	156	#else
139	# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER	157	# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
140	#endif	158	#endif
141		159
142	#define LOCK(mutex) pthread_mutex_lock (&(mutex))	160	#define LOCK(mutex) pthread_mutex_lock (&(mutex))
143	#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex))	161	#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex))
144		162
145	/* worker threasd management */	163	/* worker threads management */
146	static pthread_mutex_t wrklock = AIO_MUTEX_INIT;	164	static pthread_mutex_t wrklock = AIO_MUTEX_INIT;
147		165
148	typedef struct worker {	166	typedef struct worker {
149	/* locked by wrklock */	167	/* locked by wrklock */
150	struct worker prev, next;	168	struct worker prev, next;
…		…
180	wrk->prev->next = wrk->next;	198	wrk->prev->next = wrk->next;
181		199
182	free (wrk);	200	free (wrk);
183	}	201	}
184		202
		203	static volatile unsigned int nreqs, nready, npending;
		204	static volatile unsigned int max_idle = 4;
		205	static volatile unsigned int max_outstanding = 0xffffffff;
		206	static int respipe [2];
		207
185	static pthread_mutex_t reslock = AIO_MUTEX_INIT;	208	static pthread_mutex_t reslock = AIO_MUTEX_INIT;
186	static pthread_mutex_t reqlock = AIO_MUTEX_INIT;	209	static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
187	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;	210	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
		211
		212	#if WORDREAD_UNSAFE
		213
		214	static unsigned int get_nready ()
		215	{
		216	unsigned int retval;
		217
		218	LOCK (reqlock);
		219	retval = nready;
		220	UNLOCK (reqlock);
		221
		222	return retval;
		223	}
		224
		225	static unsigned int get_npending ()
		226	{
		227	unsigned int retval;
		228
		229	LOCK (reslock);
		230	retval = npending;
		231	UNLOCK (reslock);
		232
		233	return retval;
		234	}
		235
		236	#else
		237
		238	# define get_nready() nready
		239	# define get_npending() npending
		240
		241	#endif
188		242
189	/*	243	/*
190	* a somewhat faster data structure might be nice, but	244	* a somewhat faster data structure might be nice, but
191	* with 8 priorities this actually needs <20 insns	245	* with 8 priorities this actually needs <20 insns
192	* per shift, the most expensive operation.	246	* per shift, the most expensive operation.
…		…
238	}	292	}
239		293
240	abort ();	294	abort ();
241	}	295	}
242		296
		297	static int poll_cb (int max);
243	static void req_invoke (aio_req req);	298	static void req_invoke (aio_req req);
244	static void req_free (aio_req req);	299	static void req_free (aio_req req);
		300	static void req_cancel (aio_req req);
245		301
246	/* must be called at most once */	302	/* must be called at most once */
247	static SV req_sv (aio_req req, const char klass)	303	static SV req_sv (aio_req req, const char klass)
248	{	304	{
249	if (!req->self)	305	if (!req->self)
…		…
311	req_invoke (grp);	367	req_invoke (grp);
312	req_free (grp);	368	req_free (grp);
313	}	369	}
314	}	370	}
315		371
316	static void poll_wait ()
317	{
318	fd_set rfd;
319
320	while (nreqs)
321	{
322	int size;
323	#if !(__i386 \|\| __x86_64) /* safe without sempahore on these archs */
324	LOCK (reslock);
325	#endif
326	size = res_queue.size;
327	#if !(__i386 \|\| __x86_64) /* safe without sempahore on these archs */
328	UNLOCK (reslock);
329	#endif
330
331	if (size)
332	return;
333
334	FD_ZERO(&rfd);
335	FD_SET(respipe [0], &rfd);
336
337	select (respipe [0] + 1, &rfd, 0, 0, 0);
338	}
339	}
340
341	static void req_invoke (aio_req req)	372	static void req_invoke (aio_req req)
342	{	373	{
343	dSP;	374	dSP;
344		375
345	if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))	376	if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))
346	{	377	{
347	errno = req->errorno;
348
349	ENTER;	378	ENTER;
350	SAVETMPS;	379	SAVETMPS;
351	PUSHMARK (SP);	380	PUSHMARK (SP);
352	EXTEND (SP, 1);	381	EXTEND (SP, 1);
353		382
…		…
418	default:	447	default:
419	PUSHs (sv_2mortal (newSViv (req->result)));	448	PUSHs (sv_2mortal (newSViv (req->result)));
420	break;	449	break;
421	}	450	}
422		451
		452	errno = req->errorno;
423		453
424	PUTBACK;	454	PUTBACK;
425	call_sv (req->callback, G_VOID \| G_EVAL);	455	call_sv (req->callback, G_VOID \| G_EVAL);
426	SPAGAIN;	456	SPAGAIN;
427		457
…		…
468	free (req->data2ptr);	498	free (req->data2ptr);
469		499
470	Safefree (req);	500	Safefree (req);
471	}	501	}
472		502
		503	static void req_cancel_subs (aio_req grp)
		504	{
		505	aio_req sub;
		506
		507	if (grp->type != REQ_GROUP)
		508	return;
		509
		510	SvREFCNT_dec (grp->fh2);
		511	grp->fh2 = 0;
		512
		513	for (sub = grp->grp_first; sub; sub = sub->grp_next)
		514	req_cancel (sub);
		515	}
		516
473	static void req_cancel (aio_req req)	517	static void req_cancel (aio_req req)
474	{	518	{
475	req->flags \|= FLAG_CANCELLED;	519	req->flags \|= FLAG_CANCELLED;
476		520
477	if (req->type == REQ_GROUP)	521	req_cancel_subs (req);
478	{	522	}
479	aio_req sub;
480		523
481	for (sub = req->grp_first; sub; sub = sub->grp_next)	524	static void aio_proc(void arg);
482	req_cancel (sub);	525
		526	static void start_thread (void)
		527	{
		528	sigset_t fullsigset, oldsigset;
		529	pthread_attr_t attr;
		530
		531	worker *wrk = calloc (1, sizeof (worker));
		532
		533	if (!wrk)
		534	croak ("unable to allocate worker thread data");
		535
		536	pthread_attr_init (&attr);
		537	pthread_attr_setstacksize (&attr, STACKSIZE);
		538	pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
		539	#ifdef PTHREAD_SCOPE_PROCESS
		540	pthread_attr_setscope (&attr, PTHREAD_SCOPE_PROCESS);
		541	#endif
		542
		543	sigfillset (&fullsigset);
		544
		545	LOCK (wrklock);
		546	sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
		547
		548	if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
483	}	549	{
484	}	550	wrk->prev = &wrk_first;
		551	wrk->next = wrk_first.next;
		552	wrk_first.next->prev = wrk;
		553	wrk_first.next = wrk;
		554	++started;
		555	}
		556	else
		557	free (wrk);
485		558
		559	sigprocmask (SIG_SETMASK, &oldsigset, 0);
		560	UNLOCK (wrklock);
		561	}
		562
		563	static void maybe_start_thread ()
		564	{
		565	if (started >= wanted)
		566	return;
		567
		568	/* todo: maybe use idle here, but might be less exact */
		569	if ((int)nready <= (int)started - (int)(nreqs - get_nready () - get_npending ()))
		570	return;
		571
		572	start_thread ();
		573	}
		574
		575	static void req_send (aio_req req)
		576	{
		577	++nreqs;
		578
		579	LOCK (reqlock);
		580	++nready;
		581	reqq_push (&req_queue, req);
		582	pthread_cond_signal (&reqwait);
		583	UNLOCK (reqlock);
		584
		585	maybe_start_thread ();
		586	}
		587
		588	static void end_thread (void)
		589	{
		590	aio_req req;
		591
		592	Newz (0, req, 1, aio_cb);
		593
		594	req->type = REQ_QUIT;
		595	req->pri = PRI_MAX + PRI_BIAS;
		596
		597	LOCK (reqlock);
		598	reqq_push (&req_queue, req);
		599	pthread_cond_signal (&reqwait);
		600	UNLOCK (reqlock);
		601
		602	LOCK (wrklock);
		603	--started;
		604	UNLOCK (wrklock);
		605	}
		606
		607	static void min_parallel (int nthreads)
		608	{
		609	if (wanted < nthreads)
		610	wanted = nthreads;
		611	}
		612
		613	static void max_parallel (int nthreads)
		614	{
		615	if (wanted > nthreads)
		616	wanted = nthreads;
		617
		618	while (started > wanted)
		619	end_thread ();
		620	}
		621
		622	static void poll_wait ()
		623	{
		624	fd_set rfd;
		625
		626	while (nreqs)
		627	{
		628	int size;
		629	if (WORDREAD_UNSAFE) LOCK (reslock);
		630	size = res_queue.size;
		631	if (WORDREAD_UNSAFE) UNLOCK (reslock);
		632
		633	if (size)
		634	return;
		635
		636	maybe_start_thread ();
		637
		638	FD_ZERO(&rfd);
		639	FD_SET(respipe [0], &rfd);
		640
		641	select (respipe [0] + 1, &rfd, 0, 0, 0);
		642	}
		643	}
		644
486	static int poll_cb ()	645	static int poll_cb (int max)
487	{	646	{
488	dSP;	647	dSP;
489	int count = 0;	648	int count = 0;
490	int do_croak = 0;	649	int do_croak = 0;
491	aio_req req;	650	aio_req req;
492		651
493	for (;;)	652	for (;;)
494	{	653	{
495	LOCK (reslock);	654	while (max <= 0 \|\| count < max)
496	req = reqq_shift (&res_queue);
497
498	if (req)
499	{	655	{
		656	maybe_start_thread ();
		657
		658	LOCK (reslock);
		659	req = reqq_shift (&res_queue);
		660
500	if (!res_queue.size)	661	if (req)
501	{	662	{
		663	--npending;
		664
		665	if (!res_queue.size)
		666	{
502	/* read any signals sent by the worker threads */	667	/* read any signals sent by the worker threads */
503	char buf [32];	668	char buf [32];
504	while (read (respipe [0], buf, 32) == 32)	669	while (read (respipe [0], buf, 32) == 32)
		670	;
505	;	671	}
506	}	672	}
		673
		674	UNLOCK (reslock);
		675
		676	if (!req)
		677	break;
		678
		679	--nreqs;
		680
		681	if (req->type == REQ_GROUP && req->length)
		682	{
		683	req->fd = 1; /* mark request as delayed */
		684	continue;
		685	}
		686	else
		687	{
		688	if (req->type == REQ_READ)
		689	SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
		690
		691	if (req->data2ptr && (req->type == REQ_READ \|\| req->type == REQ_WRITE))
		692	SvREADONLY_off (req->data);
		693
		694	if (req->statdata)
		695	{
		696	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
		697	PL_laststatval = req->result;
		698	PL_statcache = *(req->statdata);
		699	}
		700
		701	req_invoke (req);
		702
		703	count++;
		704	}
		705
		706	req_free (req);
507	}	707	}
508		708
509	UNLOCK (reslock);	709	if (nreqs <= max_outstanding)
510
511	if (!req)
512	break;	710	break;
513		711
514	--nreqs;	712	poll_wait ();
515		713
516	if (req->type == REQ_QUIT)	714	max = 0;
517	started--;
518	else if (req->type == REQ_GROUP && req->length)
519	{
520	req->fd = 1; /* mark request as delayed */
521	continue;
522	}
523	else
524	{
525	if (req->type == REQ_READ)
526	SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
527
528	if (req->data2ptr && (req->type == REQ_READ \|\| req->type == REQ_WRITE))
529	SvREADONLY_off (req->data);
530
531	if (req->statdata)
532	{
533	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
534	PL_laststatval = req->result;
535	PL_statcache = *(req->statdata);
536	}
537
538	req_invoke (req);
539
540	count++;
541	}
542
543	req_free (req);
544	}	715	}
545		716
546	return count;	717	return count;
547	}
548
549	static void aio_proc(void arg);
550
551	static void start_thread (void)
552	{
553	worker *wrk = calloc (1, sizeof (worker));
554
555	if (!wrk)
556	croak ("unable to allocate worker thread data");
557
558	sigset_t fullsigset, oldsigset;
559	pthread_attr_t attr;
560
561	pthread_attr_init (&attr);
562	pthread_attr_setstacksize (&attr, STACKSIZE);
563	pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
564
565	sigfillset (&fullsigset);
566
567	LOCK (wrklock);
568	sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
569
570	if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
571	{
572	wrk->prev = &wrk_first;
573	wrk->next = wrk_first.next;
574	wrk_first.next->prev = wrk;
575	wrk_first.next = wrk;
576	started++;
577	}
578	else
579	free (wrk);
580
581	sigprocmask (SIG_SETMASK, &oldsigset, 0);
582	UNLOCK (wrklock);
583	}
584
585	static void req_send (aio_req req)
586	{
587	while (started < wanted && nreqs >= started)
588	start_thread ();
589
590	++nreqs;
591
592	LOCK (reqlock);
593	reqq_push (&req_queue, req);
594	pthread_cond_signal (&reqwait);
595	UNLOCK (reqlock);
596
597	if (nreqs > max_outstanding)
598	for (;;)
599	{
600	poll_cb ();
601
602	if (nreqs <= max_outstanding)
603	break;
604
605	poll_wait ();
606	}
607	}
608
609	static void end_thread (void)
610	{
611	aio_req req;
612
613	Newz (0, req, 1, aio_cb);
614
615	req->type = REQ_QUIT;
616	req->pri = PRI_MAX + PRI_BIAS;
617
618	req_send (req);
619	}
620
621	static void min_parallel (int nthreads)
622	{
623	if (wanted < nthreads)
624	wanted = nthreads;
625	}
626
627	static void max_parallel (int nthreads)
628	{
629	int cur = started;
630
631	if (wanted > nthreads)
632	wanted = nthreads;
633
634	while (cur > wanted)
635	{
636	end_thread ();
637	cur--;
638	}
639
640	while (started > wanted)
641	{
642	poll_wait ();
643	poll_cb ();
644	}
645	}	718	}
646		719
647	static void create_pipe ()	720	static void create_pipe ()
648	{	721	{
649	if (pipe (respipe))	722	if (pipe (respipe))
…		…
704	#if !HAVE_FDATASYNC	777	#if !HAVE_FDATASYNC
705	# define fdatasync fsync	778	# define fdatasync fsync
706	#endif	779	#endif
707		780
708	#if !HAVE_READAHEAD	781	#if !HAVE_READAHEAD
709	# define readahead aio_readahead	782	# define readahead(fd,offset,count) aio_readahead (fd, offset, count, self)
710		783
711	static ssize_t readahead (int fd, off_t offset, size_t count)	784	static ssize_t aio_readahead (int fd, off_t offset, size_t count, worker *self)
712	{	785	{
713	dBUF;	786	dBUF;
714		787
715	while (count > 0)	788	while (count > 0)
716	{	789	{
…		…
721	count -= len;	794	count -= len;
722	}	795	}
723		796
724	errno = 0;	797	errno = 0;
725	}	798	}
		799
726	#endif	800	#endif
727		801
728	#if !HAVE_READDIR_R	802	#if !HAVE_READDIR_R
729	# define readdir_r aio_readdir_r	803	# define readdir_r aio_readdir_r
730		804
…		…
775	{	849	{
776	off_t sbytes;	850	off_t sbytes;
777	res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0);	851	res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0);
778		852
779	if (res < 0 && sbytes)	853	if (res < 0 && sbytes)
780	/* maybe only on EAGAIN only: as usual, the manpage leaves you guessing */	854	/* maybe only on EAGAIN: as usual, the manpage leaves you guessing */
781	res = sbytes;	855	res = sbytes;
782	}	856	}
783		857
784	# elif __hpux	858	# elif __hpux
785	res = sendfile (ofd, ifd, offset, count, 0, 0);	859	res = sendfile (ofd, ifd, offset, count, 0, 0);
…		…
865	int errorno;	939	int errorno;
866		940
867	LOCK (wrklock);	941	LOCK (wrklock);
868	self->dirp = dirp = opendir (req->dataptr);	942	self->dirp = dirp = opendir (req->dataptr);
869	self->dbuf = u = malloc (sizeof (*u));	943	self->dbuf = u = malloc (sizeof (*u));
		944	req->data2ptr = names = malloc (memlen);
870	UNLOCK (wrklock);	945	UNLOCK (wrklock);
871
872	req->data2ptr = names = malloc (memlen);
873		946
874	if (dirp && u && names)	947	if (dirp && u && names)
875	for (;;)	948	for (;;)
876	{	949	{
877	errno = 0;	950	errno = 0;
…		…
913	/*****************************************************************************/	986	/*****************************************************************************/
914		987
915	static void aio_proc (void thr_arg)	988	static void aio_proc (void thr_arg)
916	{	989	{
917	aio_req req;	990	aio_req req;
918	int type;	991	struct timespec ts;
919	worker self = (worker )thr_arg;	992	worker self = (worker )thr_arg;
920		993
921	do	994	/* try to distribute timeouts somewhat evenly */
		995	ts.tv_nsec = (((unsigned long)self + (unsigned long)ts.tv_sec) & 1023UL)
		996	* (1000000000UL / 1024UL);
		997
		998	for (;;)
922	{	999	{
		1000	ts.tv_sec = time (0) + IDLE_TIMEOUT;
		1001
923	LOCK (reqlock);	1002	LOCK (reqlock);
924		1003
925	for (;;)	1004	for (;;)
926	{	1005	{
927	self->req = req = reqq_shift (&req_queue);	1006	self->req = req = reqq_shift (&req_queue);
928		1007
929	if (req)	1008	if (req)
930	break;	1009	break;
931		1010
		1011	++idle;
		1012
		1013	if (pthread_cond_timedwait (&reqwait, &reqlock, &ts)
		1014	== ETIMEDOUT)
		1015	{
		1016	if (idle > max_idle)
		1017	{
		1018	--idle;
		1019	UNLOCK (reqlock);
		1020	LOCK (wrklock);
		1021	--started;
		1022	UNLOCK (wrklock);
		1023	goto quit;
		1024	}
		1025
		1026	/* we are allowed to idle, so do so without any timeout */
932	pthread_cond_wait (&reqwait, &reqlock);	1027	pthread_cond_wait (&reqwait, &reqlock);
		1028	ts.tv_sec = time (0) + IDLE_TIMEOUT;
		1029	}
		1030
		1031	--idle;
933	}	1032	}
		1033
		1034	--nready;
934		1035
935	UNLOCK (reqlock);	1036	UNLOCK (reqlock);
936		1037
937	errno = 0; /* strictly unnecessary */	1038	errno = 0; /* strictly unnecessary */
938	type = req->type; /* remember type for QUIT check */
939		1039
940	if (!(req->flags & FLAG_CANCELLED))	1040	if (!(req->flags & FLAG_CANCELLED))
941	switch (type)	1041	switch (req->type)
942	{	1042	{
943	case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;	1043	case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
944	case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;	1044	case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
945		1045
946	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;	1046	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
…		…
955	case REQ_UNLINK: req->result = unlink (req->dataptr); break;	1055	case REQ_UNLINK: req->result = unlink (req->dataptr); break;
956	case REQ_RMDIR: req->result = rmdir (req->dataptr); break;	1056	case REQ_RMDIR: req->result = rmdir (req->dataptr); break;
957	case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break;	1057	case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break;
958	case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break;	1058	case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break;
959	case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;	1059	case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;
		1060	case REQ_MKNOD: req->result = mknod (req->data2ptr, req->mode, (dev_t)req->offset); break;
960		1061
961	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;	1062	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
962	case REQ_FSYNC: req->result = fsync (req->fd); break;	1063	case REQ_FSYNC: req->result = fsync (req->fd); break;
963	case REQ_READDIR: scandir_ (req, self); break;	1064	case REQ_READDIR: scandir_ (req, self); break;
964		1065
…		…
972	req->result = select (0, 0, 0, 0, &tv);	1073	req->result = select (0, 0, 0, 0, &tv);
973	}	1074	}
974		1075
975	case REQ_GROUP:	1076	case REQ_GROUP:
976	case REQ_NOP:	1077	case REQ_NOP:
		1078	break;
		1079
977	case REQ_QUIT:	1080	case REQ_QUIT:
978	break;	1081	goto quit;
979		1082
980	default:	1083	default:
981	req->result = ENOSYS;	1084	req->result = ENOSYS;
982	break;	1085	break;
983	}	1086	}
984		1087
985	req->errorno = errno;	1088	req->errorno = errno;
986		1089
987	LOCK (reslock);	1090	LOCK (reslock);
988		1091
		1092	++npending;
		1093
989	if (!reqq_push (&res_queue, req))	1094	if (!reqq_push (&res_queue, req))
990	/* write a dummy byte to the pipe so fh becomes ready */	1095	/* write a dummy byte to the pipe so fh becomes ready */
991	write (respipe [1], &respipe, 1);	1096	write (respipe [1], &respipe, 1);
992		1097
993	self->req = 0;	1098	self->req = 0;
994	worker_clear (self);	1099	worker_clear (self);
995		1100
996	UNLOCK (reslock);	1101	UNLOCK (reslock);
997	}	1102	}
998	while (type != REQ_QUIT);
999		1103
		1104	quit:
1000	LOCK (wrklock);	1105	LOCK (wrklock);
1001	worker_free (self);	1106	worker_free (self);
1002	UNLOCK (wrklock);	1107	UNLOCK (wrklock);
1003		1108
1004	return 0;	1109	return 0;
…		…
1051		1156
1052	worker_clear (wrk);	1157	worker_clear (wrk);
1053	worker_free (wrk);	1158	worker_free (wrk);
1054	}	1159	}
1055		1160
1056	started = 0;	1161	started = 0;
		1162	idle = 0;
1057	nreqs = 0;	1163	nreqs = 0;
		1164	nready = 0;
		1165	npending = 0;
1058		1166
1059	close (respipe [0]);	1167	close (respipe [0]);
1060	close (respipe [1]);	1168	close (respipe [1]);
1061	create_pipe ();	1169	create_pipe ();
1062		1170
…		…
1089	PROTOTYPES: ENABLE	1197	PROTOTYPES: ENABLE
1090		1198
1091	BOOT:	1199	BOOT:
1092	{	1200	{
1093	HV *stash = gv_stashpv ("IO::AIO", 1);	1201	HV *stash = gv_stashpv ("IO::AIO", 1);
		1202
1094	newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));	1203	newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));
1095	newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));	1204	newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));
1096	newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));	1205	newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));
		1206	newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT));
		1207	newCONSTSUB (stash, "O_TRUNC", newSViv (O_TRUNC));
		1208	newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO));
1097		1209
1098	create_pipe ();	1210	create_pipe ();
1099	pthread_atfork (atfork_prepare, atfork_parent, atfork_child);	1211	pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
1100	}	1212	}
1101		1213
1102	void	1214	void
1103	min_parallel (nthreads)	1215	min_parallel (int nthreads)
1104	int nthreads
1105	PROTOTYPE: $	1216	PROTOTYPE: $
1106		1217
1107	void	1218	void
1108	max_parallel (nthreads)	1219	max_parallel (int nthreads)
1109	int nthreads
1110	PROTOTYPE: $	1220	PROTOTYPE: $
1111		1221
1112	int	1222	int
1113	max_outstanding (nreqs)	1223	max_outstanding (int maxreqs)
1114	int nreqs	1224	PROTOTYPE: $
1115	PROTOTYPE: $
1116	CODE:	1225	CODE:
1117	RETVAL = max_outstanding;	1226	RETVAL = max_outstanding;
1118	max_outstanding = nreqs;	1227	max_outstanding = maxreqs;
		1228	OUTPUT:
		1229	RETVAL
1119		1230
1120	void	1231	void
1121	aio_open (pathname,flags,mode,callback=&PL_sv_undef)	1232	aio_open (pathname,flags,mode,callback=&PL_sv_undef)
1122	SV * pathname	1233	SV * pathname
1123	int flags	1234	int flags
…		…
1338		1449
1339	REQ_SEND;	1450	REQ_SEND;
1340	}	1451	}
1341		1452
1342	void	1453	void
		1454	aio_mknod (pathname,mode,dev,callback=&PL_sv_undef)
		1455	SV * pathname
		1456	SV * callback
		1457	UV mode
		1458	UV dev
		1459	PPCODE:
		1460	{
		1461	dREQ;
		1462
		1463	req->type = REQ_MKNOD;
		1464	req->data = newSVsv (pathname);
		1465	req->dataptr = SvPVbyte_nolen (req->data);
		1466	req->mode = (mode_t)mode;
		1467	req->offset = dev;
		1468
		1469	REQ_SEND;
		1470	}
		1471
		1472	void
1343	aio_busy (delay,callback=&PL_sv_undef)	1473	aio_busy (delay,callback=&PL_sv_undef)
1344	double delay	1474	double delay
1345	SV * callback	1475	SV * callback
1346	PPCODE:	1476	PPCODE:
1347	{	1477	{
…		…
1378	req->type = REQ_NOP;	1508	req->type = REQ_NOP;
1379		1509
1380	REQ_SEND;	1510	REQ_SEND;
1381	}	1511	}
1382		1512
1383	void	1513	int
1384	aioreq_pri (int pri = DEFAULT_PRI)	1514	aioreq_pri (int pri = 0)
1385	CODE:	1515	PROTOTYPE: ;$
		1516	CODE:
		1517	RETVAL = next_pri - PRI_BIAS;
		1518	if (items > 0)
		1519	{
1386	if (pri < PRI_MIN) pri = PRI_MIN;	1520	if (pri < PRI_MIN) pri = PRI_MIN;
1387	if (pri > PRI_MAX) pri = PRI_MAX;	1521	if (pri > PRI_MAX) pri = PRI_MAX;
1388	next_pri = pri + PRI_BIAS;	1522	next_pri = pri + PRI_BIAS;
		1523	}
		1524	OUTPUT:
		1525	RETVAL
1389		1526
1390	void	1527	void
1391	aioreq_nice (int nice = 0)	1528	aioreq_nice (int nice = 0)
1392	CODE:	1529	CODE:
1393	nice = next_pri - nice;	1530	nice = next_pri - nice;
1394	if (nice < PRI_MIN) nice = PRI_MIN;	1531	if (nice < PRI_MIN) nice = PRI_MIN;
1395	if (nice > PRI_MAX) nice = PRI_MAX;	1532	if (nice > PRI_MAX) nice = PRI_MAX;
1396	next_pri = nice + PRI_BIAS;	1533	next_pri = nice + PRI_BIAS;
1397		1534
1398	void	1535	void
1399	flush ()	1536	flush ()
1400	PROTOTYPE:	1537	PROTOTYPE:
1401	CODE:	1538	CODE:
1402	while (nreqs)	1539	while (nreqs)
1403	{	1540	{
1404	poll_wait ();	1541	poll_wait ();
1405	poll_cb ();	1542	poll_cb (0);
1406	}	1543	}
1407		1544
1408	void	1545	void
1409	poll()	1546	poll()
1410	PROTOTYPE:	1547	PROTOTYPE:
1411	CODE:	1548	CODE:
1412	if (nreqs)	1549	if (nreqs)
1413	{	1550	{
1414	poll_wait ();	1551	poll_wait ();
1415	poll_cb ();	1552	poll_cb (0);
1416	}	1553	}
1417		1554
1418	int	1555	int
1419	poll_fileno()	1556	poll_fileno()
1420	PROTOTYPE:	1557	PROTOTYPE:
…		…
1425		1562
1426	int	1563	int
1427	poll_cb(...)	1564	poll_cb(...)
1428	PROTOTYPE:	1565	PROTOTYPE:
1429	CODE:	1566	CODE:
1430	RETVAL = poll_cb ();	1567	RETVAL = poll_cb (0);
		1568	OUTPUT:
		1569	RETVAL
		1570
		1571	int
		1572	poll_some(int max = 0)
		1573	PROTOTYPE: $
		1574	CODE:
		1575	RETVAL = poll_cb (max);
1431	OUTPUT:	1576	OUTPUT:
1432	RETVAL	1577	RETVAL
1433		1578
1434	void	1579	void
1435	poll_wait()	1580	poll_wait()
…		…
1444	CODE:	1589	CODE:
1445	RETVAL = nreqs;	1590	RETVAL = nreqs;
1446	OUTPUT:	1591	OUTPUT:
1447	RETVAL	1592	RETVAL
1448		1593
		1594	int
		1595	nready()
		1596	PROTOTYPE:
		1597	CODE:
		1598	RETVAL = get_nready ();
		1599	OUTPUT:
		1600	RETVAL
		1601
		1602	int
		1603	npending()
		1604	PROTOTYPE:
		1605	CODE:
		1606	RETVAL = get_npending ();
		1607	OUTPUT:
		1608	RETVAL
		1609
1449	PROTOTYPES: DISABLE	1610	PROTOTYPES: DISABLE
1450		1611
1451	MODULE = IO::AIO PACKAGE = IO::AIO::REQ	1612	MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1452		1613
1453	void	1614	void
…		…
1495	}	1656	}
1496	}	1657	}
1497	}	1658	}
1498		1659
1499	void	1660	void
		1661	cancel_subs (aio_req_ornot req)
		1662	CODE:
		1663	req_cancel_subs (req);
		1664
		1665	void
1500	result (aio_req grp, ...)	1666	result (aio_req grp, ...)
1501	CODE:	1667	CODE:
1502	{	1668	{
1503	int i;	1669	int i;
		1670	AV *av;
		1671
		1672	grp->errorno = errno;
		1673
1504	AV *av = newAV ();	1674	av = newAV ();
1505		1675
1506	for (i = 1; i < items; ++i )	1676	for (i = 1; i < items; ++i )
1507	av_push (av, newSVsv (ST (i)));	1677	av_push (av, newSVsv (ST (i)));
1508		1678
1509	SvREFCNT_dec (grp->data);	1679	SvREFCNT_dec (grp->data);
1510	grp->data = (SV *)av;	1680	grp->data = (SV *)av;
1511	}	1681	}
		1682
		1683	void
		1684	errno (aio_req grp, int errorno = errno)
		1685	CODE:
		1686	grp->errorno = errorno;
1512		1687
1513	void	1688	void
1514	limit (aio_req grp, int limit)	1689	limit (aio_req grp, int limit)
1515	CODE:	1690	CODE:
1516	grp->fd2 = limit;	1691	grp->fd2 = limit;

Diff Legend

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

Comparing IO-AIO/AIO.xs (file contents): Revision 1.71 by root, Tue Oct 24 16:35:04 2006 UTC vs. Revision 1.84 by root, Sat Oct 28 01:24:19 2006 UTC

Diff Legend

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.71 by root, Tue Oct 24 16:35:04 2006 UTC vs.
Revision 1.84 by root, Sat Oct 28 01:24:19 2006 UTC