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

Comparing AnyEvent-Fork/Fork.pm (file contents):
Revision 1.6 by root, Wed Apr 3 08:47:44 2013 UTC vs.
Revision 1.12 by root, Thu Apr 4 07:27:09 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
111	time, and the memory is not shared with anything else.	171	time, and the memory is not shared with anything else.
112		172
113	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
114	option of starting and stipping it on demand.	174	option of starting and stipping it on demand.
115		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
116	=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
117	it and run the code	186	it and run the code
118		187
119	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
120	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
…		…
128	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
129	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
130	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
131	the template process.	200	the template process.
132		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
133	=item execute a new perl interpreter, load some code, run it	215	=item execute a new perl interpreter, load some code, run it
134		216
135	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
136	multiple processes.	218	multiple processes.
137		219
138	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
139	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
140	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
141	needed.	223	needed.
142		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
143	=back	234	=back
144		235
145	=head1 FUNCTIONS	236	=head1 FUNCTIONS
146		237
147	=over 4	238	=over 4
…		…
156		247
157	use AnyEvent;	248	use AnyEvent;
158	use AnyEvent::Fork::Util;	249	use AnyEvent::Fork::Util;
159	use AnyEvent::Util ();	250	use AnyEvent::Util ();
160		251
		252	our $VERSION = $AnyEvent::Fork::Util::VERSION;
		253
