[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.6 by root, Mon Jul 11 01:03:17 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	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 *	277	static void *
240	aio_proc (void *thr_arg)	278	aio_proc (void *thr_arg)
241	{	279	{
242	aio_req req;	280	aio_req req;
		281	int type;
243		282
244	/* then loop */	283	do
245	while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
246	{	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
247	errno = 0; /* strictly unnecessary */	305	errno = 0; /* strictly unnecessary */
248		306
		307	type = req->type;
		308
249	switch (req->type)	309	switch (type)
250	{	310	{
251	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;
252	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;
253	#if SYS_readahead	313	#if SYS_readahead
254	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;
…		…
266		326
267	case REQ_FSYNC: req->result = fsync (req->fd); break;	327	case REQ_FSYNC: req->result = fsync (req->fd); break;
268	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;	328	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
269		329
270	case REQ_QUIT:	330	case REQ_QUIT:
271	write (respipe[1], (void *)&req, sizeof (req));	331	break;
272	return 0;
273		332
274	default:	333	default:
275	req->result = ENOSYS;	334	req->result = ENOSYS;
276	break;	335	break;
277	}	336	}
278		337
279	req->errorno = errno;	338	req->errorno = errno;
280	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);
281	}	358	}
		359	while (type != REQ_QUIT);
282		360
283	return 0;	361	return 0;
284	}	362	}
285		363
286	MODULE = IO::AIO PACKAGE = IO::AIO	364	MODULE = IO::AIO PACKAGE = IO::AIO
287		365
288	BOOT:	366	BOOT:
289	{	367	{
290	if (pipe (reqpipe) \|\| pipe (respipe))	368	if (pipe (respipe))
291	croak ("unable to initialize request or result pipe");	369	croak ("unable to initialize result pipe");
292		370
293	if (fcntl (reqpipe[1], F_SETFL, O_NONBLOCK))	371	if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
294	croak ("cannot set result pipe to nonblocking mode");	372	croak ("cannot set result pipe to nonblocking mode");
295		373
296	if (fcntl (respipe[0], F_SETFL, O_NONBLOCK))	374	if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
297	croak ("cannot set result pipe to nonblocking mode");	375	croak ("cannot set result pipe to nonblocking mode");
298	}	376	}
299		377
300	void	378	void
301	min_parallel(nthreads)	379	min_parallel(nthreads)
…		…
319	}	397	}
320		398
321	while (started > nthreads)	399	while (started > nthreads)
322	{	400	{
323	poll_wait ();	401	poll_wait ();
324	poll_cb (aTHX);	402	poll_cb ();
325	}	403	}
326	}	404	}
		405
		406	int
		407	max_outstanding(nreqs)
		408	int nreqs
		409	PROTOTYPE: $
		410	CODE:
		411	RETVAL = max_outstanding;
		412	max_outstanding = nreqs;
327		413
328	void	414	void
329	aio_open(pathname,flags,mode,callback)	415	aio_open(pathname,flags,mode,callback)
330	SV * pathname	416	SV * pathname
331	int flags	417	int flags
…		…
384	SV * data	470	SV * data
385	IV dataoffset	471	IV dataoffset
386	SV * callback	472	SV * callback
387	PROTOTYPE: $$$$$$	473	PROTOTYPE: $$$$$$
388	CODE:	474	CODE:
389	read_write (aTHX_ 0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);	475	read_write (0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
390		476
391	void	477	void
392	aio_write(fh,offset,length,data,dataoffset,callback)	478	aio_write(fh,offset,length,data,dataoffset,callback)
393	OutputStream fh	479	OutputStream fh
394	UV offset	480	UV offset
…		…
396	SV * data	482	SV * data
397	IV dataoffset	483	IV dataoffset
398	SV * callback	484	SV * callback
399	PROTOTYPE: $$$$$$	485	PROTOTYPE: $$$$$$
400	CODE:	486	CODE:
401	read_write (aTHX_ 1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);	487	read_write (1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
402		488
403	void	489	void
404	aio_readahead(fh,offset,length,callback)	490	aio_readahead(fh,offset,length,callback)
405	InputStream fh	491	InputStream fh
406	UV offset	492	UV offset
…		…
486	req->callback = SvREFCNT_inc (callback);	572	req->callback = SvREFCNT_inc (callback);
487		573
488	send_req (req);	574	send_req (req);
489	}	575	}
490		576
		577	void
		578	flush()
		579	PROTOTYPE:
		580	CODE:
		581	while (nreqs)
		582	{
		583	poll_wait ();
		584	poll_cb ();
		585	}
		586
491	int	587	int
492	poll_fileno()	588	poll_fileno()
493	PROTOTYPE:	589	PROTOTYPE:
494	CODE:	590	CODE:
495	RETVAL = respipe[0];	591	RETVAL = respipe [0];
496	OUTPUT:	592	OUTPUT:
497	RETVAL	593	RETVAL
498		594
499	int	595	int
500	poll_cb(...)	596	poll_cb(...)
501	PROTOTYPE:	597	PROTOTYPE:
502	CODE:	598	CODE:
503	RETVAL = poll_cb (aTHX);	599	RETVAL = poll_cb ();
504	OUTPUT:	600	OUTPUT:
505	RETVAL	601	RETVAL
506		602
507	void	603	void
508	poll_wait()	604	poll_wait()
509	PROTOTYPE:	605	PROTOTYPE:
510	CODE:	606	CODE:
		607	if (nreqs)
511	poll_wait ();	608	poll_wait ();
512		609
513	int	610	int
514	nreqs()	611	nreqs()
515	PROTOTYPE:	612	PROTOTYPE:
516	CODE:	613	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.6 by root, Mon Jul 11 01:03:17 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.6 by root, Mon Jul 11 01:03:17 2005 UTC