--- IO-AIO/AIO.xs	2005/08/17 03:52:20	1.26
+++ IO-AIO/AIO.xs	2005/08/17 04:47:02	1.27
@@ -64,7 +64,6 @@
 
 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 */
@@ -78,7 +77,7 @@
   if (req->fh)
     SvREFCNT_dec (req->fh);
 
-  if (req->type == REQ_STAT || req->type == REQ_LSTAT || req->type == REQ_FSTAT)
+  if (req->statdata)
     Safefree (req->statdata);
 
   if (req->callback)
@@ -106,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)
@@ -134,10 +142,9 @@
           errno = req->errorno;
 
           if (req->type == REQ_READ)
-            SvCUR_set (req->data, req->dataoffset
-                                  + req->result > 0 ? req->result : 0);
+            SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
 
-          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;
@@ -168,22 +175,21 @@
               PUTBACK;
               call_sv (req->callback, G_VOID | G_EVAL);
               SPAGAIN;
+
+              if (SvTRUE (ERRSV))
+                {
+                  free_req (req);
+                  croak (0);
+                }
             }
 
           LEAVE;
 
-          do_croak = SvTRUE (ERRSV);
-          
           errno = errorno;
           count++;
         }
 
-      prv = req;
-      req = req->next;
-      free_req (prv);
-
-      if (do_croak)
-        croak (0);
+      free_req (req);
     }
 
   return count;
@@ -231,11 +237,16 @@
   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
@@ -248,7 +259,6 @@
   send_req (req);
 }
 
-
 static void min_parallel (int nthreads)
 {
   while (nthreads > started)
@@ -258,6 +268,7 @@
 static void max_parallel (int nthreads)
 {
   int cur = started;
+
   while (cur > nthreads)
     {          
       end_thread ();
@@ -285,16 +296,35 @@
 
 static void atfork_prepare (void)
 {
-  pthread_mutex_lock (&frklock);
-  pthread_mutex_lock (&reqlock);
+  for (;;) {
+
+    for (;;)
+      {
+        poll_cb ();
+
+        if (!nreqs)
+          break;
+
+        poll_wait ();
+      }
+
+    pthread_mutex_lock (&reqlock);
+
+    if (!nreqs)
+      break;
+
+    pthread_mutex_unlock (&reqlock);
+  }
+
   pthread_mutex_lock (&reslock);
+
+  assert (!nreqs && !reqs && !ress);
 }
 
 static void atfork_parent (void)
 {
   pthread_mutex_unlock (&reslock);
   pthread_mutex_unlock (&reqlock);
-  pthread_mutex_unlock (&frklock);
 }
 
 static void atfork_child (void)
@@ -302,27 +332,6 @@
   int restart = started;
   started = 0;
 
-  while (reqs)
-    {
-      free_req (reqs);
-      reqs = reqs->next;
-    }
-
-  reqs = reqe = 0;
-
-  while (ress)
-    {
-      free_req (ress);
-      ress = ress->next;
-    }
-
-  ress = rese = 0;
-
-  close (respipe [0]);
-  close (respipe [1]);
-
-  create_pipe ();
-
   atfork_parent ();
 
   min_parallel (restart);
@@ -498,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
 
@@ -621,7 +630,6 @@
           req->length = length;
           req->data = SvREFCNT_inc (data);
           req->dataptr = (char *)svptr + dataoffset;
-          req->callback = SvREFCNT_inc (callback);
 
           send_req (req);
         }
@@ -660,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))
           {