--- IO-AIO/AIO.xs	2005/07/10 20:07:11	1.3
+++ IO-AIO/AIO.xs	2005/07/11 01:03:17	1.6
@@ -1,5 +1,3 @@
-#define PERL_NO_GET_CONTEXT
-
 #include "EXTERN.h"
 #include "perl.h"
 #include "XSUB.h"
@@ -19,12 +17,10 @@
 typedef void *OutputStream; /* hack, but 5.6.1 is simply toooo old ;) */
 typedef void *InOutStream;  /* hack, but 5.6.1 is simply toooo old ;) */
 
-#if __i386 || __amd64
-# define STACKSIZE ( 256 * sizeof (long))
-#elif __ia64
-# define STACKSIZE (8192 * sizeof (long))
+#if __ia64
+# define STACKSIZE 65536
 #else
-# define STACKSIZE ( 512 * sizeof (long))
+# define STACKSIZE  4096
 #endif
 
 enum {
@@ -57,6 +53,7 @@
 
 static int started;
 static int nreqs;
+static int max_outstanding = 1<<30;
 static int respipe [2];
 
 static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER;
@@ -66,108 +63,6 @@
 static volatile aio_req reqs, reqe; /* queue start, queue end */
 static volatile aio_req ress, rese; /* queue start, queue end */
 
-static void *aio_proc(void *arg);
-
-static void
-start_thread (void)
-{
-  sigset_t fullsigset, oldsigset;
-  pthread_t tid;
-  pthread_attr_t attr;
-
-  pthread_attr_init (&attr);
-  pthread_attr_setstacksize (&attr, STACKSIZE);
-  pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
-
-  sigfillset (&fullsigset);
-  sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
-
-  if (pthread_create (&tid, &attr, aio_proc, 0) == 0)
-    started++;
-
-  sigprocmask (SIG_SETMASK, &oldsigset, 0);
-}
-
-static void
-send_req (aio_req req)
-{
-  nreqs++;
-
-  pthread_mutex_lock (&reqlock);
-
-  req->next = 0;
-
-  if (reqe)
-    {
-      reqe->next = req;
-      reqe = req;
-    }
-  else
-    reqe = reqs = req;
-
-  pthread_cond_signal (&reqwait);
-  pthread_mutex_unlock (&reqlock);
-}
-
-static void
-end_thread (void)
-{
-  aio_req req;
-  New (0, req, 1, aio_cb);
-  req->type = REQ_QUIT;
-
-  send_req (req);
-}
-
-static void
-read_write (pTHX_
-            int dowrite, int fd, off_t offset, size_t length,
-            SV *data, STRLEN dataoffset, SV *callback)
-{
-  aio_req req;
-  STRLEN svlen;
-  char *svptr = SvPV (data, svlen);
-
-  SvUPGRADE (data, SVt_PV);
-  SvPOK_on (data);
-
-  if (dataoffset < 0)
-    dataoffset += svlen;
-
-  if (dataoffset < 0 || dataoffset > svlen)
-    croak ("data offset outside of string");
-
-  if (dowrite)
-    {
-      /* write: check length and adjust. */
-      if (length < 0 || length + dataoffset > svlen)
-        length = svlen - dataoffset;
-    }
-  else
-    {
-      /* read: grow scalar as necessary */
-      svptr = SvGROW (data, length + dataoffset);
-    }
-
-  if (length < 0)
-    croak ("length must not be negative");
-
-  Newz (0, req, 1, aio_cb);
-
-  if (!req)
-    croak ("out of memory during aio_req allocation");
-
-  req->type = dowrite ? REQ_WRITE : REQ_READ;
-  req->fd = fd;
-  req->offset = offset;
-  req->length = length;
-  req->data = SvREFCNT_inc (data);
-  req->dataptr = (char *)svptr + dataoffset;
-  req->callback = SvREFCNT_inc (callback);
-
-  send_req (req);
-}
-
 static void
 poll_wait ()
 {
@@ -182,7 +77,7 @@
 }
 
 static int
-poll_cb (pTHX)
+poll_cb ()
 {
   dSP;
   int count = 0;
@@ -272,6 +167,113 @@
   return count;
 }
 
+static void *aio_proc(void *arg);
+
+static void
+start_thread (void)
+{
+  sigset_t fullsigset, oldsigset;
+  pthread_t tid;
+  pthread_attr_t attr;
+
+  pthread_attr_init (&attr);
+  pthread_attr_setstacksize (&attr, STACKSIZE);
+  pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
+
+  sigfillset (&fullsigset);
+  sigprocmask (SIG_SETMASK, &fullsigset, &oldsigset);
+
+  if (pthread_create (&tid, &attr, aio_proc, 0) == 0)
+    started++;
+
+  sigprocmask (SIG_SETMASK, &oldsigset, 0);
+}
+
+static void
+send_req (aio_req req)
+{
+  nreqs++;
+
+  pthread_mutex_lock (&reqlock);
+
+  req->next = 0;
+
+  if (reqe)
+    {
+      reqe->next = req;
+      reqe = req;
+    }
+  else
+    reqe = reqs = req;
+
+  pthread_cond_signal (&reqwait);
+  pthread_mutex_unlock (&reqlock);
+
+  while (nreqs > max_outstanding)
+    {
+      poll_wait ();
+      poll_cb ();
+    }
+}
+
+static void
+end_thread (void)
+{
+  aio_req req;
+  New (0, req, 1, aio_cb);
+  req->type = REQ_QUIT;
+
+  send_req (req);
+}
+
+static void
+read_write (int dowrite, int fd, off_t offset, size_t length,
+            SV *data, STRLEN dataoffset, SV *callback)
+{
+  aio_req req;
+  STRLEN svlen;
+  char *svptr = SvPV (data, svlen);
+
+  SvUPGRADE (data, SVt_PV);
+  SvPOK_on (data);
+
+  if (dataoffset < 0)
+    dataoffset += svlen;
+
+  if (dataoffset < 0 || dataoffset > svlen)
+    croak ("data offset outside of string");
+
+  if (dowrite)
+    {
+      /* write: check length and adjust. */
+      if (length < 0 || length + dataoffset > svlen)
+        length = svlen - dataoffset;
+    }
+  else
+    {
+      /* read: grow scalar as necessary */
+      svptr = SvGROW (data, length + dataoffset);
+    }
+
+  if (length < 0)
+    croak ("length must not be negative");
+
+  Newz (0, req, 1, aio_cb);
+
+  if (!req)
+    croak ("out of memory during aio_req allocation");
+
+  req->type = dowrite ? REQ_WRITE : REQ_READ;
+  req->fd = fd;
+  req->offset = offset;
+  req->length = length;
+  req->data = SvREFCNT_inc (data);
+  req->dataptr = (char *)svptr + dataoffset;
+  req->callback = SvREFCNT_inc (callback);
+
+  send_req (req);
+}
+
 static void *
 aio_proc (void *thr_arg)
 {
@@ -397,10 +399,18 @@
         while (started > nthreads)
           {
             poll_wait ();
-            poll_cb (aTHX);
+            poll_cb ();
           }
 }
 
+int
+max_outstanding(nreqs)
+	int nreqs
+        PROTOTYPE: $
+        CODE:
+        RETVAL = max_outstanding;
+        max_outstanding = nreqs;
+
 void
 aio_open(pathname,flags,mode,callback)
 	SV *	pathname
@@ -462,7 +472,7 @@
         SV *		callback
 	PROTOTYPE: $$$$$$
         CODE:
-        read_write (aTHX_ 0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
+        read_write (0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
 
 void
 aio_write(fh,offset,length,data,dataoffset,callback)
@@ -474,7 +484,7 @@
         SV *		callback
 	PROTOTYPE: $$$$$$
         CODE:
-        read_write (aTHX_ 1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
+        read_write (1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
 
 void
 aio_readahead(fh,offset,length,callback)
@@ -564,6 +574,16 @@
 	send_req (req);
 }
 
+void
+flush()
+	PROTOTYPE:
+	CODE:
+        while (nreqs)
+          {
+            poll_wait ();
+            poll_cb ();
+          }
+
 int
 poll_fileno()
 	PROTOTYPE:
@@ -576,7 +596,7 @@
 poll_cb(...)
 	PROTOTYPE:
 	CODE:
-        RETVAL = poll_cb (aTHX);
+        RETVAL = poll_cb ();
 	OUTPUT:
 	RETVAL