[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.4 by root, Sun Jul 10 20:57:00 2005 UTC

…		…
17		17
18	typedef void InputStream; / hack, but 5.6.1 is simply toooo old ;) */	18	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 ;) */	19	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 ;) */	20	typedef void InOutStream; / hack, but 5.6.1 is simply toooo old ;) */
21		21
22	#if __i386 \|\| __amd64
23	# define STACKSIZE ( 256 * sizeof (long))
24	#elif __ia64	22	#if __ia64
25	# define STACKSIZE (8192 * sizeof (long))	23	# define STACKSIZE 65536
26	#else	24	#else
27	# define STACKSIZE ( 512 * sizeof (long))	25	# define STACKSIZE 4096
28	#endif	26	#endif
29		27
30	enum {	28	enum {
31	REQ_QUIT,	29	REQ_QUIT,
32	REQ_OPEN, REQ_CLOSE,	30	REQ_OPEN, REQ_CLOSE,
…		…
34	REQ_STAT, REQ_LSTAT, REQ_FSTAT, REQ_UNLINK,	32	REQ_STAT, REQ_LSTAT, REQ_FSTAT, REQ_UNLINK,
35	REQ_FSYNC, REQ_FDATASYNC,	33	REQ_FSYNC, REQ_FDATASYNC,
36	};	34	};
37		35
38	typedef struct aio_cb {	36	typedef struct aio_cb {
39	struct aio_cb *next;	37	struct aio_cb *volatile next;
40		38
41	int type;	39	int type;
42		40
43	int fd;	41	int fd;
44	off_t offset;	42	off_t offset;
…		…
55		53
56	typedef aio_cb *aio_req;	54	typedef aio_cb *aio_req;
57		55
58	static int started;	56	static int started;
59	static int nreqs;	57	static int nreqs;
		58	static int max_outstanding = 1<<30;
60	static int reqpipe[2], respipe[2];	59	static int respipe [2];
61		60
		61	static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER;
		62	static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER;
		63	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
		64
62	static aio_req qs, qe; /* queue start, queue end */	65	static volatile aio_req reqs, reqe; /* queue start, queue end */
		66	static volatile aio_req ress, rese; /* queue start, queue end */
		67
		68	static void
		69	poll_wait ()
		70	{
		71	if (!nreqs)
		72	return;
		73
		74	fd_set rfd;
		75	FD_ZERO(&rfd);
		76	FD_SET(respipe [0], &rfd);
		77
		78	select (respipe [0] + 1, &rfd, 0, 0, 0);
		79	}
		80
		81	static int
		82	poll_cb (pTHX)
		83	{
		84	dSP;
		85	int count = 0;
		86	aio_req req;
		87
		88	{
		89	/* read and signals sent by the worker threads */
		90	char buf [32];
		91	while (read (respipe [0], buf, 32) > 0)
		92	;
		93	}
		94
		95	for (;;)
		96	{
		97	pthread_mutex_lock (&reslock);
		98
		99	req = ress;
		100
		101	if (ress)
		102	{
		103	ress = ress->next;
		104	if (!ress) rese = 0;
		105	}
		106
		107	pthread_mutex_unlock (&reslock);
		108
		109	if (!req)
		110	break;
		111
		112	nreqs--;
		113
		114	if (req->type == REQ_QUIT)
		115	started--;
		116	else
		117	{
		118	int errorno = errno;
		119	errno = req->errorno;
		120
		121	if (req->type == REQ_READ)
		122	SvCUR_set (req->data, req->dataoffset
		123	+ req->result > 0 ? req->result : 0);
		124
		125	if (req->data)
		126	SvREFCNT_dec (req->data);
		127
		128	if (req->type == REQ_STAT \|\| req->type == REQ_LSTAT \|\| req->type == REQ_FSTAT)
		129	{
		130	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
		131	PL_laststatval = req->result;
		132	PL_statcache = *(req->statdata);
		133
		134	Safefree (req->statdata);
		135	}
		136
		137	PUSHMARK (SP);
		138	XPUSHs (sv_2mortal (newSViv (req->result)));
		139
		140	if (req->type == REQ_OPEN)
		141	{
		142	/* convert fd to fh */
		143	SV *fh;
		144
		145	PUTBACK;
		146	call_pv ("IO::AIO::_fd2fh", G_SCALAR \| G_EVAL);
		147	SPAGAIN;
		148
		149	fh = POPs;
		150
		151	PUSHMARK (SP);
		152	XPUSHs (fh);
		153	}
		154
		155	PUTBACK;
		156	call_sv (req->callback, G_VOID \| G_EVAL);
		157	SPAGAIN;
		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	{
…		…
168	req->callback = SvREFCNT_inc (callback);	275	req->callback = SvREFCNT_inc (callback);
169		276
170	send_req (req);	277	send_req (req);
171	}	278	}
172		279
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 *	280	static void *
256	aio_proc (void *thr_arg)	281	aio_proc (void *thr_arg)
257	{	282	{
258	aio_req req;	283	aio_req req;
		284	int type;
259		285
260	/* then loop */	286	do
261	while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
262	{	287	{
		288	pthread_mutex_lock (&reqlock);
		289
		290	for (;;)
		291	{
		292	req = reqs;
		293
		294	if (reqs)
		295	{
		296	reqs = reqs->next;
		297	if (!reqs) reqe = 0;
		298	}
		299
		300	if (req)
		301	break;
		302
		303	pthread_cond_wait (&reqwait, &reqlock);
		304	}
		305
		306	pthread_mutex_unlock (&reqlock);
		307
263	errno = 0; /* strictly unnecessary */	308	errno = 0; /* strictly unnecessary */
264		309
		310	type = req->type;
		311
265	switch (req->type)	312	switch (type)
266	{	313	{
267	case REQ_READ: req->result = pread64 (req->fd, req->dataptr, req->length, req->offset); break;	314	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;	315	case REQ_WRITE: req->result = pwrite64 (req->fd, req->dataptr, req->length, req->offset); break;
269	#if SYS_readahead	316	#if SYS_readahead
270	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;	317	case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
…		…
282		329
283	case REQ_FSYNC: req->result = fsync (req->fd); break;	330	case REQ_FSYNC: req->result = fsync (req->fd); break;
284	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;	331	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
285		332
286	case REQ_QUIT:	333	case REQ_QUIT:
287	write (respipe[1], (void *)&req, sizeof (req));	334	break;
288	return 0;
289		335
290	default:	336	default:
291	req->result = ENOSYS;	337	req->result = ENOSYS;
292	break;	338	break;
293	}	339	}
294		340
295	req->errorno = errno;	341	req->errorno = errno;
296	write (respipe[1], (void *)&req, sizeof (req));	342
		343	pthread_mutex_lock (&reslock);
		344
		345	req->next = 0;
		346
		347	if (rese)
		348	{
		349	rese->next = req;
		350	rese = req;
		351	}
		352	else
		353	{
		354	rese = ress = req;
		355
		356	/* write a dummy byte to the pipe so fh becomes ready */
		357	write (respipe [1], &respipe, 1);
		358	}
		359
		360	pthread_mutex_unlock (&reslock);
297	}	361	}
		362	while (type != REQ_QUIT);
298		363
299	return 0;	364	return 0;
300	}	365	}
301		366
302	MODULE = IO::AIO PACKAGE = IO::AIO	367	MODULE = IO::AIO PACKAGE = IO::AIO
303		368
304	BOOT:	369	BOOT:
305	{	370	{
306	if (pipe (reqpipe) \|\| pipe (respipe))	371	if (pipe (respipe))
307	croak ("unable to initialize request or result pipe");	372	croak ("unable to initialize result pipe");
308		373
309	if (fcntl (reqpipe[1], F_SETFL, O_NONBLOCK))	374	if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
310	croak ("cannot set result pipe to nonblocking mode");	375	croak ("cannot set result pipe to nonblocking mode");
311		376
312	if (fcntl (respipe[0], F_SETFL, O_NONBLOCK))	377	if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
313	croak ("cannot set result pipe to nonblocking mode");	378	croak ("cannot set result pipe to nonblocking mode");
314	}	379	}
315		380
316	void	381	void
317	min_parallel(nthreads)	382	min_parallel(nthreads)
…		…
338	{	403	{
339	poll_wait ();	404	poll_wait ();
340	poll_cb (aTHX);	405	poll_cb (aTHX);
341	}	406	}
342	}	407	}
		408
		409	int
		410	max_outstanding(nreqs)
		411	int nreqs
		412	PROTOTYPE: $
		413	CODE:
		414	RETVAL = max_outstanding;
		415	max_outstanding = nreqs;
343		416
344	void	417	void
345	aio_open(pathname,flags,mode,callback)	418	aio_open(pathname,flags,mode,callback)
346	SV * pathname	419	SV * pathname
347	int flags	420	int flags
…		…
506		579
507	int	580	int
508	poll_fileno()	581	poll_fileno()
509	PROTOTYPE:	582	PROTOTYPE:
510	CODE:	583	CODE:
511	RETVAL = respipe[0];	584	RETVAL = respipe [0];
512	OUTPUT:	585	OUTPUT:
513	RETVAL	586	RETVAL
514		587
515	int	588	int
516	poll_cb(...)	589	poll_cb(...)
…		…
522		595
523	void	596	void
524	poll_wait()	597	poll_wait()
525	PROTOTYPE:	598	PROTOTYPE:
526	CODE:	599	CODE:
		600	if (nreqs)
527	poll_wait ();	601	poll_wait ();
528		602
529	int	603	int
530	nreqs()	604	nreqs()
531	PROTOTYPE:	605	PROTOTYPE:
532	CODE:	606	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.4 by root, Sun Jul 10 20:57:00 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.4 by root, Sun Jul 10 20:57:00 2005 UTC