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.78 by root, Thu Oct 26 14:35:34 2006 UTC vs.
Revision 1.80 by root, Fri Oct 27 19:17:23 2006 UTC

… …
61# define STACKSIZE PTHREAD_STACK_MIN 61# define STACKSIZE PTHREAD_STACK_MIN
62#else 62#else
63# define STACKSIZE 16384 63# define STACKSIZE 16384
64#endif 64#endif
65 65
66/* wether word reads are potentially non-atomic.
67 * this is conservatice, likely most arches this runs
68 * on have atomic word read/writes.
69 */
70#ifndef WORDREAD_UNSAFE
71# if __i386 || __x86_64
72# define WORDREAD_UNSAFE 0
73# else
74# define WORDREAD_UNSAFE 1
75# endif
76#endif
77
66/* buffer size for various temporary buffers */ 78/* buffer size for various temporary buffers */
67#define AIO_BUFSIZE 65536 79#define AIO_BUFSIZE 65536
68 80
69#define dBUF \ 81#define dBUF \
70 char *aio_buf; \ 82 char *aio_buf; \
… …
133}; 145};
134 146
135static int next_pri = DEFAULT_PRI + PRI_BIAS; 147static int next_pri = DEFAULT_PRI + PRI_BIAS;
136 148
137static unsigned int started, wanted; 149static unsigned int started, wanted;
138static volatile unsigned int nreqs;
139static volatile unsigned int max_outstanding = 0xffffffff;
140static int respipe [2];
141 150
142#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP) 151#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
143# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP 152# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
144#else 153#else
145# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER 154# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
… …
186 wrk->prev->next = wrk->next; 195 wrk->prev->next = wrk->next;
187 196
188 free (wrk); 197 free (wrk);
189} 198}
190 199
200static volatile unsigned int nreqs, nready, npending;
201static volatile unsigned int max_outstanding = 0xffffffff;
202static int respipe [2];
203
191static pthread_mutex_t reslock = AIO_MUTEX_INIT; 204static pthread_mutex_t reslock = AIO_MUTEX_INIT;
192static pthread_mutex_t reqlock = AIO_MUTEX_INIT; 205static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
193static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 206static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
207
208#if WORDREAD_UNSAFE
209
210static unsigned int get_nready ()
211{
212 unsigned int retval;
213
214 LOCK (reqlock);
215 retval = nready;
216 UNLOCK (reqlock);
217
218 return retval;
219}
220
221static unsigned int get_npending ()
222{
223 unsigned int retval;
224
225 LOCK (reslock);
226 retval = npending;
227 UNLOCK (reslock);
228
229 return retval;
230}
231
232#else
233
234# define get_nready() nready
235# define get_npending() npending
236
237#endif
194 238
195/* 239/*
196 * a somewhat faster data structure might be nice, but 240 * a somewhat faster data structure might be nice, but
197 * with 8 priorities this actually needs <20 insns 241 * with 8 priorities this actually needs <20 insns
198 * per shift, the most expensive operation. 242 * per shift, the most expensive operation.
… …
319 req_invoke (grp); 363 req_invoke (grp);
320 req_free (grp); 364 req_free (grp);
321 } 365 }
322} 366}
323 367
324static void poll_wait ()
325{
326 fd_set rfd;
327
328 while (nreqs)
329 {
330 int size;
331#if !(__i386 || __x86_64) /* safe without sempahore on these archs */
332 LOCK (reslock);
333#endif
334 size = res_queue.size;
335#if !(__i386 || __x86_64) /* safe without sempahore on these archs */
336 UNLOCK (reslock);
337#endif
338
339 if (size)
340 return;
341
342 FD_ZERO(&rfd);
343 FD_SET(respipe [0], &rfd);
344
345 select (respipe [0] + 1, &rfd, 0, 0, 0);
346 }
347}
348
349static void req_invoke (aio_req req) 368static void req_invoke (aio_req req)
350{ 369{
351 dSP; 370 dSP;
352 371
353 if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback)) 372 if (!(req->flags & FLAG_CANCELLED) && SvOK (req->callback))
354 { 373 {
355 errno = req->errorno;
356
357 ENTER; 374 ENTER;
358 SAVETMPS; 375 SAVETMPS;
359 PUSHMARK (SP); 376 PUSHMARK (SP);
360 EXTEND (SP, 1); 377 EXTEND (SP, 1);
361 378
… …
426 default: 443 default:
427 PUSHs (sv_2mortal (newSViv (req->result))); 444 PUSHs (sv_2mortal (newSViv (req->result)));
428 break; 445 break;
429 } 446 }
430 447
448 errno = req->errorno;
431 449
432 PUTBACK; 450 PUTBACK;
433 call_sv (req->callback, G_VOID | G_EVAL); 451 call_sv (req->callback, G_VOID | G_EVAL);
434 SPAGAIN; 452 SPAGAIN;
435 453
… …
497 req->flags |= FLAG_CANCELLED; 515 req->flags |= FLAG_CANCELLED;
498 516
499 req_cancel_subs (req); 517 req_cancel_subs (req);
500} 518}
501 519
520static void *aio_proc(void *arg);
521
522static void start_thread (void)
523{
524 sigset_t fullsigset, oldsigset;
525 pthread_attr_t attr;
526
527 worker *wrk = calloc (1, sizeof (worker));
528
529 if (!wrk)
530 croak ("unable to allocate worker thread data");
531
532 pthread_attr_init (&attr);
533 pthread_attr_setstacksize (&attr, STACKSIZE);
534 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
535#ifdef PTHREAD_SCOPE_PROCESS
536 pthread_attr_setscope (&attr, PTHREAD_SCOPE_PROCESS);
537#endif
538
539 sigfillset (&fullsigset);
540
541 LOCK (wrklock);
542 sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
543
544 if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
545 {
546 wrk->prev = &wrk_first;
547 wrk->next = wrk_first.next;
548 wrk_first.next->prev = wrk;
549 wrk_first.next = wrk;
550 ++started;
551 }
552 else
553 free (wrk);
554
555 sigprocmask (SIG_SETMASK, &oldsigset, 0);
556 UNLOCK (wrklock);
557}
558
559static void maybe_start_thread ()
560{
561#if 0
562 static struct timeval last;
563 struct timeval diff, now;
564#endif
565
566 if (started >= wanted)
567 return;
568
569 if (nready <= nreqs - get_nready () - get_npending ())
570 return;
571
572#if 0
573 gettimeofday (&now, 0);
574
575 diff.tv_sec = now.tv_sec - last.tv_sec;
576 diff.tv_usec = now.tv_usec - last.tv_usec;
577
578 if (diff.tv_usec < 0)
579 {
580 --diff.tv_sec;
581 diff.tv_usec += 1000000;
582 }
583
584 if (!diff.tv_sec && diff.tv_usec < 10000)
585 return;
586
587 last = now;
588#endif
589
590 start_thread ();
591}
592
593static void req_send (aio_req req)
594{
595 ++nreqs;
596
597 LOCK (reqlock);
598 ++nready;
599 reqq_push (&req_queue, req);
600 pthread_cond_signal (&reqwait);
601 UNLOCK (reqlock);
602
603 maybe_start_thread ();
604}
605
606static void end_thread (void)
607{
608 aio_req req;
609
610 Newz (0, req, 1, aio_cb);
611
612 req->type = REQ_QUIT;
613 req->pri = PRI_MAX + PRI_BIAS;
614
615 req_send (req);
616
617 LOCK (wrklock);
618 --started;
619 UNLOCK (wrklock);
620}
621
622static void min_parallel (int nthreads)
623{
624 if (wanted < nthreads)
625 wanted = nthreads;
626}
627
628static void max_parallel (int nthreads)
629{
630 if (wanted > nthreads)
631 wanted = nthreads;
632
633 while (started > wanted)
634 end_thread ();
635}
636
637static void poll_wait ()
638{
639 fd_set rfd;
640
641 while (nreqs)
642 {
643 int size;
644 if (WORDREAD_UNSAFE) LOCK (reslock);
645 size = res_queue.size;
646 if (WORDREAD_UNSAFE) UNLOCK (reslock);
647
648 if (size)
649 return;
650
651 maybe_start_thread ();
652
653 FD_ZERO(&rfd);
654 FD_SET(respipe [0], &rfd);
655
656 select (respipe [0] + 1, &rfd, 0, 0, 0);
657 }
658}
659
502static int poll_cb (int max) 660static int poll_cb (int max)
503{ 661{
504 dSP; 662 dSP;
505 int count = 0; 663 int count = 0;
506 int do_croak = 0; 664 int do_croak = 0;
… …
508 666
509 for (;;) 667 for (;;)
510 { 668 {
511 while (max <= 0 || count < max) 669 while (max <= 0 || count < max)
512 { 670 {
671 maybe_start_thread ();
672
513 LOCK (reslock); 673 LOCK (reslock);
514 req = reqq_shift (&res_queue); 674 req = reqq_shift (&res_queue);
515 675
516 if (req) 676 if (req)
517 { 677 {
678 --npending;
679
518 if (!res_queue.size) 680 if (!res_queue.size)
519 { 681 {
520 /* read any signals sent by the worker threads */ 682 /* read any signals sent by the worker threads */
521 char buf [32]; 683 char buf [32];
522 while (read (respipe [0], buf, 32) == 32) 684 while (read (respipe [0], buf, 32) == 32)
… …
529 if (!req) 691 if (!req)
530 break; 692 break;
531 693
532 --nreqs; 694 --nreqs;
533 695
534 if (req->type == REQ_QUIT)
535 --started;
536 else if (req->type == REQ_GROUP && req->length) 696 if (req->type == REQ_GROUP && req->length)
537 { 697 {
538 req->fd = 1; /* mark request as delayed */ 698 req->fd = 1; /* mark request as delayed */
539 continue; 699 continue;
540 } 700 }
541 else 701 else
… …
568 728
569 max = 0; 729 max = 0;
570 } 730 }
571 731
572 return count; 732 return count;
573}
574
575static void *aio_proc(void *arg);
576
577static void start_thread (void)
578{
579 sigset_t fullsigset, oldsigset;
580 pthread_attr_t attr;
581
582 worker *wrk = calloc (1, sizeof (worker));
583
584 if (!wrk)
585 croak ("unable to allocate worker thread data");
586
587 pthread_attr_init (&attr);
588 pthread_attr_setstacksize (&attr, STACKSIZE);
589 pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
590
591 sigfillset (&fullsigset);
592
593 LOCK (wrklock);
594 sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
595
596 if (pthread_create (&wrk->tid, &attr, aio_proc, (void *)wrk) == 0)
597 {
598 wrk->prev = &wrk_first;
599 wrk->next = wrk_first.next;
600 wrk_first.next->prev = wrk;
601 wrk_first.next = wrk;
602 ++started;
603 }
604 else
605 free (wrk);
606
607 sigprocmask (SIG_SETMASK, &oldsigset, 0);
608 UNLOCK (wrklock);
609}
610
611static void req_send (aio_req req)
612{
613 while (started < wanted && nreqs >= started)
614 start_thread ();
615
616 ++nreqs;
617
618 LOCK (reqlock);
619 reqq_push (&req_queue, req);
620 pthread_cond_signal (&reqwait);
621 UNLOCK (reqlock);
622}
623
624static void end_thread (void)
625{
626 aio_req req;
627
628 Newz (0, req, 1, aio_cb);
629
630 req->type = REQ_QUIT;
631 req->pri = PRI_MAX + PRI_BIAS;
632
633 req_send (req);
634}
635
636static void min_parallel (int nthreads)
637{
638 if (wanted < nthreads)
639 wanted = nthreads;
640}
641
642static void max_parallel (int nthreads)
643{
644 int cur = started;
645
646 if (wanted > nthreads)
647 wanted = nthreads;
648
649 while (cur > wanted)
650 {
651 end_thread ();
652 cur--;
653 }
654
655 while (started > wanted)
656 {
657 poll_wait ();
658 poll_cb (0);
659 }
660} 733}
661 734
662static void create_pipe () 735static void create_pipe ()
663{ 736{
664 if (pipe (respipe)) 737 if (pipe (respipe))
… …
928/*****************************************************************************/ 1001/*****************************************************************************/
929 1002
930static void *aio_proc (void *thr_arg) 1003static void *aio_proc (void *thr_arg)
931{ 1004{
932 aio_req req; 1005 aio_req req;
933 int type;
934 worker *self = (worker *)thr_arg; 1006 worker *self = (worker *)thr_arg;
935 1007
936 do 1008 for (;;)
937 { 1009 {
938 LOCK (reqlock); 1010 LOCK (reqlock);
939 1011
940 for (;;) 1012 for (;;)
941 { 1013 {
… …
945 break; 1017 break;
946 1018
947 pthread_cond_wait (&reqwait, &reqlock); 1019 pthread_cond_wait (&reqwait, &reqlock);
948 } 1020 }
949 1021
1022 --nready;
1023
950 UNLOCK (reqlock); 1024 UNLOCK (reqlock);
951 1025
952 errno = 0; /* strictly unnecessary */ 1026 errno = 0; /* strictly unnecessary */
953 type = req->type; /* remember type for QUIT check */
954 1027
955 if (!(req->flags & FLAG_CANCELLED)) 1028 if (!(req->flags & FLAG_CANCELLED))
956 switch (type) 1029 switch (req->type)
957 { 1030 {
958 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 1031 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
959 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 1032 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
960 1033
961 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 1034 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
… …
987 req->result = select (0, 0, 0, 0, &tv); 1060 req->result = select (0, 0, 0, 0, &tv);
988 } 1061 }
989 1062
990 case REQ_GROUP: 1063 case REQ_GROUP:
991 case REQ_NOP: 1064 case REQ_NOP:
1065 break;
1066
992 case REQ_QUIT: 1067 case REQ_QUIT:
1068 LOCK (wrklock);
1069 worker_free (self);
1070 --started;
1071 UNLOCK (wrklock);
993 break; 1072 return 0;
994 1073
995 default: 1074 default:
996 req->result = ENOSYS; 1075 req->result = ENOSYS;
997 break; 1076 break;
998 } 1077 }
999 1078
1000 req->errorno = errno; 1079 req->errorno = errno;
1001 1080
1002 LOCK (reslock); 1081 LOCK (reslock);
1003 1082
1083 ++npending;
1084
1004 if (!reqq_push (&res_queue, req)) 1085 if (!reqq_push (&res_queue, req))
1005 /* write a dummy byte to the pipe so fh becomes ready */ 1086 /* write a dummy byte to the pipe so fh becomes ready */
1006 write (respipe [1], &respipe, 1); 1087 write (respipe [1], &respipe, 1);
1007 1088
1008 self->req = 0; 1089 self->req = 0;
1009 worker_clear (self); 1090 worker_clear (self);
1010 1091
1011 UNLOCK (reslock); 1092 UNLOCK (reslock);
1012 } 1093 }
1013 while (type != REQ_QUIT);
1014
1015 LOCK (wrklock);
1016 worker_free (self);
1017 UNLOCK (wrklock);
1018
1019 return 0;
1020} 1094}
1021 1095
1022/*****************************************************************************/ 1096/*****************************************************************************/
1023 1097
1024static void atfork_prepare (void) 1098static void atfork_prepare (void)
… …
1110 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY)); 1184 newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY));
1111 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY)); 1185 newCONSTSUB (stash, "O_WRONLY", newSViv (O_WRONLY));
1112 1186
1113 create_pipe (); 1187 create_pipe ();
1114 pthread_atfork (atfork_prepare, atfork_parent, atfork_child); 1188 pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
1189
1190 start_thread ();
1115} 1191}
1116 1192
1117void 1193void
1118min_parallel (int nthreads) 1194min_parallel (int nthreads)
1119 PROTOTYPE: $ 1195 PROTOTYPE: $
… …
1392 req->type = REQ_NOP; 1468 req->type = REQ_NOP;
1393 1469
1394 REQ_SEND; 1470 REQ_SEND;
1395} 1471}
1396 1472
1397void 1473int
1398aioreq_pri (int pri = DEFAULT_PRI) 1474aioreq_pri (int pri = 0)
1399 CODE: 1475 PROTOTYPE: ;$
1476 CODE:
1477 RETVAL = next_pri - PRI_BIAS;
1478 if (items > 0)
1479 {
1400 if (pri < PRI_MIN) pri = PRI_MIN; 1480 if (pri < PRI_MIN) pri = PRI_MIN;
1401 if (pri > PRI_MAX) pri = PRI_MAX; 1481 if (pri > PRI_MAX) pri = PRI_MAX;
1402 next_pri = pri + PRI_BIAS; 1482 next_pri = pri + PRI_BIAS;
1483 }
1484 OUTPUT:
1485 RETVAL
1403 1486
1404void 1487void
1405aioreq_nice (int nice = 0) 1488aioreq_nice (int nice = 0)
1406 CODE: 1489 CODE:
1407 nice = next_pri - nice; 1490 nice = next_pri - nice;
1408 if (nice < PRI_MIN) nice = PRI_MIN; 1491 if (nice < PRI_MIN) nice = PRI_MIN;
1409 if (nice > PRI_MAX) nice = PRI_MAX; 1492 if (nice > PRI_MAX) nice = PRI_MAX;
1410 next_pri = nice + PRI_BIAS; 1493 next_pri = nice + PRI_BIAS;
1411 1494
1412void 1495void
1413flush () 1496flush ()
1414 PROTOTYPE: 1497 PROTOTYPE:
1415 CODE: 1498 CODE:
… …
1466 CODE: 1549 CODE:
1467 RETVAL = nreqs; 1550 RETVAL = nreqs;
1468 OUTPUT: 1551 OUTPUT:
1469 RETVAL 1552 RETVAL
1470 1553
1554int
1555nready()
1556 PROTOTYPE:
1557 CODE:
1558 RETVAL = get_nready ();
1559 OUTPUT:
1560 RETVAL
1561
1562int
1563npending()
1564 PROTOTYPE:
1565 CODE:
1566 RETVAL = get_npending ();
1567 OUTPUT:
1568 RETVAL
1569
1471PROTOTYPES: DISABLE 1570PROTOTYPES: DISABLE
1472 1571
1473MODULE = IO::AIO PACKAGE = IO::AIO::REQ 1572MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1474 1573
1475void 1574void
… …
1526void 1625void
1527result (aio_req grp, ...) 1626result (aio_req grp, ...)
1528 CODE: 1627 CODE:
1529{ 1628{
1530 int i; 1629 int i;
1630 AV *av;
1631
1632 grp->errorno = errno;
1633
1531 AV *av = newAV (); 1634 av = newAV ();
1532 1635
1533 for (i = 1; i < items; ++i ) 1636 for (i = 1; i < items; ++i )
1534 av_push (av, newSVsv (ST (i))); 1637 av_push (av, newSVsv (ST (i)));
1535 1638
1536 SvREFCNT_dec (grp->data); 1639 SvREFCNT_dec (grp->data);
1537 grp->data = (SV *)av; 1640 grp->data = (SV *)av;
1538} 1641}
1642
1643void
1644errno (aio_req grp, int errorno = errno)
1645 CODE:
1646 grp->errorno = errorno;
1539 1647
1540void 1648void
1541limit (aio_req grp, int limit) 1649limit (aio_req grp, int limit)
1542 CODE: 1650 CODE:
1543 grp->fd2 = limit; 1651 grp->fd2 = limit;

Diff Legend

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