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

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.104 by root, Sun Jul 8 11:05:36 2007 UTC vs.
Revision 1.111 by root, Sun Mar 30 06:31:49 2008 UTC

…		…
97	# define SvVAL64 SvIV	97	# define SvVAL64 SvIV
98	#else	98	#else
99	# define SvVAL64 SvNV	99	# define SvVAL64 SvNV
100	#endif	100	#endif
101		101
		102	static HV *stash;
		103
102	#define dBUF \	104	#define dBUF \
103	char *aio_buf; \	105	char *aio_buf; \
104	X_LOCK (wrklock); \	106	X_LOCK (wrklock); \
105	self->dbuf = aio_buf = malloc (AIO_BUFSIZE); \	107	self->dbuf = aio_buf = malloc (AIO_BUFSIZE); \
106	X_UNLOCK (wrklock); \	108	X_UNLOCK (wrklock); \
…		…
117	REQ_STAT, REQ_LSTAT, REQ_FSTAT,	119	REQ_STAT, REQ_LSTAT, REQ_FSTAT,
118	REQ_TRUNCATE, REQ_FTRUNCATE,	120	REQ_TRUNCATE, REQ_FTRUNCATE,
119	REQ_UTIME, REQ_FUTIME,	121	REQ_UTIME, REQ_FUTIME,
120	REQ_CHMOD, REQ_FCHMOD,	122	REQ_CHMOD, REQ_FCHMOD,
121	REQ_CHOWN, REQ_FCHOWN,	123	REQ_CHOWN, REQ_FCHOWN,
122	REQ_FSYNC, REQ_FDATASYNC,	124	REQ_SYNC, REQ_FSYNC, REQ_FDATASYNC,
123	REQ_UNLINK, REQ_RMDIR, REQ_MKDIR, REQ_RENAME,	125	REQ_UNLINK, REQ_RMDIR, REQ_MKDIR, REQ_RENAME,
124	REQ_MKNOD, REQ_READDIR,	126	REQ_MKNOD, REQ_READDIR,
125	REQ_LINK, REQ_SYMLINK, REQ_READLINK,	127	REQ_LINK, REQ_SYMLINK, REQ_READLINK,
126	REQ_GROUP, REQ_NOP,	128	REQ_GROUP, REQ_NOP,
127	REQ_BUSY,	129	REQ_BUSY,
…		…
187		189
188	static thread_t main_tid;	190	static thread_t main_tid;
189	static int main_sig;	191	static int main_sig;
190	static int block_sig_level;	192	static int block_sig_level;
191		193
192	void block_sig ()	194	void block_sig (void)
193	{	195	{
194	sigset_t ss;	196	sigset_t ss;
195		197
196	if (block_sig_level++)	198	if (block_sig_level++)
197	return;	199	return;
…		…
202	sigemptyset (&ss);	204	sigemptyset (&ss);
203	sigaddset (&ss, main_sig);	205	sigaddset (&ss, main_sig);
204	pthread_sigmask (SIG_BLOCK, &ss, 0);	206	pthread_sigmask (SIG_BLOCK, &ss, 0);
205	}	207	}
206		208
207	void unblock_sig ()	209	void unblock_sig (void)
208	{	210	{
209	sigset_t ss;	211	sigset_t ss;
210		212
211	if (--block_sig_level)	213	if (--block_sig_level)
212	return;	214	return;
…		…
264	}	266	}
265		267
266	static volatile unsigned int nreqs, nready, npending;	268	static volatile unsigned int nreqs, nready, npending;
267	static volatile unsigned int max_idle = 4;	269	static volatile unsigned int max_idle = 4;
268	static volatile unsigned int max_outstanding = 0xffffffff;	270	static volatile unsigned int max_outstanding = 0xffffffff;
269	static int respipe [2], respipe_osf [2];	271	static int respipe_osf [2], respipe [2] = { -1, -1 };
270		272
271	static mutex_t reslock = X_MUTEX_INIT;	273	static mutex_t reslock = X_MUTEX_INIT;
272	static mutex_t reqlock = X_MUTEX_INIT;	274	static mutex_t reqlock = X_MUTEX_INIT;
273	static cond_t reqwait = X_COND_INIT;	275	static cond_t reqwait = X_COND_INIT;
274		276
275	#if WORDACCESS_UNSAFE	277	#if WORDACCESS_UNSAFE
276		278
277	static unsigned int get_nready ()	279	static unsigned int get_nready (void)
278	{	280	{
279	unsigned int retval;	281	unsigned int retval;
280		282
281	X_LOCK (reqlock);	283	X_LOCK (reqlock);
282	retval = nready;	284	retval = nready;
283	X_UNLOCK (reqlock);	285	X_UNLOCK (reqlock);
284		286
285	return retval;	287	return retval;
286	}	288	}
287		289
288	static unsigned int get_npending ()	290	static unsigned int get_npending (void)
289	{	291	{
290	unsigned int retval;	292	unsigned int retval;
291		293
292	X_LOCK (reslock);	294	X_LOCK (reslock);
293	retval = npending;	295	retval = npending;
294	X_UNLOCK (reslock);	296	X_UNLOCK (reslock);
295		297
296	return retval;	298	return retval;
297	}	299	}
298		300
299	static unsigned int get_nthreads ()	301	static unsigned int get_nthreads (void)
300	{	302	{
301	unsigned int retval;	303	unsigned int retval;
302		304
303	X_LOCK (wrklock);	305	X_LOCK (wrklock);
304	retval = started;	306	retval = started;
…		…
367	}	369	}
368		370
369	abort ();	371	abort ();
370	}	372	}
371		373
372	static int poll_cb ();	374	static int poll_cb (void);
373	static int req_invoke (aio_req req);	375	static int req_invoke (aio_req req);
374	static void req_destroy (aio_req req);	376	static void req_destroy (aio_req req);
375	static void req_cancel (aio_req req);	377	static void req_cancel (aio_req req);
376		378
377	/* must be called at most once */	379	/* must be called at most once */
…		…
501	break;	503	break;
502		504
503	case REQ_OPEN:	505	case REQ_OPEN:
504	{	506	{
505	/* convert fd to fh */	507	/* convert fd to fh */
506	SV *fh;	508	SV *fh = &PL_sv_undef;
507		509
508	PUSHs (sv_2mortal (newSViv (req->result)));	510	if (req->result >= 0)
509	PUTBACK;	511	{
510	call_pv ("IO::AIO::_fd2fh", G_SCALAR \| G_EVAL);	512	GV gv = (GV )sv_newmortal ();
511	SPAGAIN;	513	int flags = req->int1 & (O_RDONLY \| O_WRONLY \| O_RDWR);
		514	char sym [64];
		515	int symlen;
		516
		517	symlen = snprintf (sym, sizeof (sym), "fd#%d", req->result);
		518	gv_init (gv, stash, sym, symlen, 0);
512		519
513	fh = POPs;	520	symlen = snprintf (
514	PUSHMARK (SP);	521	sym,
		522	sizeof (sym),
		523	"%s&=%d",
		524	flags == O_RDONLY ? "<" : flags == O_WRONLY ? ">" : "+<",
		525	req->result
		526	);
		527
		528	if (do_open (gv, sym, symlen, 0, 0, 0, 0))
		529	fh = (SV *)gv;
		530	}
		531
515	XPUSHs (fh);	532	PUSHs (fh);
516	}	533	}
517	break;	534	break;
518		535
519	case REQ_GROUP:	536	case REQ_GROUP:
520	req->int1 = 2; /* mark group as finished */	537	req->int1 = 2; /* mark group as finished */
…		…
564	}	581	}
565		582
566	errno = req->errorno;	583	errno = req->errorno;
567		584
568	PUTBACK;	585	PUTBACK;
569	call_sv (req->callback, G_VOID \| G_EVAL);	586	call_sv (req->callback, G_VOID \| G_EVAL \| G_DISCARD);
570	SPAGAIN;	587	SPAGAIN;
571		588
572	FREETMPS;	589	FREETMPS;
573	LEAVE;	590	LEAVE;
		591
		592	PUTBACK;
