--- IO-AIO/AIO.xs 2005/08/16 22:22:18 1.22 +++ IO-AIO/AIO.xs 2005/08/22 23:20:37 1.32 @@ -7,6 +7,8 @@ #include "autoconf/config.h" +#include + #include #include @@ -15,11 +17,16 @@ #include #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 HAVE_SENDFILE +# if __linux +# include +# elif __freebsd +# include +# include +# elif __hpux +# include +# endif +#endif #if __ia64 # define STACKSIZE 65536 @@ -31,9 +38,10 @@ REQ_QUIT, REQ_OPEN, REQ_CLOSE, REQ_READ, REQ_WRITE, REQ_READAHEAD, + REQ_SENDFILE, REQ_STAT, REQ_LSTAT, REQ_FSTAT, REQ_FSYNC, REQ_FDATASYNC, - REQ_UNLINK, REQ_RMDIR, REQ_SYMLINK, + REQ_UNLINK, REQ_RMDIR, REQ_SYMLINK, }; @@ -42,13 +50,14 @@ int type; - int fd; + int fd, fd2; off_t offset; size_t length; ssize_t result; mode_t mode; /* open */ int errorno; - SV *data, *callback, *fh; + SV *data, *callback; + SV *fh, *fh2; void *dataptr, *data2ptr; STRLEN dataoffset; @@ -57,19 +66,38 @@ typedef aio_cb *aio_req; -static int started; +static int started, wanted; 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_mutex_t frklock = 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 free_req (aio_req req) +{ + if (req->data) + SvREFCNT_dec (req->data); + + if (req->fh) + SvREFCNT_dec (req->fh); + + if (req->fh2) + SvREFCNT_dec (req->fh2); + + if (req->statdata) + Safefree (req->statdata); + + if (req->callback) + SvREFCNT_dec (req->callback); + + Safefree (req); +} + static void poll_wait () { @@ -88,24 +116,34 @@ { dSP; int count = 0; - aio_req req, prv; + int do_croak = 0; + aio_req req; - pthread_mutex_lock (&reslock); + for (;;) + { + pthread_mutex_lock (&reslock); + req = ress; - { - /* read any signals sent by the worker threads */ - char buf [32]; - while (read (respipe [0], buf, 32) == 32) - ; - } + if (req) + { + ress = req->next; - req = ress; - ress = rese = 0; + if (!ress) + { + /* read any signals sent by the worker threads */ + char buf [32]; + while (read (respipe [0], buf, 32) == 32) + ; - pthread_mutex_unlock (&reslock); + rese = 0; + } + } + + pthread_mutex_unlock (&reslock); + + if (!req) + break; - while (req) - { nreqs--; if (req->type == REQ_QUIT) @@ -116,22 +154,16 @@ 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); + SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0)); - if (req->fh) - SvREFCNT_dec (req->fh); + if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE)) + SvREADONLY_off (req->data); - if (req->type == REQ_STAT || req->type == REQ_LSTAT || req->type == REQ_FSTAT) + if (req->statdata) { PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT; PL_laststatval = req->result; PL_statcache = *(req->statdata); - - Safefree (req->statdata); } ENTER; @@ -158,22 +190,21 @@ PUTBACK; call_sv (req->callback, G_VOID | G_EVAL); SPAGAIN; + + if (SvTRUE (ERRSV)) + { + free_req (req); + croak (0); + } } LEAVE; - - if (req->callback) - SvREFCNT_dec (req->callback); errno = errorno; count++; } - prv = req; - req = req->next; - Safefree (prv); - - /* TODO: croak on errors? */ + free_req (req); } return count; @@ -204,6 +235,9 @@ static void send_req (aio_req req) { + while (started < wanted && nreqs >= started) + start_thread (); + nreqs++; pthread_mutex_lock (&reqlock); @@ -221,92 +255,105 @@ pthread_cond_signal (&reqwait); pthread_mutex_unlock (&reqlock); - while (nreqs > max_outstanding) - { - poll_wait (); - poll_cb (); - } + if (nreqs > max_outstanding) + for (;;) + { + poll_cb (); + + if (nreqs <= max_outstanding) + break; + + poll_wait (); + } } static void end_thread (void) { aio_req req; - New (0, req, 1, aio_cb); + Newz (0, req, 1, aio_cb); req->type = REQ_QUIT; send_req (req); } - static void min_parallel (int nthreads) { - while (nthreads > started) - start_thread (); + if (wanted < nthreads) + wanted = nthreads; } static void max_parallel (int nthreads) { int cur = started; - while (cur > nthreads) - { + + if (wanted > nthreads) + wanted = nthreads; + + while (cur > wanted) + { end_thread (); cur--; } - while (started > nthreads) + while (started > wanted) { poll_wait (); poll_cb (); } } -static int fork_started; - -static void atfork_prepare (void) +static void create_pipe () { - pthread_mutex_lock (&frklock); - - fork_started = started; - - for (;;) { - while (nreqs) - { - poll_wait (); - poll_cb (); - } + if (pipe (respipe)) + croak ("unable to initialize result pipe"); - max_parallel (0); - - pthread_mutex_lock (&reqlock); + if (fcntl (respipe [0], F_SETFL, O_NONBLOCK)) + croak ("cannot set result pipe to nonblocking mode"); - if (!nreqs && !started) - break; - - pthread_mutex_unlock (&reqlock); - - min_parallel (fork_started); - } + if (fcntl (respipe [1], F_SETFL, O_NONBLOCK)) + croak ("cannot set result pipe to nonblocking mode"); +} +static void atfork_prepare (void) +{ + pthread_mutex_lock (&reqlock); pthread_mutex_lock (&reslock); - - assert (!started); - assert (!nreqs); - assert (!reqs && !reqe); - assert (!ress && !rese); } static void atfork_parent (void) { pthread_mutex_unlock (&reslock); - min_parallel (fork_started); pthread_mutex_unlock (&reqlock); - pthread_mutex_unlock (&frklock); } static void atfork_child (void) { + aio_req prv; + + started = 0; + + while (reqs) + { + prv = reqs; + reqs = prv->next; + free_req (prv); + } + reqs = reqe = 0; + + while (ress) + { + prv = ress; + ress = prv->next; + free_req (prv); + } + + ress = rese = 0; + + close (respipe [0]); + close (respipe [1]); + create_pipe (); atfork_parent (); } @@ -383,6 +430,79 @@ } #endif +/* sendfile always needs emulation */ +static ssize_t +sendfile_ (int ofd, int ifd, off_t offset, size_t count) +{ + ssize_t res; + + if (!count) + return 0; + +#if __linux + res = sendfile (ofd, ifd, &offset, count); + +#elif __freebsd + /* + * Of course, the freebsd sendfile is a dire hack with no thoughts + * wasted on making it similar to other i/o functions. + */ + { + off_t sbytes; + res = sendfile (ifd, ofd, offset, count, 0, &sbytes, 0); + + if (!res && errno == EAGAIN) + res = sbytes; + } + +#elif __hpux + res = sendfile (ofd, ifd, offset, count, 0, 0); + +#else + res = -1; + errno = ENOSYS; +#endif + + if (res < 0 && (errno == ENOSYS || errno == EINVAL || errno == ENOTSOCK)) + { + /* emulate sendfile. this is a major pain in the ass */ + char *buf = malloc (4096); + res = 0; + + for (;;) + { + ssize_t cnt; + + cnt = pread (ifd, buf, 4096, offset); + + if (cnt <= 0) + { + if (cnt && !res) res = -1; + break; + } + + cnt = write (ofd, buf, cnt); + + if (cnt <= 0) + { + if (cnt && !res) res = -1; + break; + } + + offset += cnt; + res += cnt; + } + + { + int errorno = errno; + free (buf); + errno = errorno; + } + } + + return res; +} + /*****************************************************************************/ static void * @@ -423,6 +543,7 @@ case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; + case REQ_SENDFILE: req->result = sendfile_ (req->fd, req->fd2, req->offset, req->length); break; case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break; case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break; @@ -481,7 +602,7 @@ if (!req) \ croak ("out of memory during aio_req allocation"); \ \ - req->callback = SvREFCNT_inc (callback); + req->callback = newSVsv (callback); MODULE = IO::AIO PACKAGE = IO::AIO @@ -489,15 +610,7 @@ BOOT: { - if (pipe (respipe)) - croak ("unable to initialize result pipe"); - - if (fcntl (respipe [0], F_SETFL, O_NONBLOCK)) - croak ("cannot set result pipe to nonblocking mode"); - - if (fcntl (respipe [1], F_SETFL, O_NONBLOCK)) - croak ("cannot set result pipe to nonblocking mode"); - + create_pipe (); pthread_atfork (atfork_prepare, atfork_parent, atfork_child); } @@ -563,9 +676,9 @@ aio_read(fh,offset,length,data,dataoffset,callback=&PL_sv_undef) SV * fh UV offset - IV length + UV length SV * data - IV dataoffset + UV dataoffset SV * callback ALIAS: aio_read = REQ_READ @@ -612,13 +725,41 @@ req->length = length; req->data = SvREFCNT_inc (data); req->dataptr = (char *)svptr + dataoffset; - req->callback = SvREFCNT_inc (callback); + + if (!SvREADONLY (data)) + { + SvREADONLY_on (data); + req->data2ptr = (void *)data; + } send_req (req); } } void +aio_sendfile(out_fh,in_fh,in_offset,length,callback=&PL_sv_undef) + SV * out_fh + SV * in_fh + UV in_offset + UV length + SV * callback + PROTOTYPE: $$$$;$ + CODE: +{ + dREQ; + + req->type = REQ_SENDFILE; + req->fh = newSVsv (out_fh); + req->fd = PerlIO_fileno (IoIFP (sv_2io (out_fh))); + req->fh2 = newSVsv (in_fh); + req->fd2 = PerlIO_fileno (IoIFP (sv_2io (in_fh))); + req->offset = in_offset; + req->length = length; + + send_req (req); +} + +void aio_readahead(fh,offset,length,callback=&PL_sv_undef) SV * fh UV offset @@ -651,7 +792,10 @@ New (0, req->statdata, 1, Stat_t); if (!req->statdata) - croak ("out of memory during aio_req->statdata allocation (sorry, i just leaked memory, too)"); + { + free_req (req); + croak ("out of memory during aio_req->statdata allocation"); + } if (SvPOK (fh_or_path)) {