--- IO-AIO/AIO.xs	2006/07/21 07:35:31	1.42
+++ IO-AIO/AIO.xs	2006/10/21 23:06:04	1.43
@@ -57,27 +57,35 @@
   REQ_LINK, REQ_SYMLINK,
 };
 
-typedef struct aio_cb {
+#define AIO_CB_KLASS "IO::AIO::CB"
+
+typedef struct aio_cb
+{
+  struct aio_cb *grp_prev, *grp_next;
+  struct aio_grp *grp;
+
   struct aio_cb *volatile next;
 
-  int type;
+  SV *self; /* the perl counterpart of this request, if any */
 
-  /* should receive a cleanup, with unions */
-  int fd, fd2;
+  SV *data, *callback;
+  SV *fh, *fh2;
+  void *dataptr, *data2ptr;
+  Stat_t *statdata;
   off_t offset;
   size_t length;
   ssize_t result;
-  mode_t mode; /* open */
+
+  int type;
+  int fd, fd2;
   int errorno;
-  SV *data, *callback;
-  SV *fh, *fh2;
-  void *dataptr, *data2ptr;
   STRLEN dataoffset;
-
-  Stat_t *statdata;
+  mode_t mode; /* open */
+  unsigned char cancelled;
 } aio_cb;
 
 typedef aio_cb *aio_req;
+typedef aio_cb *aio_req_ornot;
 
 static int started, wanted;
 static volatile int nreqs;
@@ -91,8 +99,66 @@
 static volatile aio_req reqs, reqe; /* queue start, queue end */
 static volatile aio_req ress, rese; /* queue start, queue end */
 
-static void free_req (aio_req req)
+typedef struct aio_grp
+{
+  struct aio_cb *first, *last;
+  SV *callback;
+  int busycount;
+} aio_grp;
+
+static void aio_grp_begin (aio_grp *grp)
+{
+  ++grp->busycount;
+}
+
+static void aio_grp_end (aio_grp *grp)
+{
+  --grp->busycount;
+
+  if (grp->busycount)
+    return;
+
+  SvREFCNT_dec (grp->callback);
+  grp->callback = 0;
+}
+
+static aio_grp *aio_grp_new ()
+{
+  aio_grp *grp;
+
+  Newz (0, grp, 1, aio_grp);
+  aio_grp_begin (grp);
+
+  return grp;
+}
+
+/* must be called at most once */
+static SV *req_sv (aio_req req)
+{
+  req->self = (SV *)newHV ();
+  sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0);
+
+  return sv_bless (newRV_noinc (req->self), gv_stashpv (AIO_CB_KLASS, 1));
+}
+
+static aio_req SvAIO_REQ (SV *sv)
 {
+  if (!sv_derived_from (sv, AIO_CB_KLASS) || !SvROK (sv))
+    croak ("object of class " AIO_CB_KLASS " expected");
+
+  MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext);
+
+  return mg ? (aio_req)mg->mg_ptr : 0;
+}
+
+static void req_free (aio_req req)
+{
+  if (req->self)
+    {
+      sv_unmagic (req->self, PERL_MAGIC_ext);
+      SvREFCNT_dec (req->self);
+    }
+
   if (req->data)
     SvREFCNT_dec (req->data);
 
@@ -228,7 +294,7 @@
                 }
             }
 
-          if (SvOK (req->callback))
+          if (SvOK (req->callback) && !req->cancelled)
             {
               PUTBACK;
               call_sv (req->callback, G_VOID | G_EVAL);
@@ -236,7 +302,7 @@
 
               if (SvTRUE (ERRSV))
                 {
-                  free_req (req);
+                  req_free (req);
                   croak (0);
                 }
             }
@@ -247,7 +313,7 @@
           count++;
         }
 
-      free_req (req);
+      req_free (req);
     }
 
   return count;
@@ -276,7 +342,7 @@
 }
 
 static void
