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

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.66 by root, Tue Oct 24 00:34:47 2006 UTC vs.
Revision 1.70 by root, Tue Oct 24 15:15:56 2006 UTC

…		…
		1	/* solaris */
		2	#define _POSIX_PTHREAD_SEMANTICS 1
		3
1	#if __linux	4	#if __linux
2	# define _GNU_SOURCE	5	# define _GNU_SOURCE
3	#endif	6	#endif
4		7
5	#define _REENTRANT 1	8	#define _REENTRANT 1
73	REQ_FSYNC, REQ_FDATASYNC,	76	REQ_FSYNC, REQ_FDATASYNC,
74	REQ_UNLINK, REQ_RMDIR, REQ_RENAME,	77	REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
75	REQ_READDIR,	78	REQ_READDIR,
76	REQ_LINK, REQ_SYMLINK,	79	REQ_LINK, REQ_SYMLINK,
77	REQ_GROUP, REQ_NOP,	80	REQ_GROUP, REQ_NOP,
78	REQ_SLEEP,	81	REQ_BUSY,
79	};	82	};
80		83
81	#define AIO_REQ_KLASS "IO::AIO::REQ"	84	#define AIO_REQ_KLASS "IO::AIO::REQ"
82	#define AIO_GRP_KLASS "IO::AIO::GRP"	85	#define AIO_GRP_KLASS "IO::AIO::GRP"
83		86
…		…
117	PRI_MIN = -4,	120	PRI_MIN = -4,
118	PRI_MAX = 4,	121	PRI_MAX = 4,
119		122
120	DEFAULT_PRI = 0,	123	DEFAULT_PRI = 0,
121	PRI_BIAS = -PRI_MIN,	124	PRI_BIAS = -PRI_MIN,
		125	NUM_PRI = PRI_MAX + PRI_BIAS + 1,
122	};	126	};
123		127
124	static int next_pri = DEFAULT_PRI + PRI_BIAS;	128	static int next_pri = DEFAULT_PRI + PRI_BIAS;
125		129
126	static int started, wanted;	130	static int started, wanted;
…		…
136		140
137	static pthread_mutex_t reslock = AIO_MUTEX_INIT;	141	static pthread_mutex_t reslock = AIO_MUTEX_INIT;
138	static pthread_mutex_t reqlock = AIO_MUTEX_INIT;	142	static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
139	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;	143	static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
140		144
141	static volatile aio_req reqs, reqe; /* queue start, queue end */	145	/*
142	static volatile aio_req ress, rese; /* queue start, queue end */	146	* a somewhat faster data structure might be nice, but
		147	* with 8 priorities this actually needs <20 insns
		148	* per shift, the most expensive operation.
		149	*/
		150	typedef struct {
		151	aio_req qs[NUM_PRI], qe[NUM_PRI]; /* qstart, qend */
		152	int size;
		153	} reqq;
		154
		155	static reqq req_queue;
		156	static reqq res_queue;
		157
		158	int reqq_push (reqq *q, aio_req req)
		159	{
		160	int pri = req->pri;
		161	req->next = 0;
		162
		163	if (q->qe[pri])
		164	{
		165	q->qe[pri]->next = req;
		166	q->qe[pri] = req;
		167	}
		168	else
		169	q->qe[pri] = q->qs[pri] = req;
		170
		171	return q->size++;
		172	}
		173
		174	aio_req reqq_shift (reqq *q)
		175	{
		176	int pri;
		177
		178	if (!q->size)
		179	return 0;
		180
		181	--q->size;
		182
		183	for (pri = NUM_PRI; pri--; )
		184	{
		185	aio_req req = q->qs[pri];
		186
		187	if (req)
		188	{
		189	if (!(q->qs[pri] = req->next))
		190	q->qe[pri] = 0;
		191
		192	return req;
		193	}
		194	}
		195
		196	abort ();
		197	}
143		198
144	static void req_invoke (aio_req req);	199	static void req_invoke (aio_req req);
145	static void req_free (aio_req req);	200	static void req_free (aio_req req);
146		201
147	/* must be called at most once */	202	/* must be called at most once */
…		…
181	ENTER;	236	ENTER;
182	SAVETMPS;	237	SAVETMPS;
183	PUSHMARK (SP);	238	PUSHMARK (SP);
184	XPUSHs (req_sv (grp, AIO_GRP_KLASS));	239	XPUSHs (req_sv (grp, AIO_GRP_KLASS));
185	PUTBACK;	240	PUTBACK;
186	call_sv (grp->fh2, G_VOID \| G_EVAL);	241	call_sv (grp->fh2, G_VOID \| G_EVAL \| G_KEEPERR);
187	SPAGAIN;	242	SPAGAIN;
188	FREETMPS;	243	FREETMPS;
189	LEAVE;	244	LEAVE;
190	}	245	}
191		246
…		…
218	{	273	{
219	fd_set rfd;	274	fd_set rfd;
220		275
221	while (nreqs)	276	while (nreqs)
222	{	277	{
223	aio_req req;	278	int size;
224	#if !(__i386 \|\| __x86_64) /* safe without sempahore on this archs */	279	#if !(__i386 \|\| __x86_64) /* safe without sempahore on this archs */
225	pthread_mutex_lock (&reslock);	280	pthread_mutex_lock (&reslock);
226	#endif	281	#endif
227	req = ress;	282	size = res_queue.size;
228	#if !(__i386 \|\| __x86_64) /* safe without sempahore on this archs */	283	#if !(__i386 \|\| __x86_64) /* safe without sempahore on this archs */
229	pthread_mutex_unlock (&reslock);	284	pthread_mutex_unlock (&reslock);
230	#endif	285	#endif
231		286
232	if (req)	287	if (size)
233	return;	288	return;
234		289
235	FD_ZERO(&rfd);	290	FD_ZERO(&rfd);
236	FD_SET(respipe [0], &rfd);	291	FD_SET(respipe [0], &rfd);
237		292
…		…
240	}	295	}
241		296
242	static void req_invoke (aio_req req)	297	static void req_invoke (aio_req req)
243	{	298	{
244	dSP;	299	dSP;
245	int errorno = errno;
246		300
247	if (req->flags & FLAG_CANCELLED \|\| !SvOK (req->callback))	301	if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))
248	return;	302	{
249
250	errno = req->errorno;	303	errno = req->errorno;
251		304
252	ENTER;	305	ENTER;
253	SAVETMPS;	306	SAVETMPS;
254	PUSHMARK (SP);	307	PUSHMARK (SP);
255	EXTEND (SP, 1);	308	EXTEND (SP, 1);
256		309
257	switch (req->type)	310	switch (req->type)
258	{
259	case REQ_READDIR:
260	{	311	{
261	SV *rv = &PL_sv_undef;	312	case REQ_READDIR:
262
263	if (req->result >= 0)
264	{	313	{
265	char *buf = req->data2ptr;	314	SV *rv = &PL_sv_undef;
266	AV *av = newAV ();
267		315
268	while (req->result)	316	if (req->result >= 0)
269	{	317	{
		318	char *buf = req->data2ptr;
		319	AV *av = newAV ();
		320
		321	while (req->result)
		322	{
270	SV *sv = newSVpv (buf, 0);	323	SV *sv = newSVpv (buf, 0);
271		324
272	av_push (av, sv);	325	av_push (av, sv);
273	buf += SvCUR (sv) + 1;	326	buf += SvCUR (sv) + 1;
274	req->result--;	327	req->result--;
		328	}
		329
		330	rv = sv_2mortal (newRV_noinc ((SV *)av));
275	}	331	}
276		332
277	rv = sv_2mortal (newRV_noinc ((SV *)av));	333	PUSHs (rv);
278	}	334	}
		335	break;
279		336
280	PUSHs (rv);	337	case REQ_OPEN:
		338	{
		339	/* convert fd to fh */
		340	SV *fh;
		341
		342	PUSHs (sv_2mortal (newSViv (req->result)));
		343	PUTBACK;
		344	call_pv ("IO::AIO::_fd2fh", G_SCALAR \| G_EVAL);
		345	SPAGAIN;
		346
		347	fh = SvREFCNT_inc (POPs);
		348
		349	PUSHMARK (SP);
		350	XPUSHs (sv_2mortal (fh));
		351	}
		352	break;
		353
		354	case REQ_GROUP:
		355	req->fd = 2; /* mark group as finished */
		356
		357	if (req->data)
		358	{
		359	int i;
		360	AV av = (AV )req->data;
		361
		362	EXTEND (SP, AvFILL (av) + 1);
		363	for (i = 0; i <= AvFILL (av); ++i)
		364	PUSHs (*av_fetch (av, i, 0));
		365	}
		366	break;
		367
		368	case REQ_NOP:
		369	case REQ_BUSY:
		370	break;
		371
		372	default:
		373	PUSHs (sv_2mortal (newSViv (req->result)));
		374	break;
281	}	375	}
282	break;
283		376
284	case REQ_OPEN:
285	{
286	/* convert fd to fh */
287	SV *fh;
288		377
289	PUSHs (sv_2mortal (newSViv (req->result)));
290	PUTBACK;	378	PUTBACK;
291	call_pv ("IO::AIO::_fd2fh", G_SCALAR \| G_EVAL);
292	SPAGAIN;
293
294	fh = SvREFCNT_inc (POPs);
295
296	PUSHMARK (SP);
297	XPUSHs (sv_2mortal (fh));
298	}
299	break;
300
301	case REQ_GROUP:
302	req->fd = 2; /* mark group as finished */
303
304	if (req->data)
305	{
306	int i;
307	AV av = (AV )req->data;
308
309	EXTEND (SP, AvFILL (av) + 1);
310	for (i = 0; i <= AvFILL (av); ++i)
311	PUSHs (*av_fetch (av, i, 0));
312	}
313	break;
314
315	case REQ_NOP:
316	case REQ_SLEEP:
317	break;
318
319	default:
320	PUSHs (sv_2mortal (newSViv (req->result)));
321	break;
322	}
323
324
325	PUTBACK;
326	call_sv (req->callback, G_VOID \| G_EVAL);	379	call_sv (req->callback, G_VOID \| G_EVAL);
327	SPAGAIN;	380	SPAGAIN;
328		381
329	FREETMPS;	382	FREETMPS;
330	LEAVE;	383	LEAVE;
331
332	errno = errorno;
333
334	if (SvTRUE (ERRSV))
335	{	384	}
336	req_free (req);
337	croak (0);
338	}
339	}
340		385
341	static void req_free (aio_req req)
342	{
343	if (req->grp)	386	if (req->grp)
344	{	387	{
345	aio_req grp = req->grp;	388	aio_req grp = req->grp;
346		389
347	/* unlink request */	390	/* unlink request */
…		…
352	grp->grp_first = req->grp_next;	395	grp->grp_first = req->grp_next;
353		396
354	aio_grp_dec (grp);	397	aio_grp_dec (grp);
355	}	398	}
356		399
		400	if (SvTRUE (ERRSV))
		401	{
		402	req_free (req);
		403	croak (0);
		404	}
		405	}
		406
		407	static void req_free (aio_req req)
		408	{
357	if (req->self)	409	if (req->self)
358	{	410	{
359	sv_unmagic (req->self, PERL_MAGIC_ext);	411	sv_unmagic (req->self, PERL_MAGIC_ext);
360	SvREFCNT_dec (req->self);	412	SvREFCNT_dec (req->self);
361	}	413	}
…		…
393	aio_req req;	445	aio_req req;
394		446
395	for (;;)	447	for (;;)
396	{	448	{
397	pthread_mutex_lock (&reslock);	449	pthread_mutex_lock (&reslock);
398	req = ress;	450	req = reqq_shift (&res_queue);
399		451
400	if (req)	452	if (req)
401	{	453	{
402	ress = req->next;
403
404	if (!ress)	454	if (!res_queue.size)
405	{	455	{
406	/* read any signals sent by the worker threads */	456	/* read any signals sent by the worker threads */
407	char buf [32];	457	char buf [32];
408	while (read (respipe [0], buf, 32) == 32)	458	while (read (respipe [0], buf, 32) == 32)
409	;	459	;
410
411	rese = 0;
412	}	460	}
413	}	461	}
414		462
415	pthread_mutex_unlock (&reslock);	463	pthread_mutex_unlock (&reslock);
416		464
…		…
479	start_thread ();	527	start_thread ();
480		528
481	++nreqs;	529	++nreqs;
482		530
483	pthread_mutex_lock (&reqlock);	531	pthread_mutex_lock (&reqlock);
484		532	reqq_push (&req_queue, req);
485	req->next = 0;
486
487	if (reqe)
488	{
489	reqe->next = req;
490	reqe = req;
491	}
492	else
493	reqe = reqs = req;
494
495	pthread_cond_signal (&reqwait);	533	pthread_cond_signal (&reqwait);
496	pthread_mutex_unlock (&reqlock);	534	pthread_mutex_unlock (&reqlock);
497		535
498	if (nreqs > max_outstanding)	536	if (nreqs > max_outstanding)
499	for (;;)	537	for (;;)
…		…
508	}	546	}
509		547
510	static void end_thread (void)	548	static void end_thread (void)
511	{	549	{
512	aio_req req;	550	aio_req req;
		551
513	Newz (0, req, 1, aio_cb);	552	Newz (0, req, 1, aio_cb);
		553
514	req->type = REQ_QUIT;	554	req->type = REQ_QUIT;
		555	req->pri = PRI_MAX + PRI_BIAS;
515		556
516	req_send (req);	557	req_send (req);
517	}	558	}
518		559
519	static void min_parallel (int nthreads)	560	static void min_parallel (int nthreads)
…		…
829	{	870	{
830	pthread_mutex_lock (&reqlock);	871	pthread_mutex_lock (&reqlock);
831		872
832	for (;;)	873	for (;;)
833	{	874	{
834	req = reqs;	875	req = reqq_shift (&req_queue);
835
836	if (reqs)
837	{
838	reqs = reqs->next;
839	if (!reqs) reqe = 0;
840	}
841		876
842	if (req)	877	if (req)
843	break;	878	break;
844		879
845	pthread_cond_wait (&reqwait, &reqlock);	880	pthread_cond_wait (&reqwait, &reqlock);
…		…
873		908
874	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;	909	case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
875	case REQ_FSYNC: req->result = fsync (req->fd); break;	910	case REQ_FSYNC: req->result = fsync (req->fd); break;
876	case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break;	911	case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break;
877		912
878	case REQ_SLEEP:	913	case REQ_BUSY:
879	{	914	{
880	struct timeval tv;	915	struct timeval tv;
881		916
882	tv.tv_sec = req->fd;	917	tv.tv_sec = req->fd;
883	tv.tv_usec = req->fd2;	918	tv.tv_usec = req->fd2;
…		…
897		932
898	req->errorno = errno;	933	req->errorno = errno;
899		934
900	pthread_mutex_lock (&reslock);	935	pthread_mutex_lock (&reslock);
901		936
902	req->next = 0;	937	if (!reqq_push (&res_queue, req))
903
904	if (rese)
905	{
906	rese->next = req;
907	rese = req;
908	}
909	else
910	{
911	rese = ress = req;
912
913	/* write a dummy byte to the pipe so fh becomes ready */	938	/* write a dummy byte to the pipe so fh becomes ready */
914	write (respipe [1], &respipe, 1);	939	write (respipe [1], &respipe, 1);
915	}
916		940
917	pthread_mutex_unlock (&reslock);	941	pthread_mutex_unlock (&reslock);
918	}	942	}
919	while (type != REQ_QUIT);	943	while (type != REQ_QUIT);
920		944
…		…
951	{	975	{
952	aio_req prv;	976	aio_req prv;
953		977
954	started = 0;	978	started = 0;
955		979
956	while (reqs)	980	while (prv = reqq_shift (&req_queue))
957	{
958	prv = reqs;
959	reqs = prv->next;
960	req_free (prv);	981	req_free (prv);
961	}
962		982
963	reqs = reqe = 0;	983	while (prv = reqq_shift (&res_queue))
964
965	while (ress)
966	{
967	prv = ress;
968	ress = prv->next;
969	req_free (prv);	984	req_free (prv);
970	}	985
971
972	ress = rese = 0;
973
974	close (respipe [0]);	986	close (respipe [0]);
975	close (respipe [1]);	987	close (respipe [1]);
976	create_pipe ();	988	create_pipe ();
977		989
978	atfork_parent ();	990	atfork_parent ();
…		…
1253		1265
1254	REQ_SEND;	1266	REQ_SEND;
1255	}	1267	}
1256		1268
1257	void	1269	void
1258	aio_sleep (delay,callback=&PL_sv_undef)	1270	aio_busy (delay,callback=&PL_sv_undef)
1259	double delay	1271	double delay
1260	SV * callback	1272	SV * callback
1261	PPCODE:	1273	PPCODE:
1262	{	1274	{
1263	dREQ;	1275	dREQ;
1264		1276
1265	req->type = REQ_SLEEP;	1277	req->type = REQ_BUSY;
1266	req->fd = delay < 0. ? 0 : delay;	1278	req->fd = delay < 0. ? 0 : delay;
1267	req->fd2 = delay < 0. ? 0 : 1000. * (delay - req->fd);	1279	req->fd2 = delay < 0. ? 0 : 1000. * (delay - req->fd);
1268		1280
1269	REQ_SEND;	1281	REQ_SEND;
1270	}	1282	}
…		…
1293	req->type = REQ_NOP;	1305	req->type = REQ_NOP;
1294		1306
1295	REQ_SEND;	1307	REQ_SEND;
1296	}	1308	}
1297		1309
1298	#if 0
1299
1300	void	1310	void
1301	aio_pri (int pri = DEFAULT_PRI)	1311	aioreq_pri (int pri = DEFAULT_PRI)
1302	CODE:	1312	CODE:
1303	if (pri < PRI_MIN) pri = PRI_MIN;	1313	if (pri < PRI_MIN) pri = PRI_MIN;
1304	if (pri > PRI_MAX) pri = PRI_MAX;	1314	if (pri > PRI_MAX) pri = PRI_MAX;
1305	next_pri = pri + PRI_BIAS;	1315	next_pri = pri + PRI_BIAS;
1306		1316
1307	#endif	1317	void
		1318	aioreq_nice (int nice = 0)
		1319	CODE:
		1320	nice = next_pri - nice;
		1321	if (nice < PRI_MIN) nice = PRI_MIN;
		1322	if (nice > PRI_MAX) nice = PRI_MAX;
		1323	next_pri = nice + PRI_BIAS;
1308		1324
1309	void	1325	void
1310	flush ()	1326	flush ()
1311	PROTOTYPE:	1327	PROTOTYPE:
1312	CODE:	1328	CODE:
…		…
1420	SvREFCNT_dec (grp->data);	1436	SvREFCNT_dec (grp->data);
1421	grp->data = (SV *)av;	1437	grp->data = (SV *)av;
1422	}	1438	}
1423		1439
1424	void	1440	void
1425	feed_limit (aio_req grp, int limit)	1441	limit (aio_req grp, int limit)
1426	CODE:	1442	CODE:
1427	grp->fd2 = limit;	1443	grp->fd2 = limit;
1428	aio_grp_feed (grp);	1444	aio_grp_feed (grp);
1429		1445
1430	void	1446	void

Diff Legend

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

Comparing IO-AIO/AIO.xs (file contents): Revision 1.66 by root, Tue Oct 24 00:34:47 2006 UTC vs. Revision 1.70 by root, Tue Oct 24 15:15:56 2006 UTC

Diff Legend

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.66 by root, Tue Oct 24 00:34:47 2006 UTC vs.
Revision 1.70 by root, Tue Oct 24 15:15:56 2006 UTC