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

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents):
Revision 1.3 by root, Wed Apr 17 17:16:48 2013 UTC vs.
Revision 1.6 by root, Wed Apr 17 19:43:48 2013 UTC

…		…
34	concurrently in the child, using AnyEvent.	34	concurrently in the child, using AnyEvent.
35		35
36	It also implements an asynchronous event mechanism from the child to the	36	It also implements an asynchronous event mechanism from the child to the
37	parent, that could be used for progress indications or other information.	37	parent, that could be used for progress indications or other information.
38		38
		39	=head1 EXAMPLES
		40
		41	=head2 Synchronous Backend
		42
		43	Here is a simple example that implements a backend that executes C<unlink>
		44	and C<rmdir> calls, and reports their status back. It also reports the
		45	number of requests it has processed every three requests, which is clearly
		46	silly, but illustrates the use of events.
		47
		48	First the parent process:
		49
		50	use AnyEvent;
		51	use AnyEvent::Fork;
		52	use AnyEvent::Fork::RPC;
		53
		54	my $done = AE::cv;
		55
		56	my $rpc = AnyEvent::Fork
		57	->new
		58	->require ("MyWorker")
		59	->AnyEvent::Fork::RPC::run ("MyWorker::run",
		60	on_error => sub { warn "FATAL: $_[0]"; exit 1 },
		61	on_event => sub { warn "$_[0] requests handled\n" },
		62	on_destroy => $done,
		63	);
		64
		65	for my $id (1..6) {
		66	$rpc->(rmdir => "/tmp/somepath/$id", sub {
		67	$_[0]
		68	or warn "/tmp/somepath/$id: $_[1]\n";
		69	});
		70	}
		71
		72	undef $rpc;
		73
		74	$done->recv;
		75
		76	The parent creates the process, queues a few rmdir's. It then forgets
		77	about the C<$rpc> object, so that the child exits after it has handled the
		78	requests, and then it waits till the requests have been handled.
		79
		80	The child is implemented using a separate module, C<MyWorker>, shown here:
		81
		82	package MyWorker;
		83
		84	my $count;
		85
		86	sub run {
		87	my ($cmd, $path) = @_;
		88
		89	AnyEvent::Fork::RPC::event ($count)
		90	unless ++$count % 3;
		91
		92	my $status = $cmd eq "rmdir" ? rmdir $path
		93	: $cmd eq "unlink" ? unlink $path
		94	: die "fatal error, illegal command '$cmd'";
		95
		96	$status or (0, "$!")
		97	}
		98
		99	1
		100
		101	The C<run> function first sends a "progress" event every three calls, and
		102	then executes C<rmdir> or C<unlink>, depending on the first parameter (or
		103	dies with a fatal error - obviously, you must never let this happen :).
		104
		105	Eventually it returns the status value true if the command was successful,
		106	or the status value 0 and the stringified error message.
		107
		108	On my system, running the first code fragment with the given
		109	F<MyWorker.pm> in the current directory yields:
		110
		111	/tmp/somepath/1: No such file or directory
		112	/tmp/somepath/2: No such file or directory
		113	3 requests handled
		114	/tmp/somepath/3: No such file or directory
		115	/tmp/somepath/4: No such file or directory
		116	/tmp/somepath/5: No such file or directory
		117	6 requests handled
		118	/tmp/somepath/6: No such file or directory
		119
		120	Obviously, none of the directories I am trying to delete even exist. Also,
		121	the events and responses are processed in exactly the same order as
		122	they were created in the child, which is true for both synchronous and
		123	asynchronous backends.
		124
		125	Note that the parentheses in the call to C<AnyEvent::Fork::RPC::event> are
		126	not optional. That is because the function isn't defined when the code is
		127	compiled. You can make sure it is visible by pre-loading the correct
		128	backend module in the call to C<require>:
		129
		130	->require ("AnyEvent::Fork::RPC::Sync", "MyWorker")
		131
		132	Since the backend module declares the C<event> function, loading it first
		133	ensures that perl will correctly interpret calls to it.
		134
		135	And as a final remark, there is a fine module on CPAN that can
		136	asynchronously C<rmdir> and C<unlink> and a lot more, and more efficiently
		137	than this example, namely L<IO::AIO>.
		138
