--- IO-AIO/AIO.xs	2006/10/22 10:33:19	1.48
+++ IO-AIO/AIO.xs	2006/10/22 22:14:33	1.51
@@ -66,10 +66,10 @@
 
 typedef struct aio_cb
 {
-  struct aio_cb *grp, *grp_prev, *grp_next;
-
   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;
@@ -103,13 +103,17 @@
 static volatile aio_req reqs, reqe; /* queue start, queue end */
 static volatile aio_req ress, rese; /* queue start, queue end */
 
+static void req_invoke (aio_req req);
 static void req_free (aio_req req);
 
 /* must be called at most once */
 static SV *req_sv (aio_req req, const char *klass)
 {
-  req->self = (SV *)newHV ();
-  sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0);
+  if (!req->self)
+    {
+      req->self = (SV *)newHV ();
+      sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0);
+    }
 
   return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1)));
 }
@@ -124,6 +128,52 @@
   return mg ? (aio_req)mg->mg_ptr : 0;
 }
 
+static void aio_grp_feed (aio_req grp)
+{
+  while (grp->length < grp->fd2)
+    {
+      int old_len = grp->length;
+
+      if (grp->fh2 && SvOK (grp->fh2))
+        {
+          dSP;
+
+          ENTER;
+          SAVETMPS;
+          PUSHMARK (SP);
+          XPUSHs (req_sv (grp, AIO_GRP_KLASS));
+          PUTBACK;
+          call_sv (grp->fh2, G_VOID | G_EVAL);
+          SPAGAIN;
+          FREETMPS;
+          LEAVE;
+        }
+
+      /* stop if no progress has been made */
+      if (old_len == grp->length)
+        {
+          SvREFCNT_dec (grp->fh2);
+          grp->fh2 = 0;
+          break;
+        }
+    }
+}
+
+static void aio_grp_dec (aio_req grp)
+{
+  --grp->length;
+
+  /* call feeder, if applicable */
+  aio_grp_feed (grp);
+
+  /* finish, if done */
+  if (!grp->length && grp->fd)
+    {
+      req_invoke (grp);
+      req_free (grp);
+    }
+}
+
 static void poll_wait ()
 {
   if (nreqs && !ress)
@@ -147,6 +197,7 @@
   errno = req->errorno;
 
   ENTER;
+  SAVETMPS;
   PUSHMARK (SP);
   EXTEND (SP, 1);
 
@@ -195,6 +246,8 @@
         break;
 
       case REQ_GROUP:
+        req->fd = 2; /* mark group as finished */
+
         if (req->data)
           {
             int i;
@@ -219,15 +272,16 @@
   call_sv (req->callback, G_VOID | G_EVAL);
   SPAGAIN;
 
+  FREETMPS;
+  LEAVE;
+
+  errno = errorno;
+
   if (SvTRUE (ERRSV))
     {
       req_free (req);
       croak (0);
     }
-
-  LEAVE;
-
-  errno = errorno;
 }
 
 static void req_free (aio_req req)
@@ -237,14 +291,13 @@
       aio_req grp = req->grp;
 
       /* unlink request */
-      req->grp_next->grp_prev = req->grp_prev;
-      req->grp_prev->grp_next = req->grp_next;
+      if (req->grp_next) req->grp_next->grp_prev = req->grp_prev;
+      if (req->grp_prev) req->grp_prev->grp_next = req->grp_next;
 
-      if (grp->grp_next == grp && grp->fd)
-        {
-          req_invoke (grp);
-          req_free (grp);
-        }
+      if (grp->grp_first == req)
+        grp->grp_first = req->grp_next;
+
+      aio_grp_dec (grp);
     }
 
   if (req->self)
@@ -253,20 +306,11 @@
       SvREFCNT_dec (req->self);
     }
 
