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