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

Comparing AnyEvent-Fork/Fork.pm (file contents):
Revision 1.4 by root, Wed Apr 3 07:35:57 2013 UTC vs.
Revision 1.10 by root, Thu Apr 4 06:09:15 2013 UTC

…		…
1	=head1 NAME	1	=head1 NAME
2		2
3	AnyEvent::Fork - everything you wanted to use fork() for, but couldn't	3	AnyEvent::Fork - everything you wanted to use fork() for, but couldn't
4		4
		5	ATTENTION, this is a very early release, and very untested. Consider it a
		6	technology preview.
		7
5	=head1 SYNOPSIS	8	=head1 SYNOPSIS
6		9
7	use AnyEvent::Fork;	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 new processes, without actually forking	71	This module allows you to create new processes, without actually forking
12	them from your current process (avoiding the problems of forking), but	72	them from your current process (avoiding the problems of forking), but
15	It can be used to create new worker processes or new independent	75	It can be used to create new worker processes or new independent
16	subprocesses for short- and long-running jobs, process pools (e.g. for use	76	subprocesses for short- and long-running jobs, process pools (e.g. for use
17	in pre-forked servers) but also to spawn new external processes (such as	77	in pre-forked servers) but also to spawn new external processes (such as
18	CGI scripts from a webserver), which can be faster (and more well behaved)	78	CGI scripts from a webserver), which can be faster (and more well behaved)
19	than using fork+exec in big processes.	79	than using fork+exec in big processes.
		80
		81	Special care has been taken to make this module useful from other modules,
		82	while still supporting specialised environments such as L<App::Staticperl>
		83	or L<PAR::Packer>.
20		84
21	=head1 PROBLEM STATEMENT	85	=head1 PROBLEM STATEMENT
22		86
23	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
24	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
…		…
107	time, and the memory is not shared with anything else.	171	time, and the memory is not shared with anything else.
108		172
109	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
110	option of starting and stipping it on demand.	174	option of starting and stipping it on demand.
111		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
112	=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
113	it and run the code	186	it and run the code
114		187
115	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
116	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
…		…
124	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
125	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
126	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
127	the template process.	200	the template process.
128		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
129	=item execute a new perl interpreter, load some code, run it	215	=item execute a new perl interpreter, load some code, run it
130		216
131	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
132	multiple processes.	218	multiple processes.
133		219
134	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
135	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
136	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
137	needed.	223	needed.
138		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
139	=back	234	=back
140		235
141	=head1 FUNCTIONS	236	=head1 FUNCTIONS
142		237
143	=over 4	238	=over 4
…		…
164		259
165	=back	260	=back
166		261
167	=cut	262	=cut
168		263
		264	# the early fork template process
		265	our $EARLY;
		266
169	# the empty template process	267	# the empty template process
170	our $TEMPLATE;	268	our $TEMPLATE;
171		269
172	sub _cmd {	270	sub _cmd {
173	my $self = shift;	271	my $self = shift;
174		272
		273	#TODO: maybe append the packet to any existing string command already in the queue
		274
175	# ideally, we would want to use "a (w/a)*" as format string, but perl versions	275	# ideally, we would want to use "a (w/a)*" as format string, but perl versions
176	# form at least 5.8.9 to 5.16.3 are all buggy and can't unpack it.	276	# from at least 5.8.9 to 5.16.3 are all buggy and can't unpack it.
177	push @{ $self->[2] }, pack "N/a", pack "(w/a)*", @_;	277	push @{ $self->[2] }, pack "N/a", pack "(w/a)*", @_;
178		278
179	$self->[3] \|\|= AE::io $self->[1], 1, sub {	279	$self->[3] \|\|= AE::io $self->[1], 1, sub {
		280	# send the next "thing" in the queue - either a reference to an fh,
		281	# or a plain string.
		282
180	if (ref $self->[2][0]) {	283	if (ref $self->[2][0]) {
		284	# send fh
181	AnyEvent::Fork::Util::fd_send fileno $self->[1], fileno ${ $self->[2][0] }	285	AnyEvent::Fork::Util::fd_send fileno $self->[1], fileno ${ $self->[2][0] }
182	and shift @{ $self->[2] };	286	and shift @{ $self->[2] };
		287
183	} else {	288	} else {
		289	# send string
184	my $len = syswrite $self->[1], $self->[2][0]	290	my $len = syswrite $self->[1], $self->[2][0]
185	or do { undef $self->[3]; die "AnyEvent::Fork: command write failure: $!" };	291	or do { undef $self->[3]; die "AnyEvent::Fork: command write failure: $!" };
		292
186	substr $self->[2][0], 0, $len, "";	293	substr $self->[2][0], 0, $len, "";
187	shift @{ $self->[2] } unless length $self->[2][0];	294	shift @{ $self->[2] } unless length $self->[2][0];
188	}	295	}
189		296
190	unless (@{ $self->[2] }) {	297	unless (@{ $self->[2] }) {
191	undef $self->[3];	298	undef $self->[3];
		299	# invoke run callback
192	$self->[0]->($self->[1]) if $self->[0];	300	$self->[0]->($self->[1]) if $self->[0];
193	}	301	}
194	};	302	};
195	}	303	}
196		304
197	sub _new {	305	sub _new {
198	my ($self, $fh) = @_;	306	my ($self, $fh) = @_;
		307
		308	AnyEvent::Util::fh_nonblocking $fh, 1;
199		309
200	$self = bless [	310	$self = bless [
201	undef, # run callback	311	undef, # run callback
202	$fh,	312	$fh,
203	[], # write queue - strings or fd's	313	[], # write queue - strings or fd's
204	undef, # AE watcher	314	undef, # AE watcher
205	], $self;	315	], $self;
206		316
207	# my ($a, $b) = AnyEvent::Util::portable_socketpair;
208
209	# queue_cmd $template, "Iabc";
210	# push @{ $template->[2] }, \$b;
211
212	# use Coro::AnyEvent; Coro::AnyEvent::sleep 1;
213	# undef $b;
214	# die "x" . <$a>;
215
216	$self	317	$self
		318	}
		319
		320	# fork template from current process, used by AnyEvent::Fork::Early/Template
		321	sub _new_fork {
		322	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
		323	my $parent = $$;
		324
		325	my $pid = fork;
		326
		327	if ($pid eq 0) {
		328	require AnyEvent::Fork::Serve;
		329	$AnyEvent::Fork::Serve::OWNER = $parent;
		330	close $fh;
		331	$0 = "$_[1] of $parent";
		332	AnyEvent::Fork::Serve::serve ($slave);
		333	AnyEvent::Fork::Util::_exit 0;
		334	} elsif (!$pid) {
		335	die "AnyEvent::Fork::Early/Template: unable to fork template process: $!";
		336	}
		337
		338	AnyEvent::Fork->_new ($fh)
217	}	339	}
218		340
219	=item my $proc = new AnyEvent::Fork	341	=item my $proc = new AnyEvent::Fork
220		342
221	Create a new "empty" perl interpreter process and returns its process	343	Create a new "empty" perl interpreter process and returns its process
222	object for further manipulation.	344	object for further manipulation.
223		345
224	The new process is forked from a template process that is kept around	346	The new process is forked from a template process that is kept around
225	for this purpose. When it doesn't exist yet, it is created by a call to	347	for this purpose. When it doesn't exist yet, it is created by a call to
226	C<new_exec> and kept around for future calls.	348	C<new_exec> and kept around for future calls.
		349
		350	When the process object is destroyed, it will release the file handle
		351	that connects it with the new process. When the new process has not yet
		352	called C<run>, then the process will exit. Otherwise, what happens depends
		353	entirely on the code that is executed.
