ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-Fork-Pool/Pool.pm
(Generate patch)

Comparing AnyEvent-Fork-Pool/Pool.pm (file contents):
Revision 1.1 by root, Thu Apr 18 14:24:01 2013 UTC vs.
Revision 1.8 by root, Sun Apr 21 12:28:34 2013 UTC

1=head1 NAME 1=head1 NAME
2 2
3AnyEvent::Fork::Pool - simple process pool manager on top of AnyEvent::Fork 3AnyEvent::Fork::Pool - simple process pool manager on top of AnyEvent::Fork
4
5THE API IS NOT FINISHED, CONSIDER THIS AN ALPHA RELEASE
4 6
5=head1 SYNOPSIS 7=head1 SYNOPSIS
6 8
7 use AnyEvent; 9 use AnyEvent;
8 use AnyEvent::Fork::Pool; 10 use AnyEvent::Fork::Pool;
9 # use AnyEvent::Fork is not needed 11 # use AnyEvent::Fork is not needed
10 12
11 # all parameters with default values 13 # all possible parameters shown, with default values
12 my $pool = new AnyEvent::Fork::Pool 14 my $pool = AnyEvent::Fork
13 "MyWorker::run", 15 ->new
16 ->require ("MyWorker")
17 ->AnyEvent::Fork::Pool::run (
18 "MyWorker::run", # the worker function
14 19
15 # pool management 20 # pool management
16 min => 0, # minimum # of processes
17 max => 8, # maximum # of processes 21 max => 4, # absolute maximum # of processes
22 idle => 0, # minimum # of idle processes
23 load => 2, # queue at most this number of jobs per process
18 busy_time => 0, # wait this before starting a new process 24 start => 0.1, # wait this many seconds before starting a new process
19 max_idle => 1, # wait this before killing an idle process 25 stop => 10, # wait this many seconds before stopping an idle process
20 idle_time => 1, # at most this many idle processes 26 on_destroy => (my $finish = AE::cv), # called when object is destroyed
21 27
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 28 # parameters passed to AnyEvent::Fork::RPC
29 async => 0, 29 async => 0,
30 on_error => sub { die "FATAL: $_[0]\n" }, 30 on_error => sub { die "FATAL: $_[0]\n" },
31 on_event => sub { my @ev = @_ }, 31 on_event => sub { my @ev = @_ },
32 init => "MyWorker::init", 32 init => "MyWorker::init",
33 serialiser => $AnyEvent::Fork::RPC::STRING_SERIALISER, 33 serialiser => $AnyEvent::Fork::RPC::STRING_SERIALISER,
34 ; 34 );
35 35
36 for (1..10) { 36 for (1..10) {
37 $pool->call (doit => $_, sub { 37 $pool->(doit => $_, sub {
38 print "MyWorker::run returned @_\n"; 38 print "MyWorker::run returned @_\n";
39 }); 39 });
40 } 40 }
41 41
42 undef $pool; 42 undef $pool;
50pool of processes that handles jobs. 50pool of processes that handles jobs.
51 51
52Understanding of L<AnyEvent::Fork> is helpful but not critical to be able 52Understanding of L<AnyEvent::Fork> is helpful but not critical to be able
53to use this module, but a thorough understanding of L<AnyEvent::Fork::RPC> 53to use this module, but a thorough understanding of L<AnyEvent::Fork::RPC>
54is, as it defines the actual API that needs to be implemented in the 54is, as it defines the actual API that needs to be implemented in the
55children. 55worker processes.
56 56
57=head1 EXAMPLES 57=head1 EXAMPLES
58 58
59=head1 API 59=head1 PARENT USAGE
60
61To create a pool, you first have to create a L<AnyEvent::Fork> object -
62this object becomes your template process. Whenever a new worker process
63is needed, it is forked from this template process. Then you need to
64"hand off" this template process to the C<AnyEvent::Fork::Pool> module by
65calling its run method on it:
66
67 my $template = AnyEvent::Fork
68 ->new
69 ->require ("SomeModule", "MyWorkerModule");
70
71 my $pool = $template->AnyEvent::Fork::Pool::run ("MyWorkerModule::myfunction");
72
73The pool "object" is not a regular Perl object, but a code reference that
74you can call and that works roughly like calling the worker function
75directly, except that it returns nothing but instead you need to specify a
76callback to be invoked once results are in:
77
78 $pool->(1, 2, 3, sub { warn "myfunction(1,2,3) returned @_" });
60 79
61=over 4 80=over 4
62 81
63=cut 82=cut
64 83
65package AnyEvent::Fork::Pool; 84package AnyEvent::Fork::Pool;
66 85
67use common::sense; 86use common::sense;
68 87
88use Scalar::Util ();
89
69use Guard (); 90use Guard ();
91use Array::Heap ();
70 92
71use AnyEvent; 93use AnyEvent;
72use AnyEvent::Fork; # we don't actually depend on it, this is for convenience 94use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
73use AnyEvent::Fork::RPC; 95use AnyEvent::Fork::RPC;
74 96
97# these are used for the first and last argument of events
98# in the hope of not colliding. yes, I don't like it either,
99# but didn't come up with an obviously better alternative.
100my $magic0 = ':t6Z@HK1N%Dx@_7?=~-7NQgWDdAs6a,jFN=wLO0*jD*1%P';
101my $magic1 = '<~53rexz.U`!]X[A235^"fyEoiTF\T~oH1l/N6+Djep9b~bI9`\1x%B~vWO1q*';
102
75our $VERSION = 0.1; 103our $VERSION = 0.1;
76 104
77=item my $rpc = new AnyEvent::Fork::RPC::pool $function, [key => value...] 105=item my $pool = AnyEvent::Fork::Pool::run $fork, $function, [key => value...]
106
107The traditional way to call the pool creation function. But it is way
108cooler to call it in the following way:
109
110=item my $pool = $fork->AnyEvent::Fork::Pool::run ($function, [key => value...])
111
112Creates a new pool object with the specified C<$function> as function
113(name) to call for each request. The pool uses the C<$fork> object as the
114template when creating worker processes.
115
116You can supply your own template process, or tell C<AnyEvent::Fork::Pool>
117to create one.
118
119A relatively large number of key/value pairs can be specified to influence
120the behaviour. They are grouped into the categories "pool management",
121"template process" and "rpc parameters".
78 122
79=over 4 123=over 4
80 124
81=item on_error => $cb->($msg) 125=item Pool Management
82 126
83Called on (fatal) errors, with a descriptive (hopefully) message. If 127The pool consists of a certain number of worker processes. These options
84this callback is not provided, but C<on_event> is, then the C<on_event> 128decide how many of these processes exist and when they are started and
85callback is called with the first argument being the string C<error>, 129stopped.
86followed by the error message.
87 130
88If neither handler is provided it prints the error to STDERR and will 131The worker pool is dynamically resized, according to (perceived :)
89start failing badly. 132load. The minimum size is given by the C<idle> parameter and the maximum
133size is given by the C<max> parameter. A new worker is started every
134C<start> seconds at most, and an idle worker is stopped at most every
135C<stop> second.
90 136
91=item on_event => $cb->(...) 137You can specify the amount of jobs sent to a worker concurrently using the
138C<load> parameter.
92 139
93Called for every call to the C<AnyEvent::Fork::RPC::event> function in the 140=over 4
94child, with the arguments of that function passed to the callback.
95 141
96Also called on errors when no C<on_error> handler is provided. 142=item idle => $count (default: 0)
97 143
98=item on_destroy => $cb->() 144The minimum amount of idle processes in the pool - when there are fewer
145than this many idle workers, C<AnyEvent::Fork::Pool> will try to start new
146ones, subject to the limits set by C<max> and C<start>.
99 147
100Called when the C<$rpc> object has been destroyed and all requests have 148This is also the initial amount of workers in the pool. The default of
101been successfully handled. This is useful when you queue some requests and 149zero means that the pool starts empty and can shrink back to zero workers
102want the child to go away after it has handled them. The problem is that 150over time.
103the parent must not exit either until all requests have been handled, and
104this can be accomplished by waiting for this callback.
105 151
106=item init => $function (default none) 152=item max => $count (default: 4)
107 153
108When specified (by name), this function is called in the child as the very 154The maximum number of processes in the pool, in addition to the template
109first thing when taking over the process, with all the arguments normally 155process. C<AnyEvent::Fork::Pool> will never have more than this number of
110passed to the C<AnyEvent::Fork::run> function, except the communications 156worker processes, although there can be more temporarily when a worker is
111socket. 157shut down and hasn't exited yet.
112 158
113It can be used to do one-time things in the child such as storing passed 159=item load => $count (default: 2)
114parameters or opening database connections.
115 160
116It is called very early - before the serialisers are created or the 161The maximum number of concurrent jobs sent to a single worker process.
117C<$function> name is resolved into a function reference, so it could be 162
118used to load any modules that provide the serialiser or function. It can 163Jobs that cannot be sent to a worker immediately (because all workers are
119not, however, create events. 164busy) will be queued until a worker is available.
165
166Setting this low improves latency. For example, at C<1>, every job that
167is sent to a worker is sent to a completely idle worker that doesn't run
168any other jobs. The downside is that throughput is reduced - a worker that
169finishes a job needs to wait for a new job from the parent.
170
171The default of C<2> is usually a good compromise.
172
173=item start => $seconds (default: 0.1)
174
175When there are fewer than C<idle> workers (or all workers are completely
176busy), then a timer is started. If the timer elapses and there are still
177jobs that cannot be queued to a worker, a new worker is started.
178
179This sets the minimum time that all workers must be busy before a new
180worker is started. Or, put differently, the minimum delay between starting
181new workers.
182
183The delay is small by default, which means new workers will be started
184relatively quickly. A delay of C<0> is possible, and ensures that the pool
185will grow as quickly as possible under load.
186
187Non-zero values are useful to avoid "exploding" a pool because a lot of
188jobs are queued in an instant.
189
190Higher values are often useful to improve efficiency at the cost of
191latency - when fewer processes can do the job over time, starting more and
192more is not necessarily going to help.
193
194=item stop => $seconds (default: 10)
195
196When a worker has no jobs to execute it becomes idle. An idle worker that
197hasn't executed a job within this amount of time will be stopped, unless
198the other parameters say otherwise.
199
200Setting this to a very high value means that workers stay around longer,
201even when they have nothing to do, which can be good as they don't have to
202be started on the netx load spike again.
203
204Setting this to a lower value can be useful to avoid memory or simply
205process table wastage.
206
207Usually, setting this to a time longer than the time between load spikes
208is best - if you expect a lot of requests every minute and little work
209in between, setting this to longer than a minute avoids having to stop
210and start workers. On the other hand, you have to ask yourself if letting
211workers run idle is a good use of your resources. Try to find a good
212balance between resource usage of your workers and the time to start new
213workers - the processes created by L<AnyEvent::Fork> itself is fats at
214creating workers while not using much memory for them, so most of the
215overhead is likely from your own code.
216
217=item on_destroy => $callback->() (default: none)
218
219When a pool object goes out of scope, the outstanding requests are still
220handled till completion. Only after handling all jobs will the workers
221be destroyed (and also the template process if it isn't referenced
222otherwise).
223
224To find out when a pool I<really> has finished its work, you can set this
225callback, which will be called when the pool has been destroyed.
226
227=back
228
229=item AnyEvent::Fork::RPC Parameters
230
231These parameters are all passed more or less directly to
232L<AnyEvent::Fork::RPC>. They are only briefly mentioned here, for
233their full documentation please refer to the L<AnyEvent::Fork::RPC>
234documentation. Also, the default values mentioned here are only documented
235as a best effort - the L<AnyEvent::Fork::RPC> documentation is binding.
236
237=over 4
120 238
121=item async => $boolean (default: 0) 239=item async => $boolean (default: 0)
122 240
123The default server used in the child does all I/O blockingly, and only 241Whether to use the synchronous or asynchronous RPC backend.
124allows a single RPC call to execute concurrently.
125 242
126Setting C<async> to a true value switches to another implementation that 243=item on_error => $callback->($message) (default: die with message)
127uses L<AnyEvent> in the child and allows multiple concurrent RPC calls.
128 244
129The actual API in the child is documented in the section that describes 245The callback to call on any (fatal) errors.
130the calling semantics of the returned C<$rpc> function.
131 246
132If you want to pre-load the actual back-end modules to enable memory 247=item on_event => $callback->(...) (default: C<sub { }>, unlike L<AnyEvent::Fork::RPC>)
133sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for
134synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
135 248
136If you use a template process and want to fork both sync and async 249The callback to invoke on events.
137children, then it is permissible to load both modules.
138 250
139=item serialiser => $string (default: '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })') 251=item init => $initfunction (default: none)
140 252
141All arguments, result data and event data have to be serialised to be 253The function to call in the child, once before handling requests.
142transferred between the processes. For this, they have to be frozen and
143thawed in both parent and child processes.
144 254
145By default, only octet strings can be passed between the processes, which 255=item serialiser => $serialiser (defailt: $AnyEvent::Fork::RPC::STRING_SERIALISER)
146is reasonably fast and efficient.
147 256
148For more complicated use cases, you can provide your own freeze and thaw 257The serialiser to use.
149functions, by specifying a string with perl source code. It's supposed to
150return two code references when evaluated: the first receives a list of
151perl values and must return an octet string. The second receives the octet
152string and must return the original list of values.
153
154If you need an external module for serialisation, then you can either
155pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>
156or C<require> statement into the serialiser string. Or both.
157 258
158=back 259=back
159 260
160See the examples section earlier in this document for some actual 261=back
161examples.
162 262
163=cut 263=cut
164 264
165sub new { 265sub run {
166 my ($self, $function, %arg) = @_; 266 my ($template, $function, %arg) = @_;
167 267
168 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER; 268 my $max = $arg{max} || 4;
269 my $idle = $arg{idle} || 0,
270 my $load = $arg{load} || 2,
271 my $start = $arg{start} || 0.1,
272 my $stop = $arg{stop} || 10,
169 my $on_event = delete $arg{on_event}; 273 my $on_event = $arg{on_event} || sub { },
170 my $on_error = delete $arg{on_error};
171 my $on_destroy = delete $arg{on_destroy}; 274 my $on_destroy = $arg{on_destroy};
275
276 my @rpc = (
277 async => $arg{async},
278 init => $arg{init},
279 serialiser => delete $arg{serialiser},
280 on_error => $arg{on_error},
172 281 );
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 282
178 # default for on_event is to raise an error 283 my (@pool, @queue, $nidle, $start_w, $stop_w, $shutdown);
179 $on_event ||= sub { $on_error->("event received, but no on_event handler") }; 284 my ($start_worker, $stop_worker, $want_start, $want_stop, $scheduler);
180 285
181 my ($f, $t) = eval $serialiser; die $@ if $@; 286 my $destroy_guard = Guard::guard {
287 $on_destroy->()
288 if $on_destroy;
289 };
182 290
183 my (@rcb, %rcb, $fh, $shutdown, $wbuf, $ww); 291 $template
184 my ($rlen, $rbuf, $rw) = 512 - 16; 292 ->require ("AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync"))
185 293 ->eval ('
186 my $wcb = sub { 294 my ($magic0, $magic1) = @_;
187 my $len = syswrite $fh, $wbuf; 295 sub AnyEvent::Fork::Pool::retire() {
188 296 AnyEvent::Fork::RPC::event $magic0, "quit", $magic1;
189 unless (defined $len) {
190 if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
191 undef $rw; undef $ww; # it ends here
192 $on_error->("$!");
193 } 297 }
194 } 298 ', $magic0, $magic1)
195
196 substr $wbuf, 0, $len, "";
197
198 unless (length $wbuf) {
199 undef $ww;
200 $shutdown and shutdown $fh, 1;
201 }
202 }; 299 ;
203 300
204 my $module = "AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync"); 301 $start_worker = sub {
302 my $proc = [0, 0, undef]; # load, index, rpc
205 303
206 $self->require ($module) 304 $proc->[2] = $template
207 ->send_arg ($function, $arg{init}, $serialiser) 305 ->fork
208 ->run ("$module\::run", sub { 306 ->AnyEvent::Fork::RPC::run ($function,
209 $fh = shift; 307 @rpc,
308 on_event => sub {
309 if (@_ == 3 && $_[0] eq $magic0 && $_[2] eq $magic1) {
310 $destroy_guard if 0; # keep it alive
210 311
211 my ($id, $len); 312 $_[1] eq "quit" and $stop_worker->($proc);
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; 313 return;
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 } 314 }
234 } else { 315
235 $on_event->(@r); 316 &$on_event;
236 } 317 },
237 } 318 )
238 } elsif (defined $len) { 319 ;
239 undef $rw; undef $ww; # it ends here
240 320
241 if (@rcb || %rcb) { 321 ++$nidle;
242 $on_error->("unexpected eof"); 322 Array::Heap::push_heap_idx @pool, $proc;
323
324 Scalar::Util::weaken $proc;
325 };
326
327 $stop_worker = sub {
328 my $proc = shift;
329
330 $proc->[0]
331 or --$nidle;
332
333 Array::Heap::splice_heap_idx @pool, $proc->[1]
334 if defined $proc->[1];
335
336 @$proc = 0; # tell others to leave it be
337 };
338
339 $want_start = sub {
340 undef $stop_w;
341
342 $start_w ||= AE::timer $start, $start, sub {
343 if (($nidle < $idle || @queue) && @pool < $max) {
344 $start_worker->();
345 $scheduler->();
243 } else { 346 } else {
244 $on_destroy->(); 347 undef $start_w;
245 }
246 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
247 undef $rw; undef $ww; # it ends here
248 $on_error->("read: $!");
249 } 348 }
250 }; 349 };
251
252 $ww ||= AE::io $fh, 1, $wcb;
253 }); 350 };
254 351
352 $want_stop = sub {
353 $stop_w ||= AE::timer $stop, $stop, sub {
354 $stop_worker->($pool[0])
355 if $nidle;
356
357 undef $stop_w
358 if $nidle <= $idle;
359 };
360 };
361
362 $scheduler = sub {
363 if (@queue) {
364 while (@queue) {
365 @pool or $start_worker->();
366
367 my $proc = $pool[0];
368
369 if ($proc->[0] < $load) {
370 # found free worker, increase load
371 unless ($proc->[0]++) {
372 # worker became busy
373 --$nidle
374 or undef $stop_w;
375
376 $want_start->()
377 if $nidle < $idle && @pool < $max;
378 }
379
380 Array::Heap::adjust_heap_idx @pool, 0;
381
382 my $job = shift @queue;
383 my $ocb = pop @$job;
384
385 $proc->[2]->(@$job, sub {
386 # reduce load
387 --$proc->[0] # worker still busy?
388 or ++$nidle > $idle # not too many idle processes?
389 or $want_stop->();
390
391 Array::Heap::adjust_heap_idx @pool, $proc->[1]
392 if defined $proc->[1];
393
394 &$ocb;
395
396 $scheduler->();
397 });
398 } else {
399 $want_start->()
400 unless @pool >= $max;
401
402 last;
403 }
404 }
405 } elsif ($shutdown) {
406 @pool = ();
407 undef $start_w;
408 undef $start_worker; # frees $destroy_guard reference
409
410 $stop_worker->($pool[0])
411 while $nidle;
412 }
413 };
414
255 my $guard = Guard::guard { 415 my $shutdown_guard = Guard::guard {
256 $shutdown = 1; 416 $shutdown = 1;
257 $ww ||= $fh && AE::io $fh, 1, $wcb; 417 $scheduler->();
418 };
419
420 $start_worker->()
421 while @pool < $idle;
422
423 sub {
424 $shutdown_guard if 0; # keep it alive
425
426 $start_worker->()
427 unless @pool;
428
429 push @queue, [@_];
430 $scheduler->();
258 }; 431 }
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} 432}
283 433
284=item $pool->call (..., $cb->(...)) 434=item $pool->(..., $cb->(...))
435
436Call the RPC function of a worker with the given arguments, and when the
437worker is done, call the C<$cb> with the results, just like calling the
438RPC object durectly - see the L<AnyEvent::Fork::RPC> documentation for
439details on the RPC API.
440
441If there is no free worker, the call will be queued until a worker becomes
442available.
443
444Note that there can be considerable time between calling this method and
445the call actually being executed. During this time, the parameters passed
446to this function are effectively read-only - modifying them after the call
447and before the callback is invoked causes undefined behaviour.
448
449=cut
285 450
286=back 451=back
452
453=head1 CHILD USAGE
454
455In addition to the L<AnyEvent::Fork::RPC> API, this module implements one
456more child-side function:
457
458=over 4
459
460=item AnyEvent::Fork::Pool::retire ()
461
462This function sends an event to the parent process to request retirement:
463the worker is removed from the pool and no new jobs will be sent to it,
464but it has to handle the jobs that are already queued.
465
466The parentheses are part of the syntax: the function usually isn't defined
467when you compile your code (because that happens I<before> handing the
468template process over to C<AnyEvent::Fork::Pool::run>, so you need the
469empty parentheses to tell Perl that the function is indeed a function.
470
471Retiring a worker can be useful to gracefully shut it down when the worker
472deems this useful. For example, after executing a job, one could check
473the process size or the number of jobs handled so far, and if either is
474too high, the worker could ask to get retired, to avoid memory leaks to
475accumulate.
476
477=back
478
479=head1 POOL PARAMETERS RECIPES
480
481This section describes some recipes for pool paramaters. These are mostly
482meant for the synchronous RPC backend, as the asynchronous RPC backend
483changes the rules considerably, making workers themselves responsible for
484their scheduling.
485
486=over 4
487
488=item low latency - set load = 1
489
490If you need a deterministic low latency, you should set the C<load>
491parameter to C<1>. This ensures that never more than one job is sent to
492each worker. This avoids having to wait for a previous job to finish.
493
494This makes most sense with the synchronous (default) backend, as the
495asynchronous backend can handle multiple requests concurrently.
496
497=item lowest latency - set load = 1 and idle = max
498
499To achieve the lowest latency, you additionally should disable any dynamic
500resizing of the pool by setting C<idle> to the same value as C<max>.
501
502=item high throughput, cpu bound jobs - set load >= 2, max = #cpus
503
504To get high throughput with cpu-bound jobs, you should set the maximum
505pool size to the number of cpus in your system, and C<load> to at least
506C<2>, to make sure there can be another job waiting for the worker when it
507has finished one.
508
509The value of C<2> for C<load> is the minimum value that I<can> achieve
510100% throughput, but if your parent process itself is sometimes busy, you
511might need higher values. Also there is a limit on the amount of data that
512can be "in flight" to the worker, so if you send big blobs of data to your
513worker, C<load> might have much less of an effect.
514
515=item high throughput, I/O bound jobs - set load >= 2, max = 1, or very high
516
517When your jobs are I/O bound, using more workers usually boils down to
518higher throughput, depending very much on your actual workload - sometimes
519having only one worker is best, for example, when you read or write big
520files at maixmum speed, as a second worker will increase seek times.
521
522=back
523
524=head1 EXCEPTIONS
525
526The same "policy" as with L<AnyEvent::Fork::RPC> applies - exceptins will
527not be caught, and exceptions in both worker and in callbacks causes
528undesirable or undefined behaviour.
287 529
288=head1 SEE ALSO 530=head1 SEE ALSO
289 531
290L<AnyEvent::Fork>, to create the processes in the first place. 532L<AnyEvent::Fork>, to create the processes in the first place.
291 533

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines