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.7 by root, Wed Apr 17 20:19:41 2013 UTC vs.
Revision 1.16 by root, Thu Apr 18 14:07:15 2013 UTC

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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines