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

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents):
Revision 1.2 by root, Wed Apr 17 17:08:16 2013 UTC vs.
Revision 1.10 by root, Wed Apr 17 22:04:49 2013 UTC

…		…
2		2
3	AnyEvent::Fork::RPC - simple RPC extension for AnyEvent::Fork	3	AnyEvent::Fork::RPC - simple RPC extension for AnyEvent::Fork
4		4
5	=head1 SYNOPSIS	5	=head1 SYNOPSIS
6		6
7	use AnyEvent::Fork;
8	use AnyEvent::Fork::RPC;	7	use AnyEvent::Fork::RPC;
		8	# use AnyEvent::Fork is not needed
9		9
10	my $rpc = AnyEvent::Fork	10	my $rpc = AnyEvent::Fork
11	->new	11	->new
12	->require ("MyModule")	12	->require ("MyModule")
13	->AnyEvent::Fork::RPC::run (	13	->AnyEvent::Fork::RPC::run (
…		…
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	Loading this module also always loads L<AnyEvent::Fork>, so you can make a
		40	separate C<use AnyEvent::Fork> if you wish, but you don't have to.
		41
		42	=head1 EXAMPLES
		43
		44	=head2 Example 1: Synchronous Backend
		45
		46	Here is a simple example that implements a backend that executes C<unlink>
		47	and C<rmdir> calls, and reports their status back. It also reports the
		48	number of requests it has processed every three requests, which is clearly
		49	silly, but illustrates the use of events.
		50
		51	First the parent process:
		52
		53	use AnyEvent;
		54	use AnyEvent::Fork;
		55	use AnyEvent::Fork::RPC;
		56
		57	my $done = AE::cv;
		58
		59	my $rpc = AnyEvent::Fork
		60	->new
		61	->require ("MyWorker")
		62	->AnyEvent::Fork::RPC::run ("MyWorker::run",
		63	on_error => sub { warn "FATAL: $_[0]"; exit 1 },
		64	on_event => sub { warn "$_[0] requests handled\n" },
		65	on_destroy => $done,
		66	);
		67
		68	for my $id (1..6) {
		69	$rpc->(rmdir => "/tmp/somepath/$id", sub {
		70	$_[0]
		71	or warn "/tmp/somepath/$id: $_[1]\n";
		72	});
		73	}
		74
		75	undef $rpc;
		76
		77	$done->recv;
		78
		79	The parent creates the process, queues a few rmdir's. It then forgets
		80	about the C<$rpc> object, so that the child exits after it has handled the
		81	requests, and then it waits till the requests have been handled.
		82
		83	The child is implemented using a separate module, C<MyWorker>, shown here:
		84
		85	package MyWorker;
		86
		87	my $count;
		88
		89	sub run {
		90	my ($cmd, $path) = @_;
		91
		92	AnyEvent::Fork::RPC::event ($count)
		93	unless ++$count % 3;
		94
		95	my $status = $cmd eq "rmdir" ? rmdir $path
		96	: $cmd eq "unlink" ? unlink $path
		97	: die "fatal error, illegal command '$cmd'";
		98
		99	$status or (0, "$!")
		100	}
		101
		102	1
		103
		104	The C<run> function first sends a "progress" event every three calls, and
		105	then executes C<rmdir> or C<unlink>, depending on the first parameter (or
		106	dies with a fatal error - obviously, you must never let this happen :).
		107
		108	Eventually it returns the status value true if the command was successful,
		109	or the status value 0 and the stringified error message.
		110
		111	On my system, running the first code fragment with the given
		112	F<MyWorker.pm> in the current directory yields:
		113
		114	/tmp/somepath/1: No such file or directory
		115	/tmp/somepath/2: No such file or directory
		116	3 requests handled
		117	/tmp/somepath/3: No such file or directory
		118	/tmp/somepath/4: No such file or directory
		119	/tmp/somepath/5: No such file or directory
		120	6 requests handled
		121	/tmp/somepath/6: No such file or directory
		122
		123	Obviously, none of the directories I am trying to delete even exist. Also,
		124	the events and responses are processed in exactly the same order as
		125	they were created in the child, which is true for both synchronous and
		126	asynchronous backends.
		127
		128	Note that the parentheses in the call to C<AnyEvent::Fork::RPC::event> are
		129	not optional. That is because the function isn't defined when the code is
		130	compiled. You can make sure it is visible by pre-loading the correct
		131	backend module in the call to C<require>:
		132
		133	->require ("AnyEvent::Fork::RPC::Sync", "MyWorker")
		134
		135	Since the backend module declares the C<event> function, loading it first
		136	ensures that perl will correctly interpret calls to it.
		137
		138	And as a final remark, there is a fine module on CPAN that can
		139	asynchronously C<rmdir> and C<unlink> and a lot more, and more efficiently
		140	than this example, namely L<IO::AIO>.
		141
		142	=head3 Example 1a: the same with the asynchronous backend
		143
		144	This example only shows what needs to be changed to use the async backend
		145	instead. Doing this is not very useful, the purpose of this example is
		146	to show the minimum amount of change that is required to go from the
		147	synchronous to the asynchronous backend.
		148
		149	To use the async backend in the previous example, you need to add the
		150	C<async> parameter to the C<AnyEvent::Fork::RPC::run> call:
		151
		152	->AnyEvent::Fork::RPC::run ("MyWorker::run",
		153	async => 1,
		154	...
		155
		156	And since the function call protocol is now changed, you need to adopt
		157	C<MyWorker::run> to the async API.
		158
		159	First, you need to accept the extra initial C<$done> callback:
		160
		161	sub run {
		162	my ($done, $cmd, $path) = @_;
		163
		164	And since a response is now generated when C<$done> is called, as opposed
		165	to when the function returns, we need to call the C<$done> function with
		166	the status:
		167
		168	$done->($status or (0, "$!"));
		169
		170	A few remarks are in order. First, it's quite pointless to use the async
		171	backend for this example - but it I<is> possible. Second, you can call
		172	C<$done> before or after returning from the function. Third, having both
		173	returned from the function and having called the C<$done> callback, the
		174	child process may exit at any time, so you should call C<$done> only when
		175	you really I<are> done.
		176
		177	=head2 Example 2: Asynchronous Backend
		178
		179	#TODO
		180
39	=head1 PARENT PROCESS USAGE	181	=head1 PARENT PROCESS USAGE
40		182
41	This module exports nothing, and only implements a single function:	183	This module exports nothing, and only implements a single function:
42		184
43	=over 4	185	=over 4
…		…
50		192
51	use Errno ();	193	use Errno ();
52	use Guard ();	194	use Guard ();
53		195
54	use AnyEvent;	196	use AnyEvent;
55	#use AnyEvent::Fork;	197	use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
56		198
57	our $VERSION = 0.1;	199	our $VERSION = 0.1;
58		200
59	=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]	201	=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
60		202
…		…
92	Called for every call to the C<AnyEvent::Fork::RPC::event> function in the	234	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.	235	child, with the arguments of that function passed to the callback.
94		236
95	Also called on errors when no C<on_error> handler is provided.	237	Also called on errors when no C<on_error> handler is provided.
96		238
		239	=item on_destroy => $cb->()
		240
		241	Called when the C<$rpc> object has been destroyed and all requests have
		242	been successfully handled. This is useful when you queue some requests and
		243	want the child to go away after it has handled them. The problem is that
		244	the parent must not exit either until all requests have been handled, and
		245	this can be accomplished by waiting for this callback.
		246
97	=item init => $function (default none)	247	=item init => $function (default none)
98		248
99	When specified (by name), this function is called in the child as the very	249	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	250	first thing when taking over the process, with all the arguments normally
101	passed to the C<AnyEvent::Fork::run> function, except the communications	251	passed to the C<AnyEvent::Fork::run> function, except the communications
102	socket.	252	socket.
103		253
104	It can be used to do one-time things in the child such as storing passed	254	It can be used to do one-time things in the child such as storing passed
105	parameters or opening database connections.	255	parameters or opening database connections.
106		256
		257	It is called very early - before the serialisers are created or the
		258	C<$function> name is resolved into a function reference, so it could be
		259	used to load any modules that provide the serialiser or function. It can
		260	not, however, create events.
		261
107	=item async => $boolean (default: 0)	262	=item async => $boolean (default: 0)
108		263
109	The default server used in the child does all I/O blockingly, and only	264	The default server used in the child does all I/O blockingly, and only
110	allows a single RPC call to execute concurrently.	265	allows a single RPC call to execute concurrently.
111		266
…		…
116	the calling semantics of the returned C<$rpc> function.	271	the calling semantics of the returned C<$rpc> function.
117		272
118	If you want to pre-load the actual back-end modules to enable memory	273	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	274	sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for
120	synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.	275	synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
		276
		277	If you use a template process and want to fork both sync and async
		278	children, then it is permissible to load both modules.
121		279
122	=item serialiser => $string (default: '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })')	280	=item serialiser => $string (default: '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })')
123		281
124	All arguments, result data and event data have to be serialised to be	282	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	283	transferred between the processes. For this, they have to be frozen and
…		…
138	pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>	296	pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>
139	or C<require> statement into the serialiser string. Or both.	297	or C<require> statement into the serialiser string. Or both.
140		298
141	=back	299	=back
142		300
		301	See the examples section earlier in this document for some actual
		302	examples.
		303
143	=cut	304	=cut
144		305
145	our $STRING_SERIALISER = '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })';	306	our $STRING_SERIALISER = '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })';
146
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		307
150	sub run {	308	sub run {
151	my ($self, $function, %arg) = @_;	309	my ($self, $function, %arg) = @_;
152		310
153	my $serialiser = delete $arg{serialiser} \|\| $STRING_SERIALISER;	311	my $serialiser = delete $arg{serialiser} \|\| $STRING_SERIALISER;
154	my $on_event = delete $arg{on_event};	312	my $on_event = delete $arg{on_event};
155	my $on_error = delete $arg{on_error};	313	my $on_error = delete $arg{on_error};
		314	my $on_destroy = delete $arg{on_destroy};
156		315
157	# default for on_error is to on_event, if specified	316	# default for on_error is to on_event, if specified
158	$on_error \|\|= $on_event	317	$on_error \|\|= $on_event
159	? sub { $on_event->(error => shift) }	318	? sub { $on_event->(error => shift) }
160	: sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" };	319	: sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" };
…		…
162	# default for on_event is to raise an error	321	# default for on_event is to raise an error
163	$on_event \|\|= sub { $on_error->("event received, but no on_event handler") };	322	$on_event \|\|= sub { $on_error->("event received, but no on_event handler") };
164		323
165	my ($f, $t) = eval $serialiser; die $@ if $@;	324	my ($f, $t) = eval $serialiser; die $@ if $@;
166		325
167	my (@rcb, $fh, $shutdown, $wbuf, $ww, $rbuf, $rw);	326	my (@rcb, %rcb, $fh, $shutdown, $wbuf, $ww);
		327	my ($rlen, $rbuf, $rw) = 512 - 16;
168		328
169	my $wcb = sub {	329	my $wcb = sub {
170	my $len = syswrite $fh, $wbuf;	330	my $len = syswrite $fh, $wbuf;
171		331
172	if (!defined $len) {	332	unless (defined $len) {
173	if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {	333	if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
174	undef $rw; undef $ww; # it ends here	334	undef $rw; undef $ww; # it ends here
175	$on_error->("$!");	335	$on_error->("$!");
176	}	336	}
177	}	337	}
…		…
188		348
189	$self->require ($module)	349	$self->require ($module)
190	->send_arg ($function, $arg{init}, $serialiser)	350	->send_arg ($function, $arg{init}, $serialiser)
191	->run ("$module\::run", sub {	351	->run ("$module\::run", sub {
192	$fh = shift;	352	$fh = shift;
		353
		354	my ($id, $len);
193	$rw = AE::io $fh, 0, sub {	355	$rw = AE::io $fh, 0, sub {
		356	$rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
194	my $len = sysread $fh, $rbuf, rlen length $rbuf, length $rbuf;	357	$len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
195		358
196	if ($len) {	359	if ($len) {
197	while (5 <= length $rbuf) {	360	while (8 <= length $rbuf) {
198	$len = unpack "L", $rbuf;	361	($id, $len) = unpack "LL", $rbuf;
199	4 + $len <= length $rbuf	362	8 + $len <= length $rbuf
200	or last;	363	or last;
201		364
202	my @r = $t->(substr $rbuf, 4, $len);	365	my @r = $t->(substr $rbuf, 8, $len);
203	substr $rbuf, 0, $len + 4, "";	366	substr $rbuf, 0, 8 + $len, "";
		367
		368	if ($id) {
		369	if (@rcb) {
		370	(shift @rcb)->(@r);
		371	} elsif (my $cb = delete $rcb{$id}) {
		372	$cb->(@r);
		373	} else {
		374	undef $rw; undef $ww;
		375	$on_error->("unexpected data from child");
204		376	}
205	if (pop @r) {	377	} else {
206	$on_event->(@r);	378	$on_event->(@r);
207	} elsif (@rcb) {
208	(shift @rcb)->(@r);
209	} else {
210	undef $rw; undef $ww;
211	$on_error->("unexpected data from child");
212	}	379	}
213	}	380	}
214	} elsif (defined $len) {	381	} elsif (defined $len) {
215	undef $rw; undef $ww; # it ends here	382	undef $rw; undef $ww; # it ends here
		383
		384	if (@rcb \|\| %rcb) {
		385	use Data::Dump;ddx[\@rcb,\%rcb];#d#
216	$on_error->("unexpected eof")	386	$on_error->("unexpected eof");
217	if @rcb;	387	} else {
		388	$on_destroy->();
		389	}
218	} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {	390	} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
219	undef $rw; undef $ww; # it ends here	391	undef $rw; undef $ww; # it ends here
220	$on_error->("read: $!");	392	$on_error->("read: $!");
221	}	393	}
222	};	394	};
…		…
227	my $guard = Guard::guard {	399	my $guard = Guard::guard {
228	$shutdown = 1;	400	$shutdown = 1;
229	$ww \|\|= $fh && AE::io $fh, 1, $wcb;	401	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
230	};	402	};
231		403
		404	my $id;
		405
		406	$arg{async}
232	sub {	407	? sub {
233	push @rcb, pop;	408	$id = ($id == 0xffffffff ? 0 : $id) + 1;
		409	$id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops
234		410
		411	$rcb{$id} = pop;
		412
235	$guard; # keep it alive	413	$guard; # keep it alive
236		414
237	$wbuf .= pack "L/a*", &$f;	415	$wbuf .= pack "LL/a*", $id, &$f;
238	$ww \|\|= $fh && AE::io $fh, 1, $wcb;	416	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
239	}	417	}
		418	: sub {
		419	push @rcb, pop;
		420
		421	$guard; # keep it alive
		422
		423	$wbuf .= pack "L/a*", &$f;
		424	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
		425	}
240	}	426	}
241		427
		428	=item $rpc->(..., $cb->(...))
		429
		430	The RPC object returned by C<AnyEvent::Fork::RPC::run> is actually a code
		431	reference. There are two things you can do with it: call it, and let it go
		432	out of scope (let it get destroyed).
		433
		434	If C<async> was false when C<$rpc> was created (the default), then, if you
		435	call C<$rpc>, the C<$function> is invoked with all arguments passed to
		436	C<$rpc> except the last one (the callback). When the function returns, the
		437	callback will be invoked with all the return values.
		438
		439	If C<async> was true, then the C<$function> receives an additional
		440	initial argument, the result callback. In this case, returning from
		441	C<$function> does nothing - the function only counts as "done" when the
		442	result callback is called, and any arguments passed to it are considered
		443	the return values. This makes it possible to "return" from event handlers
		444	or e.g. Coro threads.
		445
		446	The other thing that can be done with the RPC object is to destroy it. In
		447	this case, the child process will execute all remaining RPC calls, report
		448	their results, and then exit.
		449
		450	See the examples section earlier in this document for some actual
		451	examples.
		452
242	=back	453	=back
243		454
244	=head1 CHILD PROCESS USAGE	455	=head1 CHILD PROCESS USAGE
245		456
246	These functions are not available in this module. They are only available	457	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	458	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	459	without having to load any extra modules. They are part of the child-side
249	L<AnyEvent::Fork::RPC>.	460	API of L<AnyEvent::Fork::RPC>.
250		461
251	=over 4	462	=over 4
252
253	=item AnyEvent::Fork::RPC::quit
254
255	This function can be called to gracefully stop the child process when it
256	is idle.
257
258	After this function is called, the process stops handling incoming RPC
259	requests, but outstanding events and function return values will be sent
260	to the parent. When all data has been sent, the process calls C<exit>.
261
262	Since the parent might not expect the child to exit at random points in
263	time, it is often better to signal the parent by sending an C<event> and
264	letting the parent close down the child process.
265		463
266	=item AnyEvent::Fork::RPC::event ...	464	=item AnyEvent::Fork::RPC::event ...
267		465
268	Send an event to the parent. Events are a bit like RPC calls made by the	466	Send an event to the parent. Events are a bit like RPC calls made by the
269	child process to the parent, except that there is no notion of return	467	child process to the parent, except that there is no notion of return
270	values.	468	values.
271		469
		470	See the examples section earlier in this document for some actual
		471	examples.
		472
272	=back	473	=back
273		474
274	=head1 SEE ALSO	475	=head1 SEE ALSO
275		476
276	L<AnyEvent::Fork> (to create the processes in the first place),	477	L<AnyEvent::Fork> (to create the processes in the first place),

Diff Legend

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

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents): Revision 1.2 by root, Wed Apr 17 17:08:16 2013 UTC vs. Revision 1.10 by root, Wed Apr 17 22:04:49 2013 UTC

Diff Legend

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents):
Revision 1.2 by root, Wed Apr 17 17:08:16 2013 UTC vs.
Revision 1.10 by root, Wed Apr 17 22:04:49 2013 UTC