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

Comparing AnyEvent-Fork/Fork.pm (file contents):
Revision 1.3 by root, Tue Apr 2 18:00:04 2013 UTC vs.
Revision 1.14 by root, Thu Apr 4 08:16:14 2013 UTC

…		…
1	=head1 NAME	1	=head1 NAME
2		2
3	AnyEvent::ProcessPool - manage pools of perl worker processes, exec'ed or fork'ed	3	AnyEvent::Fork - everything you wanted to use fork() for, but couldn't
		4
		5	ATTENTION, this is a very early release, and very untested. Consider it a
		6	technology preview.
4		7
5	=head1 SYNOPSIS	8	=head1 SYNOPSIS
6		9
7	use AnyEvent::ProcessPool;	10	use AnyEvent::Fork;
		11
		12	##################################################################
		13	# create a single new process, tell it to run your worker function
		14
		15	AnyEvent::Fork
		16	->new
		17	->require ("MyModule")
		18	->run ("MyModule::worker, sub {
		19	my ($master_filehandle) = @_;
		20
		21	# now $master_filehandle is connected to the
		22	# $slave_filehandle in the new process.
		23	});
		24
		25	# MyModule::worker might look like this
		26	sub MyModule::worker {
		27	my ($slave_filehandle) = @_;
		28
		29	# now $slave_filehandle is connected to the $master_filehandle
		30	# in the original prorcess. have fun!
		31	}
		32
		33	##################################################################
		34	# create a pool of server processes all accepting on the same socket
		35
		36	# create listener socket
		37	my $listener = ...;
		38
		39	# create a pool template, initialise it and give it the socket
		40	my $pool = AnyEvent::Fork
		41	->new
		42	->require ("Some::Stuff", "My::Server")
		43	->send_fh ($listener);
		44
		45	# now create 10 identical workers
		46	for my $id (1..10) {
		47	$pool
		48	->fork
		49	->send_arg ($id)
		50	->run ("My::Server::run");
		51	}
		52
		53	# now do other things - maybe use the filehandle provided by run
		54	# to wait for the processes to die. or whatever.
		55
		56	# My::Server::run might look like this
		57	sub My::Server::run {
		58	my ($slave, $listener, $id) = @_;
		59
		60	close $slave; # we do not use the socket, so close it to save resources
		61
		62	# we could go ballistic and use e.g. AnyEvent here, or IO::AIO,
		63	# or anything we usually couldn't do in a process forked normally.
		64	while (my $socket = $listener->accept) {
		65	# do sth. with new socket
		66	}
		67	}
8		68
9	=head1 DESCRIPTION	69	=head1 DESCRIPTION
10		70
11	This module allows you to create single worker processes but also worker	71	This module allows you to create new processes, without actually forking
12	pool that share memory, by forking from the main program, or exec'ing new	72	them from your current process (avoiding the problems of forking), but
13	perl interpreters from a module.	73	preserving most of the advantages of fork.
14		74
15	You create a new processes in a pool by specifying a function to call	75	It can be used to create new worker processes or new independent
16	with any combination of string values and file handles.	76	subprocesses for short- and long-running jobs, process pools (e.g. for use
		77	in pre-forked servers) but also to spawn new external processes (such as
		78	CGI scripts from a webserver), which can be faster (and more well behaved)
		79	than using fork+exec in big processes.