-  if (req->data)
-    SvREFCNT_dec (req->data);
-
-  if (req->fh)
-    SvREFCNT_dec (req->fh);
-
-  if (req->fh2)
-    SvREFCNT_dec (req->fh2);
-
-  if (req->statdata)
-    Safefree (req->statdata);
-
-  if (req->callback)
-    SvREFCNT_dec (req->callback);
+  SvREFCNT_dec (req->data);
+  SvREFCNT_dec (req->fh);
+  SvREFCNT_dec (req->fh2);
+  SvREFCNT_dec (req->callback);
+  Safefree (req->statdata);
 
   if (req->type == REQ_READDIR && req->result >= 0)
     free (req->data2ptr);
@@ -282,7 +326,7 @@
     {
       aio_req sub;
 
-      for (sub = req->grp_next; sub != req; sub = sub->grp_next)
+      for (sub = req->grp_first; sub; sub = sub->grp_next)
         req_cancel (sub);
     }
 }
@@ -319,11 +363,11 @@
       if (!req)
         break;
 
-      nreqs--;
+      --nreqs;
 
       if (req->type == REQ_QUIT)
         started--;
-      else if (req->type == REQ_GROUP && req->grp_next != req)
+      else if (req->type == REQ_GROUP && req->length)
         {
           req->fd = 1; /* mark request as delayed */
           continue;
@@ -380,7 +424,7 @@
   while (started < wanted && nreqs >= started)
     start_thread ();
 
-  nreqs++;
+  ++nreqs;
 
   pthread_mutex_lock (&reqlock);
 
@@ -742,7 +786,7 @@
       errno = 0; /* strictly unnecessary */
 
       if (!req->cancelled)
-        switch (req->type)
+        switch (type = req->type) /* remember type for QUIT check */
           {
             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;
@@ -790,6 +834,7 @@
 
       req->next = 0;
 
+      printf ("queue rese %p\n", rese);//D
       if (rese)
         {
           rese->next = req;
@@ -1163,9 +1208,6 @@
 {
 	dREQ;
         req->type = REQ_GROUP;
-        req->grp_next = req;
-        req->grp_prev = req;
-
         req_send (req);
         XPUSHs (req_sv (req, AIO_GRP_KLASS));
 }
@@ -1239,6 +1281,9 @@
 {
 	int i;
 
+        if (grp->fd == 2)
+          croak ("cannot add requests to IO::AIO::GRP after the group finished");
+
 	for (i = 1; i < items; ++i )
           {
             if (GIMME_V != G_VOID)
@@ -1248,29 +1293,60 @@
 
             if (req)
               {
-                req->grp_prev = grp;
-                req->grp_next = grp->grp_next;
-                grp->grp_next->grp_prev = req;
-                grp->grp_next = req;
-
+                ++grp->length;
                 req->grp = grp;
+
+                req->grp_prev = 0;
+                req->grp_next = grp->grp_first;
+
+                if (grp->grp_first)
+                  grp->grp_first->grp_prev = req;
+
+                grp->grp_first = req;
               }
           }
 }
 
 void
 result (aio_req grp, ...)
-	CODE:
+        CODE:
 {
-	int i;
-	AV *av = newAV ();
-
-        if (grp->data)
-          SvREFCNT_dec (grp->data);
+        int i;
+        AV *av = newAV ();
 
-	for (i = 1; i < items; ++i )
+        for (i = 1; i < items; ++i )
           av_push (av, newSVsv (ST (i)));
 
+        SvREFCNT_dec (grp->data);
         grp->data = (SV *)av;
 }
 
+void
+lock (aio_req grp)
+	CODE:
+        ++grp->length;
+
+void
+unlock (aio_req grp)
+	CODE:
+        aio_grp_dec (grp);
+
+void
+feeder_limit (aio_req grp, int limit)
+	CODE:
+        grp->fd2 = limit;
+        aio_grp_feed (grp);
+
+void
+set_feeder (aio_req grp, SV *callback=&PL_sv_undef)
+	CODE:
+{
+        SvREFCNT_dec (grp->fh2);
+        grp->fh2 = newSVsv (callback);
+
+        if (grp->fd2 <= 0)
+          grp->fd2 = 2;
+
+        aio_grp_feed (grp);
+}
+