--- IO-AIO/AIO.xs 2005/08/17 03:52:20 1.26 +++ IO-AIO/AIO.xs 2005/08/23 01:16:50 1.35 @@ -7,6 +7,8 @@ #include "autoconf/config.h" +#include + #include #include @@ -15,11 +17,20 @@ #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 @@ -31,6 +42,7 @@ 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, @@ -42,13 +54,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,14 +70,13 @@ 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 */ @@ -78,7 +90,10 @@ if (req->fh) SvREFCNT_dec (req->fh); - if (req->type == REQ_STAT || req->type == REQ_LSTAT || req->type == REQ_FSTAT) + if (req->fh2) + SvREFCNT_dec (req->fh2); + + if (req->statdata) Safefree (req->statdata); if (req->callback) @@ -106,24 +121,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) @@ -134,10 +158,12 @@ 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->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; @@ -168,22 +194,21 @@ PUTBACK; call_sv (req->callback, G_VOID | G_EVAL); SPAGAIN; + + if (SvTRUE (ERRSV)) + { + free_req (req); + croak (0); + } } LEAVE; - do_croak = SvTRUE (ERRSV); - errno = errorno; count++; } - prv = req; - req = req->next; - free_req (prv); - - if (do_croak) - croak (0); + free_req (req); } return count; @@ -214,6 +239,9 @@ static void send_req (aio_req req) { + while (started < wanted && nreqs >= started) + start_thread (); + nreqs++; pthread_mutex_lock (&reqlock); @@ -231,11 +259,16 @@ 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 @@ -248,23 +281,26 @@ 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 (); @@ -285,7 +321,6 @@ static void atfork_prepare (void) { - pthread_mutex_lock (&frklock); pthread_mutex_lock (&reqlock); pthread_mutex_lock (&reslock); } @@ -294,40 +329,43 @@ { pthread_mutex_unlock (&reslock); pthread_mutex_unlock (&reqlock); - pthread_mutex_unlock (&frklock); } static void atfork_child (void) { - int restart = started; + aio_req prv; + started = 0; while (reqs) { - free_req (reqs); - reqs = reqs->next; + prv = reqs; + reqs = prv->next; + free_req (prv); } reqs = reqe = 0; - + while (ress) { - free_req (ress); - ress = ress->next; + prv = ress; + ress = prv->next; + free_req (prv); } - + ress = rese = 0; close (respipe [0]); close (respipe [1]); - create_pipe (); atfork_parent (); - - min_parallel (restart); } +/* currently noops */ +#define LOCK_FD(fd) do { } while (0) +#define UNLOCK_FD(fd) do { } while (0) + /*****************************************************************************/ /* work around various missing functions */ @@ -348,12 +386,14 @@ ssize_t res; off_t ooffset; - pthread_mutex_lock (&iolock); + LOCK_FD (fd); + pthread_mutex_lock (&iolock); /* replace by LOCK_FD and private buffer */ 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); + UNLOCK_FD (d); return res; } @@ -364,12 +404,14 @@ ssize_t res; off_t ooffset; - pthread_mutex_lock (&iolock); + LOCK_FD (fd); + pthread_mutex_lock (&iolock); /* replace by LOCK_FD and private buffer */ 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); + UNLOCK_FD (d); return res; } @@ -400,6 +442,108 @@ } #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; + + LOCK_FD (ofd); + +#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; + } + +# else + res = -1; + errno = ENOSYS; +# endif +#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 = 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; + } + } + + UNLOCK_FD (ofd); + + return res; +} + /*****************************************************************************/ static void * @@ -440,6 +584,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; @@ -498,7 +643,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 @@ -572,9 +717,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 @@ -621,13 +766,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 @@ -660,7 +833,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)) {