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

Comparing AnyEvent-Fork-Pool/Pool.pm (file contents):
Revision 1.2 by root, Thu Apr 18 21:37:27 2013 UTC vs.
Revision 1.4 by root, Sat Apr 20 16:39:14 2013 UTC

…		…
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 parameters with default values
12	my $pool = new AnyEvent::Fork::Pool	12	my $pool = AnyEvent::Fork
13	"MyWorker::run",	13	->new
		14	->require ("MyWorker")
		15	->AnyEvent::Fork::Pool::run (
		16	"MyWorker::run", # the worker function
14		17
15	# pool management	18	# pool management
16	min => 0, # minimum # of processes
17	max => 4, # maximum # of processes	19	max => 4, # absolute maximum # of processes
18	max_queue => 2, # queue at most this number of jobs per process	20	idle => 2, # minimum # of idle processes
		21	load => 2, # queue at most this number of jobs per process
19	min_delay => 0, # wait this many seconds before starting a new process	22	start => 0.1, # wait this many seconds before starting a new process
20	min_idle => 0, # try to have at least this amount of idle processes
21	max_idle => 1, # at most this many idle processes
22	idle_time => 1, # wait this many seconds before killing an idle process	23	stop => 1, # wait this many seconds before stopping an idle process
23	on_destroy => (my $finish = AE::cv),	24	on_destroy => (my $finish = AE::cv), # called when object is destroyed
24		25
25	# template process
26	template => AnyEvent::Fork->new, # the template process to use
27	require => [MyWorker::], # module(s) to load
28	eval => "# perl code to execute in template",
29
30	# parameters passed to AnyEvent::Fork::RPC	26	# parameters passed to AnyEvent::Fork::RPC
31	async => 0,	27	async => 0,
32	on_error => sub { die "FATAL: $_[0]\n" },	28	on_error => sub { die "FATAL: $_[0]\n" },
33	on_event => sub { my @ev = @_ },	29	on_event => sub { my @ev = @_ },
34	init => "MyWorker::init",	30	init => "MyWorker::init",
35	serialiser => $AnyEvent::Fork::RPC::STRING_SERIALISER,	31	serialiser => $AnyEvent::Fork::RPC::STRING_SERIALISER,
36	;	32	);
37		33
38	for (1..10) {	34	for (1..10) {
39	$pool->call (doit => $_, sub {	35	$pool->(doit => $_, sub {
40	print "MyWorker::run returned @_\n";	36	print "MyWorker::run returned @_\n";
41	});	37	});
42	}	38	}
43		39
44	undef $pool;	40	undef $pool;
…		…
68		64
69	use common::sense;	65	use common::sense;
70		66
71	use Scalar::Util ();	67	use Scalar::Util ();
72		68
		69	use Guard ();
73	use Array::Heap ();	70	use Array::Heap ();
74		71
75	use AnyEvent;	72	use AnyEvent;
76	use AnyEvent::Fork; # we don't actually depend on it, this is for convenience	73	use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
77	use AnyEvent::Fork::RPC;	74	use AnyEvent::Fork::RPC;
78		75
		76	# 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,
		78	# but didn't come up with an obviously better alternative.
79	my $magic0 = ':t6Z@HK1N%Dx@_7?=~-7NQgWDdAs6a,jFN=wLO0jD1%P';	79	my $magic0 = ':t6Z@HK1N%Dx@_7?=~-7NQgWDdAs6a,jFN=wLO0jD1%P';
80	my $magic2 = '<~53rexz.U`!]X[A235^"fyEoiTF\T~oH1l/N6+Djep9b~bI9`\1x%B~vWO1q*';	80	my $magic1 = '<~53rexz.U`!]X[A235^"fyEoiTF\T~oH1l/N6+Djep9b~bI9`\1x%B~vWO1q*';
81		81
82	our $VERSION = 0.1;	82	our $VERSION = 0.1;
83		83
84	=item my $rpc = new AnyEvent::Fork::Pool $function, [key => value...]	84	=item my $rpc = AnyEvent::Fork::Pool::run $fork, $function, [key => value...]
		85
		86	The traditional way to call it. But it is way cooler to call it in the
		87	following way:
		88
		89	=item my $rpc = $fork->AnyEvent::Fork::Pool::run ($function, [key => value...])
