[ViewVC] Diff of: cvs/IO-AIO/AIO.pm

Comparing IO-AIO/AIO.pm (file contents):
Revision 1.57 by root, Sun Oct 22 01:28:31 2006 UTC vs.
Revision 1.62 by root, Sun Oct 22 21:13:47 2006 UTC

       print "read $_[0] bytes: <$buffer>\n";
    };
 =item aio_move $srcpath, $dstpath, $callback->($status)
+[EXPERIMENTAL due to internal aio_group use]
 Try to move the I<file> (directories not supported as either source or
 destination) from C<$srcpath> to C<$dstpath> and call the callback with
 the C<0> (error) or C<-1> ok.
 This is a composite request that tries to rename(2) the file first. If
 =cut
 sub aio_move($$$) {
    my ($src, $dst, $cb) = @_;
-   my $grp = aio_group;
+   my $grp = aio_group $cb;
    add $grp aio_rename $src, $dst, sub {
       if ($_[0] && $! == EXDEV) {
          add $grp aio_open $src, O_RDONLY, 0, sub {
             if (my $src_fh = $_[0]) {
                            chmod $stat[2] & 07777, $dst_fh;
                            chown $stat[4], $stat[5], $dst_fh;
                            close $dst_fh;
                            add $grp aio_unlink $src, sub {
-                              $cb->($_[0]);
+                              $grp->result ($_[0]);
                            };
                         } else {
                            my $errno = $!;
                            add $grp aio_unlink $dst, sub {
                               $! = $errno;
-                              $cb->(-1);
+                              $grp->result (-1);
                            };
                         }
                      };
                   } else {
-                     $cb->(-1);
+                     $grp->result (-1);
                   }
                },
             } else {
-               $cb->(-1);
+               $grp->result (-1);
             }
          };
       } else {
-         $cb->($_[0]);
+         $grp->result ($_[0]);
       }
    };
    $grp
 }
 The callback a single argument which is either C<undef> or an array-ref
 with the filenames.
 =item aio_scandir $path, $maxreq, $callback->($dirs, $nondirs)
+[EXPERIMENTAL due to internal aio_group use]
 Scans a directory (similar to C<aio_readdir>) but additionally tries to
 separate the entries of directory C<$path> into two sets of names, ones
 you can recurse into (directories or links to them), and ones you cannot
 recurse into (everything else).