39	=head1 PARENT PROCESS USAGE	139	=head1 PARENT PROCESS USAGE
40		140
41	This module exports nothing, and only implements a single function:	141	This module exports nothing, and only implements a single function:
42		142
43	=over 4	143	=over 4
…		…
92	Called for every call to the C<AnyEvent::Fork::RPC::event> function in the	192	Called for every call to the C<AnyEvent::Fork::RPC::event> function in the
93	child, with the arguments of that function passed to the callback.	193	child, with the arguments of that function passed to the callback.
94		194
95	Also called on errors when no C<on_error> handler is provided.	195	Also called on errors when no C<on_error> handler is provided.
96		196
		197	=item on_destroy => $cb->()
		198
		199	Called when the C<$rpc> object has been destroyed and all requests have
		200	been successfully handled. This is useful when you queue some requests and
		201	want the child to go away after it has handled them. The problem is that
		202	the parent must not exit either until all requests have been handled, and
		203	this can be accomplished by waiting for this callback.
		204
97	=item init => $function (default none)	205	=item init => $function (default none)
98		206
99	When specified (by name), this function is called in the child as the very	207	When specified (by name), this function is called in the child as the very
100	first thing when taking over the process, with all the arguments normally	208	first thing when taking over the process, with all the arguments normally
101	passed to the C<AnyEvent::Fork::run> function, except the communications	209	passed to the C<AnyEvent::Fork::run> function, except the communications
102	socket.	210	socket.
103		211
104	It can be used to do one-time things in the child such as storing passed	212	It can be used to do one-time things in the child such as storing passed
105	parameters or opening database connections.	213	parameters or opening database connections.
106		214
		215	It is called very early - before the serialisers are created or the
		216	C<$function> name is resolved into a function reference, so it could be
		217	used to load any modules that provide the serialiser or function. It can
		218	not, however, create events.
		219
107	=item async => $boolean (default: 0)	220	=item async => $boolean (default: 0)
108		221
109	The default server used in the child does all I/O blockingly, and only	222	The default server used in the child does all I/O blockingly, and only
110	allows a single RPC call to execute concurrently.	223	allows a single RPC call to execute concurrently.
111		224
…		…
116	the calling semantics of the returned C<$rpc> function.	229	the calling semantics of the returned C<$rpc> function.
117		230
118	If you want to pre-load the actual back-end modules to enable memory	231	If you want to pre-load the actual back-end modules to enable memory
119	sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for	232	sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for
120	synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.	233	synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
		234
		235	If you use a template process and want to fork both sync and async
		236	children, then it is permissible to load both modules.
