--- IO-AIO/AIO.xs	2006/10/22 10:33:19	1.48
+++ IO-AIO/AIO.xs	2006/10/22 13:33:28	1.49
@@ -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;
@@ -108,8 +108,11 @@
 /* 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 +127,37 @@
   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 poll_wait ()
 {
   if (nreqs && !ress)
@@ -147,6 +181,7 @@
   errno = req->errorno;
 
   ENTER;
+  SAVETMPS;
   PUSHMARK (SP);
   EXTEND (SP, 1);
 
@@ -195,6 +230,8 @@
         break;
 
       case REQ_GROUP:
+        req->fd = 2; /* mark group as finished */
+
         if (req->data)
           {
             int i;
@@ -225,6 +262,7 @@
       croak (0);
     }
 
+  FREETMPS;
   LEAVE;
 
   errno = errorno;
@@ -237,10 +275,19 @@
       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_first == req)
+        grp->grp_first = req->grp_next;
+
+      --grp->length;
 
-      if (grp->grp_next == grp && grp->fd)
+      /* call feeder, if applicable */
+      aio_grp_feed (grp);
+
+      /* finish, if done */
+      if (!grp->length && grp->fd)
         {
           req_invoke (grp);
           req_free (grp);
@@ -253,20 +300,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 +320,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);
     }
 }
@@ -323,7 +361,7 @@
 
       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;
@@ -1163,9 +1201,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 +1274,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,12 +1286,16 @@
 
             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;
               }
           }
 }
@@ -1265,12 +1307,29 @@
 	int i;
 	AV *av = newAV ();
 
-        if (grp->data)
-          SvREFCNT_dec (grp->data);
-
 	for (i = 1; i < items; ++i )
           av_push (av, newSVsv (ST (i)));
 
+        SvREFCNT_dec (grp->data);
         grp->data = (SV *)av;
 }
 
+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);
+}
+