161	our $PERL; # the path to the perl interpreter, deduces with various forms of magic	254	our $PERL; # the path to the perl interpreter, deduces with various forms of magic
162		255
163	=item my $pool = new AnyEvent::Fork key => value...	256	=item my $pool = new AnyEvent::Fork key => value...
164		257
165	Create a new process pool. The following named parameters are supported:	258	Create a new process pool. The following named parameters are supported:
…		…
177	our $TEMPLATE;	270	our $TEMPLATE;
178		271
179	sub _cmd {	272	sub _cmd {
180	my $self = shift;	273	my $self = shift;
181		274
		275	#TODO: maybe append the packet to any existing string command already in the queue
		276
182	# ideally, we would want to use "a (w/a)*" as format string, but perl versions	277	# ideally, we would want to use "a (w/a)*" as format string, but perl versions
183	# from at least 5.8.9 to 5.16.3 are all buggy and can't unpack it.	278	# from at least 5.8.9 to 5.16.3 are all buggy and can't unpack it.
184	push @{ $self->[2] }, pack "N/a", pack "(w/a)*", @_;	279	push @{ $self->[2] }, pack "N/a", pack "(w/a)*", @_;
185		280
186	$self->[3] \|\|= AE::io $self->[1], 1, sub {	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
187	if (ref $self->[2][0]) {	285	if (ref $self->[2][0]) {
		286	# send fh
188	AnyEvent::Fork::Util::fd_send fileno $self->[1], fileno ${ $self->[2][0] }	287	AnyEvent::Fork::Util::fd_send fileno $self->[1], fileno ${ $self->[2][0] }
189	and shift @{ $self->[2] };	288	and shift @{ $self->[2] };
190		289
191	} else {	290	} else {
		291	# send string
192	my $len = syswrite $self->[1], $self->[2][0]	292	my $len = syswrite $self->[1], $self->[2][0]
193	or do { undef $self->[3]; die "AnyEvent::Fork: command write failure: $!" };	293	or do { undef $self->[3]; die "AnyEvent::Fork: command write failure: $!" };
194		294
195	substr $self->[2][0], 0, $len, "";	295	substr $self->[2][0], 0, $len, "";
196	shift @{ $self->[2] } unless length $self->[2][0];	296	shift @{ $self->[2] } unless length $self->[2][0];
197	}	297	}
198		298
199	unless (@{ $self->[2] }) {	299	unless (@{ $self->[2] }) {
200	undef $self->[3];	300	undef $self->[3];
		301	# invoke run callback
201	$self->[0]->($self->[1]) if $self->[0];	302	$self->[0]->($self->[1]) if $self->[0];
202	}	303	}
203	};	304	};
204	}	305	}
205		306
…		…
213	$fh,	314	$fh,
214	[], # write queue - strings or fd's	315	[], # write queue - strings or fd's
215	undef, # AE watcher	316	undef, # AE watcher
216	], $self;	317	], $self;
217		318
218	# my ($a, $b) = AnyEvent::Util::portable_socketpair;
219
220	# queue_cmd $template, "Iabc";
221	# push @{ $template->[2] }, \$b;
222
223	# use Coro::AnyEvent; Coro::AnyEvent::sleep 1;
224	# undef $b;
225	# die "x" . <$a>;
226
227	$self	319	$self
228	}	320	}
229		321
230	# fork template from current process, used by AnyEvent::Fork::Early/Template	322	# fork template from current process, used by AnyEvent::Fork::Early/Template
231	sub _new_fork {	323	sub _new_fork {
232	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;	324	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
		325	my $parent = $$;
		326
233	my $pid = fork;	327	my $pid = fork;
234		328
235	if ($pid eq 0) {	329	if ($pid eq 0) {
236	require AnyEvent::Fork::Serve;	330	require AnyEvent::Fork::Serve;
		331	$AnyEvent::Fork::Serve::OWNER = $parent;
237	close $fh;	332	close $fh;
		333	$0 = "$_[1] of $parent";
238	AnyEvent::Fork::Serve::serve ($slave);	334	AnyEvent::Fork::Serve::serve ($slave);
239	AnyEvent::Fork::Util::_exit 0;	335	AnyEvent::Fork::Util::_exit 0;
240	} elsif (!$pid) {	336	} elsif (!$pid) {
241	die "AnyEvent::Fork::Early/Template: unable to fork template process: $!";	337	die "AnyEvent::Fork::Early/Template: unable to fork template process: $!";
242	}	338	}
…		…
250	object for further manipulation.	346	object for further manipulation.
251		347
252	The new process is forked from a template process that is kept around	348	The new process is forked from a template process that is kept around
253	for this purpose. When it doesn't exist yet, it is created by a call to	349	for this purpose. When it doesn't exist yet, it is created by a call to
254	C<new_exec> and kept around for future calls.	350	C<new_exec> and kept around for future calls.
		351
		352	When the process object is destroyed, it will release the file handle
		353	that connects it with the new process. When the new process has not yet
		354	called C<run>, then the process will exit. Otherwise, what happens depends
		355	entirely on the code that is executed.
255		356
256	=cut	357	=cut
257		358
258	sub new {	359	sub new {
259	my $class = shift;	360	my $class = shift;
…		…
328	require Proc::FastSpawn;	429	require Proc::FastSpawn;
329		430
330	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;	431	my ($fh, $slave) = AnyEvent::Util::portable_socketpair;
331	Proc::FastSpawn::fd_inherit (fileno $slave);	432	Proc::FastSpawn::fd_inherit (fileno $slave);
332		433
		434	# new fh's should always be set cloexec (due to $^F),
		435	# but hey, not on win32, so we always clear the inherit flag.
		436	Proc::FastSpawn::fd_inherit (fileno $fh, 0);
		437
333	# quick. also doesn't work in win32. of course. what did you expect	438	# quick. also doesn't work in win32. of course. what did you expect
334	#local $ENV{PERL5LIB} = join ":", grep !ref, @INC;	439	#local $ENV{PERL5LIB} = join ":", grep !ref, @INC;
335	my %env = %ENV;	440	my %env = %ENV;
336	$env{PERL5LIB} = join ":", grep !ref, @INC;	441	$env{PERL5LIB} = join +(AnyEvent::Fork::Util::WIN32 ? ";" : ":"), grep !ref, @INC;
337		442
338	Proc::FastSpawn::spawn (	443	Proc::FastSpawn::spawn (
339	$perl,	444	$perl,
340	["perl", "-MAnyEvent::Fork::Serve", "-e", "AnyEvent::Fork::Serve::me", fileno $slave],	445	["perl", "-MAnyEvent::Fork::Serve", "-e", "AnyEvent::Fork::Serve::me", fileno $slave, $$],
341	[map "$_=$env{$_}", keys %env],	446	[map "$_=$env{$_}", keys %env],
342	) or die "unable to spawn AnyEvent::Fork server: $!";	447	) or die "unable to spawn AnyEvent::Fork server: $!";
343		448
344	$self->_new ($fh)	449	$self->_new ($fh)
345	}	450	}
346		451
		452	=item $proc = $proc->eval ($perlcode, @args)
		453
		454	Evaluates the given C<$perlcode> as ... perl code, while setting C<@_> to
		455	the strings specified by C<@args>.
		456
		457	This call is meant to do any custom initialisation that might be required
		458	(for example, the C<require> method uses it). It's not supposed to be used
		459	to completely take over the process, use C<run> for that.
		460
		461	The code will usually be executed after this call returns, and there is no
		462	way to pass anything back to the calling process. Any evaluation errors
		463	will be reported to stderr and cause the process to exit.
		464
		465	Returns the process object for easy chaining of method calls.
		466
		467	=cut
		468
		469	sub eval {
		470	my ($self, $code, @args) = @_;
		471
		472	$self->_cmd (e => $code, @args);
		473
		474	$self
		475	}
		476
347	=item $proc = $proc->require ($module, ...)	477	=item $proc = $proc->require ($module, ...)
348		478
349	Tries to load the given modules into the process	479	Tries to load the given module(s) into the process
350		480
351	Returns the process object for easy chaining of method calls.	481	Returns the process object for easy chaining of method calls.
		482
		483	=cut
		484
		485	sub require {
		486	my ($self, @modules) = @_;
		487
		488	s%::%/%g for @modules;
		489	$self->eval ('require "$_.pm" for @_', @modules);
		490
		491	$self
		492	}
352		493
353	=item $proc = $proc->send_fh ($handle, ...)	494	=item $proc = $proc->send_fh ($handle, ...)
354		495
355	Send one or more file handles (I<not> file descriptors) to the process,	496	Send one or more file handles (I<not> file descriptors) to the process,
356	to prepare a call to C<run>.	497	to prepare a call to C<run>.
…		…
360	accomplished by simply not storing the file handles anywhere after passing	501	accomplished by simply not storing the file handles anywhere after passing
361	them to this method.	502	them to this method.
362		503
363	Returns the process object for easy chaining of method calls.	504	Returns the process object for easy chaining of method calls.
364		505
		506	Example: pass an fh to a process, and release it without closing. it will
		507	be closed automatically when it is no longer used.
		508
		509	$proc->send_fh ($my_fh);
		510	undef $my_fh; # free the reference if you want, but DO NOT CLOSE IT
		511
365	=cut	512	=cut
366		513
367	sub send_fh {	514	sub send_fh {
368	my ($self, @fh) = @_;	515	my ($self, @fh) = @_;
369		516
…		…
410	to save on kernel memory.	557	to save on kernel memory.
411		558
412	The socket is non-blocking in the parent, and blocking in the newly	559	The socket is non-blocking in the parent, and blocking in the newly
413	created process. The close-on-exec flag is set on both. Even if not used	560	created process. The close-on-exec flag is set on both. Even if not used
414	otherwise, the socket can be a good indicator for the existance of the	561	otherwise, the socket can be a good indicator for the existance of the
415	process - if the othe rprocess exits, you get a readable event on it,	562	process - if the other process exits, you get a readable event on it,
416	because exiting the process closes the socket (if it didn't create any	563	because exiting the process closes the socket (if it didn't create any
417	children using fork).	564	children using fork).
418		565
		566	Example: create a template for a process pool, pass a few strings, some
		567	file handles, then fork, pass one more string, and run some code.
		568
		569	my $pool = AnyEvent::Fork
		570	->new
		571	->send_arg ("str1", "str2")
		572	->send_fh ($fh1, $fh2);
		573
		574	for (1..2) {
		575	$pool
		576	->fork
		577	->send_arg ("str3")
		578	->run ("Some::function", sub {
		579	my ($fh) = @_;
		580
		581	# fh is nonblocking, but we trust that the OS can accept these
		582	# extra 3 octets anyway.
		583	syswrite $fh, "hi #$_\n";
		584
		585	# $fh is being closed here, as we don't store it anywhere
		586	});
		587	}
		588
		589	# Some::function might look like this - all parameters passed before fork
		590	# and after will be passed, in order, after the communications socket.
		591	sub Some::function {
		592	my ($fh, $str1, $str2, $fh1, $fh2, $str3) = @_;
		593
		594	print scalar <$fh>; # prints "hi 1\n" and "hi 2\n"
		595	}
		596
419	=cut	597	=cut
420		598
421	sub run {	599	sub run {
422	my ($self, $func, $cb) = @_;	600	my ($self, $func, $cb) = @_;
423		601
424	$self->[0] = $cb;	602	$self->[0] = $cb;
425	$self->_cmd ("r", $func);	603	$self->_cmd (r => $func);
426	}	604	}
427		605
428	=back	606	=back
		607
		608	=head1 PORTABILITY NOTES
		609
		610	Native win32 perls are somewhat supported (AnyEvent::Fork::Early is a nop,
		611	and ::Template is not going to work), and it cost a lot of blood and sweat
		612	to make it so, mostly due to the bloody broken perl that nobody seems to
		613	care about. The fork emulation is a bad joke - I have yet to see something
		614	useful that you cna do with it without running into memory corruption
		615	issues or other braindamage. Hrrrr.
		616
		617	Cygwin perl is not supported at the moment, as it should implement fd
		618	passing, but doesn't, and rolling my own is hard, as cygwin doesn't
		619	support enough functionality to do it.
429		620
430	=head1 AUTHOR	621	=head1 AUTHOR
431		622
432	Marc Lehmann <schmorp@schmorp.de>	623	Marc Lehmann <schmorp@schmorp.de>
433	http://home.schmorp.de/	624	http://home.schmorp.de/

Diff Legend

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

Comparing AnyEvent-Fork/Fork.pm (file contents): Revision 1.6 by root, Wed Apr 3 08:47:44 2013 UTC vs. Revision 1.12 by root, Thu Apr 4 07:27:09 2013 UTC

Diff Legend

Comparing AnyEvent-Fork/Fork.pm (file contents):
Revision 1.6 by root, Wed Apr 3 08:47:44 2013 UTC vs.
Revision 1.12 by root, Thu Apr 4 07:27:09 2013 UTC