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

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents):
Revision 1.11 by root, Thu Apr 18 07:59:46 2013 UTC vs.
Revision 1.28 by root, Sun Apr 28 15:48:31 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
…		…
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
…		…
301		303
302	use Errno ();	304	use Errno ();
303	use Guard ();	305	use Guard ();
304		306
305	use AnyEvent;	307	use AnyEvent;
306	use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
307		308
308	our $VERSION = 0.1;	309	our $VERSION = 1.1;
309		310
310	=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]	311	=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
311		312
312	The traditional way to call it. But it is way cooler to call it in the	313	The traditional way to call it. But it is way cooler to call it in the
313	following way:	314	following way:
…		…
372		373
373	The default server used in the child does all I/O blockingly, and only	374	The default server used in the child does all I/O blockingly, and only
374	allows a single RPC call to execute concurrently.	375	allows a single RPC call to execute concurrently.
375		376
376	Setting C<async> to a true value switches to another implementation that	377	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.	378	uses L<AnyEvent> in the child and allows multiple concurrent RPC calls (it
		379	does not support recursion in the event loop however, blocking condvar
		380	calls will fail).
378		381
379	The actual API in the child is documented in the section that describes	382	The actual API in the child is documented in the section that describes
380	the calling semantics of the returned C<$rpc> function.	383	the calling semantics of the returned C<$rpc> function.
381		384
382	If you want to pre-load the actual back-end modules to enable memory	385	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.	387	synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
385		388
386	If you use a template process and want to fork both sync and async	389	If you use a template process and want to fork both sync and async
387	children, then it is permissible to load both modules.	390	children, then it is permissible to load both modules.
388		391
389	=item serialiser => $string (default: '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })')	392	=item serialiser => $string (default: $AnyEvent::Fork::RPC::STRING_SERIALISER)
390		393
391	All arguments, result data and event data have to be serialised to be	394	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	395	transferred between the processes. For this, they have to be frozen and
393	thawed in both parent and child processes.	396	thawed in both parent and child processes.
394		397
395	By default, only octet strings can be passed between the processes, which	398	By default, only octet strings can be passed between the processes, which
396	is reasonably fast and efficient.	399	is reasonably fast and efficient and requires no extra modules.
397		400
398	For more complicated use cases, you can provide your own freeze and thaw	401	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	402	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	403	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	404	perl values and must return an octet string. The second receives the octet
…		…
403		406
404	If you need an external module for serialisation, then you can either	407	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>	408	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.	409	or C<require> statement into the serialiser string. Or both.
407		410
		411	Here are some examples - some of them are also available as global
		412	variables that make them easier to use.
		413
		414	=over 4
		415
		416	=item octet strings - C<$AnyEvent::Fork::RPC::STRING_SERIALISER>
		417
		418	This serialiser concatenates length-prefixes octet strings, and is the
		419	default.
		420
		421	Implementation:
		422
		423	(
		424	sub { pack "(w/a)", @_ },
		425	sub { unpack "(w/a)", shift }
		426	)
		427
		428	=item json - C<$AnyEvent::Fork::RPC::JSON_SERIALISER>
		429
		430	This serialiser creates JSON arrays - you have to make sure the L<JSON>
		431	module is installed for this serialiser to work. It can be beneficial for
		432	sharing when you preload the L<JSON> module in a template process.
		433
		434	L<JSON> (with L<JSON::XS> installed) is slower than the octet string
		435	serialiser, but usually much faster than L<Storable>, unless big chunks of
		436	binary data need to be transferred.
		437
		438	Implementation:
		439
		440	use JSON ();
		441	(
		442	sub { JSON::encode_json \@_ },
		443	sub { @{ JSON::decode_json shift } }
		444	)
		445
		446	=item storable - C<$AnyEvent::Fork::RPC::STORABLE_SERIALISER>
		447
		448	This serialiser uses L<Storable>, which means it has high chance of
		449	serialising just about anything you throw at it, at the cost of having
		450	very high overhead per operation. It also comes with perl.
		451
		452	Implementation:
		453
		454	use Storable ();
		455	(
		456	sub { Storable::freeze \@_ },
		457	sub { @{ Storable::thaw shift } }
		458	)
		459
		460	=item portable storable - C<$AnyEvent::Fork::RPC::NSTORABLE_SERIALISER>
		461
		462	This serialiser also uses L<Storable>, but uses it's "network" format
		463	to serialise data, which makes it possible to talk to incompatible
		464	perl versions (for example, when talking to a process created with
		465	L<AnyEvent::Fork::Remote>).
		466
		467	Implementation:
		468
		469	use Storable ();
		470	(
		471	sub { Storable::nfreeze \@_ },
		472	sub { @{ Storable::thaw shift } }
		473	)
		474
		475	=back
		476
