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.17 by root, Thu Apr 18 20:04:51 2013 UTC vs.
Revision 1.27 by root, Sun Apr 28 14:34:40 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
… …
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
… …
305 303
306use Errno (); 304use Errno ();
307use Guard (); 305use Guard ();
308 306
309use AnyEvent; 307use AnyEvent;
310use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
311 308
312our $VERSION = 0.1; 309our $VERSION = 1.1;
313 310
314=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...] 311=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
315 312
316The traditional way to call it. But it is way cooler to call it in the 313The traditional way to call it. But it is way cooler to call it in the
317following way: 314following way:
… …
524 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf; 521 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
525 $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf; 522 $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
526 523
527 if ($len) { 524 if ($len) {
528 while (8 <= length $rbuf) { 525 while (8 <= length $rbuf) {
529 ($id, $len) = unpack "LL", $rbuf; 526 ($id, $len) = unpack "NN", $rbuf;
530 8 + $len <= length $rbuf 527 8 + $len <= length $rbuf
531 or last; 528 or last;
532 529
533 my @r = $t->(substr $rbuf, 8, $len); 530 my @r = $t->(substr $rbuf, 8, $len);
534 substr $rbuf, 0, 8 + $len, ""; 531 substr $rbuf, 0, 8 + $len, "";
… …
550 undef $rw; undef $ww; # it ends here 547 undef $rw; undef $ww; # it ends here
551 548
552 if (@rcb || %rcb) { 549 if (@rcb || %rcb) {
553 $on_error->("unexpected eof"); 550 $on_error->("unexpected eof");
554 } else { 551 } else {
555 $on_destroy->(); 552 $on_destroy->()
553 if $on_destroy;
556 } 554 }
557 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) { 555 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
558 undef $rw; undef $ww; # it ends here 556 undef $rw; undef $ww; # it ends here
559 $on_error->("read: $!"); 557 $on_error->("read: $!");
560 } 558 }
… …
563 $ww ||= AE::io $fh, 1, $wcb; 561 $ww ||= AE::io $fh, 1, $wcb;
564 }); 562 });
565 563
566 my $guard = Guard::guard { 564 my $guard = Guard::guard {
567 $shutdown = 1; 565 $shutdown = 1;
568 $ww ||= $fh && AE::io $fh, 1, $wcb; 566
567 shutdown $fh, 1 if $fh && !$ww;
569 }; 568 };
570 569
571 my $id; 570 my $id;
572 571
573 $arg{async} 572 $arg{async}
… …
575 $id = ($id == 0xffffffff ? 0 : $id) + 1; 574 $id = ($id == 0xffffffff ? 0 : $id) + 1;
576 $id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops 575 $id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops
577 576
578 $rcb{$id} = pop; 577 $rcb{$id} = pop;
579 578
580 $guard; # keep it alive 579 $guard if 0; # keep it alive
581 580
582 $wbuf .= pack "LL/a*", $id, &$f; 581 $wbuf .= pack "NN/a*", $id, &$f;
583 $ww ||= $fh && AE::io $fh, 1, $wcb; 582 $ww ||= $fh && AE::io $fh, 1, $wcb;
584 } 583 }
585 : sub { 584 : sub {
586 push @rcb, pop; 585 push @rcb, pop;
587 586
588 $guard; # keep it alive 587 $guard; # keep it alive
589 588
590 $wbuf .= pack "L/a*", &$f; 589 $wbuf .= pack "N/a*", &$f;
591 $ww ||= $fh && AE::io $fh, 1, $wcb; 590 $ww ||= $fh && AE::io $fh, 1, $wcb;
592 } 591 }
593} 592}
594 593
595=item $rpc->(..., $cb->(...)) 594=item $rpc->(..., $cb->(...))
… …
726half it has passed earlier. 725half it has passed earlier.
727 726
728Here is some (untested) pseudocode to that effect: 727Here is some (untested) pseudocode to that effect:
729 728
730 use AnyEvent::Util; 729 use AnyEvent::Util;
730 use AnyEvent::Fork;
731 use AnyEvent::Fork::RPC; 731 use AnyEvent::Fork::RPC;
732 use IO::FDPass; 732 use IO::FDPass;
733 733
734 my ($s1, $s2) = AnyEvent::Util::portable_socketpair; 734 my ($s1, $s2) = AnyEvent::Util::portable_socketpair;
735 735
… …
771 771
772Of course, this might be blocking if you pass a lot of file descriptors, 772Of course, this might be blocking if you pass a lot of file descriptors,
773so you might want to look into L<AnyEvent::FDpasser> which can handle the 773so you might want to look into L<AnyEvent::FDpasser> which can handle the
774gory details. 774gory details.
775 775
776=head1 EXCEPTIONS
777
778There are no provisions whatsoever for catching exceptions at this time -
779in the child, exeptions might kill the process, causing calls to be lost
780and the parent encountering a fatal error. In the parent, exceptions in
781the result callback will not be caught and cause undefined behaviour.
782
776=head1 SEE ALSO 783=head1 SEE ALSO
777 784
778L<AnyEvent::Fork>, to create the processes in the first place. 785L<AnyEvent::Fork>, to create the processes in the first place.
786
787L<AnyEvent::Fork::Remote>, likewise, but helpful for remote processes.
779 788
780L<AnyEvent::Fork::Pool>, to manage whole pools of processes. 789L<AnyEvent::Fork::Pool>, to manage whole pools of processes.
781 790
782=head1 AUTHOR AND CONTACT INFORMATION 791=head1 AUTHOR AND CONTACT INFORMATION
783 792

Diff Legend

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