17		80
18	A pool can have initialisation code which is executed before forking. The	81	Special care has been taken to make this module useful from other modules,
19	initialisation code is only executed once and the resulting process is	82	while still supporting specialised environments such as L<App::Staticperl>
20	cached, to be used as a template.	83	or L<PAR::Packer>.
21
22	Pools without such initialisation code don't cache an extra process.
23		84
24	=head1 PROBLEM STATEMENT	85	=head1 PROBLEM STATEMENT
25		86
26	There are two ways to implement parallel processing on UNIX like operating	87	There are two ways to implement parallel processing on UNIX like operating
27	systems - fork and process, and fork+exec and process. They have different	88	systems - fork and process, and fork+exec and process. They have different
110	time, and the memory is not shared with anything else.	171	time, and the memory is not shared with anything else.
111		172
112	This is ideal for when you only need one extra process of a kind, with the	173	This is ideal for when you only need one extra process of a kind, with the
113	option of starting and stipping it on demand.	174	option of starting and stipping it on demand.
114		175
		176	Example:
		177
		178	AnyEvent::Fork
		179	->new
		180	->require ("Some::Module")
		181	->run ("Some::Module::run", sub {
		182	my ($fork_fh) = @_;
		183	});
		184
115	=item fork a new template process, load code, then fork processes off of	185	=item fork a new template process, load code, then fork processes off of
116	it and run the code	186	it and run the code
117		187
118	When you need to have a bunch of processes that all execute the same (or	188	When you need to have a bunch of processes that all execute the same (or
119	very similar) tasks, then a good way is to create a new template process	189	very similar) tasks, then a good way is to create a new template process
…		…
127	The disadvantage of this approach is that you need to create a template	197	The disadvantage of this approach is that you need to create a template
128	process for the sole purpose of forking new processes from it, but if you	198	process for the sole purpose of forking new processes from it, but if you
129	only need a fixed number of proceses you can create them, and then destroy	199	only need a fixed number of proceses you can create them, and then destroy
130	the template process.	200	the template process.
131		201
		202	Example:
		203
		204	my $template = AnyEvent::Fork->new->require ("Some::Module");
		205
		206	for (1..10) {
		207	$template->fork->run ("Some::Module::run", sub {
		208	my ($fork_fh) = @_;
		209	});
		210	}
		211
		212	# at this point, you can keep $template around to fork new processes
		213	# later, or you can destroy it, which causes it to vanish.
		214
132	=item execute a new perl interpreter, load some code, run it	215	=item execute a new perl interpreter, load some code, run it
133		216
134	This is relatively slow, and doesn't allow you to share memory between	217	This is relatively slow, and doesn't allow you to share memory between
135	multiple processes.	218	multiple processes.
136		219
137	The only advantage is that you don't have to have a template process	220	The only advantage is that you don't have to have a template process
138	hanging around all the time to fork off some new processes, which might be	221	hanging around all the time to fork off some new processes, which might be
139	an advantage when there are long time spans where no extra processes are	222	an advantage when there are long time spans where no extra processes are
140	needed.	223	needed.
141		224
		225	Example:
		226
		227	AnyEvent::Fork
		228	->new_exec
		229	->require ("Some::Module")
		230	->run ("Some::Module::run", sub {
		231	my ($fork_fh) = @_;
		232	});
		233
142	=back	234	=back
143		235
144	=head1 FUNCTIONS	236	=head1 FUNCTIONS
145		237
146	=over 4	238	=over 4
147		239
148	=cut	240	=cut
149		241
150	package AnyEvent::ProcessPool;	242	package AnyEvent::Fork;
151		243
152	use common::sense;	244	use common::sense;
153		245
154	use Socket ();	246	use Socket ();
155		247
156	use Proc::FastSpawn;
157	use AnyEvent;	248	use AnyEvent;
158	use AnyEvent::ProcessPool::Util;	249	use AnyEvent::Fork::Util;
159	use AnyEvent::Util ();	250	use AnyEvent::Util ();
160		251
161	BEGIN {	252	our $VERSION = $AnyEvent::Fork::Util::VERSION;
162	# require Exporter;
163	}
164		253
		254	our $PERL; # the path to the perl interpreter, deduces with various forms of magic
		255
165	=item my $pool = new AnyEvent::ProcessPool key => value...	256	=item my $pool = new AnyEvent::Fork key => value...
166		257
167	Create a new process pool. The following named parameters are supported:	258	Create a new process pool. The following named parameters are supported:
168		259
169	=over 4	260	=over 4
170		261
171	=back	262	=back
172		263
173	=cut	264	=cut
174		265
		266	# the early fork template process
		267	our $EARLY;
		268
