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.8 by root, Wed Apr 17 20:24:36 2013 UTC vs.
Revision 1.21 by root, Sun Apr 21 12:01:54 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 TECHNOLOGY DEMO
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
224 376
225The default server used in the child does all I/O blockingly, and only 377The default server used in the child does all I/O blockingly, and only
226allows a single RPC call to execute concurrently. 378allows a single RPC call to execute concurrently.
227 379
228Setting C<async> to a true value switches to another implementation that 380Setting C<async> to a true value switches to another implementation that
229uses L<AnyEvent> in the child and allows multiple concurrent RPC calls. 381uses L<AnyEvent> in the child and allows multiple concurrent RPC calls (it
382does not support recursion in the event loop however, blocking condvar
383calls will fail).
230 384
231The actual API in the child is documented in the section that describes 385The actual API in the child is documented in the section that describes
232the calling semantics of the returned C<$rpc> function. 386the calling semantics of the returned C<$rpc> function.
233 387
234If you want to pre-load the actual back-end modules to enable memory 388If you want to pre-load the actual back-end modules to enable memory
236synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode. 390synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
237 391
238If you use a template process and want to fork both sync and async 392If you use a template process and want to fork both sync and async
239children, then it is permissible to load both modules. 393children, then it is permissible to load both modules.
240 394
241=item serialiser => $string (default: '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })') 395=item serialiser => $string (default: $AnyEvent::Fork::RPC::STRING_SERIALISER)
242 396
243All arguments, result data and event data have to be serialised to be 397All arguments, result data and event data have to be serialised to be
244transferred between the processes. For this, they have to be frozen and 398transferred between the processes. For this, they have to be frozen and
245thawed in both parent and child processes. 399thawed in both parent and child processes.
246 400
247By default, only octet strings can be passed between the processes, which 401By default, only octet strings can be passed between the processes, which
248is reasonably fast and efficient. 402is reasonably fast and efficient and requires no extra modules.
249 403
250For more complicated use cases, you can provide your own freeze and thaw 404For 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 405functions, by specifying a string with perl source code. It's supposed to
252return two code references when evaluated: the first receives a list of 406return two code references when evaluated: the first receives a list of
253perl values and must return an octet string. The second receives the octet 407perl values and must return an octet string. The second receives the octet
255 409
256If you need an external module for serialisation, then you can either 410If 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> 411pre-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. 412or C<require> statement into the serialiser string. Or both.
259 413
414Here are some examples - some of them are also available as global
415variables that make them easier to use.
416
417=over 4
418
419=item octet strings - C<$AnyEvent::Fork::RPC::STRING_SERIALISER>
420
421This serialiser concatenates length-prefixes octet strings, and is the
422default.
423
424Implementation:
425
426 (
427 sub { pack "(w/a*)*", @_ },
428 sub { unpack "(w/a*)*", shift }
429 )
430
431=item json - C<$AnyEvent::Fork::RPC::JSON_SERIALISER>
432
433This serialiser creates JSON arrays - you have to make sure the L<JSON>
434module is installed for this serialiser to work. It can be beneficial for
435sharing when you preload the L<JSON> module in a template process.
436
437L<JSON> (with L<JSON::XS> installed) is slower than the octet string
438serialiser, but usually much faster than L<Storable>, unless big chunks of
439binary data need to be transferred.
440
441Implementation:
442
443 use JSON ();
444 (
445 sub { JSON::encode_json \@_ },
446 sub { @{ JSON::decode_json shift } }
447 )
448
449=item storable - C<$AnyEvent::Fork::RPC::STORABLE_SERIALISER>
450
451This serialiser uses L<Storable>, which means it has high chance of
452serialising just about anything you throw at it, at the cost of having
453very high overhead per operation. It also comes with perl.
454
455Implementation:
456
457 use Storable ();
458 (
459 sub { Storable::freeze \@_ },
460 sub { @{ Storable::thaw shift } }
461 )
462
260=back 463=back
261 464
465=back
466
262See the examples section earlier in this document for some actual examples. 467See the examples section earlier in this document for some actual
468examples.
263 469
264=cut 470=cut
265 471
266our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })'; 472our $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 } })';
474our $STORABLE_SERIALISER = 'use Storable (); (sub { Storable::freeze \@_ }, sub { @{ Storable::thaw shift } })';
267 475
268sub run { 476sub run {
269 my ($self, $function, %arg) = @_; 477 my ($self, $function, %arg) = @_;
270 478
271 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER; 479 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER;
281 # default for on_event is to raise an error 489 # default for on_event is to raise an error
282 $on_event ||= sub { $on_error->("event received, but no on_event handler") }; 490 $on_event ||= sub { $on_error->("event received, but no on_event handler") };
283 491
284 my ($f, $t) = eval $serialiser; die $@ if $@; 492 my ($f, $t) = eval $serialiser; die $@ if $@;
285 493
286 my (@rcb, $fh, $shutdown, $wbuf, $ww, $rw); 494 my (@rcb, %rcb, $fh, $shutdown, $wbuf, $ww);
287 my ($rlen, $rbuf) = 512 - 16; 495 my ($rlen, $rbuf, $rw) = 512 - 16;
288 496
289 my $wcb = sub { 497 my $wcb = sub {
290 my $len = syswrite $fh, $wbuf; 498 my $len = syswrite $fh, $wbuf;
291 499
292 if (!defined $len) { 500 unless (defined $len) {
293 if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) { 501 if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
294 undef $rw; undef $ww; # it ends here 502 undef $rw; undef $ww; # it ends here
295 $on_error->("$!"); 503 $on_error->("$!");
296 } 504 }
297 } 505 }
308 516
309 $self->require ($module) 517 $self->require ($module)
310 ->send_arg ($function, $arg{init}, $serialiser) 518 ->send_arg ($function, $arg{init}, $serialiser)
311 ->run ("$module\::run", sub { 519 ->run ("$module\::run", sub {
312 $fh = shift; 520 $fh = shift;
521
522 my ($id, $len);
313 $rw = AE::io $fh, 0, sub { 523 $rw = AE::io $fh, 0, sub {
314 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf; 524 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
315 my $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf; 525 $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
316 526
317 if ($len) { 527 if ($len) {
318 while (4 <= length $rbuf) { 528 while (8 <= length $rbuf) {
319 $len = unpack "L", $rbuf; 529 ($id, $len) = unpack "LL", $rbuf;
320 4 + $len <= length $rbuf 530 8 + $len <= length $rbuf
321 or last; 531 or last;
322 532
323 my @r = $t->(substr $rbuf, 4, $len); 533 my @r = $t->(substr $rbuf, 8, $len);
324 substr $rbuf, 0, $len + 4, ""; 534 substr $rbuf, 0, 8 + $len, "";
535
536 if ($id) {
537 if (@rcb) {
538 (shift @rcb)->(@r);
539 } elsif (my $cb = delete $rcb{$id}) {
540 $cb->(@r);
541 } else {
542 undef $rw; undef $ww;
543 $on_error->("unexpected data from child");
325 544 }
326 if (pop @r) { 545 } else {
327 $on_event->(@r); 546 $on_event->(@r);
328 } elsif (@rcb) {
329 (shift @rcb)->(@r);
330 } else {
331 undef $rw; undef $ww;
332 $on_error->("unexpected data from child");
333 } 547 }
334 } 548 }
335 } elsif (defined $len) { 549 } elsif (defined $len) {
336 undef $rw; undef $ww; # it ends here 550 undef $rw; undef $ww; # it ends here
337 551
338 if (@rcb) { 552 if (@rcb || %rcb) {
339 $on_error->("unexpected eof"); 553 $on_error->("unexpected eof");
340 } else { 554 } else {
341 $on_destroy->(); 555 $on_destroy->()
556 if $on_destroy;
342 } 557 }
343 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) { 558 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
344 undef $rw; undef $ww; # it ends here 559 undef $rw; undef $ww; # it ends here
345 $on_error->("read: $!"); 560 $on_error->("read: $!");
346 } 561 }
349 $ww ||= AE::io $fh, 1, $wcb; 564 $ww ||= AE::io $fh, 1, $wcb;
350 }); 565 });
351 566
352 my $guard = Guard::guard { 567 my $guard = Guard::guard {
353 $shutdown = 1; 568 $shutdown = 1;
354 $ww ||= $fh && AE::io $fh, 1, $wcb; 569
570 shutdown $fh, 1 if $fh && !$ww;
355 }; 571 };
356 572
573 my $id;
574
575 $arg{async}
357 sub { 576 ? sub {
358 push @rcb, pop; 577 $id = ($id == 0xffffffff ? 0 : $id) + 1;
578 $id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops
359 579
580 $rcb{$id} = pop;
581
360 $guard; # keep it alive 582 $guard if 0; # keep it alive
361 583
362 $wbuf .= pack "L/a*", &$f; 584 $wbuf .= pack "LL/a*", $id, &$f;
363 $ww ||= $fh && AE::io $fh, 1, $wcb; 585 $ww ||= $fh && AE::io $fh, 1, $wcb;
364 } 586 }
587 : sub {
588 push @rcb, pop;
589
590 $guard; # keep it alive
591
592 $wbuf .= pack "L/a*", &$f;
593 $ww ||= $fh && AE::io $fh, 1, $wcb;
594 }
365} 595}
366 596
367=item $rpc->(..., $cb->(...)) 597=item $rpc->(..., $cb->(...))
368 598
369The RPC object returned by C<AnyEvent::Fork::RPC::run> is actually a code 599The RPC object returned by C<AnyEvent::Fork::RPC::run> is actually a code
409See the examples section earlier in this document for some actual 639See the examples section earlier in this document for some actual
410examples. 640examples.
411 641
412=back 642=back
413 643
644=head1 ADVANCED TOPICS
645
646=head2 Choosing a backend
647
648So how do you decide which backend to use? Well, that's your problem to
649solve, but here are some thoughts on the matter:
650
651=over 4
652
653=item Synchronous
654
655The synchronous backend does not rely on any external modules (well,
656except L<common::sense>, which works around a bug in how perl's warning
657system works). This keeps the process very small, for example, on my
658system, an empty perl interpreter uses 1492kB RSS, which becomes 2020kB
659after C<use warnings; use strict> (for people who grew up with C64s around
660them this is probably shocking every single time they see it). The worker
661process in the first example in this document uses 1792kB.
662
663Since the calls are done synchronously, slow jobs will keep newer jobs
664from executing.
665
666The synchronous backend also has no overhead due to running an event loop
667- reading requests is therefore very efficient, while writing responses is
668less so, as every response results in a write syscall.
669
670If the parent process is busy and a bit slow reading responses, the child
671waits instead of processing further requests. This also limits the amount
672of memory needed for buffering, as never more than one response has to be
673buffered.
674
675The API in the child is simple - you just have to define a function that
676does something and returns something.
677
678It's hard to use modules or code that relies on an event loop, as the
679child cannot execute anything while it waits for more input.
680
681=item Asynchronous
682
683The asynchronous backend relies on L<AnyEvent>, which tries to be small,
684but still comes at a price: On my system, the worker from example 1a uses
6853420kB RSS (for L<AnyEvent>, which loads L<EV>, which needs L<XSLoader>
686which in turn loads a lot of other modules such as L<warnings>, L<strict>,
687L<vars>, L<Exporter>...).
688
689It batches requests and responses reasonably efficiently, doing only as
690few reads and writes as needed, but needs to poll for events via the event
691loop.
692
693Responses are queued when the parent process is busy. This means the child
694can continue to execute any queued requests. It also means that a child
695might queue a lot of responses in memory when it generates them and the
696parent process is slow accepting them.
697
698The API is not a straightforward RPC pattern - you have to call a
699"done" callback to pass return values and signal completion. Also, more
700importantly, the API starts jobs as fast as possible - when 1000 jobs
701are queued and the jobs are slow, they will all run concurrently. The
702child must implement some queueing/limiting mechanism if this causes
703problems. Alternatively, the parent could limit the amount of rpc calls
704that are outstanding.
705
706Blocking use of condvars is not supported.
707
708Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is
709easy.
710
711=back
712
713=head2 Passing file descriptors
714
715Unlike L<AnyEvent::Fork>, this module has no in-built file handle or file
716descriptor passing abilities.
717
718The reason is that passing file descriptors is extraordinary tricky
719business, and conflicts with efficient batching of messages.
720
721There still is a method you can use: Create a
722C<AnyEvent::Util::portable_socketpair> and C<send_fh> one half of it to
723the process before you pass control to C<AnyEvent::Fork::RPC::run>.
724
725Whenever you want to pass a file descriptor, send an rpc request to the
726child process (so it expects the descriptor), then send it over the other
727half of the socketpair. The child should fetch the descriptor from the
728half it has passed earlier.
729
730Here is some (untested) pseudocode to that effect:
731
732 use AnyEvent::Util;
733 use AnyEvent::Fork::RPC;
734 use IO::FDPass;
735
736 my ($s1, $s2) = AnyEvent::Util::portable_socketpair;
737
738 my $rpc = AnyEvent::Fork
739 ->new
740 ->send_fh ($s2)
741 ->require ("MyWorker")
742 ->AnyEvent::Fork::RPC::run ("MyWorker::run"
743 init => "MyWorker::init",
744 );
745
746 undef $s2; # no need to keep it around
747
748 # pass an fd
749 $rpc->("i'll send some fd now, please expect it!", my $cv = AE::cv);
750
751 IO::FDPass fileno $s1, fileno $handle_to_pass;
752
753 $cv->recv;
754
755The MyWorker module could look like this:
756
757 package MyWorker;
758
759 use IO::FDPass;
760
761 my $s2;
762
763 sub init {
764 $s2 = $_[0];
765 }
766
767 sub run {
768 if ($_[0] eq "i'll send some fd now, please expect it!") {
769 my $fd = IO::FDPass::recv fileno $s2;
770 ...
771 }
772 }
773
774Of course, this might be blocking if you pass a lot of file descriptors,
775so you might want to look into L<AnyEvent::FDpasser> which can handle the
776gory details.
777
778=head1 EXCEPTIONS
779
780There are no provisions whatsoever for catching exceptions at this time -
781in the child, exeptions might kill the process, causing calls to be lost
782and the parent encountering a fatal error. In the parent, exceptions in
783the result callback will not be caught and cause undefined behaviour.
784
414=head1 SEE ALSO 785=head1 SEE ALSO
415 786
416L<AnyEvent::Fork> (to create the processes in the first place), 787L<AnyEvent::Fork>, to create the processes in the first place.
788
417L<AnyEvent::Fork::Pool> (to manage whole pools of processes). 789L<AnyEvent::Fork::Pool>, to manage whole pools of processes.
418 790
419=head1 AUTHOR AND CONTACT INFORMATION 791=head1 AUTHOR AND CONTACT INFORMATION
420 792
421 Marc Lehmann <schmorp@schmorp.de> 793 Marc Lehmann <schmorp@schmorp.de>
422 http://software.schmorp.de/pkg/AnyEvent-Fork-RPC 794 http://software.schmorp.de/pkg/AnyEvent-Fork-RPC

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines