ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/IO-AIO/AIO.xs
(Generate patch)

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.66 by root, Tue Oct 24 00:34:47 2006 UTC vs.
Revision 1.67 by root, Tue Oct 24 03:17:39 2006 UTC

… …
117 PRI_MIN = -4, 117 PRI_MIN = -4,
118 PRI_MAX = 4, 118 PRI_MAX = 4,
119 119
120 DEFAULT_PRI = 0, 120 DEFAULT_PRI = 0,
121 PRI_BIAS = -PRI_MIN, 121 PRI_BIAS = -PRI_MIN,
122 NUM_PRI = PRI_MAX + PRI_BIAS + 1,
122}; 123};
123 124
124static int next_pri = DEFAULT_PRI + PRI_BIAS; 125static int next_pri = DEFAULT_PRI + PRI_BIAS;
125 126
126static int started, wanted; 127static int started, wanted;
… …
136 137
137static pthread_mutex_t reslock = AIO_MUTEX_INIT; 138static pthread_mutex_t reslock = AIO_MUTEX_INIT;
138static pthread_mutex_t reqlock = AIO_MUTEX_INIT; 139static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
139static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 140static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
140 141
141static volatile aio_req reqs, reqe; /* queue start, queue end */ 142/*
142static volatile aio_req ress, rese; /* queue start, queue end */ 143 * a somewhat faster data structure might be nice, but
144 * with 8 priorities this actually needs <20 insns
145 * per shift, the most expensive operation.
146 */
147typedef struct {
148 aio_req qs[NUM_PRI], qe[NUM_PRI]; /* qstart, qend */
149 int size;
150} reqq;
151
152static reqq req_queue;
153static reqq res_queue;
154
155int reqq_push (reqq *q, aio_req req)
156{
157 int pri = req->pri;
158 req->next = 0;
159
160 if (q->qe[pri])
161 {
162 q->qe[pri]->next = req;
163 q->qe[pri] = req;
164 }
165 else
166 q->qe[pri] = q->qs[pri] = req;
167
168 return q->size++;
169}
170
171aio_req reqq_shift (reqq *q)
172{
173 int pri;
174
175 if (!q->size)
176 return 0;
177
178 --q->size;
179
180 for (pri = NUM_PRI; pri--; )
181 {
182 aio_req req = q->qs[pri];
183
184 if (req)
185 {
186 if (!(q->qs[pri] = req->next))
187 q->qe[pri] = 0;
188
189 return req;
190 }
191 }
192
193 abort ();
194}
143 195
144static void req_invoke (aio_req req); 196static void req_invoke (aio_req req);
145static void req_free (aio_req req); 197static void req_free (aio_req req);
146 198
147/* must be called at most once */ 199/* must be called at most once */
… …
181 ENTER; 233 ENTER;
182 SAVETMPS; 234 SAVETMPS;
183 PUSHMARK (SP); 235 PUSHMARK (SP);
184 XPUSHs (req_sv (grp, AIO_GRP_KLASS)); 236 XPUSHs (req_sv (grp, AIO_GRP_KLASS));
185 PUTBACK; 237 PUTBACK;
186 call_sv (grp->fh2, G_VOID | G_EVAL); 238 call_sv (grp->fh2, G_VOID | G_EVAL | G_KEEPERR);
187 SPAGAIN; 239 SPAGAIN;
188 FREETMPS; 240 FREETMPS;
189 LEAVE; 241 LEAVE;
190 } 242 }
191 243
… …
218{ 270{
219 fd_set rfd; 271 fd_set rfd;
220 272
221 while (nreqs) 273 while (nreqs)
222 { 274 {
223 aio_req req; 275 int size;
224#if !(__i386 || __x86_64) /* safe without sempahore on this archs */ 276#if !(__i386 || __x86_64) /* safe without sempahore on this archs */
225 pthread_mutex_lock (&reslock); 277 pthread_mutex_lock (&reslock);
226#endif 278#endif
227 req = ress; 279 size = res_queue.size;
228#if !(__i386 || __x86_64) /* safe without sempahore on this archs */ 280#if !(__i386 || __x86_64) /* safe without sempahore on this archs */
229 pthread_mutex_unlock (&reslock); 281 pthread_mutex_unlock (&reslock);
230#endif 282#endif
231 283
232 if (req) 284 if (size)
233 return; 285 return;
234 286
235 FD_ZERO(&rfd); 287 FD_ZERO(&rfd);
236 FD_SET(respipe [0], &rfd); 288 FD_SET(respipe [0], &rfd);
237 289
… …
240} 292}
241 293
242static void req_invoke (aio_req req) 294static void req_invoke (aio_req req)
243{ 295{
244 dSP; 296 dSP;
245 int errorno = errno;
246 297
247 if (req->flags & FLAG_CANCELLED || !SvOK (req->callback)) 298 if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))
248 return; 299 {
249
250 errno = req->errorno; 300 errno = req->errorno;
251 301
252 ENTER; 302 ENTER;
253 SAVETMPS; 303 SAVETMPS;
254 PUSHMARK (SP); 304 PUSHMARK (SP);
255 EXTEND (SP, 1); 305 EXTEND (SP, 1);
256 306
257 switch (req->type) 307 switch (req->type)
258 {
259 case REQ_READDIR:
260 { 308 {
261 SV *rv = &PL_sv_undef; 309 case REQ_READDIR:
262
263 if (req->result >= 0)
264 { 310 {
265 char *buf = req->data2ptr; 311 SV *rv = &PL_sv_undef;
266 AV *av = newAV ();
267 312
268 while (req->result) 313 if (req->result >= 0)
269 { 314 {
315 char *buf = req->data2ptr;
316 AV *av = newAV ();
317
318 while (req->result)
319 {
270 SV *sv = newSVpv (buf, 0); 320 SV *sv = newSVpv (buf, 0);
271 321
272 av_push (av, sv); 322 av_push (av, sv);
273 buf += SvCUR (sv) + 1; 323 buf += SvCUR (sv) + 1;
274 req->result--; 324 req->result--;
325 }
326
327 rv = sv_2mortal (newRV_noinc ((SV *)av));
275 } 328 }
276 329
277 rv = sv_2mortal (newRV_noinc ((SV *)av)); 330 PUSHs (rv);
278 } 331 }
332 break;
279 333
280 PUSHs (rv); 334 case REQ_OPEN:
335 {
336 /* convert fd to fh */
337 SV *fh;
338
339 PUSHs (sv_2mortal (newSViv (req->result)));
340 PUTBACK;
341 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
342 SPAGAIN;
343
344 fh = SvREFCNT_inc (POPs);
345
346 PUSHMARK (SP);
347 XPUSHs (sv_2mortal (fh));
348 }
349 break;
350
351 case REQ_GROUP:
352 req->fd = 2; /* mark group as finished */
353
354 if (req->data)
355 {
356 int i;
357 AV *av = (AV *)req->data;
358
359 EXTEND (SP, AvFILL (av) + 1);
360 for (i = 0; i <= AvFILL (av); ++i)
361 PUSHs (*av_fetch (av, i, 0));
362 }
363 break;
364
365 case REQ_NOP:
366 case REQ_SLEEP:
367 break;
368
369 default:
370 PUSHs (sv_2mortal (newSViv (req->result)));
371 break;
281 } 372 }
282 break;
283 373
284 case REQ_OPEN:
285 {
286 /* convert fd to fh */
287 SV *fh;
288 374
289 PUSHs (sv_2mortal (newSViv (req->result)));
290 PUTBACK; 375 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); 376 call_sv (req->callback, G_VOID | G_EVAL);
327 SPAGAIN; 377 SPAGAIN;
328 378
329 FREETMPS; 379 FREETMPS;
330 LEAVE; 380 LEAVE;
331
332 errno = errorno;
333
334 if (SvTRUE (ERRSV))
335 { 381 }
336 req_free (req);
337 croak (0);
338 }
339}
340 382
341static void req_free (aio_req req)
342{
343 if (req->grp) 383 if (req->grp)
344 { 384 {
345 aio_req grp = req->grp; 385 aio_req grp = req->grp;
346 386
347 /* unlink request */ 387 /* unlink request */
… …
352 grp->grp_first = req->grp_next; 392 grp->grp_first = req->grp_next;
353 393
354 aio_grp_dec (grp); 394 aio_grp_dec (grp);
355 } 395 }
356 396
397 if (SvTRUE (ERRSV))
398 {
399 req_free (req);
400 croak (0);
401 }
402}
403
404static void req_free (aio_req req)
405{
357 if (req->self) 406 if (req->self)
358 { 407 {
359 sv_unmagic (req->self, PERL_MAGIC_ext); 408 sv_unmagic (req->self, PERL_MAGIC_ext);
360 SvREFCNT_dec (req->self); 409 SvREFCNT_dec (req->self);
361 } 410 }
… …
393 aio_req req; 442 aio_req req;
394 443
395 for (;;) 444 for (;;)
396 { 445 {
397 pthread_mutex_lock (&reslock); 446 pthread_mutex_lock (&reslock);
398 req = ress; 447 req = reqq_shift (&res_queue);
399 448
400 if (req) 449 if (req)
401 { 450 {
402 ress = req->next;
403
404 if (!ress) 451 if (!res_queue.size)
405 { 452 {
406 /* read any signals sent by the worker threads */ 453 /* read any signals sent by the worker threads */
407 char buf [32]; 454 char buf [32];
408 while (read (respipe [0], buf, 32) == 32) 455 while (read (respipe [0], buf, 32) == 32)
409 ; 456 ;
410
411 rese = 0;
412 } 457 }
413 } 458 }
414 459
415 pthread_mutex_unlock (&reslock); 460 pthread_mutex_unlock (&reslock);
416 461
… …
479 start_thread (); 524 start_thread ();
480 525
481 ++nreqs; 526 ++nreqs;
482 527
483 pthread_mutex_lock (&reqlock); 528 pthread_mutex_lock (&reqlock);
484 529 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); 530 pthread_cond_signal (&reqwait);
496 pthread_mutex_unlock (&reqlock); 531 pthread_mutex_unlock (&reqlock);
497 532
498 if (nreqs > max_outstanding) 533 if (nreqs > max_outstanding)
499 for (;;) 534 for (;;)
… …
508} 543}
509 544
510static void end_thread (void) 545static void end_thread (void)
511{ 546{
512 aio_req req; 547 aio_req req;
548
513 Newz (0, req, 1, aio_cb); 549 Newz (0, req, 1, aio_cb);
550
514 req->type = REQ_QUIT; 551 req->type = REQ_QUIT;
552 req->pri = PRI_MAX + PRI_BIAS;
515 553
516 req_send (req); 554 req_send (req);
517} 555}
518 556
519static void min_parallel (int nthreads) 557static void min_parallel (int nthreads)
… …
829 { 867 {
830 pthread_mutex_lock (&reqlock); 868 pthread_mutex_lock (&reqlock);
831 869
832 for (;;) 870 for (;;)
833 { 871 {
834 req = reqs; 872 req = reqq_shift (&req_queue);
835
836 if (reqs)
837 {
838 reqs = reqs->next;
839 if (!reqs) reqe = 0;
840 }
841 873
842 if (req) 874 if (req)
843 break; 875 break;
844 876
845 pthread_cond_wait (&reqwait, &reqlock); 877 pthread_cond_wait (&reqwait, &reqlock);
… …
897 929
898 req->errorno = errno; 930 req->errorno = errno;
899 931
900 pthread_mutex_lock (&reslock); 932 pthread_mutex_lock (&reslock);
901 933
902 req->next = 0; 934 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 */ 935 /* write a dummy byte to the pipe so fh becomes ready */
914 write (respipe [1], &respipe, 1); 936 write (respipe [1], &respipe, 1);
915 }
916 937
917 pthread_mutex_unlock (&reslock); 938 pthread_mutex_unlock (&reslock);
918 } 939 }
919 while (type != REQ_QUIT); 940 while (type != REQ_QUIT);
920 941
… …
951{ 972{
952 aio_req prv; 973 aio_req prv;
953 974
954 started = 0; 975 started = 0;
955 976
956 while (reqs) 977 while (prv = reqq_shift (&req_queue))
957 {
958 prv = reqs;
959 reqs = prv->next;
960 req_free (prv); 978 req_free (prv);
961 }
962 979
963 reqs = reqe = 0; 980 while (prv = reqq_shift (&res_queue))
964
965 while (ress)
966 {
967 prv = ress;
968 ress = prv->next;
969 req_free (prv); 981 req_free (prv);
970 } 982
971
972 ress = rese = 0;
973
974 close (respipe [0]); 983 close (respipe [0]);
975 close (respipe [1]); 984 close (respipe [1]);
976 create_pipe (); 985 create_pipe ();
977 986
978 atfork_parent (); 987 atfork_parent ();
… …
1293 req->type = REQ_NOP; 1302 req->type = REQ_NOP;
1294 1303
1295 REQ_SEND; 1304 REQ_SEND;
1296} 1305}
1297 1306
1298#if 0
1299
1300void 1307void
1301aio_pri (int pri = DEFAULT_PRI) 1308aioreq_pri (int pri = DEFAULT_PRI)
1302 CODE: 1309 CODE:
1303 if (pri < PRI_MIN) pri = PRI_MIN; 1310 if (pri < PRI_MIN) pri = PRI_MIN;
1304 if (pri > PRI_MAX) pri = PRI_MAX; 1311 if (pri > PRI_MAX) pri = PRI_MAX;
1305 next_pri = pri + PRI_BIAS; 1312 next_pri = pri + PRI_BIAS;
1306
1307#endif
1308 1313
1309void 1314void
1310flush () 1315flush ()
1311 PROTOTYPE: 1316 PROTOTYPE:
1312 CODE: 1317 CODE:
… …
1420 SvREFCNT_dec (grp->data); 1425 SvREFCNT_dec (grp->data);
1421 grp->data = (SV *)av; 1426 grp->data = (SV *)av;
1422} 1427}
1423 1428
1424void 1429void
1425feed_limit (aio_req grp, int limit) 1430limit (aio_req grp, int limit)
1426 CODE: 1431 CODE:
1427 grp->fd2 = limit; 1432 grp->fd2 = limit;
1428 aio_grp_feed (grp); 1433 aio_grp_feed (grp);
1429 1434
1430void 1435void

Diff Legend

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