[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.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, "";
…		…
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	$on_error->("unexpected eof");	566	$on_error->("unexpected eof");
495	} else {	567	} else {
496	$on_destroy->();	568	$on_destroy->()
		569	if $on_destroy;
497	}	570	}
498	} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {	571	} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
499	undef $rw; undef $ww; # it ends here	572	undef $rw; undef $ww; # it ends here
500	$on_error->("read: $!");	573	$on_error->("read: $!");
501	}	574	}
…		…
504	$ww \|\|= AE::io $fh, 1, $wcb;	577	$ww \|\|= AE::io $fh, 1, $wcb;
505	});	578	});
506		579
507	my $guard = Guard::guard {	580	my $guard = Guard::guard {
508	$shutdown = 1;	581	$shutdown = 1;
509	$ww \|\|= $fh && AE::io $fh, 1, $wcb;	582
		583	shutdown $fh, 1 if $fh && !$ww;
510	};	584	};
511		585
512	my $id;	586	my $id;
513		587
514	$arg{async}	588	$arg{async}
…		…
516	$id = ($id == 0xffffffff ? 0 : $id) + 1;	590	$id = ($id == 0xffffffff ? 0 : $id) + 1;
517	$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
518		592
519	$rcb{$id} = pop;	593	$rcb{$id} = pop;
520		594
521	$guard; # keep it alive	595	$guard if 0; # keep it alive
522		596
523	$wbuf .= pack "LL/a*", $id, &$f;	597	$wbuf .= pack "NN/a*", $id, &$f;
524	$ww \|\|= $fh && AE::io $fh, 1, $wcb;	598	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
525	}	599	}
526	: sub {	600	: sub {
527	push @rcb, pop;	601	push @rcb, pop;
528		602
529	$guard; # keep it alive	603	$guard; # keep it alive
530		604
531	$wbuf .= pack "L/a*", &$f;	605	$wbuf .= pack "N/a*", &$f;
532	$ww \|\|= $fh && AE::io $fh, 1, $wcb;	606	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
533	}	607	}
534	}	608	}
535		609
536	=item $rpc->(..., $cb->(...))	610	=item $rpc->(..., $cb->(...))
…		…
640	are queued and the jobs are slow, they will all run concurrently. The	714	are queued and the jobs are slow, they will all run concurrently. The
641	child must implement some queueing/limiting mechanism if this causes	715	child must implement some queueing/limiting mechanism if this causes
642	problems. Alternatively, the parent could limit the amount of rpc calls	716	problems. Alternatively, the parent could limit the amount of rpc calls
643	that are outstanding.	717	that are outstanding.
644		718
		719	Blocking use of condvars is not supported.
		720
645	Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is	721	Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is
646	easy.	722	easy.
647		723
648	=back	724	=back
649		725
…		…
665	half it has passed earlier.	741	half it has passed earlier.
666		742
667	Here is some (untested) pseudocode to that effect:	743	Here is some (untested) pseudocode to that effect:
668		744
669	use AnyEvent::Util;	745	use AnyEvent::Util;
		746	use AnyEvent::Fork;
670	use AnyEvent::Fork::RPC;	747	use AnyEvent::Fork::RPC;
671	use IO::FDPass;	748	use IO::FDPass;
672		749
673	my ($s1, $s2) = AnyEvent::Util::portable_socketpair;	750	my ($s1, $s2) = AnyEvent::Util::portable_socketpair;
674		751
…		…
710		787
711	Of course, this might be blocking if you pass a lot of file descriptors,	788	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	789	so you might want to look into L<AnyEvent::FDpasser> which can handle the
713	gory details.	790	gory details.
714		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
715	=head1 SEE ALSO	799	=head1 SEE ALSO
716		800
717	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
718	L<AnyEvent::Fork::Pool> (to manage whole pools of processes).	805	L<AnyEvent::Fork::Pool>, to manage whole pools of processes.
719		806
720	=head1 AUTHOR AND CONTACT INFORMATION	807	=head1 AUTHOR AND CONTACT INFORMATION
721		808
722	Marc Lehmann <schmorp@schmorp.de>	809	Marc Lehmann <schmorp@schmorp.de>
723	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.13 by root, Thu Apr 18 11:11:26 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.13 by root, Thu Apr 18 11:11:26 2013 UTC vs.
Revision 1.28 by root, Sun Apr 28 15:48:31 2013 UTC