-C<aio_scandir> is a composite request that consists of many sub
+C<aio_scandir> is a composite request that creates of many sub requests_
-requests. C<$maxreq> specifies the maximum number of outstanding aio
+C<$maxreq> specifies the maximum number of outstanding aio requests that
-requests that this function generates. If it is C<< <= 0 >>, then a
+this function generates. If it is C<< <= 0 >>, then a suitable default
-suitable default will be chosen (currently 8).
+will be chosen (currently 6).
 On error, the callback is called without arguments, otherwise it receives
 two array-refs with path-relative entry names.
 Example:
 =cut
 sub aio_scandir($$$) {
    my ($path, $maxreq, $cb) = @_;
-   my $grp = aio_group;
+   my $grp = aio_group $cb;
-   $maxreq = 8 if $maxreq <= 0;
+   $maxreq = 6 if $maxreq <= 0;
    # stat once
    add $grp aio_stat $path, sub {
-      return $cb->() if $_[0];
+      return $grp->result () if $_[0];
       my $now = time;
       my $hash1 = join ":", (stat _)[0,1,3,7,9];
       # read the directory entries
       add $grp aio_readdir $path, sub {
          my $entries = shift
-            or return $cb->();
+            or return $grp->result ();
          # stat the dir another time
          add $grp aio_stat $path, sub {
             my $hash2 = join ":", (stat _)[0,1,3,7,9];
                $ndirs = -1;
             } else {
                # if nlink == 2, we are finished
                # on non-posix-fs's, we rely on nlink < 2
                $ndirs = (stat _)[3] - 2
-                  or return $cb->([], $entries);
+                  or return $grp->result ([], $entries);
             }
             # sort into likely dirs and likely nondirs
             # dirs == files without ".", short entries first
             $entries = [map $_->[0],
             my (@dirs, @nondirs);
             my ($statcb, $schedcb);
             my $nreq = 0;
+            my $statgrp = add $grp aio_group;
             $schedcb = sub {
                if (@$entries) {
                   if ($nreq < $maxreq) {
                      my $ent = pop @$entries;
                      $nreq++;
-                     add $grp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
+                     add $statgrp aio_stat "$path/$ent/.", sub { $statcb->($_[0], $ent) };
                   }
                } elsif (!$nreq) {
                   # finished
+                  $statgrp->cancel;
                   undef $statcb;
                   undef $schedcb;
-                  $cb->(\@dirs, \@nondirs) if $cb;
+                  $grp->result (\@dirs, \@nondirs);
-                  undef $cb;
                }
             };
             $statcb = sub {
                my ($status, $entry) = @_;
 callback with the fdatasync result code.
 If this call isn't available because your OS lacks it or it couldn't be
 detected, it will be emulated by calling C<fsync> instead.
-=item aio_group $callback->()
+=item aio_group $callback->(...)
 [EXPERIMENTAL]
 This is a very special aio request: Instead of doing something, it is a
 container for other aio requests, which is useful if you want to bundle
 You add requests by calling the C<add> method with one or more
 C<IO::AIO::REQ> objects:
    $grp->add (aio_unlink "...");
-   add $grp aio_stat "...", sub { ... };
+   add $grp aio_stat "...", sub {
+      $_[0] or return $grp->result ("error");
+      # add another request dynamically, if first succeeded
+      add $grp aio_open "...", sub {
+         $grp->result ("ok");
+      };
+   };
 This makes it very easy to create composite requests (see the source of
 C<aio_move> for an application) that work and feel like simple requests.
+=over 4
-The IO::AIO::GRP objects will be cleaned up during calls to
+=item * The IO::AIO::GRP objects will be cleaned up during calls to
 C<IO::AIO::poll_cb>, just like any other request.
-They can be canceled like any other request. Canceling will cancel not
+=item * They can be canceled like any other request. Canceling will cancel not
-just the request itself, but also all requests it contains.
+only the request itself, but also all requests it contains.
-They can also can also be added to other IO::AIO::GRP objects.
+=item * They can also can also be added to other IO::AIO::GRP objects.
+=item * You must not add requests to a group from within the group callback (or
+any later time).
+=item * This does not harmonise well with C<max_outstanding>, so best do
+not combine C<aio_group> with it. Groups and feeders are recommended for
+this kind of concurrency-limiting.
+=back
 Their lifetime, simplified, looks like this: when they are empty, they
 will finish very quickly. If they contain only requests that are in the
 C<done> state, they will also finish. Otherwise they will continue to
 exist.
 be added, including other groups, as long as you do not create circular
 dependencies.
 Returns all its arguments.
+=item $grp->result (...)
+Set the result value(s) that will be passed to the group callback when all
+subrequests have finished. By default, no argument will be passed.
+=item $grp->set_feeder ($callback->($grp))
+[VERY EXPERIMENTAL]
+Sets a feeder/generator on this group: every group can have an attached
+generator that generates requests if idle. The idea behind this is that,
+although you could just queue as many requests as you want in a group,
+this might starve other requests for a potentially long time.  For
+example, C<aio_scandir> might generate hundreds of thousands C<aio_stat>
+requests, delaying any later requests for a long time.
+To avoid this, and allow incremental generation of requests, you can
+instead a group and set a feeder on it that generates those requests. The
+feeder will be called whenever there are few enough (see C<feeder_limit>,
+below) requests active in the group itself and is expected to queue more
+requests.
+The feeder can queue as many requests as it likes (i.e. C<add> does not
+impose any limits).
+If the feeder does not queue more requests when called, it will be
+automatically removed from the group.
+If the feeder limit is C<0>, it will be set to C<2> automatically.
+Example:
+   # stat all files in @files, but only ever use four aio requests concurrently:
+   my $grp = aio_group sub { print "finished\n" };
+   $grp->feeder_limit (4);
+   $grp->set_feeder (sub {
+      my $file = pop @files
+         or return;
+      add $grp aio_stat $file, sub { ... };
+   });
+=item $grp->feeder_limit ($num)
+Sets the feeder limit for the group: The feeder will be called whenever
+the group contains less than this many requests.
+Setting the limit to C<0> will pause the feeding process.
 =back
 =head2 SUPPORT FUNCTIONS
 =over 4
    IO::AIO::poll_wait, IO::AIO::poll_cb
       if IO::AIO::nreqs;
 =item IO::AIO::min_parallel $nthreads
-Set the minimum number of AIO threads to C<$nthreads>. The current default
+Set the minimum number of AIO threads to C<$nthreads>. The current
-is C<4>, which means four asynchronous operations can be done at one time
+default is C<8>, which means eight asynchronous operations can execute
-(the number of outstanding operations, however, is unlimited).
+concurrently at any one time (the number of outstanding requests,
+however, is unlimited).
 IO::AIO starts threads only on demand, when an AIO request is queued and
 no free thread exists.
-It is recommended to keep the number of threads low, as some Linux
+It is recommended to keep the number of threads relatively low, as some
-kernel versions will scale negatively with the number of threads (higher
+Linux kernel versions will scale negatively with the number of threads
-parallelity => MUCH higher latency). With current Linux 2.6 versions, 4-32
+(higher parallelity => MUCH higher latency). With current Linux 2.6
-threads should be fine.
+versions, 4-32 threads should be fine.
 Under most circumstances you don't need to call this function, as the
 module selects a default that is suitable for low to moderate load.
 =item IO::AIO::max_parallel $nthreads
 Under normal circumstances you don't need to call this function.
 =item $oldnreqs = IO::AIO::max_outstanding $nreqs
+[DEPRECATED]
 Sets the maximum number of outstanding requests to C<$nreqs>. If you
 try to queue up more than this number of requests, the caller will block until
 some requests have been handled.
 The default is very large, so normally there is no practical limit. If you
 queue up many requests in a loop it often improves speed if you set
 this to a relatively low number, such as C<100>.
+This function does not work well together with C<aio_group>'s, and their
+feeder interface is better suited to limiting concurrency, so do not use
+this function.
 Under normal circumstances you don't need to call this function.
 =back
       or return undef;
    *$sym
 }
-min_parallel 4;
+min_parallel 8;
 END {
    max_parallel 0;
 }
 In short: the parent will, after a short pause, continue as if fork had
 not been called, while the child will act as if IO::AIO has not been used
 yet.
+=head2 MEMORY USAGE
+Each aio request uses - depending on your architecture - around 128 bytes
+of memory.  In addition, stat requests need a stat buffer (possibly a few
+hundred bytes). Perl scalars and other data passed into aio requests will
+also be locked.
+This is now awfully much, so queuing lots of requests is not usually a
+problem.
+Each thread needs a stack area which is usually around 16k, sometimes much
+larger, depending on the OS.
 =head1 SEE ALSO
 L<Coro>, L<Linux::AIO> (obsolete).
 =head1 AUTHOR

Diff Legend

-–
+Removed lines
-+
+Added lines
-<
+Changed lines
->
+Changed lines

Comparing IO-AIO/AIO.pm (file contents): Revision 1.57 by root, Sun Oct 22 01:28:31 2006 UTC vs. Revision 1.62 by root, Sun Oct 22 21:13:47 2006 UTC

Diff Legend

Comparing IO-AIO/AIO.pm (file contents):
Revision 1.57 by root, Sun Oct 22 01:28:31 2006 UTC vs.
Revision 1.62 by root, Sun Oct 22 21:13:47 2006 UTC