[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.82 by root, Fri Oct 27 20:11:58 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"
47	/* used for struct dirent, AIX doesn't provide it */	48	/* used for struct dirent, AIX doesn't provide it */
48	#ifndef NAME_MAX	49	#ifndef NAME_MAX
49	# define NAME_MAX 4096	50	# define NAME_MAX 4096
50	#endif	51	#endif
51		52
		53	#ifndef PTHREAD_STACK_MIN
		54	/* care for broken platforms, e.g. windows */
		55	# define PTHREAD_STACK_MIN 16384
		56	#endif
		57
52	#if __ia64	58	#if __ia64
53	# define STACKSIZE 65536	59	# define STACKSIZE 65536
54	#elif __i386 \|\| __x86_64 /* 16k is unreasonably high :( */	60	#elif __i386 \|\| __x86_64 /* 16k is unreasonably high :( */
55	# define STACKSIZE PTHREAD_STACK_MIN	61	# define STACKSIZE PTHREAD_STACK_MIN
56	#else	62	#else
57	# define STACKSIZE 16384	63	# define STACKSIZE 16384
		64	#endif
		65
		66	/* wether word reads are potentially non-atomic.
		67	* this is conservatice, likely most arches this runs
		68	* on have atomic word read/writes.
		69	*/
		70	#ifndef WORDREAD_UNSAFE
		71	# if __i386 \|\| __x86_64
		72	# define WORDREAD_UNSAFE 0
		73	# else
		74	# define WORDREAD_UNSAFE 1
		75	# endif
58	#endif	76	#endif
59		77
60	/* buffer size for various temporary buffers */	78	/* buffer size for various temporary buffers */
61	#define AIO_BUFSIZE 65536	79	#define AIO_BUFSIZE 65536
62		80
…		…
74	REQ_READ, REQ_WRITE, REQ_READAHEAD,	92	REQ_READ, REQ_WRITE, REQ_READAHEAD,
75	REQ_SENDFILE,	93	REQ_SENDFILE,
76	REQ_STAT, REQ_LSTAT, REQ_FSTAT,	94	REQ_STAT, REQ_LSTAT, REQ_FSTAT,
77	REQ_FSYNC, REQ_FDATASYNC,	95	REQ_FSYNC, REQ_FDATASYNC,
78	REQ_UNLINK, REQ_RMDIR, REQ_RENAME,	96	REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
79	REQ_READDIR,	97	REQ_MKNOD, REQ_READDIR,
80	REQ_LINK, REQ_SYMLINK,	98	REQ_LINK, REQ_SYMLINK,
81	REQ_GROUP, REQ_NOP,	99	REQ_GROUP, REQ_NOP,
82	REQ_BUSY,	100	REQ_BUSY,
83	};	101	};
84		102
…		…
126	NUM_PRI = PRI_MAX + PRI_BIAS + 1,	144	NUM_PRI = PRI_MAX + PRI_BIAS + 1,
127	};	145	};
128		146
129	static int next_pri = DEFAULT_PRI + PRI_BIAS;	147	static int next_pri = DEFAULT_PRI + PRI_BIAS;
130		148
131	static int started, wanted;	149	static unsigned int started, wanted;
132	static volatile int nreqs;
133	static int max_outstanding = 1<<30;
134	static int respipe [2];
135		150
136	#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)	151	#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
137	# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP	152	# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
138	#else	153	#else
139	# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER	154	# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
140	#endif	155	#endif
141		156
142	#define LOCK(mutex) pthread_mutex_lock (&(mutex))	157	#define LOCK(mutex) pthread_mutex_lock (&(mutex))
143	#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex))	158	#define UNLOCK(mutex) pthread_mutex_unlock (&(mutex))
144		159
145	/* worker threasd management */	160	/* worker threads management */
146	static pthread_mutex_t wrklock = AIO_MUTEX_INIT;	161	static pthread_mutex_t wrklock = AIO_MUTEX_INIT;
147		162
148	typedef struct worker {	163	typedef struct worker {
149	/* locked by wrklock */	164	/* locked by wrklock */
150	struct worker prev, next;	165	struct worker prev, next;
…		…
180	wrk->prev->next = wrk->next;	195	wrk->prev->next = wrk->next;
181		196
182	free (wrk);	197	free (wrk);
183	}	198	}
184		199
		200	static volatile unsigned int nreqs, nready, npending;
		201	static volatile unsigned int max_outstanding = 0xffffffff;
		202	static int respipe [2];
		203