175	# the template process	269	# the empty template process
176	our $template;	270	our $TEMPLATE;
177		271
178	sub _queue {	272	sub _cmd {
		273	my $self = shift;
		274
		275	#TODO: maybe append the packet to any existing string command already in the queue
		276
		277	# ideally, we would want to use "a (w/a)*" as format string, but perl versions
		278	# from at least 5.8.9 to 5.16.3 are all buggy and can't unpack it.
		279	push @{ $self->[2] }, pack "N/a", pack "(w/a)*", @_;
		280
		281	$self->[3] \|\|= AE::io $self->[1], 1, sub {
		282	# send the next "thing" in the queue - either a reference to an fh,
		283	# or a plain string.
		284
		285	if (ref $self->[2][0]) {
		286	# send fh
		287	AnyEvent::Fork::Util::fd_send fileno $self->[1], fileno ${ $self->[2][0] }
		288	and shift @{ $self->[2] };
		289
		290	} else {
		291	# send string
		292	my $len = syswrite $self->[1], $self->[2][0]
		293	or do { undef $self->[3]; die "AnyEvent::Fork: command write failure: $!" };
		294
		295	substr $self->[2][0], 0, $len, "";
		296	shift @{ $self->[2] } unless length $self->[2][0];
		297	}
		298
		299	unless (@{ $self->[2] }) {
		300	undef $self->[3];
		301	# invoke run callback
		302	$self->[0]->($self->[1]) if $self->[0];
		303	}
		304	};
		305
		306	() # make sure we don't leak the watcher
		307	}
		308
		309	sub _new {
179	my ($pid, $fh) = @_;	310	my ($self, $fh) = @_;
180		311
181	[	312	AnyEvent::Util::fh_nonblocking $fh, 1;
182	$pid,	313
		314	$self = bless [
		315	undef, # run callback
183	$fh,	316	$fh,
184	[],	317	[], # write queue - strings or fd's
185	undef	318	undef, # AE watcher
186	]	319	], $self;
187	}
188		320
189	sub queue_cmd {	321	$self
190	my $queue = shift;
191
192	push @{ $queue->[2] }, pack "N/a", pack "a (w/a)*", @_;
193
194	$queue->[3] \|\|= AE::io $queue->[1], 1, sub {
195	if (ref $queue->[2][0]) {
196	AnyEvent::ProcessPool::Util::fd_send fileno $queue->[1], fileno ${ $queue->[2][0] }
197	and shift @{ $queue->[2] };
198	} else {
199	my $len = syswrite $queue->[1], $queue->[2][0]
200	or do { undef $queue->[3]; die "AnyEvent::ProcessPool::queue write failure: $!" };
201	substr $queue->[2][0], 0, $len, "";
202	shift @{ $queue->[2] } unless length $queue->[2][0];
203	}
204
205	undef $queue->[3] unless @{ $queue->[2] };
206	};
207	}	322	}
208		323
209	sub run_template {	324	# fork template from current process, used by AnyEvent::Fork::Early/Template
210	return if $template;	325	sub _new_fork {
211
212	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;	326	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
213	AnyEvent::Util::fh_nonblocking $fh, 1;	327	my $parent = $$;
214	fd_inherit fileno $slave;
215		328
216	my %env = %ENV;	329	my $pid = fork;
217	$env{PERL5LIB} = join ":", grep !ref, @INC;
218		330
219	my $pid = spawn	331	if ($pid eq 0) {
220	$^X,	332	require AnyEvent::Fork::Serve;
221	["perl", "-MAnyEvent::ProcessPool::Serve", "-e", "AnyEvent::ProcessPool::Serve::me", fileno $slave],	333	$AnyEvent::Fork::Serve::OWNER = $parent;
222	[map "$_=$env{$_}", keys %env],	334	close $fh;
223	or die "unable to spawn AnyEvent::ProcessPool server: $!";	335	$0 = "$_[1] of $parent";
		336	AnyEvent::Fork::Serve::serve ($slave);
		337	AnyEvent::Fork::Util::_exit 0;
		338	} elsif (!$pid) {
		339	die "AnyEvent::Fork::Early/Template: unable to fork template process: $!";
		340	}
224		341
225	close $slave;	342	AnyEvent::Fork->_new ($fh)
226
227	$template = _queue $pid, $fh;
228
229	my ($a, $b) = AnyEvent::Util::portable_socketpair;
230
231	queue_cmd $template, "Iabc";
232	push @{ $template->[2] }, \$b;
233
234	use Coro::AnyEvent; Coro::AnyEvent::sleep 1;
235	undef $b;
236	die "x" . <$a>;
237	}	343	}
		344
		345	=item my $proc = new AnyEvent::Fork
		346
		347	Create a new "empty" perl interpreter process and returns its process
		348	object for further manipulation.
		349
		350	The new process is forked from a template process that is kept around
		351	for this purpose. When it doesn't exist yet, it is created by a call to
		352	C<new_exec> and kept around for future calls.
		353
		354	When the process object is destroyed, it will release the file handle
		355	that connects it with the new process. When the new process has not yet
		356	called C<run>, then the process will exit. Otherwise, what happens depends
		357	entirely on the code that is executed.
		358
		359	=cut