574	}	593	}
575		594
576	if (req->grp)	595	if (req->grp)
577	{	596	{
578	aio_req grp = req->grp;	597	aio_req grp = req->grp;
…		…
634	#else	653	#else
635	# define TO_SOCKET(x) (x)	654	# define TO_SOCKET(x) (x)
636	#endif	655	#endif
637		656
638	static void	657	static void
639	create_pipe (int fd[2])	658	create_respipe (void)
640	{	659	{
		660	int old_readfd = respipe [0];
		661
		662	if (respipe [1] >= 0)
		663	respipe_close (TO_SOCKET (respipe [1]));
		664
		665	#ifdef _WIN32
		666	if (PerlSock_socketpair (AF_UNIX, SOCK_STREAM, 0, respipe))
		667	#else
		668	if (pipe (respipe))
		669	#endif
		670	croak ("unable to initialize result pipe");
		671
		672	if (old_readfd >= 0)
		673	{
		674	if (dup2 (TO_SOCKET (respipe [0]), TO_SOCKET (old_readfd)) < 0)
		675	croak ("unable to initialize result pipe(2)");
		676
		677	respipe_close (respipe [0]);
		678	respipe [0] = old_readfd;
		679	}
		680
641	#ifdef _WIN32	681	#ifdef _WIN32
642	int arg = 1;	682	int arg = 1;
643	if (PerlSock_socketpair (AF_UNIX, SOCK_STREAM, 0, fd)
644	\|\| ioctlsocket (TO_SOCKET (fd [0]), FIONBIO, &arg)	683	if (ioctlsocket (TO_SOCKET (respipe [0]), FIONBIO, &arg)
645	\|\| ioctlsocket (TO_SOCKET (fd [1]), FIONBIO, &arg))	684	\|\| ioctlsocket (TO_SOCKET (respipe [1]), FIONBIO, &arg))
646	#else	685	#else
647	if (pipe (fd)
648	\|\| fcntl (fd [0], F_SETFL, O_NONBLOCK)	686	if (fcntl (respipe [0], F_SETFL, O_NONBLOCK)
649	\|\| fcntl (fd [1], F_SETFL, O_NONBLOCK))	687	\|\| fcntl (respipe [1], F_SETFL, O_NONBLOCK))
650	#endif	688	#endif
651	croak ("unable to initialize result pipe");	689	croak ("unable to initialize result pipe(3)");
652		690
653	respipe_osf [0] = TO_SOCKET (respipe [0]);	691	respipe_osf [0] = TO_SOCKET (respipe [0]);
654	respipe_osf [1] = TO_SOCKET (respipe [1]);	692	respipe_osf [1] = TO_SOCKET (respipe [1]);
655	}	693	}
656		694
…		…
677	free (wrk);	715	free (wrk);
678		716
679	X_UNLOCK (wrklock);	717	X_UNLOCK (wrklock);
680	}	718	}
681		719
682	static void maybe_start_thread ()	720	static void maybe_start_thread (void)
683	{	721	{
684	if (get_nthreads () >= wanted)	722	if (get_nthreads () >= wanted)
685	return;	723	return;
686		724
687	/* todo: maybe use idle here, but might be less exact */	725	/* todo: maybe use idle here, but might be less exact */
…		…
747		785
748	while (started > wanted)	786	while (started > wanted)
749	end_thread ();	787	end_thread ();
750	}	788	}
751		789
752	static void poll_wait ()	790	static void poll_wait (void)
753	{	791	{
754	fd_set rfd;	792	fd_set rfd;
755		793
756	while (nreqs)	794	while (nreqs)
757	{	795	{
…		…
770		808
771	PerlSock_select (respipe [0] + 1, &rfd, 0, 0, 0);	809	PerlSock_select (respipe [0] + 1, &rfd, 0, 0, 0);
772	}	810	}
773	}	811	}
774		812
775	static int poll_cb ()	813	static int poll_cb (void)
776	{	814	{
777	dSP;	815	dSP;
778	int count = 0;	816	int count = 0;
779	int maxreqs = max_poll_reqs;	817	int maxreqs = max_poll_reqs;
780	int do_croak = 0;	818	int do_croak = 0;
…		…
801		839
802	if (!res_queue.size)	840	if (!res_queue.size)
803	{	841	{
804	/* read any signals sent by the worker threads */	842	/* read any signals sent by the worker threads */
805	char buf [4];	843	char buf [4];
806	while (PerlSock_recv (respipe [0], buf, 4, 0) == 4)	844	while (respipe_read (respipe [0], buf, 4) == 4)
807	;	845	;
808	}	846	}
809	}	847	}
810		848
811	X_UNLOCK (reslock);	849	X_UNLOCK (reslock);
…		…
1229	case REQ_LINK: req->result = link (req->ptr2, req->ptr1); break;	1267	case REQ_LINK: req->result = link (req->ptr2, req->ptr1); break;
1230	case REQ_SYMLINK: req->result = symlink (req->ptr2, req->ptr1); break;	1268	case REQ_SYMLINK: req->result = symlink (req->ptr2, req->ptr1); break;
1231	case REQ_MKNOD: req->result = mknod (req->ptr2, req->mode, (dev_t)req->offs); break;	1269	case REQ_MKNOD: req->result = mknod (req->ptr2, req->mode, (dev_t)req->offs); break;
1232	case REQ_READLINK: req->result = readlink (req->ptr2, req->ptr1, NAME_MAX); break;	1270	case REQ_READLINK: req->result = readlink (req->ptr2, req->ptr1, NAME_MAX); break;
1233		1271
		1272	case REQ_SYNC: req->result = 0; sync (); break;
		1273	case REQ_FSYNC: req->result = fsync (req->int1); break;
