--- IO-AIO/AIO.xs 2007/07/08 11:05:36 1.104 +++ IO-AIO/AIO.xs 2008/04/16 16:45:18 1.113 @@ -99,6 +99,8 @@ # define SvVAL64 SvNV #endif +static HV *stash; + #define dBUF \ char *aio_buf; \ X_LOCK (wrklock); \ @@ -119,7 +121,7 @@ REQ_UTIME, REQ_FUTIME, REQ_CHMOD, REQ_FCHMOD, REQ_CHOWN, REQ_FCHOWN, - REQ_FSYNC, REQ_FDATASYNC, + REQ_SYNC, REQ_FSYNC, REQ_FDATASYNC, REQ_UNLINK, REQ_RMDIR, REQ_MKDIR, REQ_RENAME, REQ_MKNOD, REQ_READDIR, REQ_LINK, REQ_SYMLINK, REQ_READLINK, @@ -189,7 +191,7 @@ static int main_sig; static int block_sig_level; -void block_sig () +void block_sig (void) { sigset_t ss; @@ -204,7 +206,7 @@ pthread_sigmask (SIG_BLOCK, &ss, 0); } -void unblock_sig () +void unblock_sig (void) { sigset_t ss; @@ -266,7 +268,7 @@ static volatile unsigned int nreqs, nready, npending; static volatile unsigned int max_idle = 4; static volatile unsigned int max_outstanding = 0xffffffff; -static int respipe [2], respipe_osf [2]; +static int respipe_osf [2], respipe [2] = { -1, -1 }; static mutex_t reslock = X_MUTEX_INIT; static mutex_t reqlock = X_MUTEX_INIT; @@ -274,7 +276,7 @@ #if WORDACCESS_UNSAFE -static unsigned int get_nready () +static unsigned int get_nready (void) { unsigned int retval; @@ -285,7 +287,7 @@ return retval; } -static unsigned int get_npending () +static unsigned int get_npending (void) { unsigned int retval; @@ -296,7 +298,7 @@ return retval; } -static unsigned int get_nthreads () +static unsigned int get_nthreads (void) { unsigned int retval; @@ -369,7 +371,7 @@ abort (); } -static int poll_cb (); +static int poll_cb (void); static int req_invoke (aio_req req); static void req_destroy (aio_req req); static void req_cancel (aio_req req); @@ -503,16 +505,31 @@ case REQ_OPEN: { /* convert fd to fh */ - SV *fh; + SV *fh = &PL_sv_undef; + + if (req->result >= 0) + { + GV *gv = (GV *)sv_newmortal (); + int flags = req->int1 & (O_RDONLY | O_WRONLY | O_RDWR); + char sym [64]; + int symlen; + + symlen = snprintf (sym, sizeof (sym), "fd#%d", req->result); + gv_init (gv, stash, sym, symlen, 0); + + symlen = snprintf ( + sym, + sizeof (sym), + "%s&=%d", + flags == O_RDONLY ? "<" : flags == O_WRONLY ? ">" : "+<", + req->result + ); - PUSHs (sv_2mortal (newSViv (req->result))); - PUTBACK; - call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL); - SPAGAIN; - - fh = POPs; - PUSHMARK (SP); - XPUSHs (fh); + if (do_open (gv, sym, symlen, 0, 0, 0, 0)) + fh = (SV *)gv; + } + + PUSHs (fh); } break; @@ -566,11 +583,13 @@ errno = req->errorno; PUTBACK; - call_sv (req->callback, G_VOID | G_EVAL); + call_sv (req->callback, G_VOID | G_EVAL | G_DISCARD); SPAGAIN; FREETMPS; LEAVE; + + PUTBACK; } if (req->grp) @@ -636,20 +655,39 @@ #endif static void -create_pipe (int fd[2]) +create_respipe (void) { + int old_readfd = respipe [0]; + + if (respipe [1] >= 0) + respipe_close (TO_SOCKET (respipe [1])); + #ifdef _WIN32 - int arg = 1; - if (PerlSock_socketpair (AF_UNIX, SOCK_STREAM, 0, fd) - || ioctlsocket (TO_SOCKET (fd [0]), FIONBIO, &arg) - || ioctlsocket (TO_SOCKET (fd [1]), FIONBIO, &arg)) + if (PerlSock_socketpair (AF_UNIX, SOCK_STREAM, 0, respipe)) #else - if (pipe (fd) - || fcntl (fd [0], F_SETFL, O_NONBLOCK) - || fcntl (fd [1], F_SETFL, O_NONBLOCK)) + if (pipe (respipe)) #endif croak ("unable to initialize result pipe"); + if (old_readfd >= 0) + { + if (dup2 (TO_SOCKET (respipe [0]), TO_SOCKET (old_readfd)) < 0) + croak ("unable to initialize result pipe(2)"); + + respipe_close (respipe [0]); + respipe [0] = old_readfd; + } + +#ifdef _WIN32 + int arg = 1; + if (ioctlsocket (TO_SOCKET (respipe [0]), FIONBIO, &arg) + || ioctlsocket (TO_SOCKET (respipe [1]), FIONBIO, &arg)) +#else + if (fcntl (respipe [0], F_SETFL, O_NONBLOCK) + || fcntl (respipe [1], F_SETFL, O_NONBLOCK)) +#endif + croak ("unable to initialize result pipe(3)"); + respipe_osf [0] = TO_SOCKET (respipe [0]); respipe_osf [1] = TO_SOCKET (respipe [1]); } @@ -679,7 +717,7 @@ X_UNLOCK (wrklock); } -static void maybe_start_thread () +static void maybe_start_thread (void) { if (get_nthreads () >= wanted) return; @@ -749,7 +787,7 @@ end_thread (); } -static void poll_wait () +static void poll_wait (void) { fd_set rfd; @@ -772,7 +810,7 @@ } } -static int poll_cb () +static int poll_cb (void) { dSP; int count = 0; @@ -803,7 +841,7 @@ { /* read any signals sent by the worker threads */ char buf [4]; - while (PerlSock_recv (respipe [0], buf, 4, 0) == 4) + while (respipe_read (respipe [0], buf, 4) == 4) ; } } @@ -1142,11 +1180,37 @@ req->result = res; } +static int +aio_close (int fd) +{ + static int close_pipe = -1; /* dummy fd to close fds via dup2 */ + + X_LOCK (wrklock); + + if (close_pipe < 0) + { + int pipefd [2]; + + if (pipe (pipefd) < 0 + || close (pipefd [1]) < 0 + || fcntl (pipefd [0], F_SETFD, FD_CLOEXEC) < 0) + { + X_UNLOCK (wrklock); + return -1; + } + + close_pipe = pipefd [0]; + } + + X_UNLOCK (wrklock); + + return dup2 (close_pipe, fd) < 0 ? -1 : 0; +} + /*****************************************************************************/ X_THREAD_PROC (aio_proc) { - {//D aio_req req; struct timespec ts; worker *self = (worker *)thr_arg; @@ -1221,7 +1285,7 @@ case REQ_FTRUNCATE: req->result = ftruncate (req->int1, req->offs); break; case REQ_OPEN: req->result = open (req->ptr1, req->int1, req->mode); break; - case REQ_CLOSE: req->result = close (req->int1); break; + case REQ_CLOSE: req->result = aio_close (req->int1); break; case REQ_UNLINK: req->result = unlink (req->ptr1); break; case REQ_RMDIR: req->result = rmdir (req->ptr1); break; case REQ_MKDIR: req->result = mkdir (req->ptr1, req->mode); break; @@ -1231,8 +1295,10 @@ case REQ_MKNOD: req->result = mknod (req->ptr2, req->mode, (dev_t)req->offs); break; case REQ_READLINK: req->result = readlink (req->ptr2, req->ptr1, NAME_MAX); break; - case REQ_FDATASYNC: req->result = fdatasync (req->int1); break; + case REQ_SYNC: req->result = 0; sync (); break; case REQ_FSYNC: req->result = fsync (req->int1); break; + case REQ_FDATASYNC: req->result = fdatasync (req->int1); break; + case REQ_READDIR: scandir_ (req, self); break; case REQ_BUSY: @@ -1295,7 +1361,7 @@ if (!reqq_push (&res_queue, req)) { /* write a dummy byte to the pipe so fh becomes ready */ - send (respipe_osf [1], (const void *)&respipe_osf, 1, 0); + respipe_write (respipe_osf [1], (const void *)&respipe_osf, 1); /* optionally signal the main thread asynchronously */ if (main_sig) @@ -1314,7 +1380,6 @@ X_UNLOCK (wrklock); return 0; - }//D } /*****************************************************************************/ @@ -1372,10 +1437,7 @@ nready = 0; npending = 0; - PerlSock_closesocket (respipe [0]); - PerlSock_closesocket (respipe [1]); - - create_pipe (respipe); + create_respipe (); atfork_parent (); } @@ -1407,7 +1469,7 @@ BOOT: { - HV *stash = gv_stashpv ("IO::AIO", 1); + stash = gv_stashpv ("IO::AIO", 1); newCONSTSUB (stash, "EXDEV", newSViv (EXDEV)); newCONSTSUB (stash, "O_RDONLY", newSViv (O_RDONLY)); @@ -1428,7 +1490,7 @@ newCONSTSUB (stash, "SIGIO", newSViv (SIGIO)); #endif - create_pipe (respipe); + create_respipe (); X_THREAD_ATFORK (atfork_prepare, atfork_parent, atfork_child); } @@ -1485,10 +1547,9 @@ } void -aio_close (SV *fh, SV *callback=&PL_sv_undef) +aio_fsync (SV *fh, SV *callback=&PL_sv_undef) PROTOTYPE: $;$ ALIAS: - aio_close = REQ_CLOSE aio_fsync = REQ_FSYNC aio_fdatasync = REQ_FDATASYNC PPCODE: @@ -1503,6 +1564,19 @@ } void +aio_close (SV *fh, SV *callback=&PL_sv_undef) + PROTOTYPE: $;$ + PPCODE: +{ + dREQ; + + req->type = REQ_CLOSE; + req->int1 = PerlIO_fileno (IoIFP (sv_2io (fh))); + + REQ_SEND (req); +} + +void aio_read (SV *fh, SV *offset, SV *length, SV8 *data, IV dataoffset, SV *callback=&PL_sv_undef) ALIAS: aio_read = REQ_READ @@ -1834,11 +1908,14 @@ void aio_nop (SV *callback=&PL_sv_undef) + ALIAS: + aio_nop = REQ_NOP + aio_sync = REQ_SYNC PPCODE: { dREQ; - req->type = REQ_NOP; + req->type = ix; REQ_SEND; } @@ -1934,7 +2011,6 @@ PUSHMARK (SP); PUTBACK; count = call_sv (cb, GIMME_V | G_NOARGS | G_EVAL); - SPAGAIN; unblock_sig (); if (SvTRUE (ERRSV))