-send_req (aio_req req)
+req_send (aio_req req)
 {
   while (started < wanted && nreqs >= started)
     start_thread ();
@@ -317,7 +383,7 @@
   Newz (0, req, 1, aio_cb);
   req->type = REQ_QUIT;
 
-  send_req (req);
+  req_send (req);
 }
 
 static void min_parallel (int nthreads)
@@ -650,39 +716,38 @@
      
       errno = 0; /* strictly unnecessary */
 
-      type = req->type;
-
-      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;
+      if (!req->cancelled)
+        switch (req->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;
 
-          case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
-          case REQ_SENDFILE:  req->result = sendfile_ (req->fd, req->fd2, req->offset, req->length); break;
+            case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
+            case REQ_SENDFILE:  req->result = sendfile_ (req->fd, req->fd2, req->offset, req->length); break;
 
-          case REQ_STAT:      req->result = stat      (req->dataptr, req->statdata); break;
-          case REQ_LSTAT:     req->result = lstat     (req->dataptr, req->statdata); break;
-          case REQ_FSTAT:     req->result = fstat     (req->fd     , req->statdata); break;
-
-          case REQ_OPEN:      req->result = open      (req->dataptr, req->fd, req->mode); break;
-          case REQ_CLOSE:     req->result = close     (req->fd); break;
-          case REQ_UNLINK:    req->result = unlink    (req->dataptr); break;
-          case REQ_RMDIR:     req->result = rmdir     (req->dataptr); break;
-          case REQ_RENAME:    req->result = rename    (req->data2ptr, req->dataptr); break;
-          case REQ_LINK:      req->result = link      (req->data2ptr, req->dataptr); break;
-          case REQ_SYMLINK:   req->result = symlink   (req->data2ptr, req->dataptr); break;
-
-          case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
-          case REQ_FSYNC:     req->result = fsync     (req->fd); break;
-          case REQ_READDIR:   req->result = scandir_  (req->dataptr, &req->data2ptr); break;
+            case REQ_STAT:      req->result = stat      (req->dataptr, req->statdata); break;
+            case REQ_LSTAT:     req->result = lstat     (req->dataptr, req->statdata); break;
+            case REQ_FSTAT:     req->result = fstat     (req->fd     , req->statdata); break;
+
+            case REQ_OPEN:      req->result = open      (req->dataptr, req->fd, req->mode); break;
+            case REQ_CLOSE:     req->result = close     (req->fd); break;
+            case REQ_UNLINK:    req->result = unlink    (req->dataptr); break;
+            case REQ_RMDIR:     req->result = rmdir     (req->dataptr); break;
+            case REQ_RENAME:    req->result = rename    (req->data2ptr, req->dataptr); break;
+            case REQ_LINK:      req->result = link      (req->data2ptr, req->dataptr); break;
+            case REQ_SYMLINK:   req->result = symlink   (req->data2ptr, req->dataptr); break;
+
+            case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
+            case REQ_FSYNC:     req->result = fsync     (req->fd); break;
+            case REQ_READDIR:   req->result = scandir_  (req->dataptr, &req->data2ptr); break;
 
-          case REQ_QUIT:
-            break;
+            case REQ_QUIT:
+              break;
 
-          default:
-            req->result = ENOSYS;
-            break;
-        }
+            default:
+              req->result = ENOSYS;
+              break;
+          }
 
       req->errorno = errno;
 
@@ -746,7 +811,7 @@
     {
       prv = reqs;
       reqs = prv->next;
-      free_req (prv);
+      req_free (prv);
     }
 
   reqs = reqe = 0;
@@ -755,7 +820,7 @@
     {
       prv = ress;
       ress = prv->next;
-      free_req (prv);
+      req_free (prv);
     }
       
   ress = rese = 0;
@@ -771,13 +836,19 @@
   aio_req req;							\
 								\
   if (SvOK (callback) && !SvROK (callback))			\
-    croak ("clalback must be undef or of reference type");	\
+    croak ("callback must be undef or of reference type");	\
 								\
   Newz (0, req, 1, aio_cb);					\
   if (!req)							\
     croak ("out of memory during aio_req allocation");		\
 								\