1234	case REQ_FDATASYNC: req->result = fdatasync (req->int1); break;	1274	case REQ_FDATASYNC: req->result = fdatasync (req->int1); break;
1235	case REQ_FSYNC: req->result = fsync (req->int1); break;	1275
1236	case REQ_READDIR: scandir_ (req, self); break;	1276	case REQ_READDIR: scandir_ (req, self); break;
1237		1277
1238	case REQ_BUSY:	1278	case REQ_BUSY:
1239	#ifdef _WIN32	1279	#ifdef _WIN32
1240	Sleep (req->nv1 * 1000.);	1280	Sleep (req->nv1 * 1000.);
…		…
1293	++npending;	1333	++npending;
1294		1334
1295	if (!reqq_push (&res_queue, req))	1335	if (!reqq_push (&res_queue, req))
1296	{	1336	{
1297	/* write a dummy byte to the pipe so fh becomes ready */	1337	/* write a dummy byte to the pipe so fh becomes ready */
1298	send (respipe_osf [1], (const void *)&respipe_osf, 1, 0);	1338	respipe_write (respipe_osf [1], (const void *)&respipe_osf, 1);
1299		1339
1300	/* optionally signal the main thread asynchronously */	1340	/* optionally signal the main thread asynchronously */
1301	if (main_sig)	1341	if (main_sig)
1302	pthread_kill (main_tid, main_sig);	1342	pthread_kill (main_tid, main_sig);
1303	}	1343	}
…		…
1370	idle = 0;	1410	idle = 0;
1371	nreqs = 0;	1411	nreqs = 0;
1372	nready = 0;	1412	nready = 0;
1373	npending = 0;	1413	npending = 0;
1374		1414
1375	PerlSock_closesocket (respipe [0]);
1376	PerlSock_closesocket (respipe [1]);
1377
1378	create_pipe (respipe);	1415	create_respipe ();
1379		1416
1380	atfork_parent ();	1417	atfork_parent ();
1381	}	1418	}
1382		1419
1383	#define dREQ \	1420	#define dREQ \
…		…
1405		1442
1406	PROTOTYPES: ENABLE	1443	PROTOTYPES: ENABLE
1407		1444
1408	BOOT:	1445	BOOT:
1409	{	1446	{
1410	HV *stash = gv_stashpv ("IO::AIO", 1);	1447	stash = gv_stashpv ("IO::AIO", 1);
1411		1448
1412	newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));	1449	newCONSTSUB (stash, "EXDEV", newSViv (EXDEV));
1413	newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));	1450	newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));
1414	newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));	1451	newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));
1415	newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT));	1452	newCONSTSUB (stash, "O_CREAT", newSViv (O_CREAT));
…		…
1426	#else	1463	#else
1427	newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO));	1464	newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO));
1428	newCONSTSUB (stash, "SIGIO", newSViv (SIGIO));	1465	newCONSTSUB (stash, "SIGIO", newSViv (SIGIO));
1429	#endif	1466	#endif
1430		1467
1431	create_pipe (respipe);	1468	create_respipe ();
1432		1469
1433	X_THREAD_ATFORK (atfork_prepare, atfork_parent, atfork_child);	1470	X_THREAD_ATFORK (atfork_prepare, atfork_parent, atfork_child);
1434	}	1471	}
1435		1472
1436	void	1473	void
…		…
1483		1520
1484	REQ_SEND;	1521	REQ_SEND;
1485	}	1522	}
1486		1523
1487	void	1524	void
1488	aio_close (SV fh, SV callback=&PL_sv_undef)	1525	aio_fsync (SV fh, SV callback=&PL_sv_undef)
1489	PROTOTYPE: $;$	1526	PROTOTYPE: $;$
1490	ALIAS:	1527	ALIAS:
1491	aio_close = REQ_CLOSE
1492	aio_fsync = REQ_FSYNC	1528	aio_fsync = REQ_FSYNC
1493	aio_fdatasync = REQ_FDATASYNC	1529	aio_fdatasync = REQ_FDATASYNC
1494	PPCODE:	1530	PPCODE:
1495	{	1531	{
1496	dREQ;	1532	dREQ;
1497		1533
1498	req->type = ix;	1534	req->type = ix;
1499	req->sv1 = newSVsv (fh);	1535	req->sv1 = newSVsv (fh);
1500	req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh)));	1536	req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh)));
		1537
		1538	REQ_SEND (req);
		1539	}
		1540
		1541	int
		1542	_dup (int fd)
		1543	PROTOTYPE: $
		1544	CODE:
		1545	RETVAL = dup (fd);
		1546	OUTPUT:
		1547	RETVAL
		1548
		1549	void
		1550	_aio_close (int fd, SV *callback=&PL_sv_undef)
		1551	PROTOTYPE: $;$
		1552	PPCODE:
		1553	{
		1554	dREQ;
		1555
		1556	req->type = REQ_CLOSE;
		1557	req->int1 = fd;
1501		1558
1502	REQ_SEND (req);	1559	REQ_SEND (req);
1503	}	1560	}
1504		1561
1505	void	1562	void
…		…
1832	XPUSHs (req_sv (req, AIO_GRP_KLASS));	1889	XPUSHs (req_sv (req, AIO_GRP_KLASS));
1833	}	1890	}
1834		1891
1835	void	1892	void
1836	aio_nop (SV *callback=&PL_sv_undef)	1893	aio_nop (SV *callback=&PL_sv_undef)
		1894	ALIAS:
		1895	aio_nop = REQ_NOP
		1896	aio_sync = REQ_SYNC