185	static pthread_mutex_t reslock = AIO_MUTEX_INIT;	204	static pthread_mutex_t reslock = AIO_MUTEX_INIT;
186	static pthread_mutex_t reqlock = AIO_MUTEX_INIT;	205	static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
187	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;	206	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
		207
		208	#if WORDREAD_UNSAFE
		209
		210	static unsigned int get_nready ()
		211	{
		212	unsigned int retval;
		213
		214	LOCK (reqlock);
		215	retval = nready;
		216	UNLOCK (reqlock);
		217
		218	return retval;
		219	}
		220
		221	static unsigned int get_npending ()
		222	{
		223	unsigned int retval;
		224
		225	LOCK (reslock);
		226	retval = npending;
		227	UNLOCK (reslock);
		228
		229	return retval;
		230	}
		231
		232	#else
		233
		234	# define get_nready() nready
		235	# define get_npending() npending
		236
		237	#endif
188		238
189	/*	239	/*
190	* a somewhat faster data structure might be nice, but	240	* a somewhat faster data structure might be nice, but
191	* with 8 priorities this actually needs <20 insns	241	* with 8 priorities this actually needs <20 insns
192	* per shift, the most expensive operation.	242	* per shift, the most expensive operation.
…		…
238	}	288	}
239		289
240	abort ();	290	abort ();
241	}	291	}
242		292
		293	static int poll_cb (int max);
243	static void req_invoke (aio_req req);	294	static void req_invoke (aio_req req);
244	static void req_free (aio_req req);	295	static void req_free (aio_req req);
		296	static void req_cancel (aio_req req);
245		297
246	/* must be called at most once */	298	/* must be called at most once */
247	static SV req_sv (aio_req req, const char klass)	299	static SV req_sv (aio_req req, const char klass)
248	{	300	{
249	if (!req->self)	301	if (!req->self)
…		…
311	req_invoke (grp);	363	req_invoke (grp);
312	req_free (grp);	364	req_free (grp);
313	}	365	}
314	}	366	}
315		367
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)	368	static void req_invoke (aio_req req)
342	{	369	{
343	dSP;	370	dSP;
344		371
345	if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))	372	if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))
346	{	373	{
347	errno = req->errorno;
348
349	ENTER;	374	ENTER;
350	SAVETMPS;	375	SAVETMPS;
351	PUSHMARK (SP);	376	PUSHMARK (SP);
352	EXTEND (SP, 1);	377	EXTEND (SP, 1);
353		378
…		…
418	default:	443	default:
419	PUSHs (sv_2mortal (newSViv (req->result)));	444	PUSHs (sv_2mortal (newSViv (req->result)));
420	break;	445	break;
421	}	446	}
422		447
		448	errno = req->errorno;
423		449
424	PUTBACK;	450	PUTBACK;
425	call_sv (req->callback, G_VOID \| G_EVAL);	451	call_sv (req->callback, G_VOID \| G_EVAL);
426	SPAGAIN;	452	SPAGAIN;
427		453
…		…
468	free (req->data2ptr);	494	free (req->data2ptr);
469		495
470	Safefree (req);	496	Safefree (req);
471	}	497	}
472		498
		499	static void req_cancel_subs (aio_req grp)
		500	{
		501	aio_req sub;
		502
		503	if (grp->type != REQ_GROUP)
		504	return;
		505
		506	SvREFCNT_dec (grp->fh2);
		507	grp->fh2 = 0;
		508
		509	for (sub = grp->grp_first; sub; sub = sub->grp_next)
		510	req_cancel (sub);
		511	}
		512