121		237
122	=item serialiser => $string (default: '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })')	238	=item serialiser => $string (default: '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })')
123		239
124	All arguments, result data and event data have to be serialised to be	240	All arguments, result data and event data have to be serialised to be
125	transferred between the processes. For this, they have to be frozen and	241	transferred between the processes. For this, they have to be frozen and
…		…
142		258
143	=cut	259	=cut
144		260
145	our $STRING_SERIALISER = '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })';	261	our $STRING_SERIALISER = '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })';
146		262
147	# ideally, we want (SvLEN - SvCUR) \|\| 1024 or somesuch...
148	sub rlen($) { ($_[0] < 384 ? 512 + 16 : 2 << int +(log $_[0] + 512) / log 2) - $_[0] - 16 }
149
150	sub run {	263	sub run {
151	my ($self, $function, %arg) = @_;	264	my ($self, $function, %arg) = @_;
152		265
153	my $serialiser = delete $arg{serialiser} \|\| $STRING_SERIALISER;	266	my $serialiser = delete $arg{serialiser} \|\| $STRING_SERIALISER;
154	my $on_event = delete $arg{on_event};	267	my $on_event = delete $arg{on_event};
155	my $on_error = delete $arg{on_error};	268	my $on_error = delete $arg{on_error};
		269	my $on_destroy = delete $arg{on_destroy};
156		270
157	# default for on_error is to on_event, if specified	271	# default for on_error is to on_event, if specified
158	$on_error \|\|= $on_event	272	$on_error \|\|= $on_event
159	? sub { $on_event->(error => shift) }	273	? sub { $on_event->(error => shift) }
160	: sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" };	274	: sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" };
…		…
162	# default for on_event is to raise an error	276	# default for on_event is to raise an error
163	$on_event \|\|= sub { $on_error->("event received, but no on_event handler") };	277	$on_event \|\|= sub { $on_error->("event received, but no on_event handler") };
164		278
165	my ($f, $t) = eval $serialiser; die $@ if $@;	279	my ($f, $t) = eval $serialiser; die $@ if $@;
166		280
167	my (@rcb, $fh, $shutdown, $wbuf, $ww, $rbuf, $rw);	281	my (@rcb, $fh, $shutdown, $wbuf, $ww, $rw);
		282	my ($rlen, $rbuf) = 512 - 16;
168		283
169	my $wcb = sub {	284	my $wcb = sub {
170	my $len = syswrite $fh, $wbuf;	285	my $len = syswrite $fh, $wbuf;
171		286
172	if (!defined $len) {	287	if (!defined $len) {
…		…
189	$self->require ($module)	304	$self->require ($module)
190	->send_arg ($function, $arg{init}, $serialiser)	305	->send_arg ($function, $arg{init}, $serialiser)
191	->run ("$module\::run", sub {	306	->run ("$module\::run", sub {
192	$fh = shift;	307	$fh = shift;
193	$rw = AE::io $fh, 0, sub {	308	$rw = AE::io $fh, 0, sub {
		309	$rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
194	my $len = sysread $fh, $rbuf, rlen length $rbuf, length $rbuf;	310	my $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
195		311
196	if ($len) {	312	if ($len) {
197	while (5 <= length $rbuf) {	313	while (5 <= length $rbuf) {
198	$len = unpack "L", $rbuf;	314	$len = unpack "L", $rbuf;
199	4 + $len <= length $rbuf	315	4 + $len <= length $rbuf
…		…
211	$on_error->("unexpected data from child");	327	$on_error->("unexpected data from child");
212	}	328	}
213	}	329	}
214	} elsif (defined $len) {	330	} elsif (defined $len) {
215	undef $rw; undef $ww; # it ends here	331	undef $rw; undef $ww; # it ends here
		332
		333	if (@rcb) {
216	$on_error->("unexpected eof")	334	$on_error->("unexpected eof");
217	if @rcb;	335	} else {
		336	$on_destroy->();
		337	}
218	} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {	338	} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
219	undef $rw; undef $ww; # it ends here	339	undef $rw; undef $ww; # it ends here
220	$on_error->("read: $!");	340	$on_error->("read: $!");
221	}	341	}
222	};	342	};
…		…
237	$wbuf .= pack "L/a*", &$f;	357	$wbuf .= pack "L/a*", &$f;
238	$ww \|\|= $fh && AE::io $fh, 1, $wcb;	358	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
239	}	359	}
240	}	360	}
241		361
		362	=item $rpc->(..., $cb->(...))
		363
		364	The RPC object returned by C<AnyEvent::Fork::RPC::run> is actually a code
		365	reference. There are two things you can do with it: call it, and let it go
		366	out of scope (let it get destroyed).
		367
		368	If C<async> was false when C<$rpc> was created (the default), then, if you
		369	call C<$rpc>, the C<$function> is invoked with all arguments passed to
		370	C<$rpc> except the last one (the callback). When the function returns, the
		371	callback will be invoked with all the return values.
		372
		373	If C<async> was true, then the C<$function> receives an additional
		374	initial argument, the result callback. In this case, returning from
		375	C<$function> does nothing - the function only counts as "done" when the
		376	result callback is called, and any arguments passed to it are considered
		377	the return values. This makes it possible to "return" from event handlers
		378	or e.g. Coro threads.
		379
		380	The other thing that can be done with the RPC object is to destroy it. In
		381	this case, the child process will execute all remaining RPC calls, report
		382	their results, and then exit.
		383
242	=back	384	=back
243		385
244	=head1 CHILD PROCESS USAGE	386	=head1 CHILD PROCESS USAGE
245		387
246	These functions are not available in this module. They are only available	388	The following function is not available in this module. They are only
247	in the namespace of this module when the child is running, without	389	available in the namespace of this module when the child is running,
248	having to load any extra module. They are part of the child-side API of	390	without having to load any extra modules. They are part of the child-side
249	L<AnyEvent::Fork::RPC>.	391	API of L<AnyEvent::Fork::RPC>.
250		392
251	=over 4	393	=over 4
252		394
253	=item AnyEvent::Fork::RPC::event ...	395	=item AnyEvent::Fork::RPC::event ...
254		396

Diff Legend

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

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents): Revision 1.3 by root, Wed Apr 17 17:16:48 2013 UTC vs. Revision 1.6 by root, Wed Apr 17 19:43:48 2013 UTC

Diff Legend

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents):
Revision 1.3 by root, Wed Apr 17 17:16:48 2013 UTC vs.
Revision 1.6 by root, Wed Apr 17 19:43:48 2013 UTC