--- IO-AIO/AIO.xs	2005/07/10 18:16:49	1.2
+++ IO-AIO/AIO.xs	2005/07/10 20:07:11	1.3
@@ -36,7 +36,7 @@
 };
 
 typedef struct aio_cb {
-  struct aio_cb *next;
+  struct aio_cb *volatile next;
 
   int type;
 
@@ -57,9 +57,14 @@
 
 static int started;
 static int nreqs;
-static int reqpipe[2], respipe[2];
+static int respipe [2];
 
-static aio_req qs, qe; /* queue start, queue end */
+static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER;
+static pthread_mutex_t reqlock = 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 *aio_proc(void *arg);
 
@@ -84,31 +89,24 @@
 }
 
 static void
-send_reqs (void)
-{
-  /* this write is atomic */
-  while (qs && write (reqpipe[1], &qs, sizeof qs) == sizeof qs)
-   {
-     qs = qs->next;
-     if (!qs) qe = 0;
-   }
-}
-
-static void
 send_req (aio_req req)
 {
   nreqs++;
+
+  pthread_mutex_lock (&reqlock);
+
   req->next = 0;
 
-  if (qe)
+  if (reqe)
     {
-      qe->next = req;
-      qe = req;
+      reqe->next = req;
+      reqe = req;
     }
   else
-    qe = qs = req;
+    reqe = reqs = req;
 
-  send_reqs ();
+  pthread_cond_signal (&reqwait);
+  pthread_mutex_unlock (&reqlock);
 }
 
 static void
@@ -173,11 +171,14 @@
 static void
 poll_wait ()
 {
+  if (!nreqs)
+    return;
+
   fd_set rfd;
   FD_ZERO(&rfd);
-  FD_SET(respipe[0], &rfd);
+  FD_SET(respipe [0], &rfd);
 
-  select (respipe[0] + 1, &rfd, 0, 0, 0);
+  select (respipe [0] + 1, &rfd, 0, 0, 0);
 }
 
 static int
@@ -186,9 +187,31 @@
   dSP;
   int count = 0;
   aio_req req;
+  
+  {
+    /* read and signals sent by the worker threads */
+    char buf [32];
+    while (read (respipe [0], buf, 32) > 0)
+      ;
+  }
 
-  while (read (respipe[0], (void *)&req, sizeof (req)) == sizeof (req))
+  for (;;)
     {
+      pthread_mutex_lock (&reslock);
+
+      req = ress;
+
+      if (ress)
+        {
+          ress = ress->next;
+          if (!ress) rese = 0;
+        }
+
+      pthread_mutex_unlock (&reslock);
+
+      if (!req)
+        break;
+
       nreqs--;
 
       if (req->type == REQ_QUIT)
@@ -246,9 +269,6 @@
       Safefree (req);
     }
 
-  if (qs)
-    send_reqs ();
-
   return count;
 }
 
@@ -256,13 +276,35 @@
 aio_proc (void *thr_arg)
 {
   aio_req req;
+  int type;
 
-  /* then loop */
-  while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
+  do
     {
+      pthread_mutex_lock (&reqlock);
+
+      for (;;)
+        {
+          req = reqs;
+
+          if (reqs)
+            {
+              reqs = reqs->next;
+              if (!reqs) reqe = 0;
+            }
+
+          if (req)
+            break;
+
+          pthread_cond_wait (&reqwait, &reqlock);
+        }
+
+      pthread_mutex_unlock (&reqlock);
+     
       errno = 0; /* strictly unnecessary */
 
-      switch (req->type)
+      type = req->type;
+
+      switch (type)
         {
           case REQ_READ:      req->result = pread64   (req->fd, req->dataptr, req->length, req->offset); break;
           case REQ_WRITE:     req->result = pwrite64  (req->fd, req->dataptr, req->length, req->offset); break;
@@ -284,8 +326,7 @@
           case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
 
           case REQ_QUIT:
-            write (respipe[1], (void *)&req, sizeof (req));
-	    return 0;
+            break;
 
           default:
             req->result = ENOSYS;
@@ -293,8 +334,27 @@
         }
 
       req->errorno = errno;
-      write (respipe[1], (void *)&req, sizeof (req));
+
+      pthread_mutex_lock (&reslock);
+
+      req->next = 0;
+
+      if (rese)
+        {
+          rese->next = req;
+          rese = req;
+        }
+      else
+        {
+          rese = ress = req;
+
+          /* write a dummy byte to the pipe so fh becomes ready */
+          write (respipe [1], &respipe, 1);
+        }
+
+      pthread_mutex_unlock (&reslock);
     }
+  while (type != REQ_QUIT);
 
   return 0;
 }
@@ -303,13 +363,13 @@
 
 BOOT:
 {
-        if (pipe (reqpipe) || pipe (respipe))
-          croak ("unable to initialize request or result pipe");
+        if (pipe (respipe))
+          croak ("unable to initialize result pipe");
 
-        if (fcntl (reqpipe[1], F_SETFL, O_NONBLOCK))
+        if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
           croak ("cannot set result pipe to nonblocking mode");
 
-        if (fcntl (respipe[0], F_SETFL, O_NONBLOCK))
+        if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
           croak ("cannot set result pipe to nonblocking mode");
 }
 
@@ -508,7 +568,7 @@
 poll_fileno()
 	PROTOTYPE:
 	CODE:
-        RETVAL = respipe[0];
+        RETVAL = respipe [0];
 	OUTPUT:
 	RETVAL
 
@@ -524,7 +584,8 @@
 poll_wait()
 	PROTOTYPE:
 	CODE:
-        poll_wait ();
+        if (nreqs)
+          poll_wait ();
 
 int
 nreqs()