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

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.1 by root, Sun Jul 10 17:07:44 2005 UTC vs.
Revision 1.10 by root, Wed Jul 13 00:13:09 2005 UTC

…		…
1	#define PERL_NO_GET_CONTEXT
2
3	#include "EXTERN.h"	1	#include "EXTERN.h"
4	#include "perl.h"	2	#include "perl.h"
5	#include "XSUB.h"	3	#include "XSUB.h"
6		4
		5	#define _XOPEN_SOURCE 500
		6
7	#include <sys/types.h>	7	#include <sys/types.h>
8	#include <sys/stat.h>	8	#include <sys/stat.h>
		9
9	#include <unistd.h>	10	#include <unistd.h>
10	#include <fcntl.h>	11	#include <fcntl.h>
11	#include <signal.h>	12	#include <signal.h>
12	#include <sched.h>	13	#include <sched.h>
13	#include <endian.h>	14	#if __linux
		15	#include <sys/syscall.h>
		16	#endif
14		17
15	#include <pthread.h>	18	#include <pthread.h>
16	#include <sys/syscall.h>
17		19
18	typedef void InputStream; / hack, but 5.6.1 is simply toooo old ;) */	20	typedef void InputStream; / hack, but 5.6.1 is simply toooo old ;) */
19	typedef void OutputStream; / hack, but 5.6.1 is simply toooo old ;) */	21	typedef void OutputStream; / hack, but 5.6.1 is simply toooo old ;) */
20	typedef void InOutStream; / hack, but 5.6.1 is simply toooo old ;) */	22	typedef void InOutStream; / hack, but 5.6.1 is simply toooo old ;) */
21		23
22	#if __i386 \|\| __amd64
23	# define STACKSIZE ( 256 * sizeof (long))
24	#elif __ia64	24	#if __ia64
25	# define STACKSIZE (8192 * sizeof (long))	25	# define STACKSIZE 65536
26	#else	26	#else
27	# define STACKSIZE ( 512 * sizeof (long))	27	# define STACKSIZE 4096
28	#endif	28	#endif
29		29
30	enum {	30	enum {
31	REQ_QUIT,	31	REQ_QUIT,
32	REQ_OPEN, REQ_CLOSE,	32	REQ_OPEN, REQ_CLOSE,
34	REQ_STAT, REQ_LSTAT, REQ_FSTAT, REQ_UNLINK,	34	REQ_STAT, REQ_LSTAT, REQ_FSTAT, REQ_UNLINK,
35	REQ_FSYNC, REQ_FDATASYNC,	35	REQ_FSYNC, REQ_FDATASYNC,
36	};	36	};
37		37
38	typedef struct aio_cb {	38	typedef struct aio_cb {
39	struct aio_cb *next;	39	struct aio_cb *volatile next;
40		40
41	int type;	41	int type;
42		42
43	int fd;	43	int fd;
44	off_t offset;	44	off_t offset;
…		…
55		55
56	typedef aio_cb *aio_req;	56	typedef aio_cb *aio_req;
57		57
58	static int started;	58	static int started;
59	static int nreqs;	59	static int nreqs;
		60	static int max_outstanding = 1<<30;
60	static int reqpipe[2], respipe[2];	61	static int respipe [2];
61		62
		63	static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER;
		64	static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER;
		65	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
		66
62	static aio_req qs, qe; /* queue start, queue end */	67	static volatile aio_req reqs, reqe; /* queue start, queue end */
		68	static volatile aio_req ress, rese; /* queue start, queue end */
		69
		70	static void
		71	poll_wait ()
		72	{
		73	if (!nreqs)
		74	return;
		75
		76	fd_set rfd;
		77	FD_ZERO(&rfd);
		78	FD_SET(respipe [0], &rfd);
		79
		80	select (respipe [0] + 1, &rfd, 0, 0, 0);
		81	}
		82
		83	static int
		84	poll_cb ()
		85	{
		86	dSP;
		87	int count = 0;
		88	aio_req req;
		89
		90	{
		91	/* read and signals sent by the worker threads */
		92	char buf [32];
		93	while (read (respipe [0], buf, 32) > 0)
		94	;
		95	}
		96
		97	for (;;)
		98	{
		99	pthread_mutex_lock (&reslock);
		100
		101	req = ress;
		102
		103	if (ress)
		104	{
		105	ress = ress->next;
		106	if (!ress) rese = 0;
		107	}
		108
		109	pthread_mutex_unlock (&reslock);
		110
		111	if (!req)
		112	break;
		113
		114	nreqs--;
		115
		116	if (req->type == REQ_QUIT)
		117	started--;
		118	else
		119	{
		120	int errorno = errno;
		121	errno = req->errorno;
		122
		123	if (req->type == REQ_READ)
		124	SvCUR_set (req->data, req->dataoffset
		125	+ req->result > 0 ? req->result : 0);
		126
		127	if (req->data)
		128	SvREFCNT_dec (req->data);
		129
		130	if (req->type == REQ_STAT \|\| req->type == REQ_LSTAT \|\| req->type == REQ_FSTAT)
		131	{
		132	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
		133	PL_laststatval = req->result;
		134	PL_statcache = *(req->statdata);
		135
		136	Safefree (req->statdata);
		137	}
		138
		139	PUSHMARK (SP);
		140	XPUSHs (sv_2mortal (newSViv (req->result)));
		141
		142	if (req->type == REQ_OPEN)
		143	{
		144	/* convert fd to fh */
		145	SV *fh;
		146
		147	PUTBACK;
		148	call_pv ("IO::AIO::_fd2fh", G_SCALAR \| G_EVAL);
		149	SPAGAIN;
		150
		151	fh = POPs;
		152
		153	PUSHMARK (SP);
		154	XPUSHs (fh);
		155	}
		156
		157	if (SvOK (req->callback))
		158	{
		159	PUTBACK;
		160	call_sv (req->callback, G_VOID \| G_EVAL);
		161	SPAGAIN;
		162	}
		163
		164	if (req->callback)
		165	SvREFCNT_dec (req->callback);
		166
		167	errno = errorno;
		168	count++;
		169	}
		170
		171	Safefree (req);
		172	}
		173
		174	return count;
		175	}
63		176
64	static void aio_proc(void arg);	177	static void aio_proc(void arg);
65		178
66	static void	179	static void
67	start_thread (void)	180	start_thread (void)
…		…
82		195
83	sigprocmask (SIG_SETMASK, &oldsigset, 0);	196	sigprocmask (SIG_SETMASK, &oldsigset, 0);
84	}	197	}
85		198
86	static void	199	static void
87	send_reqs (void)
88	{
89	/* this write is atomic */
90	while (qs && write (reqpipe[1], &qs, sizeof qs) == sizeof qs)
91	{
92	qs = qs->next;
93	if (!qs) qe = 0;
94	}
95	}
96
97	static void
98	send_req (aio_req req)	200	send_req (aio_req req)
99	{	201	{
100	nreqs++;	202	nreqs++;
		203
		204	pthread_mutex_lock (&reqlock);
		205
101	req->next = 0;	206	req->next = 0;
102		207
103	if (qe)	208	if (reqe)
104	{	209	{
105	qe->next = req;	210	reqe->next = req;
106	qe = req;	211	reqe = req;
107	}	212	}
108	else	213	else
109	qe = qs = req;	214	reqe = reqs = req;
110		215
111	send_reqs ();	216	pthread_cond_signal (&reqwait);
		217	pthread_mutex_unlock (&reqlock);
		218
		219	while (nreqs > max_outstanding)
		220	{
		221	poll_wait ();
		222	poll_cb ();
		223	}
112	}	224	}
113		225
114	static void	226	static void
115	end_thread (void)	227	end_thread (void)
116	{	228	{
…		…
120		232
121	send_req (req);	233	send_req (req);
122	}	234	}
123		235
124	static void	236	static void
125	read_write (pTHX_
126	int dowrite, int fd, off_t offset, size_t length,	237	read_write (int dowrite, int fd, off_t offset, size_t length,
127	SV data, STRLEN dataoffset, SV callback)	238	SV data, STRLEN dataoffset, SV callback)
128	{	239	{
129	aio_req req;	240	aio_req req;
130	STRLEN svlen;	241	STRLEN svlen;
131	char *svptr = SvPV (data, svlen);	242	char *svptr = SvPV (data, svlen);
…		…
168	req->callback = SvREFCNT_inc (callback);	279	req->callback = SvREFCNT_inc (callback);
169		280
170	send_req (req);	281	send_req (req);
171	}	282	}
172		283
173	static void
174	poll_wait ()
175	{
176	fd_set rfd;
177	FD_ZERO(&rfd);
178	FD_SET(respipe[0], &rfd);
179
180	select (respipe[0] + 1, &rfd, 0, 0, 0);
181	}
182
183	static int
184	poll_cb (pTHX)
185	{
186	dSP;
187	int count = 0;
188	aio_req req;
189
190	while (read (respipe[0], (void *)&req, sizeof (req)) == sizeof (req))
191	{
192	nreqs--;
193
194	if (req->type == REQ_QUIT)
195	started--;
196	else
197	{
198	int errorno = errno;
199	errno = req->errorno;
200
201	if (req->type == REQ_READ)
202	SvCUR_set (req->data, req->dataoffset
203	+ req->result > 0 ? req->result : 0);
204
205	if (req->data)
206	SvREFCNT_dec (req->data);
207
208	if (req->type == REQ_STAT \|\| req->type == REQ_LSTAT \|\| req->type == REQ_FSTAT)
209	{
210	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
211	PL_laststatval = req->result;
212	PL_statcache = *(req->statdata);
213
214	Safefree (req->statdata);
215	}
216
217	PUSHMARK (SP);
218	XPUSHs (sv_2mortal (newSViv (req->result)));
219	PUTBACK;
220	call_sv (req->callback, G_VOID);
221	SPAGAIN;
222
223	if (req->callback)
224	SvREFCNT_dec (req->callback);
225
226	errno = errorno;
227	count++;
228	}
229
230	Safefree (req);
231	}
232
233	if (qs)
234	send_reqs ();
235
236	return count;
237	}
238
239	static void *	284	static void *
240	aio_proc (void *thr_arg)	285	aio_proc (void *thr_arg)
241	{	286	{
242	aio_req req;	287	aio_req req;
		288	int type;
243		289
244	/* then loop */	290	do
245	while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
246	{	291	{
		292	pthread_mutex_lock (&reqlock);
		293
		294	for (;;)
		295	{
		296	req = reqs;
		297
		298	if (reqs)
		299	{
		300	reqs = reqs->next;
		301	if (!reqs) reqe = 0;
		302	}
		303
		304	if (req)
		305	break;
		306
		307	pthread_cond_wait (&reqwait, &reqlock);
		308	}
		309
		310	pthread_mutex_unlock (&reqlock);
		311
247	errno = 0; /* strictly unnecessary */	312	errno = 0; /* strictly unnecessary */
248		313
		314	type = req->type;
		315
249	switch (req->type)	316	switch (type)
250	{	317	{
251	case REQ_READ: req->result = pread64 (req->fd, req->dataptr, req->length, req->offset); break;	318	case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
252	case REQ_WRITE: req->result = pwrite64 (req->fd, req->dataptr, req->length, req->offset); break;	319	case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
253	#if SYS_readahead	320	#if SYS_readahead
254	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;	321	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
255	#else	322	#else
256	case REQ_READAHEAD: req->result = -1; errno = ENOSYS; break;	323	case REQ_READAHEAD: req->result = -1; errno = ENOSYS; break;
257	#endif	324	#endif
…		…
266		333
267	case REQ_FSYNC: req->result = fsync (req->fd); break;	334	case REQ_FSYNC: req->result = fsync (req->fd); break;
268	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;	335	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
269		336
270	case REQ_QUIT:	337	case REQ_QUIT:
271	write (respipe[1], (void *)&req, sizeof (req));	338	break;
272	return 0;
273		339
274	default:	340	default:
275	req->result = ENOSYS;	341	req->result = ENOSYS;
276	break;	342	break;
277	}	343	}
278		344
279	req->errorno = errno;	345	req->errorno = errno;
280	write (respipe[1], (void *)&req, sizeof (req));	346
		347	pthread_mutex_lock (&reslock);
		348
		349	req->next = 0;
		350
		351	if (rese)
		352	{
		353	rese->next = req;
		354	rese = req;
		355	}
		356	else
		357	{
		358	rese = ress = req;
		359
		360	/* write a dummy byte to the pipe so fh becomes ready */
		361	write (respipe [1], &respipe, 1);
		362	}
		363
		364	pthread_mutex_unlock (&reslock);
281	}	365	}
		366	while (type != REQ_QUIT);
282		367
283	return 0;	368	return 0;
284	}	369	}
285		370
286	MODULE = IO::AIO PACKAGE = IO::AIO	371	MODULE = IO::AIO PACKAGE = IO::AIO
287		372
		373	PROTOTYPES: ENABLE
		374