473	static void req_cancel (aio_req req)	513	static void req_cancel (aio_req req)
474	{	514	{
475	req->flags \|= FLAG_CANCELLED;	515	req->flags \|= FLAG_CANCELLED;
476		516
477	if (req->type == REQ_GROUP)	517	req_cancel_subs (req);
478	{	518	}
479	aio_req sub;
480		519
481	for (sub = req->grp_first; sub; sub = sub->grp_next)	520	static void aio_proc(void arg);
482	req_cancel (sub);	521
		522	static void start_thread (void)
		523	{
		524	sigset_t fullsigset, oldsigset;
		525	pthread_attr_t attr;
		526
		527	worker *wrk = calloc (1, sizeof (worker));
		528
		529	if (!wrk)
		530	croak ("unable to allocate worker thread data");
		531
		532	pthread_attr_init (&attr);
		533	pthread_attr_setstacksize (&attr, STACKSIZE);
		534	pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
		535	#ifdef PTHREAD_SCOPE_PROCESS
		536	pthread_attr_setscope (&attr, PTHREAD_SCOPE_PROCESS);
		537	#endif
		538
		539	sigfillset (&fullsigset);
		540
		541	LOCK (wrklock);
		542	sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
		543
		544	if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
483	}	545	{
484	}	546	wrk->prev = &wrk_first;
		547	wrk->next = wrk_first.next;
		548	wrk_first.next->prev = wrk;
		549	wrk_first.next = wrk;
		550	++started;
		551	}
		552	else
		553	free (wrk);
485		554
		555	sigprocmask (SIG_SETMASK, &oldsigset, 0);
		556	UNLOCK (wrklock);
		557	}
		558
		559	static void maybe_start_thread ()
		560	{
		561	#if 0
		562	static struct timeval last;
		563	struct timeval diff, now;
		564	#endif
		565
		566	if (started >= wanted)
		567	return;
		568
		569	if (nready <= nreqs - get_nready () - get_npending ())
		570	return;
		571
		572	#if 0
		573	gettimeofday (&now, 0);
		574
		575	diff.tv_sec = now.tv_sec - last.tv_sec;
		576	diff.tv_usec = now.tv_usec - last.tv_usec;
		577
		578	if (diff.tv_usec < 0)
		579	{
		580	--diff.tv_sec;
		581	diff.tv_usec += 1000000;
		582	}
		583
		584	if (!diff.tv_sec && diff.tv_usec < 10000)
		585	return;
		586
		587	last = now;
		588	#endif
		589
		590	start_thread ();
		591	}
		592
		593	static void req_send (aio_req req)
		594	{
		595	++nreqs;
		596
		597	LOCK (reqlock);
		598	++nready;
		599	reqq_push (&req_queue, req);
		600	pthread_cond_signal (&reqwait);
		601	UNLOCK (reqlock);
		602
		603	maybe_start_thread ();
		604	}
		605
		606	static void end_thread (void)
		607	{
		608	aio_req req;
		609
		610	Newz (0, req, 1, aio_cb);
		611
		612	req->type = REQ_QUIT;
		613	req->pri = PRI_MAX + PRI_BIAS;
		614
		615	req_send (req);
		616
		617	LOCK (wrklock);
		618	--started;
		619	UNLOCK (wrklock);
		620	}
		621
		622	static void min_parallel (int nthreads)
		623	{
		624	if (wanted < nthreads)
		625	wanted = nthreads;
		626	}
		627
		628	static void max_parallel (int nthreads)
		629	{
		630	if (wanted > nthreads)
		631	wanted = nthreads;
		632
		633	while (started > wanted)
		634	end_thread ();
		635	}
		636
		637	static void poll_wait ()
		638	{
		639	fd_set rfd;
		640
		641	while (nreqs)
		642	{
		643	int size;
		644	if (WORDREAD_UNSAFE) LOCK (reslock);
		645	size = res_queue.size;
		646	if (WORDREAD_UNSAFE) UNLOCK (reslock);
		647
		648	if (size)
		649	return;
		650
		651	maybe_start_thread ();
		652
		653	FD_ZERO(&rfd);
		654	FD_SET(respipe [0], &rfd);
		655
		656	select (respipe [0] + 1, &rfd, 0, 0, 0);
		657	}
		658	}
		659
