--- IO-AIO/AIO.xs	2006/10/23 14:49:51	1.56
+++ IO-AIO/AIO.xs	2006/10/23 23:54:41	1.64
@@ -1,4 +1,9 @@
+#if __linux
+# define _GNU_SOURCE
+#endif
+
 #define _REENTRANT 1
+
 #include <errno.h>
 
 #include "EXTERN.h"
@@ -68,10 +73,6 @@
 {
   struct aio_cb *volatile next;
 
-  struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
-
-  SV *self; /* the perl counterpart of this request, if any */
-
   SV *data, *callback;
   SV *fh, *fh2;
   void *dataptr, *data2ptr;
@@ -80,24 +81,49 @@
   size_t length;
   ssize_t result;
 
+  STRLEN dataoffset;
   int type;
   int fd, fd2;
   int errorno;
-  STRLEN dataoffset;
   mode_t mode; /* open */
-  unsigned char cancelled;
+
+  unsigned char flags;
+  unsigned char pri;
+
+  SV *self; /* the perl counterpart of this request, if any */
+  struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
 } aio_cb;
 
+enum {
+  FLAG_CANCELLED = 0x01,
+};
+
 typedef aio_cb *aio_req;
 typedef aio_cb *aio_req_ornot;
 
+enum {
+  PRI_MIN     = -4,
+  PRI_MAX     =  4,
+
+  DEFAULT_PRI = 0,
+  PRI_BIAS    = -PRI_MIN,
+};
+
+static int next_pri = DEFAULT_PRI + PRI_BIAS;
+
 static int started, wanted;
 static volatile int nreqs;
 static int max_outstanding = 1<<30;
 static int respipe [2];
 
-static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER;
-static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER;
+#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
+# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
+#else
+# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
+#endif
+
+static pthread_mutex_t reslock = AIO_MUTEX_INIT;
+static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
 static pthread_cond_t  reqwait = PTHREAD_COND_INITIALIZER;
 
 static volatile aio_req reqs, reqe; /* queue start, queue end */
@@ -132,7 +158,7 @@
 
 static void aio_grp_feed (aio_req grp)
 {
-  while (grp->length < grp->fd2)
+  while (grp->length < grp->fd2 && !(grp->flags & FLAG_CANCELLED))
     {
       int old_len = grp->length;
 
@@ -178,9 +204,22 @@
 
 static void poll_wait ()
 {
-  if (nreqs && !ress)
+  fd_set rfd;
+
+  while (nreqs)
     {
-      fd_set rfd;
+      aio_req req;
+#if !(__x86 || __x86_64) /* safe without sempahore on this archs */
+      pthread_mutex_lock (&reslock);
+#endif
+      req = ress;
+#if !(__x86 || __x86_64) /* safe without sempahore on this archs */
+      pthread_mutex_unlock (&reslock);
+#endif
+
+      if (req)
+        return;
+
       FD_ZERO(&rfd);
       FD_SET(respipe [0], &rfd);
 
@@ -193,7 +232,7 @@
   dSP;
   int errorno = errno;
 
-  if (req->cancelled || !SvOK (req->callback))
+  if (req->flags & FLAG_CANCELLED || !SvOK (req->callback))
     return;
 
   errno = req->errorno;
@@ -323,7 +362,7 @@
 
 static void req_cancel (aio_req req)
 {
-  req->cancelled = 1;
+  req->flags |= FLAG_CANCELLED;
 
   if (req->type == REQ_GROUP)
     {
@@ -787,9 +826,10 @@
       pthread_mutex_unlock (&reqlock);
      
       errno = 0; /* strictly unnecessary */
+      type = req->type; /* remember type for QUIT check */
 
-      if (!req->cancelled)
-        switch (type = req->type) /* remember type for QUIT check */
+      if (!(req->flags & FLAG_CANCELLED))
+        switch (type)
           {
             case REQ_READ:      req->result = pread     (req->fd, req->dataptr, req->length, req->offset); break;
             case REQ_WRITE:     req->result = pwrite    (req->fd, req->dataptr, req->length, req->offset); break;
@@ -918,6 +958,8 @@
 
 #define dREQ							\
   aio_req req;							\
+  int req_pri = next_pri;					\
+  next_pri = DEFAULT_PRI + PRI_BIAS;				\
 								\
   if (SvOK (callback) && !SvROK (callback))			\
     croak ("callback must be undef or of reference type");	\
@@ -926,7 +968,8 @@
   if (!req)							\
     croak ("out of memory during aio_req allocation");		\
 								\
-  req->callback = newSVsv (callback)
+  req->callback = newSVsv (callback);				\
+  req->pri = req_pri
 
 #define REQ_SEND						\
   req_send (req);						\
@@ -1211,8 +1254,10 @@
         PPCODE:
 {
 	dREQ;
+
         req->type = REQ_GROUP;
         req_send (req);
+
         XPUSHs (req_sv (req, AIO_GRP_KLASS));
 }
 
@@ -1228,6 +1273,17 @@
 	REQ_SEND;
 }
 
+#if 0
+
+void
+aio_pri (int pri = DEFAULT_PRI)
+	CODE:
+        if (pri < PRI_MIN) pri = PRI_MIN;
+        if (pri > PRI_MAX) pri = PRI_MAX;
+        next_pri = pri + PRI_BIAS;
+
+#endif
+
 void
 flush ()
 	PROTOTYPE:
@@ -1290,7 +1346,7 @@
         req_cancel (req);
 
 void
-cb (aio_req req, SV *callback=&PL_sv_undef)
+cb (aio_req_ornot req, SV *callback=&PL_sv_undef)
 	CODE:
         SvREFCNT_dec (req->callback);
         req->callback = newSVsv (callback);