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

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

Diff Legend

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

Comparing IO-AIO/AIO.xs (file contents): Revision 1.2 by root, Sun Jul 10 18:16:49 2005 UTC vs. Revision 1.9 by root, Tue Jul 12 11:02:54 2005 UTC

Diff Legend

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.2 by root, Sun Jul 10 18:16:49 2005 UTC vs.
Revision 1.9 by root, Tue Jul 12 11:02:54 2005 UTC