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.5 by root, Sat Apr 20 19:33:23 2013 UTC vs.
Revision 1.13 by root, Sun Apr 28 14:27:31 2013 UTC

6 6
7 use AnyEvent; 7 use AnyEvent;
8 use AnyEvent::Fork::Pool; 8 use AnyEvent::Fork::Pool;
9 # use AnyEvent::Fork is not needed 9 # use AnyEvent::Fork is not needed
10 10
11 # all parameters with default values 11 # all possible parameters shown, with default values
12 my $pool = AnyEvent::Fork 12 my $pool = AnyEvent::Fork
13 ->new 13 ->new
14 ->require ("MyWorker") 14 ->require ("MyWorker")
15 ->AnyEvent::Fork::Pool::run ( 15 ->AnyEvent::Fork::Pool::run (
16 "MyWorker::run", # the worker function 16 "MyWorker::run", # the worker function
17 17
18 # pool management 18 # pool management
19 max => 4, # absolute maximum # of processes 19 max => 4, # absolute maximum # of processes
20 idle => 2, # minimum # of idle processes 20 idle => 0, # minimum # of idle processes
21 load => 2, # queue at most this number of jobs per process 21 load => 2, # queue at most this number of jobs per process
22 start => 0.1, # wait this many seconds before starting a new process 22 start => 0.1, # wait this many seconds before starting a new process
23 stop => 1, # wait this many seconds before stopping an idle process 23 stop => 10, # wait this many seconds before stopping an idle process
24 on_destroy => (my $finish = AE::cv), # called when object is destroyed 24 on_destroy => (my $finish = AE::cv), # called when object is destroyed
25 25
26 # parameters passed to AnyEvent::Fork::RPC 26 # parameters passed to AnyEvent::Fork::RPC
27 async => 0, 27 async => 0,
28 on_error => sub { die "FATAL: $_[0]\n" }, 28 on_error => sub { die "FATAL: $_[0]\n" },
41 41
42 $finish->recv; 42 $finish->recv;
43 43
44=head1 DESCRIPTION 44=head1 DESCRIPTION
45 45
46This module uses processes created via L<AnyEvent::Fork> and the RPC 46This module uses processes created via L<AnyEvent::Fork> (or
47protocol implement in L<AnyEvent::Fork::RPC> to create a load-balanced 47L<AnyEvent::Fork::Remote>) and the RPC protocol implement in
48pool of processes that handles jobs. 48L<AnyEvent::Fork::RPC> to create a load-balanced pool of processes that
49handles jobs.
49 50
50Understanding of L<AnyEvent::Fork> is helpful but not critical to be able 51Understanding of L<AnyEvent::Fork> is helpful but not critical to be able
51to use this module, but a thorough understanding of L<AnyEvent::Fork::RPC> 52to use this module, but a thorough understanding of L<AnyEvent::Fork::RPC>
52is, as it defines the actual API that needs to be implemented in the 53is, as it defines the actual API that needs to be implemented in the
53children. 54worker processes.
54
55=head1 EXAMPLES
56 55
57=head1 PARENT USAGE 56=head1 PARENT USAGE
57
58To create a pool, you first have to create a L<AnyEvent::Fork> object -
59this object becomes your template process. Whenever a new worker process
60is needed, it is forked from this template process. Then you need to
61"hand off" this template process to the C<AnyEvent::Fork::Pool> module by
62calling its run method on it:
63
64 my $template = AnyEvent::Fork
65 ->new
66 ->require ("SomeModule", "MyWorkerModule");
67
68 my $pool = $template->AnyEvent::Fork::Pool::run ("MyWorkerModule::myfunction");
69
70The pool "object" is not a regular Perl object, but a code reference that
71you can call and that works roughly like calling the worker function
72directly, except that it returns nothing but instead you need to specify a
73callback to be invoked once results are in:
74
75 $pool->(1, 2, 3, sub { warn "myfunction(1,2,3) returned @_" });
58 76
59=over 4 77=over 4
60 78
61=cut 79=cut
62 80
68 86
69use Guard (); 87use Guard ();
70use Array::Heap (); 88use Array::Heap ();
71 89
72use AnyEvent; 90use AnyEvent;
91# explicit version on next line, as some cpan-testers test with the 0.1 version,
92# ignoring dependencies, and this line will at least give a clear indication of that.
73use AnyEvent::Fork; # we don't actually depend on it, this is for convenience 93use AnyEvent::Fork 0.6; # we don't actually depend on it, this is for convenience
74use AnyEvent::Fork::RPC; 94use AnyEvent::Fork::RPC;
75 95
76# these are used for the first and last argument of events 96# these are used for the first and last argument of events
77# in the hope of not colliding. yes, I don't like it either, 97# in the hope of not colliding. yes, I don't like it either,
78# but didn't come up with an obviously better alternative. 98# but didn't come up with an obviously better alternative.
79my $magic0 = ':t6Z@HK1N%Dx@_7?=~-7NQgWDdAs6a,jFN=wLO0*jD*1%P'; 99my $magic0 = ':t6Z@HK1N%Dx@_7?=~-7NQgWDdAs6a,jFN=wLO0*jD*1%P';
80my $magic1 = '<~53rexz.U`!]X[A235^"fyEoiTF\T~oH1l/N6+Djep9b~bI9`\1x%B~vWO1q*'; 100my $magic1 = '<~53rexz.U`!]X[A235^"fyEoiTF\T~oH1l/N6+Djep9b~bI9`\1x%B~vWO1q*';
81 101
82our $VERSION = 0.1; 102our $VERSION = 1.1;
83 103
84=item my $pool = AnyEvent::Fork::Pool::run $fork, $function, [key => value...] 104=item my $pool = AnyEvent::Fork::Pool::run $fork, $function, [key => value...]
85 105
86The traditional way to call it. But it is way cooler to call it in the 106The traditional way to call the pool creation function. But it is way
87following way: 107cooler to call it in the following way:
88 108
89=item my $pool = $fork->AnyEvent::Fork::Pool::run ($function, [key => value...]) 109=item my $pool = $fork->AnyEvent::Fork::Pool::run ($function, [key => value...])
90 110
91Creates a new pool object with the specified C<$function> as function 111Creates a new pool object with the specified C<$function> as function
92(name) to call for each request. The pool uses the C<$fork> object as the 112(name) to call for each request. The pool uses the C<$fork> object as the
105 125
106The pool consists of a certain number of worker processes. These options 126The pool consists of a certain number of worker processes. These options
107decide how many of these processes exist and when they are started and 127decide how many of these processes exist and when they are started and
108stopped. 128stopped.
109 129
130The worker pool is dynamically resized, according to (perceived :)
131load. The minimum size is given by the C<idle> parameter and the maximum
132size is given by the C<max> parameter. A new worker is started every
133C<start> seconds at most, and an idle worker is stopped at most every
134C<stop> second.
135
136You can specify the amount of jobs sent to a worker concurrently using the
137C<load> parameter.
138
110=over 4 139=over 4
111 140
112=item idle => $count (default: 0) 141=item idle => $count (default: 0)
113 142
114The minimum amount of idle processes in the pool - when there are fewer 143The minimum amount of idle processes in the pool - when there are fewer
115than this many idle workers, C<AnyEvent::Fork::Pool> will try to start new 144than this many idle workers, C<AnyEvent::Fork::Pool> will try to start new
116ones, subject to C<max> and C<start>. 145ones, subject to the limits set by C<max> and C<start>.
117 146
118This is also the initial/minimum amount of workers in the pool. The 147This is also the initial amount of workers in the pool. The default of
119default of zero means that the pool starts empty and can shrink back to 148zero means that the pool starts empty and can shrink back to zero workers
120zero workers over time. 149over time.
121 150
122=item max => $count (default: 4) 151=item max => $count (default: 4)
123 152
124The maximum number of processes in the pool, in addition to the template 153The maximum number of processes in the pool, in addition to the template
125process. C<AnyEvent::Fork::Pool> will never create more than this number 154process. C<AnyEvent::Fork::Pool> will never have more than this number of
126of worker processes, although there can be more temporarily when a worker 155worker processes, although there can be more temporarily when a worker is
127is shut down and hasn't exited yet. 156shut down and hasn't exited yet.
128 157
129=item load => $count (default: 2) 158=item load => $count (default: 2)
130 159
131The maximum number of concurrent jobs sent to a single worker 160The maximum number of concurrent jobs sent to a single worker process.
132process. Worker processes that handle this number of jobs already are
133called "busy".
134 161
135Jobs that cannot be sent to a worker immediately (because all workers are 162Jobs that cannot be sent to a worker immediately (because all workers are
136busy) will be queued until a worker is available. 163busy) will be queued until a worker is available.
137 164
165Setting this low improves latency. For example, at C<1>, every job that
166is sent to a worker is sent to a completely idle worker that doesn't run
167any other jobs. The downside is that throughput is reduced - a worker that
168finishes a job needs to wait for a new job from the parent.
169
170The default of C<2> is usually a good compromise.
171
138=item start => $seconds (default: 0.1) 172=item start => $seconds (default: 0.1)
139 173
140When a job is queued and all workers are busy, a timer is started. If the 174When there are fewer than C<idle> workers (or all workers are completely
141timer elapses and there are still jobs that cannot be queued to a worker, 175busy), then a timer is started. If the timer elapses and there are still
142a new worker is started. 176jobs that cannot be queued to a worker, a new worker is started.
143 177
144This configurs the time that all workers must be busy before a new worker 178This sets the minimum time that all workers must be busy before a new
145is started. Or, put differently, the minimum delay betwene starting new 179worker is started. Or, put differently, the minimum delay between starting
146workers. 180new workers.
147 181
148The delay is zero by default, which means new workers will be started 182The delay is small by default, which means new workers will be started
149without delay. 183relatively quickly. A delay of C<0> is possible, and ensures that the pool
184will grow as quickly as possible under load.
150 185
186Non-zero values are useful to avoid "exploding" a pool because a lot of
187jobs are queued in an instant.
188
189Higher values are often useful to improve efficiency at the cost of
190latency - when fewer processes can do the job over time, starting more and
191more is not necessarily going to help.
192
151=item stop => $seconds (default: 1) 193=item stop => $seconds (default: 10)
152 194
153When a worker has no jobs to execute it becomes idle. An idle worker that 195When a worker has no jobs to execute it becomes idle. An idle worker that
154hasn't executed a job within this amount of time will be stopped, unless 196hasn't executed a job within this amount of time will be stopped, unless
155the other parameters say otherwise. 197the other parameters say otherwise.
156 198
199Setting this to a very high value means that workers stay around longer,
200even when they have nothing to do, which can be good as they don't have to
201be started on the netx load spike again.
202
203Setting this to a lower value can be useful to avoid memory or simply
204process table wastage.
205
206Usually, setting this to a time longer than the time between load spikes
207is best - if you expect a lot of requests every minute and little work
208in between, setting this to longer than a minute avoids having to stop
209and start workers. On the other hand, you have to ask yourself if letting
210workers run idle is a good use of your resources. Try to find a good
211balance between resource usage of your workers and the time to start new
212workers - the processes created by L<AnyEvent::Fork> itself is fats at
213creating workers while not using much memory for them, so most of the
214overhead is likely from your own code.
215
157=item on_destroy => $callback->() (default: none) 216=item on_destroy => $callback->() (default: none)
158 217
159When a pool object goes out of scope, it will still handle all outstanding 218When a pool object goes out of scope, the outstanding requests are still
160jobs. After that, it will destroy all workers (and also the template 219handled till completion. Only after handling all jobs will the workers
161process if it isn't referenced otherwise). 220be destroyed (and also the template process if it isn't referenced
221otherwise).
222
223To find out when a pool I<really> has finished its work, you can set this
224callback, which will be called when the pool has been destroyed.
162 225
163=back 226=back
164 227
165=item Template Process 228=item AnyEvent::Fork::RPC Parameters
166 229
167The worker processes are all forked from a single template 230These parameters are all passed more or less directly to
168process. Ideally, all modules and all cdoe used by the worker, as well as 231L<AnyEvent::Fork::RPC>. They are only briefly mentioned here, for
169any shared data structures should be loaded into the template process, to 232their full documentation please refer to the L<AnyEvent::Fork::RPC>
170take advantage of data sharing via fork. 233documentation. Also, the default values mentioned here are only documented
171 234as a best effort - the L<AnyEvent::Fork::RPC> documentation is binding.
172You can create your own template process by creating a L<AnyEvent::Fork>
173object yourself and passing it as the C<template> parameter, but
174C<AnyEvent::Fork::Pool> can create one for you, including some standard
175options.
176 235
177=over 4 236=over 4
178 237
179=item template => $fork (default: C<< AnyEvent::Fork->new >>)
180
181The template process to use, if you want to create your own.
182
183=item require => \@modules (default: C<[]>)
184
185The modules in this list will be laoded into the template process.
186
187=item eval => "# perl code to execute in template" (default: none)
188
189This is a perl string that is evaluated after creating the template
190process and after requiring the modules. It can do whatever it wants to
191configure the process, but it must not do anything that would keep a later
192fork from working (so must not create event handlers or (real) threads for
193example).
194
195=back
196
197=item AnyEvent::Fork::RPC Parameters
198
199These parameters are all passed directly to L<AnyEvent::Fork::RPC>. They
200are only briefly mentioned here, for their full documentation
201please refer to the L<AnyEvent::Fork::RPC> documentation. Also, the
202default values mentioned here are only documented as a best effort -
203L<AnyEvent::Fork::RPC> documentation is binding.
204
205=over 4
206
207=item async => $boolean (default: 0) 238=item async => $boolean (default: 0)
208 239
209Whether to sue the synchronous or asynchronous RPC backend. 240Whether to use the synchronous or asynchronous RPC backend.
210 241
211=item on_error => $callback->($message) (default: die with message) 242=item on_error => $callback->($message) (default: die with message)
212 243
213The callback to call on any (fatal) errors. 244The callback to call on any (fatal) errors.
214 245
235 266
236 my $max = $arg{max} || 4; 267 my $max = $arg{max} || 4;
237 my $idle = $arg{idle} || 0, 268 my $idle = $arg{idle} || 0,
238 my $load = $arg{load} || 2, 269 my $load = $arg{load} || 2,
239 my $start = $arg{start} || 0.1, 270 my $start = $arg{start} || 0.1,
240 my $stop = $arg{stop} || 1, 271 my $stop = $arg{stop} || 10,
241 my $on_event = $arg{on_event} || sub { }, 272 my $on_event = $arg{on_event} || sub { },
242 my $on_destroy = $arg{on_destroy}; 273 my $on_destroy = $arg{on_destroy};
243 274
244 my @rpc = ( 275 my @rpc = (
245 async => $arg{async}, 276 async => $arg{async},
258 289
259 $template 290 $template
260 ->require ("AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync")) 291 ->require ("AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync"))
261 ->eval (' 292 ->eval ('
262 my ($magic0, $magic1) = @_; 293 my ($magic0, $magic1) = @_;
263 sub AnyEvent::Fork::Pool::quit() { 294 sub AnyEvent::Fork::Pool::retire() {
264 AnyEvent::Fork::RPC::on_event $magic0, "quit", $magic1; 295 AnyEvent::Fork::RPC::event $magic0, "quit", $magic1;
265 } 296 }
266 ', $magic0, $magic1) 297 ', $magic0, $magic1)
267 ->eval ($arg{eval}); 298 ;
268 299
269 $start_worker = sub { 300 $start_worker = sub {
270 my $proc = [0, 0, undef]; # load, index, rpc 301 my $proc = [0, 0, undef]; # load, index, rpc
271 302
272 $proc->[2] = $template 303 $proc->[2] = $template
298 $proc->[0] 329 $proc->[0]
299 or --$nidle; 330 or --$nidle;
300 331
301 Array::Heap::splice_heap_idx @pool, $proc->[1] 332 Array::Heap::splice_heap_idx @pool, $proc->[1]
302 if defined $proc->[1]; 333 if defined $proc->[1];
334
335 @$proc = 0; # tell others to leave it be
303 }; 336 };
304 337
305 $want_start = sub { 338 $want_start = sub {
306 undef $stop_w; 339 undef $stop_w;
307 340
326 }; 359 };
327 360
328 $scheduler = sub { 361 $scheduler = sub {
329 if (@queue) { 362 if (@queue) {
330 while (@queue) { 363 while (@queue) {
364 @pool or $start_worker->();
365
331 my $proc = $pool[0]; 366 my $proc = $pool[0];
332 367
333 if ($proc->[0] < $load) { 368 if ($proc->[0] < $load) {
334 # found free worker, increase load 369 # found free worker, increase load
335 unless ($proc->[0]++) { 370 unless ($proc->[0]++) {
353 or $want_stop->(); 388 or $want_stop->();
354 389
355 Array::Heap::adjust_heap_idx @pool, $proc->[1] 390 Array::Heap::adjust_heap_idx @pool, $proc->[1]
356 if defined $proc->[1]; 391 if defined $proc->[1];
357 392
393 &$ocb;
394
358 $scheduler->(); 395 $scheduler->();
359
360 &$ocb;
361 }); 396 });
362 } else { 397 } else {
363 $want_start->() 398 $want_start->()
364 unless @pool >= $max; 399 unless @pool >= $max;
365 400
397 432
398=item $pool->(..., $cb->(...)) 433=item $pool->(..., $cb->(...))
399 434
400Call the RPC function of a worker with the given arguments, and when the 435Call the RPC function of a worker with the given arguments, and when the
401worker is done, call the C<$cb> with the results, just like calling the 436worker is done, call the C<$cb> with the results, just like calling the
402L<AnyEvent::Fork::RPC> object directly. 437RPC object durectly - see the L<AnyEvent::Fork::RPC> documentation for
438details on the RPC API.
403 439
404If there is no free worker, the call will be queued. 440If there is no free worker, the call will be queued until a worker becomes
441available.
405 442
406Note that there can be considerable time between calling this method and 443Note that there can be considerable time between calling this method and
407the call actually being executed. During this time, the parameters passed 444the call actually being executed. During this time, the parameters passed
408to this function are effectively read-only - modifying them after the call 445to this function are effectively read-only - modifying them after the call
409and before the callback is invoked causes undefined behaviour. 446and before the callback is invoked causes undefined behaviour.
410 447
411=cut 448=cut
412 449
450=item $cpus = AnyEvent::Fork::Pool::ncpu [$default_cpus]
451
452=item ($cpus, $eus) = AnyEvent::Fork::Pool::ncpu [$default_cpus]
453
454Tries to detect the number of CPUs (C<$cpus> often called cpu cores
455nowadays) and execution units (C<$eus>) which include e.g. extra
456hyperthreaded units). When C<$cpus> cannot be determined reliably,
457C<$default_cpus> is returned for both values, or C<1> if it is missing.
458
459For normal CPU bound uses, it is wise to have as many worker processes
460as CPUs in the system (C<$cpus>), if nothing else uses the CPU. Using
461hyperthreading is usually detrimental to performance, but in those rare
462cases where that really helps it might be beneficial to use more workers
463(C<$eus>).
464
465Currently, F</proc/cpuinfo> is parsed on GNU/Linux systems for both
466C<$cpus> and C<$eu>, and on {Free,Net,Open}BSD, F<sysctl -n hw.ncpu> is
467used for C<$cpus>.
468
469Example: create a worker pool with as many workers as cpu cores, or C<2>,
470if the actual number could not be determined.
471
472 $fork->AnyEvent::Fork::Pool::run ("myworker::function",
473 max => (scalar AnyEvent::Fork::Pool::ncpu 2),
474 );
475
476=cut
477
478BEGIN {
479 if ($^O eq "linux") {
480 *ncpu = sub(;$) {
481 my ($cpus, $eus);
482
483 if (open my $fh, "<", "/proc/cpuinfo") {
484 my %id;
485
486 while (<$fh>) {
487 if (/^core id\s*:\s*(\d+)/) {
488 ++$eus;
489 undef $id{$1};
490 }
491 }
492
493 $cpus = scalar keys %id;
494 } else {
495 $cpus = $eus = @_ ? shift : 1;
496 }
497 wantarray ? ($cpus, $eus) : $cpus
498 };
499 } elsif ($^O eq "freebsd" || $^O eq "netbsd" || $^O eq "openbsd") {
500 *ncpu = sub(;$) {
501 my $cpus = qx<sysctl -n hw.ncpu> * 1
502 || (@_ ? shift : 1);
503 wantarray ? ($cpus, $cpus) : $cpus
504 };
505 } else {
506 *ncpu = sub(;$) {
507 my $cpus = @_ ? shift : 1;
508 wantarray ? ($cpus, $cpus) : $cpus
509 };
510 }
511}
512
413=back 513=back
414 514
515=head1 CHILD USAGE
516
517In addition to the L<AnyEvent::Fork::RPC> API, this module implements one
518more child-side function:
519
520=over 4
521
522=item AnyEvent::Fork::Pool::retire ()
523
524This function sends an event to the parent process to request retirement:
525the worker is removed from the pool and no new jobs will be sent to it,
526but it has to handle the jobs that are already queued.
527
528The parentheses are part of the syntax: the function usually isn't defined
529when you compile your code (because that happens I<before> handing the
530template process over to C<AnyEvent::Fork::Pool::run>, so you need the
531empty parentheses to tell Perl that the function is indeed a function.
532
533Retiring a worker can be useful to gracefully shut it down when the worker
534deems this useful. For example, after executing a job, one could check
535the process size or the number of jobs handled so far, and if either is
536too high, the worker could ask to get retired, to avoid memory leaks to
537accumulate.
538
539Example: retire a worker after it has handled roughly 100 requests.
540
541 my $count = 0;
542
543 sub my::worker {
544
545 ++$count == 100
546 and AnyEvent::Fork::Pool::retire ();
547
548 ... normal code goes here
549 }
550
551=back
552
553=head1 POOL PARAMETERS RECIPES
554
555This section describes some recipes for pool paramaters. These are mostly
556meant for the synchronous RPC backend, as the asynchronous RPC backend
557changes the rules considerably, making workers themselves responsible for
558their scheduling.
559
560=over 4
561
562=item low latency - set load = 1
563
564If you need a deterministic low latency, you should set the C<load>
565parameter to C<1>. This ensures that never more than one job is sent to
566each worker. This avoids having to wait for a previous job to finish.
567
568This makes most sense with the synchronous (default) backend, as the
569asynchronous backend can handle multiple requests concurrently.
570
571=item lowest latency - set load = 1 and idle = max
572
573To achieve the lowest latency, you additionally should disable any dynamic
574resizing of the pool by setting C<idle> to the same value as C<max>.
575
576=item high throughput, cpu bound jobs - set load >= 2, max = #cpus
577
578To get high throughput with cpu-bound jobs, you should set the maximum
579pool size to the number of cpus in your system, and C<load> to at least
580C<2>, to make sure there can be another job waiting for the worker when it
581has finished one.
582
583The value of C<2> for C<load> is the minimum value that I<can> achieve
584100% throughput, but if your parent process itself is sometimes busy, you
585might need higher values. Also there is a limit on the amount of data that
586can be "in flight" to the worker, so if you send big blobs of data to your
587worker, C<load> might have much less of an effect.
588
589=item high throughput, I/O bound jobs - set load >= 2, max = 1, or very high
590
591When your jobs are I/O bound, using more workers usually boils down to
592higher throughput, depending very much on your actual workload - sometimes
593having only one worker is best, for example, when you read or write big
594files at maixmum speed, as a second worker will increase seek times.
595
596=back
597
598=head1 EXCEPTIONS
599
600The same "policy" as with L<AnyEvent::Fork::RPC> applies - exceptins will
601not be caught, and exceptions in both worker and in callbacks causes
602undesirable or undefined behaviour.
603
415=head1 SEE ALSO 604=head1 SEE ALSO
416 605
417L<AnyEvent::Fork>, to create the processes in the first place. 606L<AnyEvent::Fork>, to create the processes in the first place.
607
608L<AnyEvent::Fork::Remote>, likewise, but helpful for remote processes.
418 609
419L<AnyEvent::Fork::RPC>, which implements the RPC protocol and API. 610L<AnyEvent::Fork::RPC>, which implements the RPC protocol and API.
420 611
421=head1 AUTHOR AND CONTACT INFORMATION 612=head1 AUTHOR AND CONTACT INFORMATION
422 613

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines