--- IO-AIO/AIO.xs	2005/08/17 03:01:56	1.24
+++ IO-AIO/AIO.xs	2005/08/17 04:47:02	1.27
@@ -64,12 +64,28 @@
 
 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 */
 static volatile aio_req ress, rese; /* queue start, queue end */
 
+static void free_req (aio_req req)
+{
+  if (req->data)
+    SvREFCNT_dec (req->data);
+
+  if (req->fh)
+    SvREFCNT_dec (req->fh);
+
+  if (req->statdata)
+    Safefree (req->statdata);
+
+  if (req->callback)
+    SvREFCNT_dec (req->callback);
+
+  Safefree (req);
+}
+
 static void
 poll_wait ()
 {
@@ -89,24 +105,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)
+            {
+              rese = 0;
 
-  pthread_mutex_unlock (&reslock);
+              /* read any signals sent by the worker threads */
+              char buf [32];
+              while (read (respipe [0], buf, 32) == 32)
+                ;
+            }
+        }
+
+      pthread_mutex_unlock (&reslock);
+
+      if (!req)
+        break;
 
-  while (req)
-    {
       nreqs--;
 
       if (req->type == REQ_QUIT)
@@ -117,22 +142,13 @@
           errno = req->errorno;
 
           if (req->type == REQ_READ)
-            SvCUR_set (req->data, req->dataoffset
-                                  + req->result > 0 ? req->result : 0);
-
-          if (req->data)
-            SvREFCNT_dec (req->data);
+            SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
 
-          if (req->fh)
-            SvREFCNT_dec (req->fh);
-
-          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;
               PL_statcache   = *(req->statdata);
-
-              Safefree (req->statdata);
             }
 
           ENTER;
@@ -159,25 +175,21 @@
               PUTBACK;
               call_sv (req->callback, G_VOID | G_EVAL);
               SPAGAIN;
-            }
 
-          do_croak = SvTRUE (ERRSV);
+              if (SvTRUE (ERRSV))
+                {
+                  free_req (req);
+                  croak (0);
+                }
+            }
 
           LEAVE;
-          
-          if (req->callback)
-            SvREFCNT_dec (req->callback);
 
           errno = errorno;
           count++;
         }
 
-      prv = req;
-      req = req->next;
-      Safefree (prv);
-
-      if (do_croak)
-        croak (0);
+      free_req (req);
     }
 
   return count;
@@ -225,24 +237,28 @@
   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
 end_thread (void)
 {
   aio_req req;
-  New (0, req, 1, aio_cb);
+  Newz (0, req, 1, aio_cb);
   req->type = REQ_QUIT;
 
   send_req (req);
 }
 
-
 static void min_parallel (int nthreads)
 {
   while (nthreads > started)
@@ -252,6 +268,7 @@
 static void max_parallel (int nthreads)
 {
   int cur = started;
+
   while (cur > nthreads)
     {          
       end_thread ();
@@ -265,54 +282,59 @@
     }
 }
 
-static int fork_started;
-
-static void atfork_prepare (void)
+static void create_pipe ()
 {
-  pthread_mutex_lock (&frklock);
+  if (pipe (respipe))
+    croak ("unable to initialize result pipe");
+
+  if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
+    croak ("cannot set result pipe to nonblocking mode");
 
-  fork_started = started;
+  if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
+    croak ("cannot set result pipe to nonblocking mode");
+}
 
+static void atfork_prepare (void)
+{
   for (;;) {
-    while (nreqs)
+
+    for (;;)
       {
-        poll_wait ();
         poll_cb ();
+
+        if (!nreqs)
+          break;
+
+        poll_wait ();
       }
 
-    max_parallel (0);
-  
     pthread_mutex_lock (&reqlock);
 
-    if (!nreqs && !started)
+    if (!nreqs)
       break;
 
     pthread_mutex_unlock (&reqlock);
-
-    min_parallel (fork_started);
   }
 
   pthread_mutex_lock (&reslock);
 
-  assert (!started);
-  assert (!nreqs);
-  assert (!reqs && !reqe);
-  assert (!ress && !rese);
+  assert (!nreqs && !reqs && !ress);
 }
 
 static void atfork_parent (void)
 {
   pthread_mutex_unlock (&reslock);
-  min_parallel (fork_started);
   pthread_mutex_unlock (&reqlock);
-  pthread_mutex_unlock (&frklock);
 }
 
 static void atfork_child (void)
 {
-  reqs = reqe = 0;
+  int restart = started;
+  started = 0;
 
   atfork_parent ();
+
+  min_parallel (restart);
 }
 
 /*****************************************************************************/
@@ -485,7 +507,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
 
@@ -493,15 +515,7 @@
 
 BOOT:
 {
-        if (pipe (respipe))
-          croak ("unable to initialize result pipe");
-
-        if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
-          croak ("cannot set result pipe to nonblocking mode");
-
-        if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
-          croak ("cannot set result pipe to nonblocking mode");
-
+	create_pipe ();
         pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
 }
 
@@ -616,7 +630,6 @@
           req->length = length;
           req->data = SvREFCNT_inc (data);
           req->dataptr = (char *)svptr + dataoffset;
-          req->callback = SvREFCNT_inc (callback);
 
           send_req (req);
         }
@@ -655,7 +668,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))
           {