288	BOOT:	375	BOOT:
289	{	376	{
290	if (pipe (reqpipe) \|\| pipe (respipe))	377	if (pipe (respipe))
291	croak ("unable to initialize request or result pipe");	378	croak ("unable to initialize result pipe");
292		379
293	if (fcntl (reqpipe[1], F_SETFL, O_NONBLOCK))	380	if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
294	croak ("cannot set result pipe to nonblocking mode");	381	croak ("cannot set result pipe to nonblocking mode");
295		382
296	if (fcntl (respipe[0], F_SETFL, O_NONBLOCK))	383	if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
297	croak ("cannot set result pipe to nonblocking mode");	384	croak ("cannot set result pipe to nonblocking mode");
298	}	385	}
299		386
300	void	387	void
301	min_parallel(nthreads)	388	min_parallel(nthreads)
…		…
319	}	406	}
320		407
321	while (started > nthreads)	408	while (started > nthreads)
322	{	409	{
323	poll_wait ();	410	poll_wait ();
324	poll_cb (aTHX);	411	poll_cb ();
325	}	412	}
326	}	413	}
327		414
		415	int
		416	max_outstanding(nreqs)
		417	int nreqs
		418	PROTOTYPE: $
		419	CODE:
		420	RETVAL = max_outstanding;
		421	max_outstanding = nreqs;
		422
