AnyEvent-Fork-Pool/Pool.pm

=head1 NAME

AnyEvent::Fork::Pool - simple process pool manager on top of AnyEvent::Fork

=head1 SYNOPSIS

   use AnyEvent;
   use AnyEvent::Fork::Pool;
   # use AnyEvent::Fork is not needed

   # all parameters with default values
   my $pool = new AnyEvent::Fork::Pool
      "MyWorker::run",

      # pool management
      min        => 0, # minimum # of processes
      max        => 8, # maximum # of processes
      busy_time  => 0, # wait this before starting a new process
      max_idle   => 1, # wait this before killing an idle process
      idle_time  => 1, # at most this many idle processes

      # template process
      template   => AnyEvent::Fork->new, # the template process to use
      require    => [MyWorker::], # module(s) to load
      eval       => "# perl code to execute in template",
      on_destroy => (my $finish = AE::cv),

      # parameters passed to AnyEvent::Fork::RPC
      async      => 0,
      on_error   => sub { die "FATAL: $_[0]\n" },
      on_event   => sub { my @ev = @_ },
      init       => "MyWorker::init",
      serialiser => $AnyEvent::Fork::RPC::STRING_SERIALISER,
   ;

   for (1..10) {
      $pool->call (doit => $_, sub {
         print "MyWorker::run returned @_\n";
      });
   }

   undef $pool;

   $finish->recv;

=head1 DESCRIPTION

This module uses processes created via L<AnyEvent::Fork> and the RPC
protocol implement in L<AnyEvent::Fork::RPC> to create a load-balanced
pool of processes that handles jobs.

Understanding of L<AnyEvent::Fork> is helpful but not critical to be able
to use this module, but a thorough understanding of L<AnyEvent::Fork::RPC>
is, as it defines the actual API that needs to be implemented in the
children.

=head1 EXAMPLES

=head1 API

=over 4

=cut

package AnyEvent::Fork::Pool;

use common::sense;

use Guard ();

use AnyEvent;
use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
use AnyEvent::Fork::RPC;

our $VERSION = 0.1;

=item my $rpc = new AnyEvent::Fork::RPC::pool $function, [key => value...]

=over 4

=item on_error => $cb->($msg)

Called on (fatal) errors, with a descriptive (hopefully) message. If
this callback is not provided, but C<on_event> is, then the C<on_event>
callback is called with the first argument being the string C<error>,
followed by the error message.

If neither handler is provided it prints the error to STDERR and will
start failing badly.

=item on_event => $cb->(...)

Called for every call to the C<AnyEvent::Fork::RPC::event> function in the
child, with the arguments of that function passed to the callback.

Also called on errors when no C<on_error> handler is provided.

=item on_destroy => $cb->()

Called when the C<$rpc> object has been destroyed and all requests have
been successfully handled. This is useful when you queue some requests and
want the child to go away after it has handled them. The problem is that
the parent must not exit either until all requests have been handled, and
this can be accomplished by waiting for this callback.

=item init => $function (default none)

When specified (by name), this function is called in the child as the very
first thing when taking over the process, with all the arguments normally
passed to the C<AnyEvent::Fork::run> function, except the communications
socket.

It can be used to do one-time things in the child such as storing passed
parameters or opening database connections.

It is called very early - before the serialisers are created or the
C<$function> name is resolved into a function reference, so it could be
used to load any modules that provide the serialiser or function. It can
not, however, create events.

=item async => $boolean (default: 0)

The default server used in the child does all I/O blockingly, and only
allows a single RPC call to execute concurrently.

Setting C<async> to a true value switches to another implementation that
uses L<AnyEvent> in the child and allows multiple concurrent RPC calls.

The actual API in the child is documented in the section that describes
the calling semantics of the returned C<$rpc> function.

If you want to pre-load the actual back-end modules to enable memory
sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for
synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.

If you use a template process and want to fork both sync and async
children, then it is permissible to load both modules.

=item serialiser => $string (default: '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })')

All arguments, result data and event data have to be serialised to be
transferred between the processes. For this, they have to be frozen and
thawed in both parent and child processes.

By default, only octet strings can be passed between the processes, which
is reasonably fast and efficient.

For more complicated use cases, you can provide your own freeze and thaw
functions, by specifying a string with perl source code. It's supposed to
return two code references when evaluated: the first receives a list of
perl values and must return an octet string. The second receives the octet
string and must return the original list of values.

