--- IO-AIO/AIO.xs	2007/07/08 09:09:34	1.103
+++ IO-AIO/AIO.xs	2007/10/03 21:27:51	1.107
@@ -26,6 +26,8 @@
 # undef open
 # undef read
 # undef write
+# undef send
+# undef recv
 # undef stat
 # undef fstat
 # define lstat stat
@@ -97,6 +99,8 @@
 # define SvVAL64 SvNV
 #endif
 
+static HV *stash;
+
 #define dBUF	 				\
   char *aio_buf;				\
   X_LOCK (wrklock);				\
@@ -264,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];
+static int respipe_osf [2], respipe [2] = { -1, -1 };
 
 static mutex_t reslock = X_MUTEX_INIT;
 static mutex_t reqlock = X_MUTEX_INIT;
@@ -501,15 +505,30 @@
           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;
+                  if (do_open (gv, sym, symlen, 0, 0, 0, 0))
+                    fh = gv;
+                }
 
-              fh = POPs;
-              PUSHMARK (SP);
               XPUSHs (fh);
             }
             break;
@@ -627,6 +646,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 +801,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 +839,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)
                     ;
                 }
             }
@@ -1194,7 +1257,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 = PerlIO_close ((PerlIO *)req->ptr1); 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;
@@ -1268,7 +1331,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 +1408,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 ();
 }
@@ -1381,7 +1440,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));
@@ -1395,13 +1454,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);
 }
@@ -1458,10 +1518,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:
@@ -1476,6 +1535,30 @@
 }
 
 void
+aio_close (SV *fh, SV *callback=&PL_sv_undef)
+	PROTOTYPE: $;$
+	PPCODE:
+{
+  	PerlIO *io = IoIFP (sv_2io (fh));
+        int fd = PerlIO_fileno (io);
+
+        if (fd < 0)
+          croak ("aio_close called with fd-less filehandle");
+
+        PerlIO_binmode (io, 0, 0, 0);
+
+        {
+          dREQ;
+
+          req->type = REQ_CLOSE;
+          req->sv1  = newSVsv (fh);
+          req->ptr1 = (void *)io;
+
+          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