328	void	423	void
329	aio_open(pathname,flags,mode,callback)	424	aio_open(pathname,flags,mode,callback=&PL_sv_undef)
330	SV * pathname	425	SV * pathname
331	int flags	426	int flags
332	int mode	427	int mode
333	SV * callback	428	SV * callback
334	PROTOTYPE: $$$$	429	PROTOTYPE: $$$;$
335	CODE:	430	CODE:
336	{	431	{
337	aio_req req;	432	aio_req req;
338		433
339	Newz (0, req, 1, aio_cb);	434	Newz (0, req, 1, aio_cb);
…		…
350		445
351	send_req (req);	446	send_req (req);
352	}	447	}
353		448
354	void	449	void
355	aio_close(fh,callback)	450	aio_close(fh,callback=&PL_sv_undef)
356	InputStream fh	451	InputStream fh
357	SV * callback	452	SV * callback
358	PROTOTYPE: $$	453	PROTOTYPE: $;$
359	ALIAS:	454	ALIAS:
360	aio_close = REQ_CLOSE	455	aio_close = REQ_CLOSE
361	aio_fsync = REQ_FSYNC	456	aio_fsync = REQ_FSYNC
362	aio_fdatasync = REQ_FDATASYNC	457	aio_fdatasync = REQ_FDATASYNC
363	CODE:	458	CODE:
…		…
375		470
376	send_req (req);	471	send_req (req);
377	}	472	}
378		473
379	void	474	void
380	aio_read(fh,offset,length,data,dataoffset,callback)	475	aio_read(fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
381	InputStream fh	476	InputStream fh
382	UV offset	477	UV offset
383	IV length	478	IV length
384	SV * data	479	SV * data
385	IV dataoffset	480	IV dataoffset
386	SV * callback	481	SV * callback
387	PROTOTYPE: $$$$$$	482	PROTOTYPE: $$$$$;$
388	CODE:	483	CODE:
389	read_write (aTHX_ 0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);	484	read_write (0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
390		485
391	void	486	void
392	aio_write(fh,offset,length,data,dataoffset,callback)	487	aio_write(fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
393	OutputStream fh	488	OutputStream fh
394	UV offset	489	UV offset
395	IV length	490	IV length
396	SV * data	491	SV * data
397	IV dataoffset	492	IV dataoffset
398	SV * callback	493	SV * callback
399	PROTOTYPE: $$$$$$	494	PROTOTYPE: $$$$$;$
400	CODE:	495	CODE:
401	read_write (aTHX_ 1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);	496	read_write (1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
402		497
403	void	498	void
404	aio_readahead(fh,offset,length,callback)	499	aio_readahead(fh,offset,length,callback=&PL_sv_undef)
405	InputStream fh	500	InputStream fh
406	UV offset	501	UV offset
407	IV length	502	IV length
408	SV * callback	503	SV * callback
409	PROTOTYPE: $$$$	504	PROTOTYPE: $$$;$
410	CODE:	505	CODE:
411	{	506	{
412	aio_req req;	507	aio_req req;
413		508
414	if (length < 0)	509	if (length < 0)
…		…
427		522
428	send_req (req);	523	send_req (req);
429	}	524	}
430		525
431	void	526	void
432	aio_stat(fh_or_path,callback)	527	aio_stat(fh_or_path,callback=&PL_sv_undef)
433	SV * fh_or_path	528	SV * fh_or_path
434	SV * callback	529	SV * callback
435	PROTOTYPE: $$
436	ALIAS:	530	ALIAS:
		531	aio_stat = REQ_STAT
437	aio_lstat = 1	532	aio_lstat = REQ_LSTAT
438	CODE:	533	CODE:
439	{	534	{
440	aio_req req;	535	aio_req req;
441		536
442	Newz (0, req, 1, aio_cb);	537	Newz (0, req, 1, aio_cb);
…		…
449	if (!req->statdata)	544	if (!req->statdata)
450	croak ("out of memory during aio_req->statdata allocation");	545	croak ("out of memory during aio_req->statdata allocation");
451		546
452	if (SvPOK (fh_or_path))	547	if (SvPOK (fh_or_path))
453	{	548	{
454	req->type = ix ? REQ_LSTAT : REQ_STAT;	549	req->type = ix;
455	req->data = newSVsv (fh_or_path);	550	req->data = newSVsv (fh_or_path);
456	req->dataptr = SvPV_nolen (req->data);	551	req->dataptr = SvPV_nolen (req->data);
457	}	552	}
458	else	553	else
459	{	554	{
…		…
465		560
466	send_req (req);	561	send_req (req);
467	}	562	}
468		563
469	void	564	void
470	aio_unlink(pathname,callback)	565	aio_unlink(pathname,callback=&PL_sv_undef)
471	SV * pathname	566	SV * pathname
472	SV * callback	567	SV * callback
473	PROTOTYPE: $$
474	CODE:	568	CODE:
475	{	569	{
476	aio_req req;	570	aio_req req;
477		571
478	Newz (0, req, 1, aio_cb);	572	Newz (0, req, 1, aio_cb);
…		…
486	req->callback = SvREFCNT_inc (callback);	580	req->callback = SvREFCNT_inc (callback);
487		581
488	send_req (req);	582	send_req (req);
489	}	583	}
490		584
		585	void
		586	flush()
		587	PROTOTYPE:
		588	CODE:
		589	while (nreqs)
		590	{
		591	poll_wait ();
		592	poll_cb ();
		593	}
		594
		595	void
		596	poll()
		597	PROTOTYPE:
		598	CODE:
		599	if (nreqs)
		600	{
		601	poll_wait ();
		602	poll_cb ();
		603	}
		604
491	int	605	int
492	poll_fileno()	606	poll_fileno()
493	PROTOTYPE:	607	PROTOTYPE:
494	CODE:	608	CODE:
495	RETVAL = respipe[0];	609	RETVAL = respipe [0];
496	OUTPUT:	610	OUTPUT:
497	RETVAL	611	RETVAL
498		612
499	int	613	int
500	poll_cb(...)	614	poll_cb(...)
501	PROTOTYPE:	615	PROTOTYPE:
502	CODE:	616	CODE:
503	RETVAL = poll_cb (aTHX);	617	RETVAL = poll_cb ();
504	OUTPUT:	618	OUTPUT:
505	RETVAL	619	RETVAL
506		620
507	void	621	void
508	poll_wait()	622	poll_wait()
509	PROTOTYPE:	623	PROTOTYPE:
510	CODE:	624	CODE:
		625	if (nreqs)
511	poll_wait ();	626	poll_wait ();
512		627
513	int	628	int
514	nreqs()	629	nreqs()
515	PROTOTYPE:	630	PROTOTYPE:
516	CODE:	631	CODE:

Diff Legend

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

Comparing IO-AIO/AIO.xs (file contents): Revision 1.1 by root, Sun Jul 10 17:07:44 2005 UTC vs. Revision 1.10 by root, Wed Jul 13 00:13:09 2005 UTC

Diff Legend

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.1 by root, Sun Jul 10 17:07:44 2005 UTC vs.
Revision 1.10 by root, Wed Jul 13 00:13:09 2005 UTC