85		90
86	Creates a new pool object with the specified C<$function> as function	91	Creates a new pool object with the specified C<$function> as function
87	(name) to call for each request.	92	(name) to call for each request. The pool uses the C<$fork> object as the
88		93	template when creating worker processes.
89	A pool consists of a template process that contains the code and data that
90	the worker processes need. And a number of worker processes that have been
91	forked off of that template process.
92		94
93	You can supply your own template process, or tell C<AnyEvent::Fork::Pool>	95	You can supply your own template process, or tell C<AnyEvent::Fork::Pool>
94	to create one.	96	to create one.
95		97
96	A relatively large number of key/value pairs can be specified to influence	98	A relatively large number of key/value pairs can be specified to influence
…		…
105	decide how many of these processes exist and when they are started and	107	decide how many of these processes exist and when they are started and
106	stopp.ed	108	stopp.ed
107		109
108	=over 4	110	=over 4
109		111
110	=item min => $count (default: 0)	112	=item idle => $count (default: 0)
111		113
112	The minimum number of processes in the pool, in addition to the template	114	The minimum amount of idle processes in the pool - when there are fewer
113	process. Even when idle, there will never be fewer than this number of	115	than this many idle workers, C<AnyEvent::Fork::Pool> will try to start new
114	worker processes. The default means that the pool can be empty.	116	ones, subject to C<max> and C<start>.
		117
		118	This is also the initial/minimum amount of workers in the pool. The
		119	default of zero means that the pool starts empty and can shrink back to
		120	zero workers over time.
115		121
116	=item max => $count (default: 4)	122	=item max => $count (default: 4)
117		123
118	The maximum number of processes in the pool, in addition to the template	124	The maximum number of processes in the pool, in addition to the template
119	process. C<AnyEvent::Fork::Pool> will never create more than this number	125	process. C<AnyEvent::Fork::Pool> will never create more than this number
120	of processes.	126	of worker processes, although there can be more temporarily when a worker
		127	is shut down and hasn't exited yet.
121		128
122	=item max_queue => $count (default: 2)	129	=item load => $count (default: 2)
123		130
124	The maximum number of jobs sent to a single worker process. Worker	131	The maximum number of concurrent jobs sent to a single worker
125	processes that handle this number of jobs already are called "busy".	132	process. Worker processes that handle this number of jobs already are
		133	called "busy".