486	static int poll_cb ()	660	static int poll_cb (int max)
487	{	661	{
488	dSP;	662	dSP;
489	int count = 0;	663	int count = 0;
490	int do_croak = 0;	664	int do_croak = 0;
491	aio_req req;	665	aio_req req;
492		666
493	for (;;)	667	for (;;)
494	{	668	{
495	LOCK (reslock);	669	while (max <= 0 \|\| count < max)
496	req = reqq_shift (&res_queue);
497
498	if (req)
499	{	670	{
		671	maybe_start_thread ();
		672
		673	LOCK (reslock);
		674	req = reqq_shift (&res_queue);
		675
500	if (!res_queue.size)	676	if (req)
501	{	677	{
		678	--npending;
		679
		680	if (!res_queue.size)
		681	{
502	/* read any signals sent by the worker threads */	682	/* read any signals sent by the worker threads */
503	char buf [32];	683	char buf [32];
504	while (read (respipe [0], buf, 32) == 32)	684	while (read (respipe [0], buf, 32) == 32)
		685	;
505	;	686	}
506	}	687	}
		688
		689	UNLOCK (reslock);
		690
		691	if (!req)
		692	break;
		693
		694	--nreqs;
		695
		696	if (req->type == REQ_GROUP && req->length)
		697	{
		698	req->fd = 1; /* mark request as delayed */
		699	continue;
		700	}
		701	else
		702	{
		703	if (req->type == REQ_READ)
		704	SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
		705
		706	if (req->data2ptr && (req->type == REQ_READ \|\| req->type == REQ_WRITE))
		707	SvREADONLY_off (req->data);
		708
		709	if (req->statdata)
		710	{
		711	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
		712	PL_laststatval = req->result;
		713	PL_statcache = *(req->statdata);
		714	}
		715
		716	req_invoke (req);
		717
		718	count++;
		719	}
		720
		721	req_free (req);
507	}	722	}
508		723
509	UNLOCK (reslock);	724	if (nreqs <= max_outstanding)
510
511	if (!req)
512	break;	725	break;
513		726
514	--nreqs;	727	poll_wait ();
515		728
516	if (req->type == REQ_QUIT)	729	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	}	730	}
545		731
546	return count;	732	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	}	733	}
646		734
647	static void create_pipe ()	735	static void create_pipe ()
648	{	736	{
649	if (pipe (respipe))	737	if (pipe (respipe))
…		…
704	#if !HAVE_FDATASYNC	792	#if !HAVE_FDATASYNC
705	# define fdatasync fsync	793	# define fdatasync fsync
706	#endif	794	#endif
707		795
708	#if !HAVE_READAHEAD	796	#if !HAVE_READAHEAD
709	# define readahead aio_readahead	797	# define readahead(fd,offset,count) aio_readahead (fd, offset, count, self)
710		798
711	static ssize_t readahead (int fd, off_t offset, size_t count)	799	static ssize_t aio_readahead (int fd, off_t offset, size_t count, worker *self)
712	{	800	{
713	dBUF;	801	dBUF;
714		802
715	while (count > 0)	803	while (count > 0)
716	{	804	{
…		…
721	count -= len;	809	count -= len;
722	}	810	}
723		811
724	errno = 0;	812	errno = 0;
725	}	813	}
		814
726	#endif	815	#endif
727		816
728	#if !HAVE_READDIR_R	817	#if !HAVE_READDIR_R
729	# define readdir_r aio_readdir_r	818	# define readdir_r aio_readdir_r
730		819
…		…
775	{	864	{
776	off_t sbytes;	865	off_t sbytes;
777	res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0);	866	res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0);
778		867
779	if (res < 0 && sbytes)	868	if (res < 0 && sbytes)
780	/* maybe only on EAGAIN only: as usual, the manpage leaves you guessing */	869	/* maybe only on EAGAIN: as usual, the manpage leaves you guessing */
781	res = sbytes;	870	res = sbytes;
782	}	871	}
783		872
784	# elif __hpux	873	# elif __hpux
785	res = sendfile (ofd, ifd, offset, count, 0, 0);	874	res = sendfile (ofd, ifd, offset, count, 0, 0);
…		…
865	int errorno;	954	int errorno;
866		955
867	LOCK (wrklock);	956	LOCK (wrklock);
868	self->dirp = dirp = opendir (req->dataptr);	957	self->dirp = dirp = opendir (req->dataptr);
869	self->dbuf = u = malloc (sizeof (*u));	958	self->dbuf = u = malloc (sizeof (*u));
		959	req->data2ptr = names = malloc (memlen);
