--- IO-AIO/AIO.xs 2005/07/10 20:57:00 1.4 +++ IO-AIO/AIO.xs 2005/07/20 21:57:04 1.12 @@ -1,4 +1,4 @@ -#define PERL_NO_GET_CONTEXT +#define _XOPEN_SOURCE 500 #include "EXTERN.h" #include "perl.h" @@ -6,14 +6,16 @@ #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 ;) */ @@ -54,7 +56,7 @@ typedef aio_cb *aio_req; static int started; -static int nreqs; +static volatile int nreqs; static int max_outstanding = 1<<30; static int respipe [2]; @@ -68,47 +70,39 @@ static void poll_wait () { - if (!nreqs) - return; - - fd_set rfd; - FD_ZERO(&rfd); - FD_SET(respipe [0], &rfd); + if (nreqs && !ress) + { + fd_set rfd; + FD_ZERO(&rfd); + FD_SET(respipe [0], &rfd); - select (respipe [0] + 1, &rfd, 0, 0, 0); + select (respipe [0] + 1, &rfd, 0, 0, 0); + } } static int -poll_cb (pTHX) +poll_cb () { dSP; int count = 0; - aio_req req; - + aio_req req, prv; + + pthread_mutex_lock (&reslock); + { - /* read and signals sent by the worker threads */ + /* read any signals sent by the worker threads */ char buf [32]; while (read (respipe [0], buf, 32) > 0) ; } - for (;;) - { - pthread_mutex_lock (&reslock); - - req = ress; - - if (ress) - { - ress = ress->next; - if (!ress) rese = 0; - } - - pthread_mutex_unlock (&reslock); + req = ress; + ress = rese = 0; - if (!req) - break; + pthread_mutex_unlock (&reslock); + while (req) + { nreqs--; if (req->type == REQ_QUIT) @@ -152,9 +146,12 @@ XPUSHs (fh); } - PUTBACK; - call_sv (req->callback, G_VOID | G_EVAL); - SPAGAIN; + if (SvOK (req->callback)) + { + PUTBACK; + call_sv (req->callback, G_VOID | G_EVAL); + SPAGAIN; + } if (req->callback) SvREFCNT_dec (req->callback); @@ -163,7 +160,11 @@ count++; } - Safefree (req); + prv = req; + req = req->next; + Safefree (prv); + + /* TODO: croak on errors? */ } return count; @@ -229,8 +230,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; @@ -311,8 +311,8 @@ 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 @@ -366,6 +366,8 @@ MODULE = IO::AIO PACKAGE = IO::AIO +PROTOTYPES: ENABLE + BOOT: { if (pipe (respipe)) @@ -402,7 +404,7 @@ while (started > nthreads) { poll_wait (); - poll_cb (aTHX); + poll_cb (); } } @@ -415,12 +417,12 @@ 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; @@ -441,10 +443,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 @@ -466,36 +468,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; @@ -518,12 +520,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; @@ -540,7 +542,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); } @@ -556,10 +558,9 @@ } void -aio_unlink(pathname,callback) +aio_unlink(pathname,callback=&PL_sv_undef) SV * pathname SV * callback - PROTOTYPE: $$ CODE: { aio_req req; @@ -577,6 +578,26 @@ 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: @@ -589,7 +610,7 @@ poll_cb(...) PROTOTYPE: CODE: - RETVAL = poll_cb (aTHX); + RETVAL = poll_cb (); OUTPUT: RETVAL