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.9 by root, Wed Apr 17 21:48:35 2013 UTC vs.
Revision 1.24 by root, Sat Apr 27 23:49:01 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
5THE API IS NOT FINISHED, CONSIDER THIS A BETA RELEASE
4 6
5=head1 SYNOPSIS 7=head1 SYNOPSIS
6 8
7 use AnyEvent::Fork::RPC; 9 use AnyEvent::Fork::RPC;
8 # use AnyEvent::Fork is not needed 10 # use AnyEvent::Fork is not needed
12 ->require ("MyModule") 14 ->require ("MyModule")
13 ->AnyEvent::Fork::RPC::run ( 15 ->AnyEvent::Fork::RPC::run (
14 "MyModule::server", 16 "MyModule::server",
15 ); 17 );
16 18
19 use AnyEvent;
20
17 my $cv = AE::cv; 21 my $cv = AE::cv;
18 22
19 $rpc->(1, 2, 3, sub { 23 $rpc->(1, 2, 3, sub {
20 print "MyModule::server returned @_\n"; 24 print "MyModule::server returned @_\n";
21 $cv->send; 25 $cv->send;
39Loading this module also always loads L<AnyEvent::Fork>, so you can make a 43Loading this module also always loads L<AnyEvent::Fork>, so you can make a
40separate C<use AnyEvent::Fork> if you wish, but you don't have to. 44separate C<use AnyEvent::Fork> if you wish, but you don't have to.
41 45
42=head1 EXAMPLES 46=head1 EXAMPLES
43 47
44=head2 Synchronous Backend 48=head2 Example 1: Synchronous Backend
45 49
46Here is a simple example that implements a backend that executes C<unlink> 50Here is a simple example that implements a backend that executes C<unlink>
47and C<rmdir> calls, and reports their status back. It also reports the 51and C<rmdir> calls, and reports their status back. It also reports the
48number of requests it has processed every three requests, which is clearly 52number of requests it has processed every three requests, which is clearly
49silly, but illustrates the use of events. 53silly, but illustrates the use of events.
50 54
51First the parent process: 55First the parent process:
52 56
53 use AnyEvent; 57 use AnyEvent;
54 use AnyEvent::Fork;
55 use AnyEvent::Fork::RPC; 58 use AnyEvent::Fork::RPC;
56 59
57 my $done = AE::cv; 60 my $done = AE::cv;
58 61
59 my $rpc = AnyEvent::Fork 62 my $rpc = AnyEvent::Fork
137 140
138And as a final remark, there is a fine module on CPAN that can 141And as a final remark, there is a fine module on CPAN that can
139asynchronously C<rmdir> and C<unlink> and a lot more, and more efficiently 142asynchronously C<rmdir> and C<unlink> and a lot more, and more efficiently
140than this example, namely L<IO::AIO>. 143than this example, namely L<IO::AIO>.
141 144
145=head3 Example 1a: the same with the asynchronous backend
146
147This example only shows what needs to be changed to use the async backend
148instead. Doing this is not very useful, the purpose of this example is
149to show the minimum amount of change that is required to go from the
150synchronous to the asynchronous backend.
151
152To use the async backend in the previous example, you need to add the
153C<async> parameter to the C<AnyEvent::Fork::RPC::run> call:
154
155 ->AnyEvent::Fork::RPC::run ("MyWorker::run",
156 async => 1,
157 ...
158
159And since the function call protocol is now changed, you need to adopt
160C<MyWorker::run> to the async API.
161
162First, you need to accept the extra initial C<$done> callback:
163
164 sub run {
165 my ($done, $cmd, $path) = @_;
166
167And since a response is now generated when C<$done> is called, as opposed
168to when the function returns, we need to call the C<$done> function with
169the status:
170
171 $done->($status or (0, "$!"));
172
173A few remarks are in order. First, it's quite pointless to use the async
174backend for this example - but it I<is> possible. Second, you can call
175C<$done> before or after returning from the function. Third, having both
176returned from the function and having called the C<$done> callback, the
177child process may exit at any time, so you should call C<$done> only when
178you really I<are> done.
179
180=head2 Example 2: Asynchronous Backend
181
182This example implements multiple count-downs in the child, using
183L<AnyEvent> timers. While this is a bit silly (one could use timers in te
184parent just as well), it illustrates the ability to use AnyEvent in the
185child and the fact that responses can arrive in a different order then the
186requests.
187
188It also shows how to embed the actual child code into a C<__DATA__>
189section, so it doesn't need any external files at all.
190
191And when your parent process is often busy, and you have stricter timing
192requirements, then running timers in a child process suddenly doesn't look
193so silly anymore.
194
195Without further ado, here is the code:
196
197 use AnyEvent;
198 use AnyEvent::Fork::RPC;
199
200 my $done = AE::cv;
201
202 my $rpc = AnyEvent::Fork
203 ->new
204 ->require ("AnyEvent::Fork::RPC::Async")
205 ->eval (do { local $/; <DATA> })
206 ->AnyEvent::Fork::RPC::run ("run",
207 async => 1,
208 on_error => sub { warn "FATAL: $_[0]"; exit 1 },
209 on_event => sub { print $_[0] },
210 on_destroy => $done,
211 );
212
213 for my $count (3, 2, 1) {
214 $rpc->($count, sub {
215 warn "job $count finished\n";
216 });
217 }
218
219 undef $rpc;
220
221 $done->recv;
222
223 __DATA__
224
225 # this ends up in main, as we don't use a package declaration
226
227 use AnyEvent;
228
229 sub run {
230 my ($done, $count) = @_;
231
232 my $n;
233
234 AnyEvent::Fork::RPC::event "starting to count up to $count\n";
235
236 my $w; $w = AE::timer 1, 1, sub {
237 ++$n;
238
239 AnyEvent::Fork::RPC::event "count $n of $count\n";
240
241 if ($n == $count) {
242 undef $w;
243 $done->();
244 }
245 };
246 }
247
248The parent part (the one before the C<__DATA__> section) isn't very
249different from the earlier examples. It sets async mode, preloads
250the backend module (so the C<AnyEvent::Fork::RPC::event> function is
251declared), uses a slightly different C<on_event> handler (which we use
252simply for logging purposes) and then, instead of loading a module with
253the actual worker code, it C<eval>'s the code from the data section in the
254child process.
255
256It then starts three countdowns, from 3 to 1 seconds downwards, destroys
257the rpc object so the example finishes eventually, and then just waits for
258the stuff to trickle in.
259
260The worker code uses the event function to log some progress messages, but
261mostly just creates a recurring one-second timer.
262
263The timer callback increments a counter, logs a message, and eventually,
264when the count has been reached, calls the finish callback.
265
266On my system, this results in the following output. Since all timers fire
267at roughly the same time, the actual order isn't guaranteed, but the order
268shown is very likely what you would get, too.
269
270 starting to count up to 3
271 starting to count up to 2
272 starting to count up to 1
273 count 1 of 3
274 count 1 of 2
275 count 1 of 1
276 job 1 finished
277 count 2 of 2
278 job 2 finished
279 count 2 of 3
280 count 3 of 3
281 job 3 finished
282
283While the overall ordering isn't guaranteed, the async backend still
284guarantees that events and responses are delivered to the parent process
285in the exact same ordering as they were generated in the child process.
286
287And unless your system is I<very> busy, it should clearly show that the
288job started last will finish first, as it has the lowest count.
289
290This 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
292L<AnyEvent>, but also L<IO::AIO>, or L<Tk> for example.
293
142=head1 PARENT PROCESS USAGE 294=head1 PARENT PROCESS USAGE
143 295
144This module exports nothing, and only implements a single function: 296This module exports nothing, and only implements a single function:
145 297
146=over 4 298=over 4
153 305
154use Errno (); 306use Errno ();
155use Guard (); 307use Guard ();
156 308
157use AnyEvent; 309use AnyEvent;
310# explicit version on next line, as some cpan-testers test with the 0.1 version,
311# ignoring dependencies, and this line will at least give a clear indication of that.
158use AnyEvent::Fork; # we don't actually depend on it, this is for convenience 312use AnyEvent::Fork 0.6; # we don't actually depend on it, this is for convenience
159 313
160our $VERSION = 0.1; 314our $VERSION = 0.2;
161 315
162=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...] 316=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
163 317
164The traditional way to call it. But it is way cooler to call it in the 318The traditional way to call it. But it is way cooler to call it in the
165following way: 319following way:
224 378
225The default server used in the child does all I/O blockingly, and only 379The default server used in the child does all I/O blockingly, and only
226allows a single RPC call to execute concurrently. 380allows a single RPC call to execute concurrently.
227 381
228Setting C<async> to a true value switches to another implementation that 382Setting C<async> to a true value switches to another implementation that
229uses L<AnyEvent> in the child and allows multiple concurrent RPC calls. 383uses L<AnyEvent> in the child and allows multiple concurrent RPC calls (it
384does not support recursion in the event loop however, blocking condvar
385calls will fail).
230 386
231The actual API in the child is documented in the section that describes 387The actual API in the child is documented in the section that describes
232the calling semantics of the returned C<$rpc> function. 388the calling semantics of the returned C<$rpc> function.
233 389
234If you want to pre-load the actual back-end modules to enable memory 390If you want to pre-load the actual back-end modules to enable memory
236synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode. 392synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
237 393
238If you use a template process and want to fork both sync and async 394If you use a template process and want to fork both sync and async
239children, then it is permissible to load both modules. 395children, then it is permissible to load both modules.
240 396
241=item serialiser => $string (default: '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })') 397=item serialiser => $string (default: $AnyEvent::Fork::RPC::STRING_SERIALISER)
242 398
243All arguments, result data and event data have to be serialised to be 399All arguments, result data and event data have to be serialised to be
244transferred between the processes. For this, they have to be frozen and 400transferred between the processes. For this, they have to be frozen and
245thawed in both parent and child processes. 401thawed in both parent and child processes.
246 402
247By default, only octet strings can be passed between the processes, which 403By default, only octet strings can be passed between the processes, which
248is reasonably fast and efficient. 404is reasonably fast and efficient and requires no extra modules.
249 405
250For more complicated use cases, you can provide your own freeze and thaw 406For more complicated use cases, you can provide your own freeze and thaw
251functions, by specifying a string with perl source code. It's supposed to 407functions, by specifying a string with perl source code. It's supposed to
252return two code references when evaluated: the first receives a list of 408return two code references when evaluated: the first receives a list of
253perl values and must return an octet string. The second receives the octet 409perl values and must return an octet string. The second receives the octet
255 411
256If you need an external module for serialisation, then you can either 412If you need an external module for serialisation, then you can either
257pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use> 413pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>
258or C<require> statement into the serialiser string. Or both. 414or C<require> statement into the serialiser string. Or both.
259 415
416Here are some examples - some of them are also available as global
417variables that make them easier to use.
418
419=over 4
420
421=item octet strings - C<$AnyEvent::Fork::RPC::STRING_SERIALISER>
422
423This serialiser concatenates length-prefixes octet strings, and is the
424default.
425
426Implementation:
427
428 (
429 sub { pack "(w/a*)*", @_ },
430 sub { unpack "(w/a*)*", shift }
431 )
432
433=item json - C<$AnyEvent::Fork::RPC::JSON_SERIALISER>
434
435This serialiser creates JSON arrays - you have to make sure the L<JSON>
436module is installed for this serialiser to work. It can be beneficial for
437sharing when you preload the L<JSON> module in a template process.
438
439L<JSON> (with L<JSON::XS> installed) is slower than the octet string
440serialiser, but usually much faster than L<Storable>, unless big chunks of
441binary data need to be transferred.
442
443Implementation:
444
445 use JSON ();
446 (
447 sub { JSON::encode_json \@_ },
448 sub { @{ JSON::decode_json shift } }
449 )
450
451=item storable - C<$AnyEvent::Fork::RPC::STORABLE_SERIALISER>
452
453This serialiser uses L<Storable>, which means it has high chance of
454serialising just about anything you throw at it, at the cost of having
455very high overhead per operation. It also comes with perl.
456
457Implementation:
458
459 use Storable ();
460 (
461 sub { Storable::freeze \@_ },
462 sub { @{ Storable::thaw shift } }
463 )
464
465=back
466
260=back 467=back
261 468
262See the examples section earlier in this document for some actual 469See the examples section earlier in this document for some actual
263examples. 470examples.
264 471
265=cut 472=cut
266 473
267our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })'; 474our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })';
475our $JSON_SERIALISER = 'use JSON (); (sub { JSON::encode_json \@_ }, sub { @{ JSON::decode_json shift } })';
476our $STORABLE_SERIALISER = 'use Storable (); (sub { Storable::freeze \@_ }, sub { @{ Storable::thaw shift } })';
268 477
269sub run { 478sub run {
270 my ($self, $function, %arg) = @_; 479 my ($self, $function, %arg) = @_;
271 480
272 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER; 481 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER;
317 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf; 526 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
318 $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf; 527 $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
319 528
320 if ($len) { 529 if ($len) {
321 while (8 <= length $rbuf) { 530 while (8 <= length $rbuf) {
322 ($id, $len) = unpack "LL", $rbuf; 531 ($id, $len) = unpack "NN", $rbuf;
323 8 + $len <= length $rbuf 532 8 + $len <= length $rbuf
324 or last; 533 or last;
325 534
326 my @r = $t->(substr $rbuf, 8, $len); 535 my @r = $t->(substr $rbuf, 8, $len);
327 substr $rbuf, 0, 8 + $len, ""; 536 substr $rbuf, 0, 8 + $len, "";
341 } 550 }
342 } elsif (defined $len) { 551 } elsif (defined $len) {
343 undef $rw; undef $ww; # it ends here 552 undef $rw; undef $ww; # it ends here
344 553
345 if (@rcb || %rcb) { 554 if (@rcb || %rcb) {
346 use Data::Dump;ddx[\@rcb,\%rcb];#d#
347 $on_error->("unexpected eof"); 555 $on_error->("unexpected eof");
348 } else { 556 } else {
349 $on_destroy->(); 557 $on_destroy->()
558 if $on_destroy;
350 } 559 }
351 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) { 560 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
352 undef $rw; undef $ww; # it ends here 561 undef $rw; undef $ww; # it ends here
353 $on_error->("read: $!"); 562 $on_error->("read: $!");
354 } 563 }
357 $ww ||= AE::io $fh, 1, $wcb; 566 $ww ||= AE::io $fh, 1, $wcb;
358 }); 567 });
359 568
360 my $guard = Guard::guard { 569 my $guard = Guard::guard {
361 $shutdown = 1; 570 $shutdown = 1;
362 $ww ||= $fh && AE::io $fh, 1, $wcb; 571
572 shutdown $fh, 1 if $fh && !$ww;
363 }; 573 };
364 574
365 my $id; 575 my $id;
366 576
367 $arg{async} 577 $arg{async}
369 $id = ($id == 0xffffffff ? 0 : $id) + 1; 579 $id = ($id == 0xffffffff ? 0 : $id) + 1;
370 $id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops 580 $id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops
371 581
372 $rcb{$id} = pop; 582 $rcb{$id} = pop;
373 583
374 $guard; # keep it alive 584 $guard if 0; # keep it alive
375 585
376 $wbuf .= pack "LL/a*", $id, &$f; 586 $wbuf .= pack "NN/a*", $id, &$f;
377 $ww ||= $fh && AE::io $fh, 1, $wcb; 587 $ww ||= $fh && AE::io $fh, 1, $wcb;
378 } 588 }
379 : sub { 589 : sub {
380 push @rcb, pop; 590 push @rcb, pop;
381 591
382 $guard; # keep it alive 592 $guard; # keep it alive
383 593
384 $wbuf .= pack "L/a*", &$f; 594 $wbuf .= pack "N/a*", &$f;
385 $ww ||= $fh && AE::io $fh, 1, $wcb; 595 $ww ||= $fh && AE::io $fh, 1, $wcb;
386 } 596 }
387} 597}
388 598
389=item $rpc->(..., $cb->(...)) 599=item $rpc->(..., $cb->(...))
431See the examples section earlier in this document for some actual 641See the examples section earlier in this document for some actual
432examples. 642examples.
433 643
434=back 644=back
435 645
646=head1 ADVANCED TOPICS
647
648=head2 Choosing a backend
649
650So how do you decide which backend to use? Well, that's your problem to
651solve, but here are some thoughts on the matter:
652
653=over 4
654
655=item Synchronous
656
657The synchronous backend does not rely on any external modules (well,
658except L<common::sense>, which works around a bug in how perl's warning
659system works). This keeps the process very small, for example, on my
660system, an empty perl interpreter uses 1492kB RSS, which becomes 2020kB
661after C<use warnings; use strict> (for people who grew up with C64s around
662them this is probably shocking every single time they see it). The worker
663process in the first example in this document uses 1792kB.
664
665Since the calls are done synchronously, slow jobs will keep newer jobs
666from executing.
667
668The synchronous backend also has no overhead due to running an event loop
669- reading requests is therefore very efficient, while writing responses is
670less so, as every response results in a write syscall.
671
672If the parent process is busy and a bit slow reading responses, the child
673waits instead of processing further requests. This also limits the amount
674of memory needed for buffering, as never more than one response has to be
675buffered.
676
677The API in the child is simple - you just have to define a function that
678does something and returns something.
679
680It's hard to use modules or code that relies on an event loop, as the
681child cannot execute anything while it waits for more input.
682
683=item Asynchronous
684
685The asynchronous backend relies on L<AnyEvent>, which tries to be small,
686but still comes at a price: On my system, the worker from example 1a uses
6873420kB RSS (for L<AnyEvent>, which loads L<EV>, which needs L<XSLoader>
688which in turn loads a lot of other modules such as L<warnings>, L<strict>,
689L<vars>, L<Exporter>...).
690
691It batches requests and responses reasonably efficiently, doing only as
692few reads and writes as needed, but needs to poll for events via the event
693loop.
694
695Responses are queued when the parent process is busy. This means the child
696can continue to execute any queued requests. It also means that a child
697might queue a lot of responses in memory when it generates them and the
698parent process is slow accepting them.
699
700The API is not a straightforward RPC pattern - you have to call a
701"done" callback to pass return values and signal completion. Also, more
702importantly, the API starts jobs as fast as possible - when 1000 jobs
703are queued and the jobs are slow, they will all run concurrently. The
704child must implement some queueing/limiting mechanism if this causes
705problems. Alternatively, the parent could limit the amount of rpc calls
706that are outstanding.
707
708Blocking use of condvars is not supported.
709
710Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is
711easy.
712
713=back
714
715=head2 Passing file descriptors
716
717Unlike L<AnyEvent::Fork>, this module has no in-built file handle or file
718descriptor passing abilities.
719
720The reason is that passing file descriptors is extraordinary tricky
721business, and conflicts with efficient batching of messages.
722
723There still is a method you can use: Create a
724C<AnyEvent::Util::portable_socketpair> and C<send_fh> one half of it to
725the process before you pass control to C<AnyEvent::Fork::RPC::run>.
726
727Whenever you want to pass a file descriptor, send an rpc request to the
728child process (so it expects the descriptor), then send it over the other
729half of the socketpair. The child should fetch the descriptor from the
730half it has passed earlier.
731
732Here is some (untested) pseudocode to that effect:
733
734 use AnyEvent::Util;
735 use AnyEvent::Fork::RPC;
736 use IO::FDPass;
737
738 my ($s1, $s2) = AnyEvent::Util::portable_socketpair;
739
740 my $rpc = AnyEvent::Fork
741 ->new
742 ->send_fh ($s2)
743 ->require ("MyWorker")
744 ->AnyEvent::Fork::RPC::run ("MyWorker::run"
745 init => "MyWorker::init",
746 );
747
748 undef $s2; # no need to keep it around
749
750 # pass an fd
751 $rpc->("i'll send some fd now, please expect it!", my $cv = AE::cv);
752
753 IO::FDPass fileno $s1, fileno $handle_to_pass;
754
755 $cv->recv;
756
757The MyWorker module could look like this:
758
759 package MyWorker;
760
761 use IO::FDPass;
762
763 my $s2;
764
765 sub init {
766 $s2 = $_[0];
767 }
768
769 sub run {
770 if ($_[0] eq "i'll send some fd now, please expect it!") {
771 my $fd = IO::FDPass::recv fileno $s2;
772 ...
773 }
774 }
775
776Of course, this might be blocking if you pass a lot of file descriptors,
777so you might want to look into L<AnyEvent::FDpasser> which can handle the
778gory details.
779
780=head1 EXCEPTIONS
781
782There are no provisions whatsoever for catching exceptions at this time -
783in the child, exeptions might kill the process, causing calls to be lost
784and the parent encountering a fatal error. In the parent, exceptions in
785the result callback will not be caught and cause undefined behaviour.
786
436=head1 SEE ALSO 787=head1 SEE ALSO
437 788
438L<AnyEvent::Fork> (to create the processes in the first place), 789L<AnyEvent::Fork>, to create the processes in the first place.
790
439L<AnyEvent::Fork::Pool> (to manage whole pools of processes). 791L<AnyEvent::Fork::Pool>, to manage whole pools of processes.
440 792
441=head1 AUTHOR AND CONTACT INFORMATION 793=head1 AUTHOR AND CONTACT INFORMATION
442 794
443 Marc Lehmann <schmorp@schmorp.de> 795 Marc Lehmann <schmorp@schmorp.de>
444 http://software.schmorp.de/pkg/AnyEvent-Fork-RPC 796 http://software.schmorp.de/pkg/AnyEvent-Fork-RPC

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines