--- IO-AIO/AIO.xs 2005/07/10 18:16:49 1.2 +++ IO-AIO/AIO.xs 2005/07/20 21:55:27 1.11 @@ -1,4 +1,4 @@ -#define PERL_NO_GET_CONTEXT +#define _XOPEN_SOURCE 500 #include "EXTERN.h" #include "perl.h" @@ -6,25 +6,25 @@ #include #include + #include #include #include #include -#include +#if __linux +#include +#endif #include -#include typedef void *InputStream; /* hack, but 5.6.1 is simply toooo old ;) */ 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 +36,7 @@ }; typedef struct aio_cb { - struct aio_cb *next; + struct aio_cb *volatile next; int type; @@ -56,10 +56,124 @@ typedef aio_cb *aio_req; static int started; -static int nreqs; -static int reqpipe[2], respipe[2]; +static volatile int nreqs; +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 && !ress) + { + 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, prv; + + static int rl;//D + //printf ("%d ENTER\n", ++rl);//D + + pthread_mutex_lock (&reslock); + + { + /* read any signals sent by the worker threads */ + char buf [32]; + while (read (respipe [0], buf, 32) > 0) + ; + } + + req = ress; + ress = rese = 0; + + pthread_mutex_unlock (&reslock); + + while (req) + { + nreqs--; + //printf ("%d count %d %p->%p\n", rl, count, req, req->next);//D + + 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); -static aio_req qs, qe; /* queue start, queue end */ + 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); + } + + if (SvOK (req->callback)) + { + PUTBACK; + call_sv (req->callback, G_VOID | G_EVAL); + SPAGAIN; + } + + if (req->callback) + SvREFCNT_dec (req->callback); + + errno = errorno; + count++; + } + + prv = req; + req = req->next; + Safefree (prv); + + /* TODO: croak on errors? */ + } + + //printf ("%d LEAVE %p %p\n", rl--, ress, rese);//D + return count; +} static void *aio_proc(void *arg); @@ -84,31 +198,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 +235,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,102 +282,42 @@ 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); - - 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))); + req = reqs; - if (req->type == REQ_OPEN) + if (reqs) { - /* convert fd to fh */ - SV *fh; - - PUTBACK; - call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL); - SPAGAIN; - - fh = POPs; - - PUSHMARK (SP); - XPUSHs (fh); + reqs = reqs->next; + if (!reqs) reqe = 0; } - PUTBACK; - call_sv (req->callback, G_VOID | G_EVAL); - 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; + case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; + case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; #if SYS_readahead case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; #else @@ -284,8 +336,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; @@ -293,23 +344,44 @@ } 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; } MODULE = IO::AIO PACKAGE = IO::AIO +PROTOTYPES: ENABLE + 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"); } @@ -337,17 +409,25 @@ 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) +aio_open(pathname,flags,mode,callback=&PL_sv_undef) SV * pathname int flags int mode SV * callback - PROTOTYPE: $$$$ + PROTOTYPE: $$$;$ CODE: { aio_req req; @@ -368,10 +448,10 @@ } void -aio_close(fh,callback) +aio_close(fh,callback=&PL_sv_undef) InputStream fh SV * callback - PROTOTYPE: $$ + PROTOTYPE: $;$ ALIAS: aio_close = REQ_CLOSE aio_fsync = REQ_FSYNC @@ -393,36 +473,36 @@ } void -aio_read(fh,offset,length,data,dataoffset,callback) +aio_read(fh,offset,length,data,dataoffset,callback=&PL_sv_undef) InputStream fh UV offset IV length SV * data IV dataoffset SV * callback - PROTOTYPE: $$$$$$ + 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) +aio_write(fh,offset,length,data,dataoffset,callback=&PL_sv_undef) OutputStream fh UV offset IV length SV * data IV dataoffset SV * callback - PROTOTYPE: $$$$$$ + 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) +aio_readahead(fh,offset,length,callback=&PL_sv_undef) InputStream fh UV offset IV length SV * callback - PROTOTYPE: $$$$ + PROTOTYPE: $$$;$ CODE: { aio_req req; @@ -445,12 +525,12 @@ } void -aio_stat(fh_or_path,callback) +aio_stat(fh_or_path,callback=&PL_sv_undef) SV * fh_or_path SV * callback - PROTOTYPE: $$ ALIAS: - aio_lstat = 1 + aio_stat = REQ_STAT + aio_lstat = REQ_LSTAT CODE: { aio_req req; @@ -467,7 +547,7 @@ if (SvPOK (fh_or_path)) { - req->type = ix ? REQ_LSTAT : REQ_STAT; + req->type = ix; req->data = newSVsv (fh_or_path); req->dataptr = SvPV_nolen (req->data); } @@ -483,10 +563,9 @@ } void -aio_unlink(pathname,callback) +aio_unlink(pathname,callback=&PL_sv_undef) SV * pathname SV * callback - PROTOTYPE: $$ CODE: { aio_req req; @@ -504,11 +583,31 @@ send_req (req); } +void +flush() + PROTOTYPE: + CODE: + while (nreqs) + { + poll_wait (); + poll_cb (); + } + +void +poll() + PROTOTYPE: + CODE: + if (nreqs) + { + poll_wait (); + poll_cb (); + } + int poll_fileno() PROTOTYPE: CODE: - RETVAL = respipe[0]; + RETVAL = respipe [0]; OUTPUT: RETVAL @@ -516,7 +615,7 @@ poll_cb(...) PROTOTYPE: CODE: - RETVAL = poll_cb (aTHX); + RETVAL = poll_cb (); OUTPUT: RETVAL @@ -524,7 +623,8 @@ poll_wait() PROTOTYPE: CODE: - poll_wait (); + if (nreqs) + poll_wait (); int nreqs()