-  req->callback = newSVsv (callback);
+  req->callback = newSVsv (callback)
+
+#define REQ_SEND						\
+  req_send (req);						\
+								\
+  if (GIMME_V != G_VOID)					\
+    XPUSHs (req_sv (req));
 	
 MODULE = IO::AIO                PACKAGE = IO::AIO
 
@@ -819,7 +890,7 @@
         int	mode
         SV *	callback
 	PROTOTYPE: $$$;$
-	CODE:
+	PPCODE:
 {
         dREQ;
 
@@ -829,7 +900,7 @@
         req->fd = flags;
         req->mode = mode;
 
-        send_req (req);
+        REQ_SEND;
 }
 
 void
@@ -841,7 +912,7 @@
            aio_close     = REQ_CLOSE
            aio_fsync     = REQ_FSYNC
            aio_fdatasync = REQ_FDATASYNC
-	CODE:
+	PPCODE:
 {
         dREQ;
 
@@ -849,7 +920,7 @@
         req->fh = newSVsv (fh);
         req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));
 
-        send_req (req);
+        REQ_SEND (req);
 }
 
 void
@@ -864,7 +935,7 @@
            aio_read  = REQ_READ
            aio_write = REQ_WRITE
 	PROTOTYPE: $$$$$;$
-        CODE:
+        PPCODE:
 {
         aio_req req;
         STRLEN svlen;
@@ -912,7 +983,7 @@
               req->data2ptr = (void *)data;
             }
 
-          send_req (req);
+          REQ_SEND;
         }
 }
 
@@ -924,7 +995,7 @@
         UV	length
         SV *	callback
 	PROTOTYPE: $$$$;$
-        CODE:
+        PPCODE:
 {
 	dREQ;
 
@@ -936,7 +1007,7 @@
         req->offset = in_offset;
         req->length = length;
 
-        send_req (req);
+        REQ_SEND;
 }
 
 void
@@ -946,7 +1017,7 @@
         IV	length
         SV *	callback
 	PROTOTYPE: $$$;$
-        CODE:
+        PPCODE:
 {
 	dREQ;
 
@@ -956,7 +1027,7 @@
         req->offset = offset;
         req->length = length;
 
-        send_req (req);
+        REQ_SEND;
 }
 
 void
@@ -966,14 +1037,14 @@
         ALIAS:
            aio_stat  = REQ_STAT
            aio_lstat = REQ_LSTAT
-	CODE:
+	PPCODE:
 {
 	dREQ;
 
         New (0, req->statdata, 1, Stat_t);
         if (!req->statdata)
           {
-            free_req (req);
+            req_free (req);
             croak ("out of memory during aio_req->statdata allocation");
           }
 
@@ -990,7 +1061,7 @@
             req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
           }
 
-        send_req (req);
+        REQ_SEND;
 }
 
 void
@@ -1001,7 +1072,7 @@
            aio_unlink  = REQ_UNLINK
            aio_rmdir   = REQ_RMDIR
            aio_readdir = REQ_READDIR
-	CODE:
+	PPCODE:
 {
 	dREQ;
 	
@@ -1009,7 +1080,7 @@
 	req->data = newSVsv (pathname);
 	req->dataptr = SvPVbyte_nolen (req->data);
 	
-	send_req (req);
+	REQ_SEND;
 }
 
 void
@@ -1021,7 +1092,7 @@
            aio_link    = REQ_LINK
            aio_symlink = REQ_SYMLINK
            aio_rename  = REQ_RENAME
-	CODE:
+	PPCODE:
 {
 	dREQ;
 	
@@ -1031,7 +1102,7 @@
 	req->data = newSVsv (newpath);
 	req->dataptr = SvPVbyte_nolen (req->data);
 	
-	send_req (req);
+	REQ_SEND;
 }
 
 #if 0
@@ -1108,3 +1179,11 @@
 	OUTPUT:
 	RETVAL
 
+MODULE = IO::AIO                PACKAGE = IO::AIO::CB
+
+void
+cancel (aio_req_ornot req)
+	PROTOTYPE:
+	CODE:
+	req->cancelled = 1;
+