126		134
127	Jobs that cannot be sent to a worker immediately (because all workers are	135	Jobs that cannot be sent to a worker immediately (because all workers are
128	busy) will be queued until a worker is available.	136	busy) will be queued until a worker is available.
129		137
130	=item min_delay => $seconds (default: 0)	138	=item start => $seconds (default: 0.1)
131		139
132	When a job is queued and all workers are busy, a timer is started. If the	140	When a job is queued and all workers are busy, a timer is started. If the
133	timer elapses and there are still jobs that cannot be queued to a worker,	141	timer elapses and there are still jobs that cannot be queued to a worker,
134	a new worker is started.	142	a new worker is started.
135		143
…		…
138	workers.	146	workers.
139		147
140	The delay is zero by default, which means new workers will be started	148	The delay is zero by default, which means new workers will be started
141	without delay.	149	without delay.
142		150
143	=item min_idle => $count (default: 0)
144
145	The minimum number of idle workers - when they are less, more
146	are started. The C<min_delay> is still respected though, and
147	C<min_idle>/C<min_delay> and C<max_idle>/C<idle_time> are useful to
148	dynamically adjust the pool.
149
150	=item max_idle => $count (default: 1)
151
152	The maximum number of idle workers. If a worker becomes idle and there are
153	already this many idle workers, it will be stopped immediately instead of
154	waiting for the idle timer to elapse.
155
156	=item idle_time => $seconds (default: 1)	151	=item stop => $seconds (default: 1)
157		152
158	When a worker has no jobs to execute it becomes idle. An idle worker that	153	When a worker has no jobs to execute it becomes idle. An idle worker that
159	hasn't executed a job within this amount of time will be stopped, unless	154	hasn't executed a job within this amount of time will be stopped, unless
160	the other parameters say otherwise.	155	the other parameters say otherwise.
161		156
…		…
233		228
234	=back	229	=back
235		230
236	=cut	231	=cut
237		232
238	sub new {	233	sub run {
239	my ($class, $function, %arg) = @_;	234	my ($template, $function, %arg) = @_;
240		235
241	my $self = bless {	236	my $max = $arg{max} \|\| 4;
242	min => 0,	237	my $idle = $arg{idle} \|\| 0,
243	max => 4,	238	my $load = $arg{load} \|\| 2,
244	max_queue => 2,	239	my $start = $arg{start} \|\| 0.1,
245	min_delay => 0,	240	my $stop = $arg{stop} \|\| 1,
246	max_idle => 1,	241	my $on_event = $arg{on_event} \|\| sub { },
247	idle_time => 1,	242	my $on_destroy = $arg{on_destroy};
248	on_event => sub { },
249	%arg,
250	pool => [],
251	queue => [],
252	}, $class;
253		243
254	$self->{function} = $function;	244	my @rpc = (
		245	async => $arg{async},
		246	init => $arg{init},
		247	serialiser => $arg{serialiser},
		248	on_error => $arg{on_error},
		249	);
255		250
256	($self->{template} \|\|= new AnyEvent::Fork)	251	my (@pool, @queue, $nidle, $start_w, $stop_w, $shutdown);
		252	my ($start, $stop, $want_start, $want_stop, $scheduler);
		253
		254	my $destroy_guard = Guard::guard {
		255	$on_destroy->()
		256	if $on_destroy;
		257	};
		258
		259	$template
257	->require ("AnyEvent::Fork::RPC::" . ($self->{async} ? "Async" : "Sync"))	260	->require ("AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync"))
258	->require (@{ delete $self->{require} })
259	->eval ('	261	->eval ('
260	my ($magic0, $magic2) = @_;	262	my ($magic0, $magic1) = @_;
261	sub AnyEvent::Fork::Pool::quit() {	263	sub AnyEvent::Fork::Pool::quit() {
262	AnyEvent::Fork::RPC::on_event $magic0, "quit", $magic2;	264	AnyEvent::Fork::RPC::on_event $magic0, "quit", $magic1;
263	}	265	}
264	', $magic0, $magic2)	266	', $magic0, $magic1)
265	->eval (delete $self->{eval});	267	->eval (delete $arg{eval});
266		268
267	$self->start	269	$start = sub {
268	while @{ $self->{pool} } < $self->{min};	270	my $proc = [0, 0, undef]; # load, index, rpc
269		271
270	$self
271	}
272
273	sub start {
274	my ($self) = @_;
275
276	warn "start\n";#d#	272	warn "start a worker\n";#d#
277		273
278	Scalar::Util::weaken $self;
279
280	my $proc = [0, undef, undef];
281
282	$proc->[1] = $self->{template}	274	$proc->[2] = $template
283	->fork	275	->fork
284	->AnyEvent::Fork::RPC::run ($self->{function},	276	->AnyEvent::Fork::RPC::run ($function,
285	async => $self->{async},	277	@rpc,
286	init => $self->{init},
287	serialiser => $self->{serialiser},
288	on_error => $self->{on_error},
289	on_event => sub {	278	on_event => sub {
290	if (@_ == 3 && $_[0] eq $magic0 && $_[2] eq $magic2) {	279	if (@_ == 3 && $_[0] eq $magic0 && $_[2] eq $magic1) {
291	if ($_[1] eq "quit") {	280	$destroy_guard if 0; # keep it alive
292	my $pool = $self->{pool};	281
293	for (0 .. $#$pool) {	282	$_[1] eq "quit" and $stop->($proc);
294	if ($pool->[$_] == $proc) {
295	Array::Heap::splice_heap @$pool, $_;
296	return;	283	return;
297	}
298	}
299	die;
300	}	284	}
		285
301	return;	286	&$on_event;
302	}	287	},
		288	)
		289	;
303		290
304	&{ $self->{on_event} };	291	++$nidle;
		292	Array::Heap::push_heap @pool, $proc;
		293
		294	Scalar::Util::weaken $proc;
		295	};
		296
		297	$stop = sub {
		298	my $proc = shift;
		299
		300	$proc->[0]
		301	or --$nidle;
		302
		303	Array::Heap::splice_heap_idx @pool, $proc->[1]
		304	if defined $proc->[1];
		305	};
		306
		307	$want_start = sub {
		308	undef $stop_w;
		309
		310	$start_w \|\|= AE::timer $start, 0, sub {
		311	undef $start_w;
		312
		313	if (@queue) {
		314	$start->();
		315	$scheduler->();
		316	}
		317	};
		318	};
		319
		320	$want_stop = sub {
		321	$stop_w \|\|= AE::timer $stop, 0, sub {
		322	undef $stop_w;
		323
		324	$stop->($pool[0])
		325	if $nidle;
		326	};
		327	};
		328
		329	$scheduler = sub {
		330	if (@queue) {
		331	while (@queue) {
		332	my $proc = $pool[0];
		333
		334	if ($proc->[0] < $load) {
		335	warn "free $proc $proc->[0]\n";#d#
		336	# found free worker
		337	$proc->[0]++
		338	or --$nidle >= $idle
		339	or $want_start->();
		340
		341	Array::Heap::adjust_heap @pool, 0;
		342
		343	my $job = shift @queue;
		344	my $ocb = pop @$job;
		345
		346	$proc->[2]->(@$job, sub {
		347	# reduce queue counter
		348	--$pool[$_][0]
		349	or ++$nidle > $idle
		350	or $want_stop->();
		351
		352	Array::Heap::adjust_heap @pool, $_;
		353
		354	$scheduler->();
		355
		356	&$ocb;
		357	});
		358	} else {
		359	warn "busy $proc->[0]\n";#d#
		360	# all busy, delay
		361
		362	$want_start->();
		363	last;
305	},	364	}
306	)	365	}
		366	} elsif ($shutdown) {
		367	@pool = ();
		368	undef $start_w;
		369	undef $start; # frees $destroy_guard reference
		370
		371	$stop->($pool[0])
		372	while $nidle;
		373	}
307	;	374	};
308		375
309	++$self->{idle};	376	my $shutdown_guard = Guard::guard {
310	Array::Heap::push_heap @{ $self->{pool} }, $proc;	377	$shutdown = 1;
		378	$scheduler->();
		379	};
		380
		381	$start->()
		382	while @pool < $idle;
		383
		384	sub {
		385	$shutdown_guard if 0; # keep it alive
		386
		387	$start->()
		388	unless @pool;
		389
		390	push @queue, [@_];
		391	$scheduler->();
		392	}
311	}	393	}
312		394
313	=item $pool->call (..., $cb->(...))	395	=item $pool->call (..., $cb->(...))
314		396
315	Call the RPC function of a worker with the given arguments, and when the	397	Call the RPC function of a worker with the given arguments, and when the
…		…
323	to this function are effectively read-only - modifying them after the call	405	to this function are effectively read-only - modifying them after the call
324	and before the callback is invoked causes undefined behaviour.	406	and before the callback is invoked causes undefined behaviour.
325		407
326	=cut	408	=cut
327		409
328	sub scheduler {
329	my $self = shift;
330
331	my $pool = $self->{pool};
332	my $queue = $self->{queue};
333
334	$self->start
335	unless @$pool;
336
337	while (@$queue) {
338	my $proc = $pool->[0];
339
340	if ($proc->[0] < $self->{max_queue}) {
341	warn "free $proc $proc->[0]\n";#d#
342	# found free worker
343	--$self->{idle}
344	unless $proc->[0]++;
345
346	undef $proc->[2];
347
348	Array::Heap::adjust_heap @$pool, 0;
349
350	my $job = shift @$queue;
351	my $ocb = pop @$job;
352
353	$proc->[1]->(@$job, sub {
354	for (0 .. $#$pool) {
355	if ($pool->[$_] == $proc) {
356	# reduce queue counter
357	unless (--$pool->[$_][0]) {
358	# worker becomes idle
359	my $to = ++$self->{idle} > $self->{max_idle}
360	? 0
361	: $self->{idle_time};
362
363	$proc->[2] = AE::timer $to, 0, sub {
364	undef $proc->[2];
365
366	warn "destroy $proc afzer $to\n";#d#
367
368	for (0 .. $#$pool) {
369	if ($pool->[$_] == $proc) {
370	Array::Heap::splice_heap @$pool, $_;
371	--$self->{idle};
372	last;
373	}
374	}
375	};
376	}
377
378	Array::Heap::adjust_heap @$pool, $_;
379	last;
380	}
381	}
382	&$ocb;
383	});
384	} else {
385	warn "busy $proc->[0]\n";#d#
386	# all busy, delay
387
388	$self->{min_delay_w} \|\|= AE::timer $self->{min_delay}, 0, sub {
389	delete $self->{min_delay_w};
390
391	if (@{ $self->{queue} }) {
392	$self->start;
393	$self->scheduler;
394	}
395	};
396	last;
397	}
398	}
399	warn "last\n";#d#
400	}
401
402	sub call {
403	my $self = shift;
404
405	push @{ $self->{queue} }, [@_];
406	$self->scheduler;
407	}
408
409	sub DESTROY {
410	$_[0]{on_destroy}->();
411	}
412
413	=back	410	=back
414		411
415	=head1 SEE ALSO	412	=head1 SEE ALSO
416		413
417	L<AnyEvent::Fork>, to create the processes in the first place.	414	L<AnyEvent::Fork>, to create the processes in the first place.

Diff Legend

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

Comparing AnyEvent-Fork-Pool/Pool.pm (file contents): Revision 1.2 by root, Thu Apr 18 21:37:27 2013 UTC vs. Revision 1.4 by root, Sat Apr 20 16:39:14 2013 UTC

Diff Legend

Comparing AnyEvent-Fork-Pool/Pool.pm (file contents):
Revision 1.2 by root, Thu Apr 18 21:37:27 2013 UTC vs.
Revision 1.4 by root, Sat Apr 20 16:39:14 2013 UTC