If you need an external module for serialisation, then you can either
pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>
or C<require> statement into the serialiser string. Or both.

=back

See the examples section earlier in this document for some actual
examples.

=cut

sub new {
   my ($self, $function, %arg) = @_;

   my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER;
   my $on_event   = delete $arg{on_event};
   my $on_error   = delete $arg{on_error};
   my $on_destroy = delete $arg{on_destroy};
   
   # default for on_error is to on_event, if specified
   $on_error ||= $on_event
               ? sub { $on_event->(error => shift) }
               : sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" };

   # default for on_event is to raise an error
   $on_event ||= sub { $on_error->("event received, but no on_event handler") };

   my ($f, $t) = eval $serialiser; die $@ if $@;

   my (@rcb, %rcb, $fh, $shutdown, $wbuf, $ww);
   my ($rlen, $rbuf, $rw) = 512 - 16;

   my $wcb = sub {
      my $len = syswrite $fh, $wbuf;

      unless (defined $len) {
         if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
            undef $rw; undef $ww; # it ends here
            $on_error->("$!");
         }
      }

      substr $wbuf, 0, $len, "";

      unless (length $wbuf) {
         undef $ww;
         $shutdown and shutdown $fh, 1;
      }
   };

   my $module = "AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync");

   $self->require ($module)
        ->send_arg ($function, $arg{init}, $serialiser)
        ->run ("$module\::run", sub {
      $fh = shift;

      my ($id, $len);
      $rw = AE::io $fh, 0, sub {
         $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
         $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;

         if ($len) {
            while (8 <= length $rbuf) {
               ($id, $len) = unpack "LL", $rbuf;
               8 + $len <= length $rbuf
                  or last;

               my @r = $t->(substr $rbuf, 8, $len);
               substr $rbuf, 0, 8 + $len, "";

               if ($id) {
                  if (@rcb) {
                     (shift @rcb)->(@r);
                  } elsif (my $cb = delete $rcb{$id}) {
                     $cb->(@r);
                  } else {
                     undef $rw; undef $ww;
                     $on_error->("unexpected data from child");
                  }
               } else {
                  $on_event->(@r);
               }
            }
         } elsif (defined $len) {
            undef $rw; undef $ww; # it ends here

            if (@rcb || %rcb) {
               $on_error->("unexpected eof");
            } else {
               $on_destroy->();
            }
         } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
            undef $rw; undef $ww; # it ends here
            $on_error->("read: $!");
         }
      };

      $ww ||= AE::io $fh, 1, $wcb;
   });

   my $guard = Guard::guard {
      $shutdown = 1;
      $ww ||= $fh && AE::io $fh, 1, $wcb;
   };

   my $id;

   $arg{async}
      ? sub {
           $id = ($id == 0xffffffff ? 0 : $id) + 1;
           $id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops

           $rcb{$id} = pop;

           $guard; # keep it alive

           $wbuf .= pack "LL/a*", $id, &$f;
           $ww ||= $fh && AE::io $fh, 1, $wcb;
        }
      : sub {
           push @rcb, pop;

           $guard; # keep it alive

           $wbuf .= pack "L/a*", &$f;
           $ww ||= $fh && AE::io $fh, 1, $wcb;
        }
}

=item $pool->call (..., $cb->(...))

=back

=head1 SEE ALSO

L<AnyEvent::Fork>, to create the processes in the first place.

L<AnyEvent::Fork::RPC>, which implements the RPC protocol and API.

=head1 AUTHOR AND CONTACT INFORMATION

 Marc Lehmann <schmorp@schmorp.de>
 http://software.schmorp.de/pkg/AnyEvent-Fork-Pool

=cut

1