1837	PPCODE:	1897	PPCODE:
1838	{	1898	{
1839	dREQ;	1899	dREQ;
1840		1900
1841	req->type = REQ_NOP;	1901	req->type = ix;
1842		1902
1843	REQ_SEND;	1903	REQ_SEND;
1844	}	1904	}
1845		1905
1846	int	1906	int
…		…
1932		1992
1933	block_sig ();	1993	block_sig ();
1934	PUSHMARK (SP);	1994	PUSHMARK (SP);
1935	PUTBACK;	1995	PUTBACK;
1936	count = call_sv (cb, GIMME_V \| G_NOARGS \| G_EVAL);	1996	count = call_sv (cb, GIMME_V \| G_NOARGS \| G_EVAL);
1937	SPAGAIN;
1938	unblock_sig ();	1997	unblock_sig ();
1939		1998
1940	if (SvTRUE (ERRSV))	1999	if (SvTRUE (ERRSV))
1941	croak (0);	2000	croak (0);
1942		2001

Diff Legend

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

Comparing IO-AIO/AIO.xs (file contents): Revision 1.104 by root, Sun Jul 8 11:05:36 2007 UTC vs. Revision 1.111 by root, Sun Mar 30 06:31:49 2008 UTC

Diff Legend

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.104 by root, Sun Jul 8 11:05:36 2007 UTC vs.
Revision 1.111 by root, Sun Mar 30 06:31:49 2008 UTC