--- IO-AIO/AIO.xs	2006/10/26 13:25:40	1.77
+++ IO-AIO/AIO.xs	2006/10/26 14:35:34	1.78
@@ -134,8 +134,9 @@
 
 static int next_pri = DEFAULT_PRI + PRI_BIAS;
 
-static int started, wanted;
-static volatile int nreqs;
+static unsigned int started, wanted;
+static volatile unsigned int nreqs;
+static volatile unsigned int max_outstanding = 0xffffffff;
 static int respipe [2];
 
 #if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
@@ -505,57 +506,67 @@
   int do_croak = 0;
   aio_req req;
 
-  while (max <= 0 || count < max)
+  for (;;)
     {
-      LOCK (reslock);
-      req = reqq_shift (&res_queue);
-
-      if (req)
+      while (max <= 0 || count < max)
         {
-          if (!res_queue.size)
+          LOCK (reslock);
+          req = reqq_shift (&res_queue);
+
+          if (req)
             {
-              /* read any signals sent by the worker threads */
-              char buf [32];
-              while (read (respipe [0], buf, 32) == 32)
-                ;
+              if (!res_queue.size)
+                {
+                  /* read any signals sent by the worker threads */
+                  char buf [32];
+                  while (read (respipe [0], buf, 32) == 32)
+                    ;
+                }
             }
-        }
 
-      UNLOCK (reslock);
-
-      if (!req)
-        break;
-
-      --nreqs;
+          UNLOCK (reslock);
 
-      if (req->type == REQ_QUIT)
-        started--;
-      else if (req->type == REQ_GROUP && req->length)
-        {
-          req->fd = 1; /* mark request as delayed */
-          continue;
-        }
-      else
-        {
-          if (req->type == REQ_READ)
-            SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
+          if (!req)
+            break;
 
-          if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE))
-            SvREADONLY_off (req->data);
+          --nreqs;
 
-          if (req->statdata)
+          if (req->type == REQ_QUIT)
+            --started;
+          else if (req->type == REQ_GROUP && req->length)
             {
-              PL_laststype   = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
-              PL_laststatval = req->result;
-              PL_statcache   = *(req->statdata);
+              req->fd = 1; /* mark request as delayed */
+              continue;
             }
+          else
+            {
+              if (req->type == REQ_READ)
+                SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
+
+              if (req->data2ptr && (req->type == REQ_READ || req->type == REQ_WRITE))
+                SvREADONLY_off (req->data);
+
+              if (req->statdata)
+                {
+                  PL_laststype   = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
+                  PL_laststatval = req->result;
+                  PL_statcache   = *(req->statdata);
+                }
 
-          req_invoke (req);
+              req_invoke (req);
+
+              count++;
+            }
 
-          count++;
+          req_free (req);
         }
 
-      req_free (req);
+      if (nreqs <= max_outstanding)
+        break;
+
+      poll_wait ();
+
+      max = 0;
     }
 
   return count;
@@ -588,7 +599,7 @@
       wrk->next = wrk_first.next;
       wrk_first.next->prev = wrk;
       wrk_first.next = wrk;
-      started++;
+      ++started;
     }
   else
     free (wrk);
@@ -1104,14 +1115,21 @@
 }
 
 void
-min_parallel (nthreads)
-	int	nthreads
+min_parallel (int nthreads)
 	PROTOTYPE: $
 
 void
-max_parallel (nthreads)
-	int	nthreads
+max_parallel (int nthreads)
+	PROTOTYPE: $
+
+int
+max_outstanding (int maxreqs)
 	PROTOTYPE: $
+        CODE:
+        RETVAL = max_outstanding;
+        max_outstanding = maxreqs;
+	OUTPUT:
+        RETVAL
 
 void
 aio_open (pathname,flags,mode,callback=&PL_sv_undef)