--- IO-AIO/AIO.xs 2005/07/10 17:07:44 1.1 +++ IO-AIO/AIO.xs 2005/07/11 01:03:17 1.6 @@ -1,5 +1,3 @@ -#define PERL_NO_GET_CONTEXT - #include "EXTERN.h" #include "perl.h" #include "XSUB.h" @@ -19,12 +17,10 @@ typedef void *OutputStream; /* hack, but 5.6.1 is simply toooo old ;) */ typedef void *InOutStream; /* hack, but 5.6.1 is simply toooo old ;) */ -#if __i386 || __amd64 -# define STACKSIZE ( 256 * sizeof (long)) -#elif __ia64 -# define STACKSIZE (8192 * sizeof (long)) +#if __ia64 +# define STACKSIZE 65536 #else -# define STACKSIZE ( 512 * sizeof (long)) +# define STACKSIZE 4096 #endif enum { @@ -36,7 +32,7 @@ }; typedef struct aio_cb { - struct aio_cb *next; + struct aio_cb *volatile next; int type; @@ -57,9 +53,119 @@ static int started; static int nreqs; -static int reqpipe[2], respipe[2]; +static int max_outstanding = 1<<30; +static int respipe [2]; + +static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER; +static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; + +static volatile aio_req reqs, reqe; /* queue start, queue end */ +static volatile aio_req ress, rese; /* queue start, queue end */ + +static void +poll_wait () +{ + if (!nreqs) + return; + + fd_set rfd; + FD_ZERO(&rfd); + FD_SET(respipe [0], &rfd); + + select (respipe [0] + 1, &rfd, 0, 0, 0); +} + +static int +poll_cb () +{ + dSP; + int count = 0; + aio_req req; + + { + /* read and signals sent by the worker threads */ + char buf [32]; + while (read (respipe [0], buf, 32) > 0) + ; + } + + for (;;) + { + pthread_mutex_lock (&reslock); + + req = ress; -static aio_req qs, qe; /* queue start, queue end */ + if (ress) + { + ress = ress->next; + if (!ress) rese = 0; + } + + pthread_mutex_unlock (&reslock); + + if (!req) + break; + + nreqs--; + + if (req->type == REQ_QUIT) + started--; + else + { + int errorno = errno; + errno = req->errorno; + + if (req->type == REQ_READ) + SvCUR_set (req->data, req->dataoffset + + req->result > 0 ? req->result : 0); + + if (req->data) + SvREFCNT_dec (req->data); + + if (req->type == REQ_STAT || req->type == REQ_LSTAT || req->type == REQ_FSTAT) + { + PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT; + PL_laststatval = req->result; + PL_statcache = *(req->statdata); + + Safefree (req->statdata); + } + + PUSHMARK (SP); + XPUSHs (sv_2mortal (newSViv (req->result))); + + if (req->type == REQ_OPEN) + { + /* convert fd to fh */ + SV *fh; + + PUTBACK; + call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL); + SPAGAIN; + + fh = POPs; + + PUSHMARK (SP); + XPUSHs (fh); + } + + PUTBACK; + call_sv (req->callback, G_VOID | G_EVAL); + SPAGAIN; + + if (req->callback) + SvREFCNT_dec (req->callback); + + errno = errorno; + count++; + } + + Safefree (req); + } + + return count; +} static void *aio_proc(void *arg); @@ -84,31 +190,30 @@ } static void -send_reqs (void) -{ - /* this write is atomic */ - while (qs && write (reqpipe[1], &qs, sizeof qs) == sizeof qs) - { - qs = qs->next; - if (!qs) qe = 0; - } -} - -static void send_req (aio_req req) { nreqs++; + + pthread_mutex_lock (&reqlock); + req->next = 0; - if (qe) + if (reqe) { - qe->next = req; - qe = req; + reqe->next = req; + reqe = req; } else - qe = qs = req; + reqe = reqs = req; - send_reqs (); + pthread_cond_signal (&reqwait); + pthread_mutex_unlock (&reqlock); + + while (nreqs > max_outstanding) + { + poll_wait (); + poll_cb (); + } } static void @@ -122,8 +227,7 @@ } static void -read_write (pTHX_ - int dowrite, int fd, off_t offset, size_t length, +read_write (int dowrite, int fd, off_t offset, size_t length, SV *data, STRLEN dataoffset, SV *callback) { aio_req req; @@ -170,83 +274,39 @@ send_req (req); } -static void -poll_wait () -{ - fd_set rfd; - FD_ZERO(&rfd); - FD_SET(respipe[0], &rfd); - - select (respipe[0] + 1, &rfd, 0, 0, 0); -} - -static int -poll_cb (pTHX) +static void * +aio_proc (void *thr_arg) { - dSP; - int count = 0; aio_req req; + int type; - while (read (respipe[0], (void *)&req, sizeof (req)) == sizeof (req)) + do { - nreqs--; + pthread_mutex_lock (&reqlock); - if (req->type == REQ_QUIT) - started--; - else + for (;;) { - int errorno = errno; - errno = req->errorno; - - if (req->type == REQ_READ) - SvCUR_set (req->data, req->dataoffset - + req->result > 0 ? req->result : 0); + req = reqs; - if (req->data) - SvREFCNT_dec (req->data); - - if (req->type == REQ_STAT || req->type == REQ_LSTAT || req->type == REQ_FSTAT) + if (reqs) { - PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT; - PL_laststatval = req->result; - PL_statcache = *(req->statdata); - - Safefree (req->statdata); + reqs = reqs->next; + if (!reqs) reqe = 0; } - PUSHMARK (SP); - XPUSHs (sv_2mortal (newSViv (req->result))); - PUTBACK; - call_sv (req->callback, G_VOID); - SPAGAIN; - - if (req->callback) - SvREFCNT_dec (req->callback); + if (req) + break; - errno = errorno; - count++; + pthread_cond_wait (&reqwait, &reqlock); } - Safefree (req); - } - - if (qs) - send_reqs (); - - return count; -} - -static void * -aio_proc (void *thr_arg) -{ - aio_req req; - - /* then loop */ - while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req)) - { + pthread_mutex_unlock (&reqlock); + errno = 0; /* strictly unnecessary */ - switch (req->type) + type = req->type; + + switch (type) { case REQ_READ: req->result = pread64 (req->fd, req->dataptr, req->length, req->offset); break; case REQ_WRITE: req->result = pwrite64 (req->fd, req->dataptr, req->length, req->offset); break; @@ -268,8 +328,7 @@ case REQ_FDATASYNC: req->result = fdatasync (req->fd); break; case REQ_QUIT: - write (respipe[1], (void *)&req, sizeof (req)); - return 0; + break; default: req->result = ENOSYS; @@ -277,8 +336,27 @@ } req->errorno = errno; - write (respipe[1], (void *)&req, sizeof (req)); + + pthread_mutex_lock (&reslock); + + req->next = 0; + + if (rese) + { + rese->next = req; + rese = req; + } + else + { + rese = ress = req; + + /* write a dummy byte to the pipe so fh becomes ready */ + write (respipe [1], &respipe, 1); + } + + pthread_mutex_unlock (&reslock); } + while (type != REQ_QUIT); return 0; } @@ -287,13 +365,13 @@ BOOT: { - if (pipe (reqpipe) || pipe (respipe)) - croak ("unable to initialize request or result pipe"); + if (pipe (respipe)) + croak ("unable to initialize result pipe"); - if (fcntl (reqpipe[1], F_SETFL, O_NONBLOCK)) + if (fcntl (respipe [0], F_SETFL, O_NONBLOCK)) croak ("cannot set result pipe to nonblocking mode"); - if (fcntl (respipe[0], F_SETFL, O_NONBLOCK)) + if (fcntl (respipe [1], F_SETFL, O_NONBLOCK)) croak ("cannot set result pipe to nonblocking mode"); } @@ -321,10 +399,18 @@ while (started > nthreads) { poll_wait (); - poll_cb (aTHX); + poll_cb (); } } +int +max_outstanding(nreqs) + int nreqs + PROTOTYPE: $ + CODE: + RETVAL = max_outstanding; + max_outstanding = nreqs; + void aio_open(pathname,flags,mode,callback) SV * pathname @@ -386,7 +472,7 @@ SV * callback PROTOTYPE: $$$$$$ CODE: - read_write (aTHX_ 0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback); + read_write (0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback); void aio_write(fh,offset,length,data,dataoffset,callback) @@ -398,7 +484,7 @@ SV * callback PROTOTYPE: $$$$$$ CODE: - read_write (aTHX_ 1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback); + read_write (1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback); void aio_readahead(fh,offset,length,callback) @@ -488,11 +574,21 @@ send_req (req); } +void +flush() + PROTOTYPE: + CODE: + while (nreqs) + { + poll_wait (); + poll_cb (); + } + int poll_fileno() PROTOTYPE: CODE: - RETVAL = respipe[0]; + RETVAL = respipe [0]; OUTPUT: RETVAL @@ -500,7 +596,7 @@ poll_cb(...) PROTOTYPE: CODE: - RETVAL = poll_cb (aTHX); + RETVAL = poll_cb (); OUTPUT: RETVAL @@ -508,7 +604,8 @@ poll_wait() PROTOTYPE: CODE: - poll_wait (); + if (nreqs) + poll_wait (); int nreqs()