870	UNLOCK (wrklock);	960	UNLOCK (wrklock);
871
872	req->data2ptr = names = malloc (memlen);
873		961
874	if (dirp && u && names)	962	if (dirp && u && names)
875	for (;;)	963	for (;;)
876	{	964	{
877	errno = 0;	965	errno = 0;
…		…
913	/*****************************************************************************/	1001	/*****************************************************************************/
914		1002
915	static void aio_proc (void thr_arg)	1003	static void aio_proc (void thr_arg)
916	{	1004	{
917	aio_req req;	1005	aio_req req;
918	int type;
919	worker self = (worker )thr_arg;	1006	worker self = (worker )thr_arg;
920		1007
921	do	1008	for (;;)
922	{	1009	{
923	LOCK (reqlock);	1010	LOCK (reqlock);
924		1011
925	for (;;)	1012	for (;;)
926	{	1013	{
…		…
930	break;	1017	break;
931		1018
932	pthread_cond_wait (&reqwait, &reqlock);	1019	pthread_cond_wait (&reqwait, &reqlock);
933	}	1020	}
934		1021
		1022	--nready;
		1023
935	UNLOCK (reqlock);	1024	UNLOCK (reqlock);
936		1025
937	errno = 0; /* strictly unnecessary */	1026	errno = 0; /* strictly unnecessary */
938	type = req->type; /* remember type for QUIT check */
939		1027
940	if (!(req->flags & FLAG_CANCELLED))	1028	if (!(req->flags & FLAG_CANCELLED))
941	switch (type)	1029	switch (req->type)
942	{	1030	{
943	case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;	1031	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;	1032	case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
945		1033
946	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;	1034	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
…		…
955	case REQ_UNLINK: req->result = unlink (req->dataptr); break;	1043	case REQ_UNLINK: req->result = unlink (req->dataptr); break;
956	case REQ_RMDIR: req->result = rmdir (req->dataptr); break;	1044	case REQ_RMDIR: req->result = rmdir (req->dataptr); break;
957	case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break;	1045	case REQ_RENAME: req->result = rename (req->data2ptr, req->dataptr); break;
958	case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break;	1046	case REQ_LINK: req->result = link (req->data2ptr, req->dataptr); break;
959	case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;	1047	case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;
		1048	case REQ_MKNOD: req->result = mknod (req->data2ptr, req->mode, (dev_t)req->offset); break;
960		1049
961	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;	1050	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
962	case REQ_FSYNC: req->result = fsync (req->fd); break;	1051	case REQ_FSYNC: req->result = fsync (req->fd); break;
963	case REQ_READDIR: scandir_ (req, self); break;	1052	case REQ_READDIR: scandir_ (req, self); break;
964		1053
…		…
972	req->result = select (0, 0, 0, 0, &tv);	1061	req->result = select (0, 0, 0, 0, &tv);
973	}	1062	}
974		1063
975	case REQ_GROUP:	1064	case REQ_GROUP:
976	case REQ_NOP:	1065	case REQ_NOP:
		1066	break;
		1067
977	case REQ_QUIT:	1068	case REQ_QUIT:
		1069	LOCK (wrklock);
		1070	worker_free (self);
		1071	--started;
		1072	UNLOCK (wrklock);
978	break;	1073	return 0;
979		1074
980	default:	1075	default:
981	req->result = ENOSYS;	1076	req->result = ENOSYS;
982	break;	1077	break;
983	}	1078	}
984		1079
985	req->errorno = errno;	1080	req->errorno = errno;
986		1081
987	LOCK (reslock);	1082	LOCK (reslock);
988		1083
		1084	++npending;
		1085
989	if (!reqq_push (&res_queue, req))	1086	if (!reqq_push (&res_queue, req))
990	/* write a dummy byte to the pipe so fh becomes ready */	1087	/* write a dummy byte to the pipe so fh becomes ready */
991	write (respipe [1], &respipe, 1);	1088	write (respipe [1], &respipe, 1);
992		1089
993	self->req = 0;	1090	self->req = 0;
994	worker_clear (self);	1091	worker_clear (self);
995		1092
996	UNLOCK (reslock);	1093	UNLOCK (reslock);
997	}	1094	}
998	while (type != REQ_QUIT);
999
1000	LOCK (wrklock);
1001	worker_free (self);
1002	UNLOCK (wrklock);
1003
1004	return 0;
1005	}	1095	}
1006		1096
1007	/*****************************************************************************/	1097	/*****************************************************************************/
1008		1098
1009	static void atfork_prepare (void)	1099	static void atfork_prepare (void)
…		…
1089	PROTOTYPES: ENABLE	1179	PROTOTYPES: ENABLE
1090		1180
1091	BOOT:	1181	BOOT:
1092	{	1182	{
1093	HV *stash = gv_stashpv ("IO::AIO", 1);	1183	HV *stash = gv_stashpv ("IO::AIO", 1);
		1184
1094	newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));	1185	newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));
1095	newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));	1186	newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));
1096	newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));	1187	newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));
		1188	newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT));
		1189	newCONSTSUB (stash, "O_TRUNC", newSViv (O_TRUNC));
		1190	newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO));