Revision:	1.1
Committed:	Thu Apr 18 14:24:01 2013 UTC (11 years, 1 month ago) by root
Branch:	MAIN
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	=head1 NAME
2
3			AnyEvent::Fork::Pool - simple process pool manager on top of AnyEvent::Fork
4
5			=head1 SYNOPSIS
6
7			use AnyEvent;
8			use AnyEvent::Fork::Pool;
9			# use AnyEvent::Fork is not needed
10
11			# all parameters with default values
12			my $pool = new AnyEvent::Fork::Pool
13			"MyWorker::run",
14
15			# pool management
16			min => 0, # minimum # of processes
17			max => 8, # maximum # of processes
18			busy_time => 0, # wait this before starting a new process
19			max_idle => 1, # wait this before killing an idle process
20			idle_time => 1, # at most this many idle processes
21
22			# template process
23			template => AnyEvent::Fork->new, # the template process to use
24			require => [MyWorker::], # module(s) to load
25			eval => "# perl code to execute in template",
26			on_destroy => (my $finish = AE::cv),
27
28			# parameters passed to AnyEvent::Fork::RPC
29			async => 0,
30			on_error => sub { die "FATAL: $_[0]\n" },
31			on_event => sub { my @ev = @_ },
32			init => "MyWorker::init",
33			serialiser => $AnyEvent::Fork::RPC::STRING_SERIALISER,
34			;
35
36			for (1..10) {
37			$pool->call (doit => $_, sub {
38			print "MyWorker::run returned @_\n";
39			});
40			}
41
42			undef $pool;
43
44			$finish->recv;
45
46			=head1 DESCRIPTION
47
48			This module uses processes created via L<AnyEvent::Fork> and the RPC
49			protocol implement in L<AnyEvent::Fork::RPC> to create a load-balanced
50			pool of processes that handles jobs.
51
52			Understanding of L<AnyEvent::Fork> is helpful but not critical to be able
53			to use this module, but a thorough understanding of L<AnyEvent::Fork::RPC>
54			is, as it defines the actual API that needs to be implemented in the
55			children.
56
57			=head1 EXAMPLES
58
59			=head1 API
60
61			=over 4
62
63			=cut
64
65			package AnyEvent::Fork::Pool;
66
67			use common::sense;
68
69			use Guard ();
70
71			use AnyEvent;
72			use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
73			use AnyEvent::Fork::RPC;
74
75			our $VERSION = 0.1;
76
77			=item my $rpc = new AnyEvent::Fork::RPC::pool $function, [key => value...]
78
79			=over 4
80
81			=item on_error => $cb->($msg)
82
83			Called on (fatal) errors, with a descriptive (hopefully) message. If
84			this callback is not provided, but C<on_event> is, then the C<on_event>
85			callback is called with the first argument being the string C<error>,
86			followed by the error message.
87
88			If neither handler is provided it prints the error to STDERR and will
89			start failing badly.
90
91			=item on_event => $cb->(...)
92
93			Called for every call to the C<AnyEvent::Fork::RPC::event> function in the
94			child, with the arguments of that function passed to the callback.
95
96			Also called on errors when no C<on_error> handler is provided.
97
98			=item on_destroy => $cb->()
99
100			Called when the C<$rpc> object has been destroyed and all requests have
101			been successfully handled. This is useful when you queue some requests and
102			want the child to go away after it has handled them. The problem is that
103			the parent must not exit either until all requests have been handled, and
104			this can be accomplished by waiting for this callback.
105
106			=item init => $function (default none)
107
108			When specified (by name), this function is called in the child as the very
109			first thing when taking over the process, with all the arguments normally
110			passed to the C<AnyEvent::Fork::run> function, except the communications
111			socket.
112
113			It can be used to do one-time things in the child such as storing passed
114			parameters or opening database connections.
115
116			It is called very early - before the serialisers are created or the
117			C<$function> name is resolved into a function reference, so it could be
118			used to load any modules that provide the serialiser or function. It can
119			not, however, create events.
120
121			=item async => $boolean (default: 0)
122
123			The default server used in the child does all I/O blockingly, and only
124			allows a single RPC call to execute concurrently.
125
126			Setting C<async> to a true value switches to another implementation that
127			uses L<AnyEvent> in the child and allows multiple concurrent RPC calls.
128
129			The actual API in the child is documented in the section that describes
130			the calling semantics of the returned C<$rpc> function.
131
132			If you want to pre-load the actual back-end modules to enable memory
133			sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for
134			synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
135
136			If you use a template process and want to fork both sync and async
137			children, then it is permissible to load both modules.
138
139			=item serialiser => $string (default: '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })')
140
141			All arguments, result data and event data have to be serialised to be
142			transferred between the processes. For this, they have to be frozen and
143			thawed in both parent and child processes.
144
145			By default, only octet strings can be passed between the processes, which
146			is reasonably fast and efficient.
147
148			For more complicated use cases, you can provide your own freeze and thaw
149			functions, by specifying a string with perl source code. It's supposed to
150			return two code references when evaluated: the first receives a list of
151			perl values and must return an octet string. The second receives the octet
152			string and must return the original list of values.
153
154			If you need an external module for serialisation, then you can either
155			pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>
156			or C<require> statement into the serialiser string. Or both.
157
158			=back
159
160			See the examples section earlier in this document for some actual
161			examples.
162
163			=cut
164
165			sub new {
166			my ($self, $function, %arg) = @_;
167
168			my $serialiser = delete $arg{serialiser} \|\| $STRING_SERIALISER;
169			my $on_event = delete $arg{on_event};
170			my $on_error = delete $arg{on_error};
171			my $on_destroy = delete $arg{on_destroy};
172
173			# default for on_error is to on_event, if specified
174			$on_error \|\|= $on_event
175			? sub { $on_event->(error => shift) }
176			: sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" };
177
178			# default for on_event is to raise an error
179			$on_event \|\|= sub { $on_error->("event received, but no on_event handler") };
180
181			my ($f, $t) = eval $serialiser; die $@ if $@;
182
183			my (@rcb, %rcb, $fh, $shutdown, $wbuf, $ww);
184			my ($rlen, $rbuf, $rw) = 512 - 16;
185
186			my $wcb = sub {
187			my $len = syswrite $fh, $wbuf;
188
189			unless (defined $len) {
190			if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
191			undef $rw; undef $ww; # it ends here
192			$on_error->("$!");
193			}
194			}
195
196			substr $wbuf, 0, $len, "";
197
198			unless (length $wbuf) {
199			undef $ww;
200			$shutdown and shutdown $fh, 1;
201			}
202			};
203
204			my $module = "AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync");
205
206			$self->require ($module)
207			->send_arg ($function, $arg{init}, $serialiser)
208			->run ("$module\::run", sub {
209			$fh = shift;
210
211			my ($id, $len);
212			$rw = AE::io $fh, 0, sub {
213			$rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
214			$len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
215
216			if ($len) {
217			while (8 <= length $rbuf) {
218			($id, $len) = unpack "LL", $rbuf;
219			8 + $len <= length $rbuf
220			or last;
221
222			my @r = $t->(substr $rbuf, 8, $len);
223			substr $rbuf, 0, 8 + $len, "";
224
225			if ($id) {
226			if (@rcb) {
227			(shift @rcb)->(@r);
228			} elsif (my $cb = delete $rcb{$id}) {
229			$cb->(@r);
230			} else {
231			undef $rw; undef $ww;
232			$on_error->("unexpected data from child");
233			}
234			} else {
235			$on_event->(@r);
236			}
237			}
238			} elsif (defined $len) {
239			undef $rw; undef $ww; # it ends here
240
241			if (@rcb \|\| %rcb) {
242			$on_error->("unexpected eof");
243			} else {
244			$on_destroy->();
245			}
246			} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
247			undef $rw; undef $ww; # it ends here
248			$on_error->("read: $!");
249			}
250			};
251
252			$ww \|\|= AE::io $fh, 1, $wcb;
253			});
254
255			my $guard = Guard::guard {
256			$shutdown = 1;
257			$ww \|\|= $fh && AE::io $fh, 1, $wcb;
258			};
259
260			my $id;
261
262			$arg{async}
263			? sub {
264			$id = ($id == 0xffffffff ? 0 : $id) + 1;
265			$id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops
266
267			$rcb{$id} = pop;
268
269			$guard; # keep it alive
270
271			$wbuf .= pack "LL/a*", $id, &$f;
272			$ww \|\|= $fh && AE::io $fh, 1, $wcb;
273			}
274			: sub {
275			push @rcb, pop;
276
277			$guard; # keep it alive
278
279			$wbuf .= pack "L/a*", &$f;
280			$ww \|\|= $fh && AE::io $fh, 1, $wcb;
281			}
282			}
283
284			=item $pool->call (..., $cb->(...))
285
286			=back
287
288			=head1 SEE ALSO
289
290			L<AnyEvent::Fork>, to create the processes in the first place.
291
292			L<AnyEvent::Fork::RPC>, which implements the RPC protocol and API.
293
294			=head1 AUTHOR AND CONTACT INFORMATION
295
296			Marc Lehmann <schmorp@schmorp.de>
297			http://software.schmorp.de/pkg/AnyEvent-Fork-Pool
298
299			=cut
300
301			1
302