[ViewVC] Diff of: cvs/AnyEvent-Fork-Pool/Pool.pm

Comparing AnyEvent-Fork-Pool/Pool.pm (file contents):
Revision 1.3 by root, Sat Apr 20 16:38:10 2013 UTC vs.
Revision 1.5 by root, Sat Apr 20 19:33:23 2013 UTC

 my $magic0 = ':t6Z@HK1N%Dx@_7?=~-7NQgWDdAs6a,jFN=wLO0*jD*1%P';
 my $magic1 = '<~53rexz.U`!]X[A235^"fyEoiTF\T~oH1l/N6+Djep9b~bI9`\1x%B~vWO1q*';
 our $VERSION = 0.1;
-=item my $rpc = AnyEvent::Fork::Pool::run $fork, $function, [key => value...]
+=item my $pool = AnyEvent::Fork::Pool::run $fork, $function, [key => value...]
 The traditional way to call it. But it is way cooler to call it in the
 following way:
-=item my $rpc = $fork->AnyEvent::Fork::Pool::run ($function, [key => value...])
+=item my $pool = $fork->AnyEvent::Fork::Pool::run ($function, [key => value...])
 Creates a new pool object with the specified C<$function> as function
 (name) to call for each request. The pool uses the C<$fork> object as the
 template when creating worker processes.
 =item Pool Management
 The pool consists of a certain number of worker processes. These options
 decide how many of these processes exist and when they are started and
-stopp.ed
+stopped.
 =over 4
 =item idle => $count (default: 0)
    my $stop       = $arg{stop}       || 1,
    my $on_event   = $arg{on_event}   || sub { },
    my $on_destroy = $arg{on_destroy};
    my @rpc = (
-      async      => $arg{async},
+      async      =>        $arg{async},
-      init       => $arg{init},
+      init       =>        $arg{init},
-      serialiser => $arg{serialiser},
+      serialiser => delete $arg{serialiser},
-      on_error   => $arg{on_error},
+      on_error   =>        $arg{on_error},
    );
    my (@pool, @queue, $nidle, $start_w, $stop_w, $shutdown);
-   my ($start, $stop, $want_start, $want_stop, $scheduler);
+   my ($start_worker, $stop_worker, $want_start, $want_stop, $scheduler);
    my $destroy_guard = Guard::guard {
       $on_destroy->()
          if $on_destroy;
    };
-   my $busy;#d#
    $template
       ->require ("AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync"))
       ->eval ('
            my ($magic0, $magic1) = @_;
            sub AnyEvent::Fork::Pool::quit() {
               AnyEvent::Fork::RPC::on_event $magic0, "quit", $magic1;
            }
         ', $magic0, $magic1)
-      ->eval (delete $arg{eval});
+      ->eval ($arg{eval});
-   $start = sub {
+   $start_worker = sub {
       my $proc = [0, 0, undef]; # load, index, rpc
-      warn "start a worker\n";#d#
       $proc->[2] = $template
          ->fork
          ->AnyEvent::Fork::RPC::run ($function,
               @rpc,
               on_event => sub {
                  if (@_ == 3 && $_[0] eq $magic0 && $_[2] eq $magic1) {
                     $destroy_guard if 0; # keep it alive
-                    $_[1] eq "quit" and $stop->($proc);
+                    $_[1] eq "quit" and $stop_worker->($proc);
                     return;
                  }
                  &$on_event;
               },
            )
       ;
       ++$nidle;
-      Array::Heap::push_heap @pool, $proc;
+      Array::Heap::push_heap_idx @pool, $proc;
       Scalar::Util::weaken $proc;
    };
-   $stop = sub {
+   $stop_worker = sub {
       my $proc = shift;
       $proc->[0]
          or --$nidle;
    };
    $want_start = sub {
       undef $stop_w;
-      $start_w ||= AE::timer $start, 0, sub {
+      $start_w ||= AE::timer $start, $start, sub {
-         undef $start_w;
+         if (($nidle < $idle || @queue) && @pool < $max) {
-         if (@queue) {
-            $start->();
+            $start_worker->();
             $scheduler->();
+         } else {
+            undef $start_w;
          }
       };
    };
    $want_stop = sub {
-      $stop_w ||= AE::timer $stop, 0, sub {
+      $stop_w ||= AE::timer $stop, $stop, sub {
-         undef $stop_w;
-         $stop->($pool[0])
+         $stop_worker->($pool[0])
             if $nidle;
+         undef $stop_w
+            if $nidle <= $idle;
       };
    };
    $scheduler = sub {
       if (@queue) {
          while (@queue) {
             my $proc = $pool[0];
             if ($proc->[0] < $load) {
-         warn "free $proc $proc->[0]\n";#d#
-               # found free worker
+               # found free worker, increase load
-               $proc->[0]++
+               unless ($proc->[0]++) {
+                  # worker became busy
-                  or --$nidle >= $idle
+                  --$nidle
+                     or undef $stop_w;
-                  or $want_start->();
+                  $want_start->()
+                     if $nidle < $idle && @pool < $max;
+               }
-               Array::Heap::adjust_heap @pool, 0;
+               Array::Heap::adjust_heap_idx @pool, 0;
                my $job = shift @queue;
                my $ocb = pop @$job;
                $proc->[2]->(@$job, sub {
-                  --$busy; warn "busy now $busy\n";#d#
-                  # reduce queue counter
+                  # reduce load
-                  --$pool[$_][0]
+                  --$proc->[0] # worker still busy?
-                     or ++$nidle > $idle
+                     or ++$nidle > $idle # not too many idle processes?
                      or $want_stop->();
-                  Array::Heap::adjust_heap @pool, $_;
+                  Array::Heap::adjust_heap_idx @pool, $proc->[1]
+                     if defined $proc->[1];
                   $scheduler->();
                   &$ocb;
                });
             } else {
-         warn "busy $proc->[0]\n";#d#
-               # all busy, delay
-               $want_start->();
+               $want_start->()
+                  unless @pool >= $max;
                last;
             }
          }
       } elsif ($shutdown) {
          @pool = ();
          undef $start_w;
-         undef $start; # frees $destroy_guard reference
+         undef $start_worker; # frees $destroy_guard reference
-         $stop->($pool[0])
+         $stop_worker->($pool[0])
             while $nidle;
       }
    };
    my $shutdown_guard = Guard::guard {
       $shutdown = 1;
       $scheduler->();
    };
-   $start->()
+   $start_worker->()
       while @pool < $idle;
    sub {
       $shutdown_guard if 0; # keep it alive
-      ++$busy;#d#
-      $start->()
+      $start_worker->()
          unless @pool;
       push @queue, [@_];
       $scheduler->();
    }
 }
-=item $pool->call (..., $cb->(...))
+=item $pool->(..., $cb->(...))
 Call the RPC function of a worker with the given arguments, and when the
-worker is done, call the C<$cb> with the results, like just calling the
+worker is done, call the C<$cb> with the results, just like calling the
 L<AnyEvent::Fork::RPC> object directly.
 If there is no free worker, the call will be queued.
 Note that there can be considerable time between calling this method and

Diff Legend

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

Comparing AnyEvent-Fork-Pool/Pool.pm (file contents): Revision 1.3 by root, Sat Apr 20 16:38:10 2013 UTC vs. Revision 1.5 by root, Sat Apr 20 19:33:23 2013 UTC

Diff Legend

Comparing AnyEvent-Fork-Pool/Pool.pm (file contents):
Revision 1.3 by root, Sat Apr 20 16:38:10 2013 UTC vs.
Revision 1.5 by root, Sat Apr 20 19:33:23 2013 UTC