1097		1191
1098	create_pipe ();	1192	create_pipe ();
1099	pthread_atfork (atfork_prepare, atfork_parent, atfork_child);	1193	pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
1100	}
1101		1194
		1195	start_thread ();
		1196	}
		1197
1102	void	1198	void
1103	min_parallel (nthreads)	1199	min_parallel (int nthreads)
1104	int nthreads
1105	PROTOTYPE: $	1200	PROTOTYPE: $
1106		1201
1107	void	1202	void
1108	max_parallel (nthreads)	1203	max_parallel (int nthreads)
1109	int nthreads
1110	PROTOTYPE: $	1204	PROTOTYPE: $
1111		1205
1112	int	1206	int
1113	max_outstanding (nreqs)	1207	max_outstanding (int maxreqs)
1114	int nreqs	1208	PROTOTYPE: $
1115	PROTOTYPE: $
1116	CODE:	1209	CODE:
1117	RETVAL = max_outstanding;	1210	RETVAL = max_outstanding;
1118	max_outstanding = nreqs;	1211	max_outstanding = maxreqs;
		1212	OUTPUT:
		1213	RETVAL
1119		1214
1120	void	1215	void
1121	aio_open (pathname,flags,mode,callback=&PL_sv_undef)	1216	aio_open (pathname,flags,mode,callback=&PL_sv_undef)
1122	SV * pathname	1217	SV * pathname
1123	int flags	1218	int flags
…		…
1338		1433
1339	REQ_SEND;	1434	REQ_SEND;
1340	}	1435	}
1341		1436
1342	void	1437	void
		1438	aio_mknod (pathname,mode,dev,callback=&PL_sv_undef)
		1439	SV * pathname
		1440	SV * callback
		1441	UV mode
		1442	UV dev
		1443	PPCODE:
		1444	{
		1445	dREQ;
		1446
		1447	req->type = REQ_MKNOD;
		1448	req->data = newSVsv (pathname);
		1449	req->dataptr = SvPVbyte_nolen (req->data);
		1450	req->mode = (mode_t)mode;
		1451	req->offset = dev;
		1452
		1453	REQ_SEND;
		1454	}
		1455
		1456	void
1343	aio_busy (delay,callback=&PL_sv_undef)	1457	aio_busy (delay,callback=&PL_sv_undef)
1344	double delay	1458	double delay
1345	SV * callback	1459	SV * callback
1346	PPCODE:	1460	PPCODE:
1347	{	1461	{
…		…
1378	req->type = REQ_NOP;	1492	req->type = REQ_NOP;
1379		1493
1380	REQ_SEND;	1494	REQ_SEND;
1381	}	1495	}
1382		1496
1383	void	1497	int
1384	aioreq_pri (int pri = DEFAULT_PRI)	1498	aioreq_pri (int pri = 0)
1385	CODE:	1499	PROTOTYPE: ;$
		1500	CODE:
		1501	RETVAL = next_pri - PRI_BIAS;
		1502	if (items > 0)
		1503	{
1386	if (pri < PRI_MIN) pri = PRI_MIN;	1504	if (pri < PRI_MIN) pri = PRI_MIN;
1387	if (pri > PRI_MAX) pri = PRI_MAX;	1505	if (pri > PRI_MAX) pri = PRI_MAX;
1388	next_pri = pri + PRI_BIAS;	1506	next_pri = pri + PRI_BIAS;
		1507	}
		1508	OUTPUT:
		1509	RETVAL