408	=back	477	=back
409		478
410	See the examples section earlier in this document for some actual	479	See the examples section earlier in this document for some actual
411	examples.	480	examples.
412		481
413	=cut	482	=cut
414		483
415	our $STRING_SERIALISER = '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })';	484	our $STRING_SERIALISER = '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })';
		485	our $JSON_SERIALISER = 'use JSON (); (sub { JSON::encode_json \@_ }, sub { @{ JSON::decode_json shift } })';
		486	our $STORABLE_SERIALISER = 'use Storable (); (sub { Storable::freeze \@_ }, sub { @{ Storable::thaw shift } })';
		487	our $NSTORABLE_SERIALISER = 'use Storable (); (sub { Storable::nfreeze \@_ }, sub { @{ Storable::thaw shift } })';
416		488
417	sub run {	489	sub run {
418	my ($self, $function, %arg) = @_;	490	my ($self, $function, %arg) = @_;
419		491
420	my $serialiser = delete $arg{serialiser} \|\| $STRING_SERIALISER;	492	my $serialiser = delete $arg{serialiser} \|\| $STRING_SERIALISER;
…		…
465	$rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;	537	$rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
466	$len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;	538	$len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
467		539
468	if ($len) {	540	if ($len) {
469	while (8 <= length $rbuf) {	541	while (8 <= length $rbuf) {
470	($id, $len) = unpack "LL", $rbuf;	542	($id, $len) = unpack "NN", $rbuf;
471	8 + $len <= length $rbuf	543	8 + $len <= length $rbuf
472	or last;	544	or last;
473		545
474	my @r = $t->(substr $rbuf, 8, $len);	546	my @r = $t->(substr $rbuf, 8, $len);
475	substr $rbuf, 0, 8 + $len, "";	547	substr $rbuf, 0, 8 + $len, "";
…		…
489	}	561	}
490	} elsif (defined $len) {	562	} elsif (defined $len) {
491	undef $rw; undef $ww; # it ends here	563	undef $rw; undef $ww; # it ends here
492		564
493	if (@rcb \|\| %rcb) {	565	if (@rcb \|\| %rcb) {
494	use Data::Dump;ddx[\@rcb,\%rcb];#d#
495	$on_error->("unexpected eof");	566	$on_error->("unexpected eof");
496	} else {	567	} else {
497	$on_destroy->();	568	$on_destroy->()
		569	if $on_destroy;
498	}	570	}
499	} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {	571	} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
500	undef $rw; undef $ww; # it ends here	572	undef $rw; undef $ww; # it ends here
501	$on_error->("read: $!");	573	$on_error->("read: $!");
502	}	574	}
…		…
505	$ww \|\|= AE::io $fh, 1, $wcb;	577	$ww \|\|= AE::io $fh, 1, $wcb;
506	});	578	});
507		579
508	my $guard = Guard::guard {	580	my $guard = Guard::guard {
509	$shutdown = 1;	581	$shutdown = 1;
510	$ww \|\|= $fh && AE::io $fh, 1, $wcb;	582
		583	shutdown $fh, 1 if $fh && !$ww;
511	};	584	};
512		585
513	my $id;	586	my $id;
514		587
515	$arg{async}	588	$arg{async}
…		…
517	$id = ($id == 0xffffffff ? 0 : $id) + 1;	590	$id = ($id == 0xffffffff ? 0 : $id) + 1;
518	$id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops	591	$id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops
519		592
520	$rcb{$id} = pop;	593	$rcb{$id} = pop;
521		594
522	$guard; # keep it alive	595	$guard if 0; # keep it alive
523		596
524	$wbuf .= pack "LL/a*", $id, &$f;	597	$wbuf .= pack "NN/a*", $id, &$f;
525	$ww \|\|= $fh && AE::io $fh, 1, $wcb;	598	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
526	}	599	}
527	: sub {	600	: sub {
528	push @rcb, pop;	601	push @rcb, pop;
529		602
530	$guard; # keep it alive	603	$guard; # keep it alive
531		604
532	$wbuf .= pack "L/a*", &$f;	605	$wbuf .= pack "N/a*", &$f;
533	$ww \|\|= $fh && AE::io $fh, 1, $wcb;	606	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
534	}	607	}
535	}	608	}
536		609
537	=item $rpc->(..., $cb->(...))	610	=item $rpc->(..., $cb->(...))
…		…
579	See the examples section earlier in this document for some actual	652	See the examples section earlier in this document for some actual
580	examples.	653	examples.
581		654
582	=back	655	=back
583		656
		657	=head1 ADVANCED TOPICS
		658
		659	=head2 Choosing a backend
		660
		661	So how do you decide which backend to use? Well, that's your problem to
		662	solve, but here are some thoughts on the matter:
		663
		664	=over 4
		665
		666	=item Synchronous
		667
		668	The synchronous backend does not rely on any external modules (well,
		669	except L<common::sense>, which works around a bug in how perl's warning
		670	system works). This keeps the process very small, for example, on my
		671	system, an empty perl interpreter uses 1492kB RSS, which becomes 2020kB
		672	after C<use warnings; use strict> (for people who grew up with C64s around
		673	them this is probably shocking every single time they see it). The worker
		674	process in the first example in this document uses 1792kB.
		675
		676	Since the calls are done synchronously, slow jobs will keep newer jobs
		677	from executing.
		678
		679	The synchronous backend also has no overhead due to running an event loop
		680	- reading requests is therefore very efficient, while writing responses is
		681	less so, as every response results in a write syscall.
		682
		683	If the parent process is busy and a bit slow reading responses, the child
		684	waits instead of processing further requests. This also limits the amount
		685	of memory needed for buffering, as never more than one response has to be
		686	buffered.
		687
		688	The API in the child is simple - you just have to define a function that
		689	does something and returns something.
		690
		691	It's hard to use modules or code that relies on an event loop, as the
		692	child cannot execute anything while it waits for more input.
		693
		694	=item Asynchronous
		695
		696	The asynchronous backend relies on L<AnyEvent>, which tries to be small,
		697	but still comes at a price: On my system, the worker from example 1a uses
		698	3420kB RSS (for L<AnyEvent>, which loads L<EV>, which needs L<XSLoader>
		699	which in turn loads a lot of other modules such as L<warnings>, L<strict>,
		700	L<vars>, L<Exporter>...).
		701
		702	It batches requests and responses reasonably efficiently, doing only as
		703	few reads and writes as needed, but needs to poll for events via the event
		704	loop.
		705
		706	Responses are queued when the parent process is busy. This means the child
		707	can continue to execute any queued requests. It also means that a child
		708	might queue a lot of responses in memory when it generates them and the
		709	parent process is slow accepting them.
		710
		711	The API is not a straightforward RPC pattern - you have to call a
		712	"done" callback to pass return values and signal completion. Also, more
		713	importantly, the API starts jobs as fast as possible - when 1000 jobs
		714	are queued and the jobs are slow, they will all run concurrently. The
		715	child must implement some queueing/limiting mechanism if this causes
		716	problems. Alternatively, the parent could limit the amount of rpc calls
		717	that are outstanding.
		718
		719	Blocking use of condvars is not supported.
		720
		721	Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is
		722	easy.
		723
		724	=back
		725
		726	=head2 Passing file descriptors
		727
		728	Unlike L<AnyEvent::Fork>, this module has no in-built file handle or file
		729	descriptor passing abilities.
		730
		731	The reason is that passing file descriptors is extraordinary tricky
		732	business, and conflicts with efficient batching of messages.
		733
		734	There still is a method you can use: Create a
		735	C<AnyEvent::Util::portable_socketpair> and C<send_fh> one half of it to
		736	the process before you pass control to C<AnyEvent::Fork::RPC::run>.
		737
		738	Whenever you want to pass a file descriptor, send an rpc request to the
		739	child process (so it expects the descriptor), then send it over the other
		740	half of the socketpair. The child should fetch the descriptor from the
		741	half it has passed earlier.
		742
		743	Here is some (untested) pseudocode to that effect:
		744
		745	use AnyEvent::Util;
		746	use AnyEvent::Fork;
		747	use AnyEvent::Fork::RPC;
		748	use IO::FDPass;
		749
		750	my ($s1, $s2) = AnyEvent::Util::portable_socketpair;
		751
		752	my $rpc = AnyEvent::Fork
		753	->new
		754	->send_fh ($s2)
		755	->require ("MyWorker")
		756	->AnyEvent::Fork::RPC::run ("MyWorker::run"
		757	init => "MyWorker::init",
		758	);
		759
		760	undef $s2; # no need to keep it around
		761
		762	# pass an fd
		763	$rpc->("i'll send some fd now, please expect it!", my $cv = AE::cv);
		764
		765	IO::FDPass fileno $s1, fileno $handle_to_pass;
		766
		767	$cv->recv;
		768
		769	The MyWorker module could look like this:
		770
		771	package MyWorker;
		772
		773	use IO::FDPass;
		774
		775	my $s2;
		776
		777	sub init {
		778	$s2 = $_[0];
		779	}
		780
		781	sub run {
		782	if ($_[0] eq "i'll send some fd now, please expect it!") {
		783	my $fd = IO::FDPass::recv fileno $s2;
		784	...
		785	}
		786	}
		787
		788	Of course, this might be blocking if you pass a lot of file descriptors,
		789	so you might want to look into L<AnyEvent::FDpasser> which can handle the
		790	gory details.
		791
		792	=head1 EXCEPTIONS
		793
		794	There are no provisions whatsoever for catching exceptions at this time -
		795	in the child, exeptions might kill the process, causing calls to be lost
		796	and the parent encountering a fatal error. In the parent, exceptions in
		797	the result callback will not be caught and cause undefined behaviour.
		798
584	=head1 SEE ALSO	799	=head1 SEE ALSO
585		800
586	L<AnyEvent::Fork> (to create the processes in the first place),	801	L<AnyEvent::Fork>, to create the processes in the first place.
		802
		803	L<AnyEvent::Fork::Remote>, likewise, but helpful for remote processes.
		804
587	L<AnyEvent::Fork::Pool> (to manage whole pools of processes).	805	L<AnyEvent::Fork::Pool>, to manage whole pools of processes.
588		806
589	=head1 AUTHOR AND CONTACT INFORMATION	807	=head1 AUTHOR AND CONTACT INFORMATION
590		808
591	Marc Lehmann <schmorp@schmorp.de>	809	Marc Lehmann <schmorp@schmorp.de>
592	http://software.schmorp.de/pkg/AnyEvent-Fork-RPC	810	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.11 by root, Thu Apr 18 07:59:46 2013 UTC vs. Revision 1.28 by root, Sun Apr 28 15:48:31 2013 UTC

Diff Legend

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents):
Revision 1.11 by root, Thu Apr 18 07:59:46 2013 UTC vs.
Revision 1.28 by root, Sun Apr 28 15:48:31 2013 UTC