ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-Fork-RPC/RPC.pm
(Generate patch)

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

1=head1 NAME 1=head1 NAME
2 2
3AnyEvent::Fork::RPC - simple RPC extension for AnyEvent::Fork 3AnyEvent::Fork::RPC - simple RPC extension for AnyEvent::Fork
4 4
5THE API IS NOT FINISHED, CONSIDER THIS A TECHNOLOGY DEMO
6
7=head1 SYNOPSIS 5=head1 SYNOPSIS
8 6
7 use AnyEvent::Fork;
9 use AnyEvent::Fork::RPC; 8 use AnyEvent::Fork::RPC;
10 # use AnyEvent::Fork is not needed
11 9
12 my $rpc = AnyEvent::Fork 10 my $rpc = AnyEvent::Fork
13 ->new 11 ->new
14 ->require ("MyModule") 12 ->require ("MyModule")
15 ->AnyEvent::Fork::RPC::run ( 13 ->AnyEvent::Fork::RPC::run (
… …
28 $cv->recv; 26 $cv->recv;
29 27
30=head1 DESCRIPTION 28=head1 DESCRIPTION
31 29
32This module implements a simple RPC protocol and backend for processes 30This module implements a simple RPC protocol and backend for processes
33created via L<AnyEvent::Fork>, allowing you to call a function in the 31created via L<AnyEvent::Fork> or L<AnyEvent::Fork::Remote>, allowing you
34child process and receive its return values (up to 4GB serialised). 32to call a function in the child process and receive its return values (up
33to 4GB serialised).
35 34
36It implements two different backends: a synchronous one that works like a 35It implements two different backends: a synchronous one that works like a
37normal function call, and an asynchronous one that can run multiple jobs 36normal function call, and an asynchronous one that can run multiple jobs
38concurrently in the child, using AnyEvent. 37concurrently in the child, using AnyEvent.
39 38
40It also implements an asynchronous event mechanism from the child to the 39It also implements an asynchronous event mechanism from the child to the
41parent, that could be used for progress indications or other information. 40parent, that could be used for progress indications or other information.
42
43Loading this module also always loads L<AnyEvent::Fork>, so you can make a
44separate C<use AnyEvent::Fork> if you wish, but you don't have to.
45 41
46=head1 EXAMPLES 42=head1 EXAMPLES
47 43
48=head2 Example 1: Synchronous Backend 44=head2 Example 1: Synchronous Backend
49 45
… …
53silly, but illustrates the use of events. 49silly, but illustrates the use of events.
54 50
55First the parent process: 51First the parent process:
56 52
57 use AnyEvent; 53 use AnyEvent;
54 use AnyEvent::Fork;
58 use AnyEvent::Fork::RPC; 55 use AnyEvent::Fork::RPC;
59 56
60 my $done = AE::cv; 57 my $done = AE::cv;
61 58
62 my $rpc = AnyEvent::Fork 59 my $rpc = AnyEvent::Fork
63 ->new 60 ->new
64 ->require ("MyWorker") 61 ->require ("MyWorker")
65 ->AnyEvent::Fork::RPC::run ("MyWorker::run", 62 ->AnyEvent::Fork::RPC::run ("MyWorker::run",
66 on_error => sub { warn "FATAL: $_[0]"; exit 1 }, 63 on_error => sub { warn "ERROR: $_[0]"; exit 1 },
67 on_event => sub { warn "$_[0] requests handled\n" }, 64 on_event => sub { warn "$_[0] requests handled\n" },
68 on_destroy => $done, 65 on_destroy => $done,
69 ); 66 );
70 67
71 for my $id (1..6) { 68 for my $id (1..6) {
… …
193so silly anymore. 190so silly anymore.
194 191
195Without further ado, here is the code: 192Without further ado, here is the code:
196 193
197 use AnyEvent; 194 use AnyEvent;
195 use AnyEvent::Fork;
198 use AnyEvent::Fork::RPC; 196 use AnyEvent::Fork::RPC;
199 197
200 my $done = AE::cv; 198 my $done = AE::cv;
201 199
202 my $rpc = AnyEvent::Fork 200 my $rpc = AnyEvent::Fork
203 ->new 201 ->new
204 ->require ("AnyEvent::Fork::RPC::Async") 202 ->require ("AnyEvent::Fork::RPC::Async")
205 ->eval (do { local $/; <DATA> }) 203 ->eval (do { local $/; <DATA> })
206 ->AnyEvent::Fork::RPC::run ("run", 204 ->AnyEvent::Fork::RPC::run ("run",
207 async => 1, 205 async => 1,
208 on_error => sub { warn "FATAL: $_[0]"; exit 1 }, 206 on_error => sub { warn "ERROR: $_[0]"; exit 1 },
209 on_event => sub { print $_[0] }, 207 on_event => sub { print $_[0] },
210 on_destroy => $done, 208 on_destroy => $done,
211 ); 209 );
212 210
213 for my $count (3, 2, 1) { 211 for my $count (3, 2, 1) {
… …
289 287
290This concludes the async example. Since L<AnyEvent::Fork> does not 288This concludes the async example. Since L<AnyEvent::Fork> does not
291actually fork, you are free to use about any module in the child, not just 289actually fork, you are free to use about any module in the child, not just
292L<AnyEvent>, but also L<IO::AIO>, or L<Tk> for example. 290L<AnyEvent>, but also L<IO::AIO>, or L<Tk> for example.
293 291
292=head2 Example 3: Asynchronous backend with Coro
293
294With L<Coro> you can create a nice asynchronous backend implementation by
295defining an rpc server function that creates a new Coro thread for every
296request that calls a function "normally", i.e. the parameters from the
297parent process are passed to it, and any return values are returned to the
298parent 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
318The C<run> function creates a new thread for every invocation, using the
319first argument as function name, and calls the C<$done> callback on it's
320return values. This makes it quite natural to define the C<add> and C<mul>
321functions to add or multiply two numbers and return the result.
322
323Since this is the asynchronous backend, it's quite possible to define RPC
324function that do I/O or wait for external events - their execution will
325overlap as needed.
326
327The 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
339The C<say>'s will print C<4> and C<6>.
340
294=head1 PARENT PROCESS USAGE 341=head1 PARENT PROCESS USAGE
295 342
296This module exports nothing, and only implements a single function: 343This module exports nothing, and only implements a single function:
297 344
298=over 4 345=over 4
… …
305 352
306use Errno (); 353use Errno ();
307use Guard (); 354use Guard ();
308 355
309use AnyEvent; 356use AnyEvent;
310use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
311 357
312our $VERSION = 0.1; 358our $VERSION = 1.1;
313 359
314=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...] 360=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
315 361
316The traditional way to call it. But it is way cooler to call it in the 362The traditional way to call it. But it is way cooler to call it in the
317following way: 363following way:
… …
337Called on (fatal) errors, with a descriptive (hopefully) message. If 383Called on (fatal) errors, with a descriptive (hopefully) message. If
338this callback is not provided, but C<on_event> is, then the C<on_event> 384this callback is not provided, but C<on_event> is, then the C<on_event>
339callback is called with the first argument being the string C<error>, 385callback is called with the first argument being the string C<error>,
340followed by the error message. 386followed by the error message.
341 387
342If neither handler is provided it prints the error to STDERR and will 388If neither handler is provided, then the error is reported with loglevel
343start failing badly. 389C<error> via C<AE::log>.
344 390
345=item on_event => $cb->(...) 391=item on_event => $cb->(...)
346 392
347Called for every call to the C<AnyEvent::Fork::RPC::event> function in the 393Called for every call to the C<AnyEvent::Fork::RPC::event> function in the
348child, with the arguments of that function passed to the callback. 394child, with the arguments of that function passed to the callback.
… …
417=over 4 463=over 4
418 464
419=item octet strings - C<$AnyEvent::Fork::RPC::STRING_SERIALISER> 465=item octet strings - C<$AnyEvent::Fork::RPC::STRING_SERIALISER>
420 466
421This serialiser concatenates length-prefixes octet strings, and is the 467This serialiser concatenates length-prefixes octet strings, and is the
422default. 468default. That means you can only pass (and return) strings containing
469character codes 0-255.
423 470
424Implementation: 471Implementation:
425 472
426 ( 473 (
427 sub { pack "(w/a*)*", @_ }, 474 sub { pack "(w/a*)*", @_ },
… …
448 495
449=item storable - C<$AnyEvent::Fork::RPC::STORABLE_SERIALISER> 496=item storable - C<$AnyEvent::Fork::RPC::STORABLE_SERIALISER>
450 497
451This serialiser uses L<Storable>, which means it has high chance of 498This serialiser uses L<Storable>, which means it has high chance of
452serialising just about anything you throw at it, at the cost of having 499serialising just about anything you throw at it, at the cost of having
453very high overhead per operation. It also comes with perl. 500very high overhead per operation. It also comes with perl. It should be
501used when you need to serialise complex data structures.
454 502
455Implementation: 503Implementation:
456 504
457 use Storable (); 505 use Storable ();
458 ( 506 (
459 sub { Storable::freeze \@_ }, 507 sub { Storable::freeze \@_ },
460 sub { @{ Storable::thaw shift } } 508 sub { @{ Storable::thaw shift } }
461 ) 509 )
462 510
511=item portable storable - C<$AnyEvent::Fork::RPC::NSTORABLE_SERIALISER>
512
513This serialiser also uses L<Storable>, but uses it's "network" format
514to serialise data, which makes it possible to talk to different
515perl binaries (for example, when talking to a process created with
516L<AnyEvent::Fork::Remote>).
517
518Implementation:
519
520 use Storable ();
521 (
522 sub { Storable::nfreeze \@_ },
523 sub { @{ Storable::thaw shift } }
524 )
525
463=back 526=back
464 527
465=back 528=back
466 529
467See the examples section earlier in this document for some actual 530See the examples section earlier in this document for some actual
468examples. 531examples.
469 532
470=cut 533=cut
471 534
472our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })'; 535our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })';
473our $JSON_SERIALISER = 'use JSON (); (sub { JSON::encode_json \@_ }, sub { @{ JSON::decode_json shift } })'; 536our $JSON_SERIALISER = 'use JSON (); (sub { JSON::encode_json \@_ }, sub { @{ JSON::decode_json shift } })';
474our $STORABLE_SERIALISER = 'use Storable (); (sub { Storable::freeze \@_ }, sub { @{ Storable::thaw shift } })'; 537our $STORABLE_SERIALISER = 'use Storable (); (sub { Storable::freeze \@_ }, sub { @{ Storable::thaw shift } })';
538our $NSTORABLE_SERIALISER = 'use Storable (); (sub { Storable::nfreeze \@_ }, sub { @{ Storable::thaw shift } })';
475 539
476sub run { 540sub run {
477 my ($self, $function, %arg) = @_; 541 my ($self, $function, %arg) = @_;
478 542
479 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER; 543 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER;
… …
482 my $on_destroy = delete $arg{on_destroy}; 546 my $on_destroy = delete $arg{on_destroy};
483 547
484 # default for on_error is to on_event, if specified 548 # default for on_error is to on_event, if specified
485 $on_error ||= $on_event 549 $on_error ||= $on_event
486 ? sub { $on_event->(error => shift) } 550 ? sub { $on_event->(error => shift) }
487 : sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" }; 551 : sub { AE::log die => "AnyEvent::Fork::RPC: uncaught error: $_[0]." };
488 552
489 # default for on_event is to raise an error 553 # default for on_event is to raise an error
490 $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") };
491 555
492 my ($f, $t) = eval $serialiser; die $@ if $@; 556 my ($f, $t) = eval $serialiser; die $@ if $@;
… …
524 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf; 588 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
525 $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf; 589 $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
526 590
527 if ($len) { 591 if ($len) {
528 while (8 <= length $rbuf) { 592 while (8 <= length $rbuf) {
529 ($id, $len) = unpack "LL", $rbuf; 593 ($id, $len) = unpack "NN", $rbuf;
530 8 + $len <= length $rbuf 594 8 + $len <= length $rbuf
531 or last; 595 or last;
532 596
533 my @r = $t->(substr $rbuf, 8, $len); 597 my @r = $t->(substr $rbuf, 8, $len);
534 substr $rbuf, 0, 8 + $len, ""; 598 substr $rbuf, 0, 8 + $len, "";
… …
550 undef $rw; undef $ww; # it ends here 614 undef $rw; undef $ww; # it ends here
551 615
552 if (@rcb || %rcb) { 616 if (@rcb || %rcb) {
553 $on_error->("unexpected eof"); 617 $on_error->("unexpected eof");
554 } else { 618 } else {
555 $on_destroy->(); 619 $on_destroy->()
620 if $on_destroy;
556 } 621 }
557 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) { 622 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
558 undef $rw; undef $ww; # it ends here 623 undef $rw; undef $ww; # it ends here
559 $on_error->("read: $!"); 624 $on_error->("read: $!");
560 } 625 }
… …
576 $id = ($id == 0xffffffff ? 0 : $id) + 1; 641 $id = ($id == 0xffffffff ? 0 : $id) + 1;
577 $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
578 643
579 $rcb{$id} = pop; 644 $rcb{$id} = pop;
580 645
581 $guard; # keep it alive 646 $guard if 0; # keep it alive
582 647
583 $wbuf .= pack "LL/a*", $id, &$f; 648 $wbuf .= pack "NN/a*", $id, &$f;
584 $ww ||= $fh && AE::io $fh, 1, $wcb; 649 $ww ||= $fh && AE::io $fh, 1, $wcb;
585 } 650 }
586 : sub { 651 : sub {
587 push @rcb, pop; 652 push @rcb, pop;
588 653
589 $guard; # keep it alive 654 $guard; # keep it alive
590 655
591 $wbuf .= pack "L/a*", &$f; 656 $wbuf .= pack "N/a*", &$f;
592 $ww ||= $fh && AE::io $fh, 1, $wcb; 657 $ww ||= $fh && AE::io $fh, 1, $wcb;
593 } 658 }
594} 659}
595 660
596=item $rpc->(..., $cb->(...)) 661=item $rpc->(..., $cb->(...))
… …
727half it has passed earlier. 792half it has passed earlier.
728 793
729Here is some (untested) pseudocode to that effect: 794Here is some (untested) pseudocode to that effect:
730 795
731 use AnyEvent::Util; 796 use AnyEvent::Util;
797 use AnyEvent::Fork;
732 use AnyEvent::Fork::RPC; 798 use AnyEvent::Fork::RPC;
733 use IO::FDPass; 799 use IO::FDPass;
734 800
735 my ($s1, $s2) = AnyEvent::Util::portable_socketpair; 801 my ($s1, $s2) = AnyEvent::Util::portable_socketpair;
736 802
… …
772 838
773Of course, this might be blocking if you pass a lot of file descriptors, 839Of course, this might be blocking if you pass a lot of file descriptors,
774so you might want to look into L<AnyEvent::FDpasser> which can handle the 840so you might want to look into L<AnyEvent::FDpasser> which can handle the
775gory details. 841gory details.
776 842
843=head1 EXCEPTIONS
844
845There are no provisions whatsoever for catching exceptions at this time -
846in the child, exeptions might kill the process, causing calls to be lost
847and the parent encountering a fatal error. In the parent, exceptions in
848the result callback will not be caught and cause undefined behaviour.
849
777=head1 SEE ALSO 850=head1 SEE ALSO
778 851
779L<AnyEvent::Fork>, to create the processes in the first place. 852L<AnyEvent::Fork>, to create the processes in the first place.
853
854L<AnyEvent::Fork::Remote>, likewise, but helpful for remote processes.
780 855
781L<AnyEvent::Fork::Pool>, to manage whole pools of processes. 856L<AnyEvent::Fork::Pool>, to manage whole pools of processes.
782 857
783=head1 AUTHOR AND CONTACT INFORMATION 858=head1 AUTHOR AND CONTACT INFORMATION
784 859

Diff Legend

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