1389		1510
1390	void	1511	void
1391	aioreq_nice (int nice = 0)	1512	aioreq_nice (int nice = 0)
1392	CODE:	1513	CODE:
1393	nice = next_pri - nice;	1514	nice = next_pri - nice;
1394	if (nice < PRI_MIN) nice = PRI_MIN;	1515	if (nice < PRI_MIN) nice = PRI_MIN;
1395	if (nice > PRI_MAX) nice = PRI_MAX;	1516	if (nice > PRI_MAX) nice = PRI_MAX;
1396	next_pri = nice + PRI_BIAS;	1517	next_pri = nice + PRI_BIAS;
1397		1518
1398	void	1519	void
1399	flush ()	1520	flush ()
1400	PROTOTYPE:	1521	PROTOTYPE:
1401	CODE:	1522	CODE:
1402	while (nreqs)	1523	while (nreqs)
1403	{	1524	{
1404	poll_wait ();	1525	poll_wait ();
1405	poll_cb ();	1526	poll_cb (0);
1406	}	1527	}
1407		1528
1408	void	1529	void
1409	poll()	1530	poll()
1410	PROTOTYPE:	1531	PROTOTYPE:
1411	CODE:	1532	CODE:
1412	if (nreqs)	1533	if (nreqs)
1413	{	1534	{
1414	poll_wait ();	1535	poll_wait ();
1415	poll_cb ();	1536	poll_cb (0);
1416	}	1537	}
1417		1538
1418	int	1539	int
1419	poll_fileno()	1540	poll_fileno()
1420	PROTOTYPE:	1541	PROTOTYPE:
…		…
1425		1546
1426	int	1547	int
1427	poll_cb(...)	1548	poll_cb(...)
1428	PROTOTYPE:	1549	PROTOTYPE:
1429	CODE:	1550	CODE:
1430	RETVAL = poll_cb ();	1551	RETVAL = poll_cb (0);
		1552	OUTPUT:
		1553	RETVAL
		1554
		1555	int
		1556	poll_some(int max = 0)
		1557	PROTOTYPE: $
		1558	CODE:
		1559	RETVAL = poll_cb (max);
1431	OUTPUT:	1560	OUTPUT:
1432	RETVAL	1561	RETVAL
1433		1562
1434	void	1563	void
1435	poll_wait()	1564	poll_wait()
…		…
1444	CODE:	1573	CODE:
1445	RETVAL = nreqs;	1574	RETVAL = nreqs;
1446	OUTPUT:	1575	OUTPUT:
1447	RETVAL	1576	RETVAL
1448		1577
		1578	int
		1579	nready()
		1580	PROTOTYPE:
		1581	CODE:
		1582	RETVAL = get_nready ();
		1583	OUTPUT:
		1584	RETVAL
		1585
		1586	int
		1587	npending()
		1588	PROTOTYPE:
		1589	CODE:
		1590	RETVAL = get_npending ();
		1591	OUTPUT:
		1592	RETVAL
		1593
1449	PROTOTYPES: DISABLE	1594	PROTOTYPES: DISABLE
1450		1595
1451	MODULE = IO::AIO PACKAGE = IO::AIO::REQ	1596	MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1452		1597
1453	void	1598	void
…		…
1495	}	1640	}
1496	}	1641	}
1497	}	1642	}
1498		1643
1499	void	1644	void
		1645	cancel_subs (aio_req_ornot req)
		1646	CODE:
		1647	req_cancel_subs (req);
		1648
		1649	void
1500	result (aio_req grp, ...)	1650	result (aio_req grp, ...)
1501	CODE:	1651	CODE:
1502	{	1652	{
1503	int i;	1653	int i;
		1654	AV *av;
		1655
		1656	grp->errorno = errno;
		1657
1504	AV *av = newAV ();	1658	av = newAV ();
1505		1659
1506	for (i = 1; i < items; ++i )	1660	for (i = 1; i < items; ++i )
1507	av_push (av, newSVsv (ST (i)));	1661	av_push (av, newSVsv (ST (i)));
1508		1662
1509	SvREFCNT_dec (grp->data);	1663	SvREFCNT_dec (grp->data);
1510	grp->data = (SV *)av;	1664	grp->data = (SV *)av;
1511	}	1665	}
		1666
		1667	void
		1668	errno (aio_req grp, int errorno = errno)
		1669	CODE:
		1670	grp->errorno = errorno;
1512		1671
1513	void	1672	void
1514	limit (aio_req grp, int limit)	1673	limit (aio_req grp, int limit)
1515	CODE:	1674	CODE:
1516	grp->fd2 = limit;	1675	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.82 by root, Fri Oct 27 20:11:58 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.82 by root, Fri Oct 27 20:11:58 2006 UTC