238		360
239	sub new {	361	sub new {
240	my $class = shift;	362	my $class = shift;
241		363
242	my $self = bless {	364	$TEMPLATE \|\|= $class->new_exec;
243	@_	365	$TEMPLATE->fork
244	}, $class;	366	}
245		367
246	run_template;	368	=item $new_proc = $proc->fork
		369
		370	Forks C<$proc>, creating a new process, and returns the process object
		371	of the new process.
		372
		373	If any of the C<send_> functions have been called before fork, then they
		374	will be cloned in the child. For example, in a pre-forked server, you
		375	might C<send_fh> the listening socket into the template process, and then
		376	keep calling C<fork> and C<run>.
		377
		378	=cut
		379
		380	sub fork {
		381	my ($self) = @_;
		382
		383	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
		384
		385	$self->send_fh ($slave);
		386	$self->_cmd ("f");
		387
		388	AnyEvent::Fork->_new ($fh)
		389	}
		390
		391	=item my $proc = new_exec AnyEvent::Fork
		392
		393	Create a new "empty" perl interpreter process and returns its process
		394	object for further manipulation.
		395
		396	Unlike the C<new> method, this method I<always> spawns a new perl process
		397	(except in some cases, see L<AnyEvent::Fork::Early> for details). This
		398	reduces the amount of memory sharing that is possible, and is also slower.
		399
		400	You should use C<new> whenever possible, except when having a template
		401	process around is unacceptable.
		402
		403	The path to the perl interpreter is divined usign various methods - first
		404	C<$^X> is investigated to see if the path ends with something that sounds
		405	as if it were the perl interpreter. Failing this, the module falls back to
		406	using C<$Config::Config{perlpath}>.
		407
		408	=cut
		409
		410	sub new_exec {
		411	my ($self) = @_;
		412
		413	return $EARLY->fork
		414	if $EARLY;
		415
		416	# first find path of perl
		417	my $perl = $;
		418
		419	# first we try $^X, but the path must be absolute (always on win32), and end in sth.
		420	# that looks like perl. this obviously only works for posix and win32
		421	unless (
		422	(AnyEvent::Fork::Util::WIN32 \|\| $perl =~ m%^/%)
		423	&& $perl =~ m%[/\\]perl(?:[0-9]+(\.[0-9]+)+)?(\.exe)?$%i
		424	) {
		425	# if it doesn't look perlish enough, try Config
		426	require Config;
		427	$perl = $Config::Config{perlpath};
		428	$perl =~ s/(?:\Q$Config::Config{_exe}\E)?$/$Config::Config{_exe}/;
		429	}
		430
		431	require Proc::FastSpawn;
		432
		433	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
		434	Proc::FastSpawn::fd_inherit (fileno $slave);
		435
		436	# new fh's should always be set cloexec (due to $^F),
		437	# but hey, not on win32, so we always clear the inherit flag.
		438	Proc::FastSpawn::fd_inherit (fileno $fh, 0);
		439
		440	# quick. also doesn't work in win32. of course. what did you expect
		441	#local $ENV{PERL5LIB} = join ":", grep !ref, @INC;
		442	my %env = %ENV;
		443	$env{PERL5LIB} = join +(AnyEvent::Fork::Util::WIN32 ? ";" : ":"), grep !ref, @INC;
		444
		445	Proc::FastSpawn::spawn (
		446	$perl,
		447	["perl", "-MAnyEvent::Fork::Serve", "-e", "AnyEvent::Fork::Serve::me", fileno $slave, $$],
		448	[map "$_=$env{$_}", keys %env],
		449	) or die "unable to spawn AnyEvent::Fork server: $!";
		450
		451	$self->_new ($fh)
		452	}
		453
		454	=item $proc = $proc->eval ($perlcode, @args)
		455
		456	Evaluates the given C<$perlcode> as ... perl code, while setting C<@_> to
		457	the strings specified by C<@args>.
		458
		459	This call is meant to do any custom initialisation that might be required
		460	(for example, the C<require> method uses it). It's not supposed to be used
		461	to completely take over the process, use C<run> for that.
		462
		463	The code will usually be executed after this call returns, and there is no
		464	way to pass anything back to the calling process. Any evaluation errors
		465	will be reported to stderr and cause the process to exit.
		466
		467	Returns the process object for easy chaining of method calls.
		468
		469	=cut
		470
		471	sub eval {
		472	my ($self, $code, @args) = @_;
		473
		474	$self->_cmd (e => $code, @args);
247		475
248	$self	476	$self
249	}	477	}
250		478
		479	=item $proc = $proc->require ($module, ...)
		480
		481	Tries to load the given module(s) into the process
		482
		483	Returns the process object for easy chaining of method calls.
		484
		485	=cut
		486
		487	sub require {
		488	my ($self, @modules) = @_;
		489
		490	s%::%/%g for @modules;
		491	$self->eval ('require "$_.pm" for @_', @modules);
		492
		493	$self
		494	}
		495
		496	=item $proc = $proc->send_fh ($handle, ...)
		497
		498	Send one or more file handles (I<not> file descriptors) to the process,
		499	to prepare a call to C<run>.
		500
		501	The process object keeps a reference to the handles until this is done,
		502	so you must not explicitly close the handles. This is most easily
		503	accomplished by simply not storing the file handles anywhere after passing
		504	them to this method.
		505
		506	Returns the process object for easy chaining of method calls.
		507
		508	Example: pass an fh to a process, and release it without closing. it will
		509	be closed automatically when it is no longer used.
		510
		511	$proc->send_fh ($my_fh);
		512	undef $my_fh; # free the reference if you want, but DO NOT CLOSE IT
		513
		514	=cut
		515
		516	sub send_fh {
		517	my ($self, @fh) = @_;
		518
		519	for my $fh (@fh) {
		520	$self->_cmd ("h");
		521	push @{ $self->[2] }, \$fh;
		522	}
		523
		524	$self
		525	}
		526
		527	=item $proc = $proc->send_arg ($string, ...)
		528
		529	Send one or more argument strings to the process, to prepare a call to
		530	C<run>. The strings can be any octet string.
		531
		532	Returns the process object for easy chaining of emthod calls.
		533
		534	=cut
		535
		536	sub send_arg {
		537	my ($self, @arg) = @_;
		538
		539	$self->_cmd (a => @arg);
		540
		541	$self
		542	}
		543
		544	=item $proc->run ($func, $cb->($fh))
		545
		546	Enter the function specified by the fully qualified name in C<$func> in
		547	the process. The function is called with the communication socket as first
		548	argument, followed by all file handles and string arguments sent earlier
		549	via C<send_fh> and C<send_arg> methods, in the order they were called.
		550
		551	If the called function returns, the process exits.
		552
		553	Preparing the process can take time - when the process is ready, the
		554	callback is invoked with the local communications socket as argument.
		555
		556	The process object becomes unusable on return from this function.
		557
		558	If the communication socket isn't used, it should be closed on both sides,
		559	to save on kernel memory.
		560
		561	The socket is non-blocking in the parent, and blocking in the newly
		562	created process. The close-on-exec flag is set on both. Even if not used
		563	otherwise, the socket can be a good indicator for the existance of the
		564	process - if the other process exits, you get a readable event on it,
		565	because exiting the process closes the socket (if it didn't create any
		566	children using fork).
		567
		568	Example: create a template for a process pool, pass a few strings, some
		569	file handles, then fork, pass one more string, and run some code.
		570
		571	my $pool = AnyEvent::Fork
		572	->new
		573	->send_arg ("str1", "str2")
		574	->send_fh ($fh1, $fh2);
		575
		576	for (1..2) {
		577	$pool
		578	->fork
		579	->send_arg ("str3")
		580	->run ("Some::function", sub {
		581	my ($fh) = @_;
		582
		583	# fh is nonblocking, but we trust that the OS can accept these
		584	# extra 3 octets anyway.
		585	syswrite $fh, "hi #$_\n";
		586
		587	# $fh is being closed here, as we don't store it anywhere
		588	});
		589	}
		590
		591	# Some::function might look like this - all parameters passed before fork
		592	# and after will be passed, in order, after the communications socket.
		593	sub Some::function {
		594	my ($fh, $str1, $str2, $fh1, $fh2, $str3) = @_;
		595
		596	print scalar <$fh>; # prints "hi 1\n" and "hi 2\n"
		597	}
		598
		599	=cut
		600
		601	sub run {
		602	my ($self, $func, $cb) = @_;
		603
		604	$self->[0] = $cb;
		605	$self->_cmd (r => $func);
		606	}
		607
