--- IO-AIO/AIO.xs 2007/07/08 09:09:34 1.103 +++ IO-AIO/AIO.xs 2007/09/24 18:14:00 1.106 @@ -26,6 +26,8 @@ # undef open # undef read # undef write +# undef send +# undef recv # undef stat # undef fstat # define lstat stat @@ -264,7 +266,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]; +static int respipe_osf [2], respipe [2] = { -1, -1 }; static mutex_t reslock = X_MUTEX_INIT; static mutex_t reqlock = X_MUTEX_INIT; @@ -627,6 +629,50 @@ req_cancel_subs (req); } +#ifdef USE_SOCKETS_AS_HANDLES +# define TO_SOCKET(x) (win32_get_osfhandle (x)) +#else +# define TO_SOCKET(x) (x) +#endif + +static void +create_respipe () +{ + int old_readfd = respipe [0]; + + if (respipe [1] >= 0) + respipe_close (TO_SOCKET (respipe [1])); + +#ifdef _WIN32 + if (PerlSock_socketpair (AF_UNIX, SOCK_STREAM, 0, respipe)) +#else + 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]); +} + X_THREAD_PROC (aio_proc); static void start_thread (void) @@ -738,10 +784,10 @@ maybe_start_thread (); - FD_ZERO(&rfd); - FD_SET(respipe [0], &rfd); + FD_ZERO (&rfd); + FD_SET (respipe [0], &rfd); - select (respipe [0] + 1, &rfd, 0, 0, 0); + PerlSock_select (respipe [0] + 1, &rfd, 0, 0, 0); } } @@ -776,7 +822,7 @@ { /* read any signals sent by the worker threads */ char buf [4]; - while (read (respipe [0], buf, 4) == 4) + while (respipe_read (respipe [0], buf, 4) == 4) ; } } @@ -1268,7 +1314,7 @@ if (!reqq_push (&res_queue, req)) { /* write a dummy byte to the pipe so fh becomes ready */ - write (respipe [1], &respipe, 1); + respipe_write (respipe_osf [1], (const void *)&respipe_osf, 1); /* optionally signal the main thread asynchronously */ if (main_sig) @@ -1345,11 +1391,7 @@ nready = 0; npending = 0; - close (respipe [0]); - close (respipe [1]); - - if (!create_pipe (respipe)) - croak ("cannot set result pipe to nonblocking mode"); + create_respipe (); atfork_parent (); } @@ -1395,13 +1437,14 @@ X_MUTEX_CHECK (reqwait); X_MUTEX_CHECK (preadwritelock); X_MUTEX_CHECK (readdirlock); + + X_COND_CHECK (reqwait); #else newCONSTSUB (stash, "S_IFIFO", newSViv (S_IFIFO)); newCONSTSUB (stash, "SIGIO", newSViv (SIGIO)); #endif - if (!create_pipe (respipe)) - croak ("cannot set result pipe to nonblocking mode"); + create_respipe (); X_THREAD_ATFORK (atfork_prepare, atfork_parent, atfork_child); }