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

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents):
Revision 1.13 by root, Thu Apr 18 11:11:26 2013 UTC vs.
Revision 1.29 by root, Sun Aug 25 21:52:15 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;
7	use AnyEvent::Fork::RPC;	8	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 (
14	"MyModule::server",	14	"MyModule::server",
15	);	15	);
16		16
		17	use AnyEvent;
		18
17	my $cv = AE::cv;	19	my $cv = AE::cv;
18		20
19	$rpc->(1, 2, 3, sub {	21	$rpc->(1, 2, 3, sub {
20	print "MyModule::server returned @_\n";	22	print "MyModule::server returned @_\n";
21	$cv->send;	23	$cv->send;
…		…
24	$cv->recv;	26	$cv->recv;
25		27
26	=head1 DESCRIPTION	28	=head1 DESCRIPTION
27		29
28	This module implements a simple RPC protocol and backend for processes	30	This module implements a simple RPC protocol and backend for processes
29	created via L<AnyEvent::Fork>, allowing you to call a function in the	31	created via L<AnyEvent::Fork> or L<AnyEvent::Fork::Remote>, allowing you
30	child process and receive its return values (up to 4GB serialised).	32	to call a function in the child process and receive its return values (up
		33	to 4GB serialised).
31		34
32	It implements two different backends: a synchronous one that works like a	35	It implements two different backends: a synchronous one that works like a
33	normal function call, and an asynchronous one that can run multiple jobs	36	normal function call, and an asynchronous one that can run multiple jobs
34	concurrently in the child, using AnyEvent.	37	concurrently in the child, using AnyEvent.
35		38
36	It also implements an asynchronous event mechanism from the child to the	39	It also implements an asynchronous event mechanism from the child to the
37	parent, that could be used for progress indications or other information.	40	parent, that could be used for progress indications or other information.
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		41
42	=head1 EXAMPLES	42	=head1 EXAMPLES
43		43
44	=head2 Example 1: Synchronous Backend	44	=head2 Example 1: Synchronous Backend
45		45
…		…
49	silly, but illustrates the use of events.	49	silly, but illustrates the use of events.
50		50
51	First the parent process:	51	First the parent process:
52		52
53	use AnyEvent;	53	use AnyEvent;
		54	use AnyEvent::Fork;
54	use AnyEvent::Fork::RPC;	55	use AnyEvent::Fork::RPC;
55		56
56	my $done = AE::cv;	57	my $done = AE::cv;
57		58
58	my $rpc = AnyEvent::Fork	59	my $rpc = AnyEvent::Fork
59	->new	60	->new
60	->require ("MyWorker")	61	->require ("MyWorker")
61	->AnyEvent::Fork::RPC::run ("MyWorker::run",	62	->AnyEvent::Fork::RPC::run ("MyWorker::run",
62	on_error => sub { warn "FATAL: $_[0]"; exit 1 },	63	on_error => sub { warn "ERROR: $_[0]"; exit 1 },
63	on_event => sub { warn "$_[0] requests handled\n" },	64	on_event => sub { warn "$_[0] requests handled\n" },
64	on_destroy => $done,	65	on_destroy => $done,
65	);	66	);
66		67
67	for my $id (1..6) {	68	for my $id (1..6) {
…		…
189	so silly anymore.	190	so silly anymore.
190		191
191	Without further ado, here is the code:	192	Without further ado, here is the code:
192		193
193	use AnyEvent;	194	use AnyEvent;
		195	use AnyEvent::Fork;
194	use AnyEvent::Fork::RPC;	196	use AnyEvent::Fork::RPC;
195		197
196	my $done = AE::cv;	198	my $done = AE::cv;
197		199
198	my $rpc = AnyEvent::Fork	200	my $rpc = AnyEvent::Fork
199	->new	201	->new
200	->require ("AnyEvent::Fork::RPC::Async")	202	->require ("AnyEvent::Fork::RPC::Async")
201	->eval (do { local $/; <DATA> })	203	->eval (do { local $/; <DATA> })
202	->AnyEvent::Fork::RPC::run ("run",	204	->AnyEvent::Fork::RPC::run ("run",
203	async => 1,	205	async => 1,
204	on_error => sub { warn "FATAL: $_[0]"; exit 1 },	206	on_error => sub { warn "ERROR: $_[0]"; exit 1 },
205	on_event => sub { print $_[0] },	207	on_event => sub { print $_[0] },
206	on_destroy => $done,	208	on_destroy => $done,
207	);	209	);
208		210
209	for my $count (3, 2, 1) {	211	for my $count (3, 2, 1) {
…		…
285		287
286	This concludes the async example. Since L<AnyEvent::Fork> does not	288	This concludes the async example. Since L<AnyEvent::Fork> does not
287	actually fork, you are free to use about any module in the child, not just	289	actually fork, you are free to use about any module in the child, not just
288	L<AnyEvent>, but also L<IO::AIO>, or L<Tk> for example.	290	L<AnyEvent>, but also L<IO::AIO>, or L<Tk> for example.
289		291
		292	=head2 Example 3: Asynchronous backend with Coro
		293
		294	With L<Coro> you can create a nice asynchronous backend implementation by
		295	defining an rpc server function that creates a new Coro thread for every
		296	request that calls a function "normally", i.e. the parameters from the
		297	parent process are passed to it, and any return values are returned to the
		298	parent process, e.g.:
		299
		300	package My::Arith;
		301
		302	sub add {
		303	return $_[0] + $_[1];
		304	}
		305
		306	sub mul {
		307	return $_[0] * $_[1];
		308	}
		309
		310	sub run {
		311	my ($done, $func, @arg) = @_;
		312
		313	Coro::async_pool {
		314	$done->($func->(@arg));
		315	};
		316	}
		317
		318	The C<run> function creates a new thread for every invocation, using the
		319	first argument as function name, and calls the C<$done> callback on it's
		320	return values. This makes it quite natural to define the C<add> and C<mul>
		321	functions to add or multiply two numbers and return the result.
		322
		323	Since this is the asynchronous backend, it's quite possible to define RPC
		324	function that do I/O or wait for external events - their execution will
		325	overlap as needed.
		326
		327	The above could be used like this:
		328
		329	my $rpc = AnyEvent::Fork
		330	->new
		331	->require ("MyWorker")
		332	->AnyEvent::Fork::RPC::run ("My::Arith::run",
		333	on_error => ..., on_event => ..., on_destroy => ...,
		334	);
		335
		336	$rpc->(add => 1, 3, Coro::rouse_cb); say Coro::rouse_wait;
		337	$rpc->(mul => 3, 2, Coro::rouse_cb); say Coro::rouse_wait;
		338
		339	The C<say>'s will print C<4> and C<6>.
		340
290	=head1 PARENT PROCESS USAGE	341	=head1 PARENT PROCESS USAGE
291		342
292	This module exports nothing, and only implements a single function:	343	This module exports nothing, and only implements a single function:
293		344
294	=over 4	345	=over 4
…		…
301		352
302	use Errno ();	353	use Errno ();
303	use Guard ();	354	use Guard ();
304		355
305	use AnyEvent;	356	use AnyEvent;
306	use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
307		357
308	our $VERSION = 0.1;	358	our $VERSION = 1.1;
309		359
310	=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]	360	=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
311		361
312	The traditional way to call it. But it is way cooler to call it in the	362	The traditional way to call it. But it is way cooler to call it in the
313	following way:	363	following way:
…		…
333	Called on (fatal) errors, with a descriptive (hopefully) message. If	383	Called on (fatal) errors, with a descriptive (hopefully) message. If
334	this callback is not provided, but C<on_event> is, then the C<on_event>	384	this callback is not provided, but C<on_event> is, then the C<on_event>
335	callback is called with the first argument being the string C<error>,	385	callback is called with the first argument being the string C<error>,
336	followed by the error message.	386	followed by the error message.
337		387
338	If neither handler is provided it prints the error to STDERR and will	388	If neither handler is provided, then the error is reported with loglevel
339	start failing badly.	389	C<error> via C<AE::log>.
340		390
341	=item on_event => $cb->(...)	391	=item on_event => $cb->(...)
342		392
343	Called for every call to the C<AnyEvent::Fork::RPC::event> function in the	393	Called for every call to the C<AnyEvent::Fork::RPC::event> function in the
344	child, with the arguments of that function passed to the callback.	394	child, with the arguments of that function passed to the callback.
…		…
372		422
373	The default server used in the child does all I/O blockingly, and only	423	The default server used in the child does all I/O blockingly, and only
374	allows a single RPC call to execute concurrently.	424	allows a single RPC call to execute concurrently.
375		425
376	Setting C<async> to a true value switches to another implementation that	426	Setting C<async> to a true value switches to another implementation that
377	uses L<AnyEvent> in the child and allows multiple concurrent RPC calls.	427	uses L<AnyEvent> in the child and allows multiple concurrent RPC calls (it
		428	does not support recursion in the event loop however, blocking condvar
		429	calls will fail).
378		430
379	The actual API in the child is documented in the section that describes	431	The actual API in the child is documented in the section that describes
380	the calling semantics of the returned C<$rpc> function.	432	the calling semantics of the returned C<$rpc> function.
381		433
382	If you want to pre-load the actual back-end modules to enable memory	434	If you want to pre-load the actual back-end modules to enable memory
…		…
384	synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.	436	synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
385		437
386	If you use a template process and want to fork both sync and async	438	If you use a template process and want to fork both sync and async
387	children, then it is permissible to load both modules.	439	children, then it is permissible to load both modules.
388		440
389	=item serialiser => $string (default: '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })')	441	=item serialiser => $string (default: $AnyEvent::Fork::RPC::STRING_SERIALISER)
390		442
391	All arguments, result data and event data have to be serialised to be	443	All arguments, result data and event data have to be serialised to be
392	transferred between the processes. For this, they have to be frozen and	444	transferred between the processes. For this, they have to be frozen and
393	thawed in both parent and child processes.	445	thawed in both parent and child processes.
394		446
395	By default, only octet strings can be passed between the processes, which	447	By default, only octet strings can be passed between the processes, which
396	is reasonably fast and efficient.	448	is reasonably fast and efficient and requires no extra modules.
397		449
398	For more complicated use cases, you can provide your own freeze and thaw	450	For more complicated use cases, you can provide your own freeze and thaw
399	functions, by specifying a string with perl source code. It's supposed to	451	functions, by specifying a string with perl source code. It's supposed to
400	return two code references when evaluated: the first receives a list of	452	return two code references when evaluated: the first receives a list of
401	perl values and must return an octet string. The second receives the octet	453	perl values and must return an octet string. The second receives the octet
…		…
403		455
404	If you need an external module for serialisation, then you can either	456	If you need an external module for serialisation, then you can either
405	pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>	457	pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>
406	or C<require> statement into the serialiser string. Or both.	458	or C<require> statement into the serialiser string. Or both.
407		459
		460	Here are some examples - some of them are also available as global
		461	variables that make them easier to use.
		462
		463	=over 4
		464
		465	=item octet strings - C<$AnyEvent::Fork::RPC::STRING_SERIALISER>
		466
		467	This serialiser concatenates length-prefixes octet strings, and is the
		468	default. That means you can only pass (and return) strings containing
		469	character codes 0-255.
		470
		471	Implementation:
		472
		473	(
		474	sub { pack "(w/a)", @_ },
		475	sub { unpack "(w/a)", shift }
		476	)
		477
		478	=item json - C<$AnyEvent::Fork::RPC::JSON_SERIALISER>
		479
		480	This serialiser creates JSON arrays - you have to make sure the L<JSON>
		481	module is installed for this serialiser to work. It can be beneficial for
		482	sharing when you preload the L<JSON> module in a template process.
		483
		484	L<JSON> (with L<JSON::XS> installed) is slower than the octet string
		485	serialiser, but usually much faster than L<Storable>, unless big chunks of
		486	binary data need to be transferred.
		487
		488	Implementation:
		489
		490	use JSON ();
		491	(
		492	sub { JSON::encode_json \@_ },
		493	sub { @{ JSON::decode_json shift } }
		494	)
		495
		496	=item storable - C<$AnyEvent::Fork::RPC::STORABLE_SERIALISER>
		497
		498	This serialiser uses L<Storable>, which means it has high chance of
		499	serialising just about anything you throw at it, at the cost of having
		500	very high overhead per operation. It also comes with perl. It should be
		501	used when you need to serialise complex data structures.
		502
		503	Implementation:
		504
		505	use Storable ();
		506	(
		507	sub { Storable::freeze \@_ },
		508	sub { @{ Storable::thaw shift } }
		509	)
		510
		511	=item portable storable - C<$AnyEvent::Fork::RPC::NSTORABLE_SERIALISER>
		512
		513	This serialiser also uses L<Storable>, but uses it's "network" format
		514	to serialise data, which makes it possible to talk to different
		515	perl binaries (for example, when talking to a process created with
		516	L<AnyEvent::Fork::Remote>).
		517
		518	Implementation:
		519
		520	use Storable ();
		521	(
		522	sub { Storable::nfreeze \@_ },
		523	sub { @{ Storable::thaw shift } }
		524	)
		525
		526	=back
		527
408	=back	528	=back
409		529
410	See the examples section earlier in this document for some actual	530	See the examples section earlier in this document for some actual
411	examples.	531	examples.
412		532
413	=cut	533	=cut
414		534
415	our $STRING_SERIALISER = '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })';	535	our $STRING_SERIALISER = '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })';
		536	our $JSON_SERIALISER = 'use JSON (); (sub { JSON::encode_json \@_ }, sub { @{ JSON::decode_json shift } })';
		537	our $STORABLE_SERIALISER = 'use Storable (); (sub { Storable::freeze \@_ }, sub { @{ Storable::thaw shift } })';
		538	our $NSTORABLE_SERIALISER = 'use Storable (); (sub { Storable::nfreeze \@_ }, sub { @{ Storable::thaw shift } })';
416		539
417	sub run {	540	sub run {
418	my ($self, $function, %arg) = @_;	541	my ($self, $function, %arg) = @_;
419		542
420	my $serialiser = delete $arg{serialiser} \|\| $STRING_SERIALISER;	543	my $serialiser = delete $arg{serialiser} \|\| $STRING_SERIALISER;
…		…
423	my $on_destroy = delete $arg{on_destroy};	546	my $on_destroy = delete $arg{on_destroy};
424		547
425	# default for on_error is to on_event, if specified	548	# default for on_error is to on_event, if specified
426	$on_error \|\|= $on_event	549	$on_error \|\|= $on_event
427	? sub { $on_event->(error => shift) }	550	? sub { $on_event->(error => shift) }
428	: sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" };	551	: sub { AE::log die => "AnyEvent::Fork::RPC: uncaught error: $_[0]." };
429		552
430	# default for on_event is to raise an error	553	# default for on_event is to raise an error
431	$on_event \|\|= sub { $on_error->("event received, but no on_event handler") };	554	$on_event \|\|= sub { $on_error->("event received, but no on_event handler") };
432		555
433	my ($f, $t) = eval $serialiser; die $@ if $@;	556	my ($f, $t) = eval $serialiser; die $@ if $@;
…		…
465	$rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;	588	$rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
466	$len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;	589	$len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
467		590
468	if ($len) {	591	if ($len) {
469	while (8 <= length $rbuf) {	592	while (8 <= length $rbuf) {
470	($id, $len) = unpack "LL", $rbuf;	593	($id, $len) = unpack "NN", $rbuf;
471	8 + $len <= length $rbuf	594	8 + $len <= length $rbuf
472	or last;	595	or last;
473		596
474	my @r = $t->(substr $rbuf, 8, $len);	597	my @r = $t->(substr $rbuf, 8, $len);
475	substr $rbuf, 0, 8 + $len, "";	598	substr $rbuf, 0, 8 + $len, "";
…		…
491	undef $rw; undef $ww; # it ends here	614	undef $rw; undef $ww; # it ends here
492		615
493	if (@rcb \|\| %rcb) {	616	if (@rcb \|\| %rcb) {
494	$on_error->("unexpected eof");	617	$on_error->("unexpected eof");
495	} else {	618	} else {
496	$on_destroy->();	619	$on_destroy->()
		620	if $on_destroy;
497	}	621	}
498	} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {	622	} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
499	undef $rw; undef $ww; # it ends here	623	undef $rw; undef $ww; # it ends here
500	$on_error->("read: $!");	624	$on_error->("read: $!");
501	}	625	}
…		…
504	$ww \|\|= AE::io $fh, 1, $wcb;	628	$ww \|\|= AE::io $fh, 1, $wcb;
505	});	629	});
506		630
507	my $guard = Guard::guard {	631	my $guard = Guard::guard {
508	$shutdown = 1;	632	$shutdown = 1;
509	$ww \|\|= $fh && AE::io $fh, 1, $wcb;	633
		634	shutdown $fh, 1 if $fh && !$ww;
510	};	635	};
511		636
512	my $id;	637	my $id;
513		638
514	$arg{async}	639	$arg{async}
…		…
516	$id = ($id == 0xffffffff ? 0 : $id) + 1;	641	$id = ($id == 0xffffffff ? 0 : $id) + 1;
517	$id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops	642	$id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops
518		643
519	$rcb{$id} = pop;	644	$rcb{$id} = pop;
520		645
521	$guard; # keep it alive	646	$guard if 0; # keep it alive
522		647
523	$wbuf .= pack "LL/a*", $id, &$f;	648	$wbuf .= pack "NN/a*", $id, &$f;
524	$ww \|\|= $fh && AE::io $fh, 1, $wcb;	649	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
525	}	650	}
526	: sub {	651	: sub {
527	push @rcb, pop;	652	push @rcb, pop;
528		653
529	$guard; # keep it alive	654	$guard; # keep it alive
530		655
531	$wbuf .= pack "L/a*", &$f;	656	$wbuf .= pack "N/a*", &$f;
532	$ww \|\|= $fh && AE::io $fh, 1, $wcb;	657	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
533	}	658	}
534	}	659	}
535		660
536	=item $rpc->(..., $cb->(...))	661	=item $rpc->(..., $cb->(...))
…		…
640	are queued and the jobs are slow, they will all run concurrently. The	765	are queued and the jobs are slow, they will all run concurrently. The
641	child must implement some queueing/limiting mechanism if this causes	766	child must implement some queueing/limiting mechanism if this causes
642	problems. Alternatively, the parent could limit the amount of rpc calls	767	problems. Alternatively, the parent could limit the amount of rpc calls
643	that are outstanding.	768	that are outstanding.
644		769
		770	Blocking use of condvars is not supported.
		771
645	Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is	772	Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is
646	easy.	773	easy.
647		774
648	=back	775	=back
649		776
…		…
665	half it has passed earlier.	792	half it has passed earlier.
666		793
667	Here is some (untested) pseudocode to that effect:	794	Here is some (untested) pseudocode to that effect:
668		795
669	use AnyEvent::Util;	796	use AnyEvent::Util;
		797	use AnyEvent::Fork;
670	use AnyEvent::Fork::RPC;	798	use AnyEvent::Fork::RPC;
671	use IO::FDPass;	799	use IO::FDPass;
672		800
673	my ($s1, $s2) = AnyEvent::Util::portable_socketpair;	801	my ($s1, $s2) = AnyEvent::Util::portable_socketpair;
674		802
…		…
710		838
711	Of course, this might be blocking if you pass a lot of file descriptors,	839	Of course, this might be blocking if you pass a lot of file descriptors,
712	so you might want to look into L<AnyEvent::FDpasser> which can handle the	840	so you might want to look into L<AnyEvent::FDpasser> which can handle the
713	gory details.	841	gory details.
714		842
		843	=head1 EXCEPTIONS
		844
		845	There are no provisions whatsoever for catching exceptions at this time -
		846	in the child, exeptions might kill the process, causing calls to be lost
		847	and the parent encountering a fatal error. In the parent, exceptions in
		848	the result callback will not be caught and cause undefined behaviour.
		849
715	=head1 SEE ALSO	850	=head1 SEE ALSO
716		851
717	L<AnyEvent::Fork> (to create the processes in the first place),	852	L<AnyEvent::Fork>, to create the processes in the first place.
		853
		854	L<AnyEvent::Fork::Remote>, likewise, but helpful for remote processes.
		855
718	L<AnyEvent::Fork::Pool> (to manage whole pools of processes).	856	L<AnyEvent::Fork::Pool>, to manage whole pools of processes.
719		857
720	=head1 AUTHOR AND CONTACT INFORMATION	858	=head1 AUTHOR AND CONTACT INFORMATION
721		859
722	Marc Lehmann <schmorp@schmorp.de>	860	Marc Lehmann <schmorp@schmorp.de>
723	http://software.schmorp.de/pkg/AnyEvent-Fork-RPC	861	http://software.schmorp.de/pkg/AnyEvent-Fork-RPC

Diff Legend

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

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents): Revision 1.13 by root, Thu Apr 18 11:11:26 2013 UTC vs. Revision 1.29 by root, Sun Aug 25 21:52:15 2013 UTC

Diff Legend

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents):
Revision 1.13 by root, Thu Apr 18 11:11:26 2013 UTC vs.
Revision 1.29 by root, Sun Aug 25 21:52:15 2013 UTC