--- IO-AIO/AIO.xs 2005/08/17 03:16:56 1.25 +++ IO-AIO/AIO.xs 2005/08/28 10:51:33 1.38 @@ -7,33 +7,47 @@ #include "autoconf/config.h" +#include + +#include #include #include - +#include #include #include #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 +# elif __solaris /* not yet */ +# include +# else +# error sendfile support requested but not available +# endif +#endif #if __ia64 # define STACKSIZE 65536 #else -# define STACKSIZE 4096 +# define STACKSIZE 8192 #endif enum { 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_READDIR, REQ_SYMLINK, }; @@ -42,13 +56,15 @@ int type; - int fd; + /* should receive a cleanup, with unions */ + 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 +73,41 @@ 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); + + if (req->type == REQ_READDIR && req->result >= 0) + free (req->data2ptr); + + Safefree (req); +} + static void poll_wait () { @@ -89,24 +127,33 @@ dSP; int count = 0; int do_croak = 0; - aio_req req, prv; + 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) @@ -117,41 +164,62 @@ errno = req->errorno; if (req->type == REQ_READ) - SvCUR_set (req->data, req->dataoffset - + req->result > 0 ? req->result : 0); + SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0)); - if (req->data) - SvREFCNT_dec (req->data); + if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE)) + SvREADONLY_off (req->data); - if (req->fh) - SvREFCNT_dec (req->fh); - - 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; PUSHMARK (SP); - XPUSHs (sv_2mortal (newSViv (req->result))); - if (req->type == REQ_OPEN) + if (req->type == REQ_READDIR) { - /* convert fd to fh */ - SV *fh; + SV *rv = &PL_sv_undef; - PUTBACK; - call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL); - SPAGAIN; + if (req->result >= 0) + { + char *buf = req->data2ptr; + AV *av = newAV (); + + while (req->result) + { + SV *sv = newSVpv (buf, 0); + + av_push (av, sv); + buf += SvCUR (sv) + 1; + req->result--; + } - fh = SvREFCNT_inc (POPs); + rv = sv_2mortal (newRV_noinc ((SV *)av)); + } - PUSHMARK (SP); - XPUSHs (sv_2mortal (fh)); + XPUSHs (rv); + } + else + { + 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 = SvREFCNT_inc (POPs); + + PUSHMARK (SP); + XPUSHs (sv_2mortal (fh)); + } } if (SvOK (req->callback)) @@ -159,25 +227,21 @@ PUTBACK; call_sv (req->callback, G_VOID | G_EVAL); SPAGAIN; - } - do_croak = SvTRUE (ERRSV); + 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); - - if (do_croak) - croak (0); + free_req (req); } return count; @@ -208,6 +272,9 @@ static void send_req (aio_req req) { + while (started < wanted && nreqs >= started) + start_thread (); + nreqs++; pthread_mutex_lock (&reqlock); @@ -225,93 +292,66 @@ 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 () { - int nstarted; - - for (;;) { - while (nreqs) - { - poll_wait (); - poll_cb (); - } + if (pipe (respipe)) + croak ("unable to initialize result pipe"); - nstarted = started; - max_parallel (0); - - pthread_mutex_lock (&reqlock); - - if (!nreqs && !started) - break; - - pthread_mutex_unlock (&reqlock); - - min_parallel (fork_started); - } - - pthread_mutex_lock (&frklock); - fork_started = nstarted; - pthread_mutex_lock (&reslock); + if (fcntl (respipe [0], F_SETFL, O_NONBLOCK)) + croak ("cannot set result pipe to nonblocking mode"); - assert (!started); - assert (!nreqs); - assert (!reqs && !reqe); - assert (!ress && !rese); + if (fcntl (respipe [1], F_SETFL, O_NONBLOCK)) + croak ("cannot set result pipe to nonblocking mode"); } -static void atfork_parent (void) -{ - pthread_mutex_unlock (&reslock); - pthread_mutex_unlock (&frklock); - pthread_mutex_unlock (&reqlock); - - min_parallel (fork_started); -} - -#define atfork_child atfork_parent - /*****************************************************************************/ /* work around various missing functions */ @@ -324,7 +364,7 @@ * normal read/write by using a mutex. slows down execution a lot, * but that's your problem, not mine. */ -static pthread_mutex_t iolock = PTHREAD_MUTEX_INITIALIZER; +static pthread_mutex_t preadwritelock = PTHREAD_MUTEX_INITIALIZER; static ssize_t pread (int fd, void *buf, size_t count, off_t offset) @@ -332,12 +372,12 @@ ssize_t res; off_t ooffset; - pthread_mutex_lock (&iolock); + pthread_mutex_lock (&preadwritelock); ooffset = lseek (fd, 0, SEEK_CUR); lseek (fd, offset, SEEK_SET); res = read (fd, buf, count); lseek (fd, ooffset, SEEK_SET); - pthread_mutex_unlock (&iolock); + pthread_mutex_unlock (&preadwritelock); return res; } @@ -348,12 +388,12 @@ ssize_t res; off_t ooffset; - pthread_mutex_lock (&iolock); + pthread_mutex_lock (&preadwritelock); ooffset = lseek (fd, 0, SEEK_CUR); lseek (fd, offset, SEEK_SET); res = write (fd, buf, count); lseek (fd, offset, SEEK_SET); - pthread_mutex_unlock (&iolock); + pthread_mutex_unlock (&preadwritelock); return res; } @@ -366,11 +406,11 @@ #if !HAVE_READAHEAD # define readahead aio_readahead -static char readahead_buf[4096]; - static ssize_t readahead (int fd, off_t offset, size_t count) { + char readahead_buf[4096]; + while (count > 0) { size_t len = count < sizeof (readahead_buf) ? count : sizeof (readahead_buf); @@ -384,6 +424,194 @@ } #endif +#if !HAVE_READDIR_R +# define readdir_r aio_readdir_r + +static pthread_mutex_t readdirlock = PTHREAD_MUTEX_INITIALIZER; + +static int +readdir_r (DIR *dirp, struct dirent *ent, struct dirent **res) +{ + struct dirent *e; + int errorno; + + pthread_mutex_lock (&readdirlock); + + e = readdir (dirp); + errorno = errno; + + if (e) + { + *res = ent; + strcpy (ent->d_name, e->d_name); + } + else + *res = 0; + + pthread_mutex_unlock (&readdirlock); + + errno = errorno; + return e ? 0 : -1; +} +#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 HAVE_SENDFILE +# 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 < 0 && sbytes) + /* maybe only on EAGAIN only: as usual, the manpage leaves you guessing */ + res = sbytes; + } + +# elif __hpux + res = sendfile (ofd, ifd, offset, count, 0, 0); + +# elif __solaris + { + struct sendfilevec vec; + size_t sbytes; + + vec.sfv_fd = ifd; + vec.sfv_flag = 0; + vec.sfv_off = offset; + vec.sfv_len = count; + + res = sendfilev (ofd, &vec, 1, &sbytes); + + if (res < 0 && sbytes) + res = sbytes; + } + +# endif +#else + res = -1; + errno = ENOSYS; +#endif + + if (res < 0 + && (errno == ENOSYS || errno == EINVAL || errno == ENOTSOCK +#if __solaris + || errno == EAFNOSUPPORT || errno == EPROTOTYPE +#endif + ) + ) + { + /* emulate sendfile. this is a major pain in the ass */ + char buf[4096]; + res = 0; + + while (count) + { + ssize_t cnt; + + cnt = pread (ifd, buf, count > 4096 ? 4096 : count, 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; + count -= cnt; + } + } + + return res; +} + +/* read a full directory */ +static int +scandir_ (const char *path, void **namesp) +{ + DIR *dirp = opendir (path); + union + { + struct dirent d; + char b [offsetof (struct dirent, d_name) + NAME_MAX + 1]; + } u; + struct dirent *entp; + char *name, *names; + int memlen = 4096; + int memofs = 0; + int res = 0; + int errorno; + + if (!dirp) + return -1; + + names = malloc (memlen); + + for (;;) + { + errno = 0, readdir_r (dirp, &u.d, &entp); + + if (!entp) + break; + + name = entp->d_name; + + if (name [0] != '.' || (name [1] && (name [1] != '.' || name [2]))) + { + int len = strlen (name) + 1; + + res++; + + while (memofs + len > memlen) + { + memlen *= 2; + names = realloc (names, memlen); + if (!names) + break; + } + + memcpy (names + memofs, name, len); + memofs += len; + } + } + + errorno = errno; + closedir (dirp); + + if (errorno) + { + free (names); + errno = errorno; + res = -1; + } + + *namesp = (void *)names; + return res; +} + /*****************************************************************************/ static void * @@ -424,6 +652,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; @@ -437,6 +666,7 @@ case REQ_FDATASYNC: req->result = fdatasync (req->fd); break; case REQ_FSYNC: req->result = fsync (req->fd); break; + case REQ_READDIR: req->result = scandir_ (req->dataptr, &req->data2ptr); break; case REQ_QUIT: break; @@ -472,6 +702,63 @@ return 0; } +/*****************************************************************************/ + +static void atfork_prepare (void) +{ + pthread_mutex_lock (&reqlock); + pthread_mutex_lock (&reslock); +#if !HAVE_PREADWRITE + pthread_mutex_lock (&preadwritelock); +#endif +#if !HAVE_READDIR_R + pthread_mutex_lock (&readdirlock); +#endif +} + +static void atfork_parent (void) +{ +#if !HAVE_READDIR_R + pthread_mutex_unlock (&readdirlock); +#endif +#if !HAVE_PREADWRITE + pthread_mutex_unlock (&preadwritelock); +#endif + pthread_mutex_unlock (&reslock); + pthread_mutex_unlock (&reqlock); +} + +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 (); +} + #define dREQ \ aio_req req; \ \ @@ -482,7 +769,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 @@ -490,15 +777,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); } @@ -564,9 +843,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 @@ -613,13 +892,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 @@ -652,7 +959,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)) { @@ -704,6 +1014,21 @@ req->dataptr = SvPVbyte_nolen (req->data); send_req (req); +} + +void +aio_readdir(pathname,callback=&PL_sv_undef) + SV * pathname + SV * callback + CODE: +{ + dREQ; + + req->type = REQ_READDIR; + req->data = newSVsv (pathname); + req->dataptr = SvPVbyte_nolen (req->data); + + send_req (req); } void