[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.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	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 *	279	static void *
240	aio_proc (void *thr_arg)	280	aio_proc (void *thr_arg)
241	{	281	{
242	aio_req req;	282	aio_req req;
		283	int type;
243		284
244	/* then loop */	285	do
245	while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
246	{	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
247	errno = 0; /* strictly unnecessary */	307	errno = 0; /* strictly unnecessary */
248		308
		309	type = req->type;
		310
249	switch (req->type)	311	switch (type)
250	{	312	{
251	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;
252	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;
253	#if SYS_readahead	315	#if SYS_readahead
254	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;
…		…
266		328
267	case REQ_FSYNC: req->result = fsync (req->fd); break;	329	case REQ_FSYNC: req->result = fsync (req->fd); break;
268	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;	330	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
269		331
270	case REQ_QUIT:	332	case REQ_QUIT:
271	write (respipe[1], (void *)&req, sizeof (req));	333	break;
272	return 0;
273		334
274	default:	335	default:
275	req->result = ENOSYS;	336	req->result = ENOSYS;
276	break;	337	break;
277	}	338	}
278		339
279	req->errorno = errno;	340	req->errorno = errno;
280	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);
281	}	360	}
		361	while (type != REQ_QUIT);
282		362
283	return 0;	363	return 0;
284	}	364	}
285		365
286	MODULE = IO::AIO PACKAGE = IO::AIO	366	MODULE = IO::AIO PACKAGE = IO::AIO
287		367
		368	PROTOTYPES: ENABLE
		369
288	BOOT:	370	BOOT:
289	{	371	{
290	if (pipe (reqpipe) \|\| pipe (respipe))	372	if (pipe (respipe))
291	croak ("unable to initialize request or result pipe");	373	croak ("unable to initialize result pipe");
292		374
293	if (fcntl (reqpipe[1], F_SETFL, O_NONBLOCK))	375	if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
294	croak ("cannot set result pipe to nonblocking mode");	376	croak ("cannot set result pipe to nonblocking mode");
295		377
296	if (fcntl (respipe[0], F_SETFL, O_NONBLOCK))	378	if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
297	croak ("cannot set result pipe to nonblocking mode");	379	croak ("cannot set result pipe to nonblocking mode");
298	}	380	}
299		381
300	void	382	void
301	min_parallel(nthreads)	383	min_parallel(nthreads)
…		…
319	}	401	}
320		402
321	while (started > nthreads)	403	while (started > nthreads)
322	{	404	{
323	poll_wait ();	405	poll_wait ();
324	poll_cb (aTHX);	406	poll_cb ();
325	}	407	}
326	}	408	}
327		409
		410	int
		411	max_outstanding(nreqs)
		412	int nreqs
		413	PROTOTYPE: $
		414	CODE:
		415	RETVAL = max_outstanding;
		416	max_outstanding = nreqs;
		417