251	=back	608	=back
		609
		610	=head1 PORTABILITY NOTES
		611
		612	Native win32 perls are somewhat supported (AnyEvent::Fork::Early is a nop,
		613	and ::Template is not going to work), and it cost a lot of blood and sweat
		614	to make it so, mostly due to the bloody broken perl that nobody seems to
		615	care about. The fork emulation is a bad joke - I have yet to see something
		616	useful that you cna do with it without running into memory corruption
		617	issues or other braindamage. Hrrrr.
		618
		619	Cygwin perl is not supported at the moment, as it should implement fd
		620	passing, but doesn't, and rolling my own is hard, as cygwin doesn't
		621	support enough functionality to do it.
		622
		623	=head1 SEE ALSO
		624
		625	L<AnyEvent::Fork::Early> (to avoid executing a perl interpreter),
		626	L<AnyEvent::Fork::Template> (to create a process by forking the main
		627	program at a convenient time).
252		628
253	=head1 AUTHOR	629	=head1 AUTHOR
254		630
255	Marc Lehmann <schmorp@schmorp.de>	631	Marc Lehmann <schmorp@schmorp.de>
256	http://home.schmorp.de/	632	http://home.schmorp.de/

Diff Legend

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

Comparing AnyEvent-Fork/Fork.pm (file contents): Revision 1.3 by root, Tue Apr 2 18:00:04 2013 UTC vs. Revision 1.14 by root, Thu Apr 4 08:16:14 2013 UTC

Diff Legend

Comparing AnyEvent-Fork/Fork.pm (file contents):
Revision 1.3 by root, Tue Apr 2 18:00:04 2013 UTC vs.
Revision 1.14 by root, Thu Apr 4 08:16:14 2013 UTC