227		354
228	=cut	355	=cut
229		356
230	sub new {	357	sub new {
231	my $class = shift;	358	my $class = shift;
…		…
252	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;	379	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
253		380
254	$self->send_fh ($slave);	381	$self->send_fh ($slave);
255	$self->_cmd ("f");	382	$self->_cmd ("f");
256		383
257	AnyEvent::Util::fh_nonblocking $fh, 1;
258
259	AnyEvent::Fork->_new ($fh)	384	AnyEvent::Fork->_new ($fh)
260	}	385	}
261		386
262	=item my $proc = new_exec AnyEvent::Fork	387	=item my $proc = new_exec AnyEvent::Fork
263		388
…		…
278		403
279	=cut	404	=cut
280		405
281	sub new_exec {	406	sub new_exec {
282	my ($self) = @_;	407	my ($self) = @_;
		408
		409	return $EARLY->fork
		410	if $EARLY;
283		411
284	# first find path of perl	412	# first find path of perl
285	my $perl = $;	413	my $perl = $;
286		414
287	# first we try $^X, but the path must be absolute (always on win32), and end in sth.	415	# first we try $^X, but the path must be absolute (always on win32), and end in sth.
…		…
297	}	425	}
298		426
299	require Proc::FastSpawn;	427	require Proc::FastSpawn;
300		428
301	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;	429	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
302	AnyEvent::Util::fh_nonblocking $fh, 1;
303	Proc::FastSpawn::fd_inherit (fileno $slave);	430	Proc::FastSpawn::fd_inherit (fileno $slave);
		431
		432	# new fh's should always be set cloexec (due to $^F),
		433	# but hey, not on win32, so we always clear the inherit flag.
		434	Proc::FastSpawn::fd_inherit (fileno $fh, 0);
304		435
305	# quick. also doesn't work in win32. of course. what did you expect	436	# quick. also doesn't work in win32. of course. what did you expect
306	#local $ENV{PERL5LIB} = join ":", grep !ref, @INC;	437	#local $ENV{PERL5LIB} = join ":", grep !ref, @INC;
307	my %env = %ENV;	438	my %env = %ENV;
308	$env{PERL5LIB} = join ":", grep !ref, @INC;	439	$env{PERL5LIB} = join +(AnyEvent::Fork::Util::WIN32 ? ";" : ":"), grep !ref, @INC;
309		440
310	Proc::FastSpawn::spawn (	441	Proc::FastSpawn::spawn (
311	$perl,	442	$perl,
312	["perl", "-MAnyEvent::Fork::Serve", "-e", "AnyEvent::Fork::Serve::me", fileno $slave],	443	["perl", "-MAnyEvent::Fork::Serve", "-e", "AnyEvent::Fork::Serve::me", fileno $slave, $$],
313	[map "$_=$env{$_}", keys %env],	444	[map "$_=$env{$_}", keys %env],
314	) or die "unable to spawn AnyEvent::Fork server: $!";	445	) or die "unable to spawn AnyEvent::Fork server: $!";
315		446
316	$self->_new ($fh)	447	$self->_new ($fh)
317	}	448	}
318		449
		450	=item $proc = $proc->eval ($perlcode, @args)
		451
		452	Evaluates the given C<$perlcode> as ... perl code, while setting C<@_> to
		453	the strings specified by C<@args>.
		454
		455	This call is meant to do any custom initialisation that might be required
		456	(for example, the C<require> method uses it). It's not supposed to be used
		457	to completely take over the process, use C<run> for that.
		458
		459	The code will usually be executed after this call returns, and there is no
		460	way to pass anything back to the calling process. Any evaluation errors
		461	will be reported to stderr and cause the process to exit.
		462
		463	Returns the process object for easy chaining of method calls.
		464
		465	=cut
		466
		467	sub eval {
		468	my ($self, $code, @args) = @_;
		469
		470	$self->_cmd (e => $code, @args);
		471
		472	$self
		473	}
		474
319	=item $proc = $proc->require ($module, ...)	475	=item $proc = $proc->require ($module, ...)
320		476
321	Tries to load the given modules into the process	477	Tries to load the given module(s) into the process
322		478
323	Returns the process object for easy chaining of method calls.	479	Returns the process object for easy chaining of method calls.
		480
		481	=cut
		482
		483	sub require {
		484	my ($self, @modules) = @_;
		485
		486	s%::%/%g for @modules;
		487	$self->eval ('require "$_.pm" for @_', @modules);
		488
		489	$self
		490	}
324		491
325	=item $proc = $proc->send_fh ($handle, ...)	492	=item $proc = $proc->send_fh ($handle, ...)
326		493
327	Send one or more file handles (I<not> file descriptors) to the process,	494	Send one or more file handles (I<not> file descriptors) to the process,
328	to prepare a call to C<run>.	495	to prepare a call to C<run>.
…		…
332	accomplished by simply not storing the file handles anywhere after passing	499	accomplished by simply not storing the file handles anywhere after passing
333	them to this method.	500	them to this method.
334		501
335	Returns the process object for easy chaining of method calls.	502	Returns the process object for easy chaining of method calls.
336		503
		504	Example: pass an fh to a process, and release it without closing. it will
		505	be closed automatically when it is no longer used.
		506
		507	$proc->send_fh ($my_fh);
		508	undef $my_fh; # free the reference if you want, but DO NOT CLOSE IT
		509
337	=cut	510	=cut
338		511
339	sub send_fh {	512	sub send_fh {
340	my ($self, @fh) = @_;	513	my ($self, @fh) = @_;
341		514
…		…
382	to save on kernel memory.	555	to save on kernel memory.
383		556
384	The socket is non-blocking in the parent, and blocking in the newly	557	The socket is non-blocking in the parent, and blocking in the newly
385	created process. The close-on-exec flag is set on both. Even if not used	558	created process. The close-on-exec flag is set on both. Even if not used
386	otherwise, the socket can be a good indicator for the existance of the	559	otherwise, the socket can be a good indicator for the existance of the
387	process - if the othe rprocess exits, you get a readable event on it,	560	process - if the other process exits, you get a readable event on it,
388	because exiting the process closes the socket (if it didn't create any	561	because exiting the process closes the socket (if it didn't create any
389	children using fork).	562	children using fork).
390		563
		564	Example: create a template for a process pool, pass a few strings, some
		565	file handles, then fork, pass one more string, and run some code.
		566
		567	my $pool = AnyEvent::Fork
		568	->new
		569	->send_arg ("str1", "str2")
		570	->send_fh ($fh1, $fh2);
		571
		572	for (1..2) {
		573	$pool
		574	->fork
		575	->send_arg ("str3")
		576	->run ("Some::function", sub {
		577	my ($fh) = @_;
		578
		579	# fh is nonblocking, but we trust that the OS can accept these
		580	# extra 3 octets anyway.
		581	syswrite $fh, "hi #$_\n";
		582
		583	# $fh is being closed here, as we don't store it anywhere
		584	});
		585	}
		586
		587	# Some::function might look like this - all parameters passed before fork
		588	# and after will be passed, in order, after the communications socket.
		589	sub Some::function {
		590	my ($fh, $str1, $str2, $fh1, $fh2, $str3) = @_;
		591
		592	print scalar <$fh>; # prints "hi 1\n" and "hi 2\n"
		593	}
		594
391	=cut	595	=cut
392		596
393	sub run {	597	sub run {
394	my ($self, $func, $cb) = @_;	598	my ($self, $func, $cb) = @_;
395		599
396	$self->[0] = $cb;	600	$self->[0] = $cb;
397	$self->_cmd ("r", $func);	601	$self->_cmd (r => $func);
398	}	602	}
399		603
400	=back	604	=back
		605
		606	=head1 PORTABILITY NOTES
		607
		608	Native win32 perls are somewhat supported (AnyEvent::Fork::Early is a nop,
		609	and ::Template is not going to work), and it cost a lot of blood and sweat
		610	to make it so, mostly due to the bloody broken perl that nobody seems to
		611	care about. The fork emulation is a bad joke - I have yet to see something
		612	useful that you cna do with it without running into memory corruption
		613	issues or other braindamage. Hrrrr.
		614
		615	Cygwin perl is not supported at the moment, as it should implement fd
		616	passing, but doesn't, and rolling my own is hard, as cygwin doesn't
		617	support enough functionality to do it.
401		618
402	=head1 AUTHOR	619	=head1 AUTHOR
403		620
404	Marc Lehmann <schmorp@schmorp.de>	621	Marc Lehmann <schmorp@schmorp.de>
405	http://home.schmorp.de/	622	http://home.schmorp.de/

Diff Legend

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

Comparing AnyEvent-Fork/Fork.pm (file contents): Revision 1.4 by root, Wed Apr 3 07:35:57 2013 UTC vs. Revision 1.10 by root, Thu Apr 4 06:09:15 2013 UTC

Diff Legend

Comparing AnyEvent-Fork/Fork.pm (file contents):
Revision 1.4 by root, Wed Apr 3 07:35:57 2013 UTC vs.
Revision 1.10 by root, Thu Apr 4 06:09:15 2013 UTC