328	void	418	void
329	aio_open(pathname,flags,mode,callback)	419	aio_open(pathname,flags,mode,callback=&PL_sv_undef)
330	SV * pathname	420	SV * pathname
331	int flags	421	int flags
332	int mode	422	int mode
333	SV * callback	423	SV * callback
334	PROTOTYPE: $$$$	424	PROTOTYPE: $$$;$
335	CODE:	425	CODE:
336	{	426	{
337	aio_req req;	427	aio_req req;
338		428
339	Newz (0, req, 1, aio_cb);	429	Newz (0, req, 1, aio_cb);
…		…
350		440
351	send_req (req);	441	send_req (req);
352	}	442	}
353		443
354	void	444	void
355	aio_close(fh,callback)	445	aio_close(fh,callback=&PL_sv_undef)
356	InputStream fh	446	InputStream fh
357	SV * callback	447	SV * callback
358	PROTOTYPE: $$	448	PROTOTYPE: $;$
359	ALIAS:	449	ALIAS:
360	aio_close = REQ_CLOSE	450	aio_close = REQ_CLOSE
361	aio_fsync = REQ_FSYNC	451	aio_fsync = REQ_FSYNC
362	aio_fdatasync = REQ_FDATASYNC	452	aio_fdatasync = REQ_FDATASYNC
363	CODE:	453	CODE:
…		…
375		465
376	send_req (req);	466	send_req (req);
377	}	467	}
378		468
379	void	469	void
380	aio_read(fh,offset,length,data,dataoffset,callback)	470	aio_read(fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
381	InputStream fh	471	InputStream fh
382	UV offset	472	UV offset
383	IV length	473	IV length
384	SV * data	474	SV * data
385	IV dataoffset	475	IV dataoffset
386	SV * callback	476	SV * callback
387	PROTOTYPE: $$$$$$	477	PROTOTYPE: $$$$$;$
388	CODE:	478	CODE:
389	read_write (aTHX_ 0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);	479	read_write (0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
390		480
391	void	481	void
392	aio_write(fh,offset,length,data,dataoffset,callback)	482	aio_write(fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
393	OutputStream fh	483	OutputStream fh
394	UV offset	484	UV offset
395	IV length	485	IV length
396	SV * data	486	SV * data
397	IV dataoffset	487	IV dataoffset
398	SV * callback	488	SV * callback
399	PROTOTYPE: $$$$$$	489	PROTOTYPE: $$$$$;$
400	CODE:	490	CODE:
401	read_write (aTHX_ 1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);	491	read_write (1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
402		492
403	void	493	void
404	aio_readahead(fh,offset,length,callback)	494	aio_readahead(fh,offset,length,callback=&PL_sv_undef)
405	InputStream fh	495	InputStream fh
406	UV offset	496	UV offset
407	IV length	497	IV length
408	SV * callback	498	SV * callback
409	PROTOTYPE: $$$$	499	PROTOTYPE: $$$;$
410	CODE:	500	CODE:
411	{	501	{
412	aio_req req;	502	aio_req req;
413		503
414	if (length < 0)	504	if (length < 0)
…		…
427		517
428	send_req (req);	518	send_req (req);
429	}	519	}
430		520
431	void	521	void
432	aio_stat(fh_or_path,callback)	522	aio_stat(fh_or_path,callback=&PL_sv_undef)
433	SV * fh_or_path	523	SV * fh_or_path
434	SV * callback	524	SV * callback
435	PROTOTYPE: $$
436	ALIAS:	525	ALIAS:
		526	aio_stat = REQ_STAT
437	aio_lstat = 1	527	aio_lstat = REQ_LSTAT
438	CODE:	528	CODE:
439	{	529	{
440	aio_req req;	530	aio_req req;
441		531
442	Newz (0, req, 1, aio_cb);	532	Newz (0, req, 1, aio_cb);
…		…
449	if (!req->statdata)	539	if (!req->statdata)
450	croak ("out of memory during aio_req->statdata allocation");	540	croak ("out of memory during aio_req->statdata allocation");
451		541
452	if (SvPOK (fh_or_path))	542	if (SvPOK (fh_or_path))
453	{	543	{
454	req->type = ix ? REQ_LSTAT : REQ_STAT;	544	req->type = ix;
455	req->data = newSVsv (fh_or_path);	545	req->data = newSVsv (fh_or_path);
456	req->dataptr = SvPV_nolen (req->data);	546	req->dataptr = SvPV_nolen (req->data);
457	}	547	}
458	else	548	else
459	{	549	{
…		…
465		555
466	send_req (req);	556	send_req (req);
467	}	557	}
468		558
469	void	559	void
470	aio_unlink(pathname,callback)	560	aio_unlink(pathname,callback=&PL_sv_undef)
471	SV * pathname	561	SV * pathname
472	SV * callback	562	SV * callback
473	PROTOTYPE: $$
474	CODE:	563	CODE:
475	{	564	{
476	aio_req req;	565	aio_req req;
477		566
478	Newz (0, req, 1, aio_cb);	567	Newz (0, req, 1, aio_cb);
…		…
486	req->callback = SvREFCNT_inc (callback);	575	req->callback = SvREFCNT_inc (callback);
487		576
488	send_req (req);	577	send_req (req);
489	}	578	}
490		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
491	int	600	int
492	poll_fileno()	601	poll_fileno()
493	PROTOTYPE:	602	PROTOTYPE:
494	CODE:	603	CODE:
495	RETVAL = respipe[0];	604	RETVAL = respipe [0];
496	OUTPUT:	605	OUTPUT:
497	RETVAL	606	RETVAL
498		607
499	int	608	int
500	poll_cb(...)	609	poll_cb(...)
501	PROTOTYPE:	610	PROTOTYPE:
502	CODE:	611	CODE:
503	RETVAL = poll_cb (aTHX);	612	RETVAL = poll_cb ();
504	OUTPUT:	613	OUTPUT:
505	RETVAL	614	RETVAL
506		615
507	void	616	void
508	poll_wait()	617	poll_wait()
509	PROTOTYPE:	618	PROTOTYPE:
510	CODE:	619	CODE:
		620	if (nreqs)
511	poll_wait ();	621	poll_wait ();
512		622
513	int	623	int
514	nreqs()	624	nreqs()
515	PROTOTYPE:	625	PROTOTYPE:
516	CODE:	626	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.9 by root, Tue Jul 12 11:02:54 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.9 by root, Tue Jul 12 11:02:54 2005 UTC