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.5 by root, Wed Apr 17 19:39:54 2013 UTC vs.
Revision 1.12 by root, Thu Apr 18 10:40:34 2013 UTC

2 2
3AnyEvent::Fork::RPC - simple RPC extension for AnyEvent::Fork 3AnyEvent::Fork::RPC - simple RPC extension for AnyEvent::Fork
4 4
5=head1 SYNOPSIS 5=head1 SYNOPSIS
6 6
7 use AnyEvent::Fork;
8 use AnyEvent::Fork::RPC; 7 use AnyEvent::Fork::RPC;
8 # use AnyEvent::Fork is not needed
9 9
10 my $rpc = AnyEvent::Fork 10 my $rpc = AnyEvent::Fork
11 ->new 11 ->new
12 ->require ("MyModule") 12 ->require ("MyModule")
13 ->AnyEvent::Fork::RPC::run ( 13 ->AnyEvent::Fork::RPC::run (
34concurrently in the child, using AnyEvent. 34concurrently in the child, using AnyEvent.
35 35
36It also implements an asynchronous event mechanism from the child to the 36It also implements an asynchronous event mechanism from the child to the
37parent, that could be used for progress indications or other information. 37parent, that could be used for progress indications or other information.
38 38
39Loading 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.
41
39=head1 EXAMPLES 42=head1 EXAMPLES
40 43
41=head2 Synchronous Backend 44=head2 Example 1: Synchronous Backend
42 45
43Here is a simple example that implements a backend that executes C<unlink> 46Here is a simple example that implements a backend that executes C<unlink>
44and C<rmdir> calls, and reports their status back. It also reports the 47and C<rmdir> calls, and reports their status back. It also reports the
45number of requests it has processed every three requests, which is clearly 48number of requests it has processed every three requests, which is clearly
46silly, but illustrates the use of events. 49silly, but illustrates the use of events.
47 50
48First the parent process: 51First the parent process:
49 52
50 use AnyEvent; 53 use AnyEvent;
51 use AnyEvent::Fork;
52 use AnyEvent::Fork::RPC; 54 use AnyEvent::Fork::RPC;
53 55
54 my $done = AE::cv; 56 my $done = AE::cv;
55 57
56 my $rpc = AnyEvent::Fork 58 my $rpc = AnyEvent::Fork
103dies with a fatal error - obviously, you must never let this happen :). 105dies with a fatal error - obviously, you must never let this happen :).
104 106
105Eventually it returns the status value true if the command was successful, 107Eventually it returns the status value true if the command was successful,
106or the status value 0 and the stringified error message. 108or the status value 0 and the stringified error message.
107 109
108On my system, running the first cdoe fragment with the given 110On my system, running the first code fragment with the given
109F<MyWorker.pm> in the current directory yields: 111F<MyWorker.pm> in the current directory yields:
110 112
111 /tmp/somepath/1: No such file or directory 113 /tmp/somepath/1: No such file or directory
112 /tmp/somepath/2: No such file or directory 114 /tmp/somepath/2: No such file or directory
113 3 requests handled 115 3 requests handled
134 136
135And as a final remark, there is a fine module on CPAN that can 137And as a final remark, there is a fine module on CPAN that can
136asynchronously C<rmdir> and C<unlink> and a lot more, and more efficiently 138asynchronously C<rmdir> and C<unlink> and a lot more, and more efficiently
137than this example, namely L<IO::AIO>. 139than this example, namely L<IO::AIO>.
138 140
141=head3 Example 1a: the same with the asynchronous backend
142
143This example only shows what needs to be changed to use the async backend
144instead. Doing this is not very useful, the purpose of this example is
145to show the minimum amount of change that is required to go from the
146synchronous to the asynchronous backend.
147
148To use the async backend in the previous example, you need to add the
149C<async> parameter to the C<AnyEvent::Fork::RPC::run> call:
150
151 ->AnyEvent::Fork::RPC::run ("MyWorker::run",
152 async => 1,
153 ...
154
155And since the function call protocol is now changed, you need to adopt
156C<MyWorker::run> to the async API.
157
158First, you need to accept the extra initial C<$done> callback:
159
160 sub run {
161 my ($done, $cmd, $path) = @_;
162
163And since a response is now generated when C<$done> is called, as opposed
164to when the function returns, we need to call the C<$done> function with
165the status:
166
167 $done->($status or (0, "$!"));
168
169A few remarks are in order. First, it's quite pointless to use the async
170backend for this example - but it I<is> possible. Second, you can call
171C<$done> before or after returning from the function. Third, having both
172returned from the function and having called the C<$done> callback, the
173child process may exit at any time, so you should call C<$done> only when
174you really I<are> done.
175
176=head2 Example 2: Asynchronous Backend
177
178This example implements multiple count-downs in the child, using
179L<AnyEvent> timers. While this is a bit silly (one could use timers in te
180parent just as well), it illustrates the ability to use AnyEvent in the
181child and the fact that responses can arrive in a different order then the
182requests.
183
184It also shows how to embed the actual child code into a C<__DATA__>
185section, so it doesn't need any external files at all.
186
187And when your parent process is often busy, and you have stricter timing
188requirements, then running timers in a child process suddenly doesn't look
189so silly anymore.
190
191Without further ado, here is the code:
192
193 use AnyEvent;
194 use AnyEvent::Fork::RPC;
195
196 my $done = AE::cv;
197
198 my $rpc = AnyEvent::Fork
199 ->new
200 ->require ("AnyEvent::Fork::RPC::Async")
201 ->eval (do { local $/; <DATA> })
202 ->AnyEvent::Fork::RPC::run ("run",
203 async => 1,
204 on_error => sub { warn "FATAL: $_[0]"; exit 1 },
205 on_event => sub { print $_[0] },
206 on_destroy => $done,
207 );
208
209 for my $count (3, 2, 1) {
210 $rpc->($count, sub {
211 warn "job $count finished\n";
212 });
213 }
214
215 undef $rpc;
216
217 $done->recv;
218
219 __DATA__
220
221 # this ends up in main, as we don't use a package declaration
222
223 use AnyEvent;
224
225 sub run {
226 my ($done, $count) = @_;
227
228 my $n;
229
230 AnyEvent::Fork::RPC::event "starting to count up to $count\n";
231
232 my $w; $w = AE::timer 1, 1, sub {
233 ++$n;
234
235 AnyEvent::Fork::RPC::event "count $n of $count\n";
236
237 if ($n == $count) {
238 undef $w;
239 $done->();
240 }
241 };
242 }
243
244The parent part (the one before the C<__DATA__> section) isn't very
245different from the earlier examples. It sets async mode, preloads
246the backend module (so the C<AnyEvent::Fork::RPC::event> function is
247declared), uses a slightly different C<on_event> handler (which we use
248simply for logging purposes) and then, instead of loading a module with
249the actual worker code, it C<eval>'s the code from the data section in the
250child process.
251
252It then starts three countdowns, from 3 to 1 seconds downwards, destroys
253the rpc object so the example finishes eventually, and then just waits for
254the stuff to trickle in.
255
256The worker code uses the event function to log some progress messages, but
257mostly just creates a recurring one-second timer.
258
259The timer callback increments a counter, logs a message, and eventually,
260when the count has been reached, calls the finish callback.
261
262On my system, this results in the following output. Since all timers fire
263at roughly the same time, the actual order isn't guaranteed, but the order
264shown is very likely what you would get, too.
265
266 starting to count up to 3
267 starting to count up to 2
268 starting to count up to 1
269 count 1 of 3
270 count 1 of 2
271 count 1 of 1
272 job 1 finished
273 count 2 of 2
274 job 2 finished
275 count 2 of 3
276 count 3 of 3
277 job 3 finished
278
279While the overall ordering isn't guaranteed, the async backend still
280guarantees that events and responses are delivered to the parent process
281in the exact same ordering as they were generated in the child process.
282
283And unless your system is I<very> busy, it should clearly show that the
284job started last will finish first, as it has the lowest count.
285
286This concludes the async example. Since L<AnyEvent::Fork> does not
287actually fork, you are free to use about any module in the child, not just
288L<AnyEvent>, but also L<IO::AIO>, or L<Tk> for example.
289
139=head1 PARENT PROCESS USAGE 290=head1 PARENT PROCESS USAGE
140 291
141This module exports nothing, and only implements a single function: 292This module exports nothing, and only implements a single function:
142 293
143=over 4 294=over 4
150 301
151use Errno (); 302use Errno ();
152use Guard (); 303use Guard ();
153 304
154use AnyEvent; 305use AnyEvent;
155#use AnyEvent::Fork; 306use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
156 307
157our $VERSION = 0.1; 308our $VERSION = 0.1;
158 309
159=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...] 310=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
160 311
198 349
199Called when the C<$rpc> object has been destroyed and all requests have 350Called when the C<$rpc> object has been destroyed and all requests have
200been successfully handled. This is useful when you queue some requests and 351been successfully handled. This is useful when you queue some requests and
201want the child to go away after it has handled them. The problem is that 352want the child to go away after it has handled them. The problem is that
202the parent must not exit either until all requests have been handled, and 353the parent must not exit either until all requests have been handled, and
203this cna be accomplished by waiting for this callback. 354this can be accomplished by waiting for this callback.
204 355
205=item init => $function (default none) 356=item init => $function (default none)
206 357
207When specified (by name), this function is called in the child as the very 358When specified (by name), this function is called in the child as the very
208first thing when taking over the process, with all the arguments normally 359first thing when taking over the process, with all the arguments normally
231If you want to pre-load the actual back-end modules to enable memory 382If you want to pre-load the actual back-end modules to enable memory
232sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for 383sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for
233synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode. 384synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
234 385
235If you use a template process and want to fork both sync and async 386If you use a template process and want to fork both sync and async
236children, then it is permissible to laod both modules. 387children, then it is permissible to load both modules.
237 388
238=item serialiser => $string (default: '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })') 389=item serialiser => $string (default: '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })')
239 390
240All arguments, result data and event data have to be serialised to be 391All arguments, result data and event data have to be serialised to be
241transferred between the processes. For this, they have to be frozen and 392transferred between the processes. For this, they have to be frozen and
253If you need an external module for serialisation, then you can either 404If you need an external module for serialisation, then you can either
254pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use> 405pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>
255or C<require> statement into the serialiser string. Or both. 406or C<require> statement into the serialiser string. Or both.
256 407
257=back 408=back
409
410See the examples section earlier in this document for some actual
411examples.
258 412
259=cut 413=cut
260 414
261our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })'; 415our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })';
262 416
276 # default for on_event is to raise an error 430 # default for on_event is to raise an error
277 $on_event ||= sub { $on_error->("event received, but no on_event handler") }; 431 $on_event ||= sub { $on_error->("event received, but no on_event handler") };
278 432
279 my ($f, $t) = eval $serialiser; die $@ if $@; 433 my ($f, $t) = eval $serialiser; die $@ if $@;
280 434
281 my (@rcb, $fh, $shutdown, $wbuf, $ww, $rw); 435 my (@rcb, %rcb, $fh, $shutdown, $wbuf, $ww);
282 my ($rlen, $rbuf) = 512 - 16; 436 my ($rlen, $rbuf, $rw) = 512 - 16;
283 437
284 my $wcb = sub { 438 my $wcb = sub {
285 my $len = syswrite $fh, $wbuf; 439 my $len = syswrite $fh, $wbuf;
286 440
287 if (!defined $len) { 441 unless (defined $len) {
288 if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) { 442 if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
289 undef $rw; undef $ww; # it ends here 443 undef $rw; undef $ww; # it ends here
290 $on_error->("$!"); 444 $on_error->("$!");
291 } 445 }
292 } 446 }
303 457
304 $self->require ($module) 458 $self->require ($module)
305 ->send_arg ($function, $arg{init}, $serialiser) 459 ->send_arg ($function, $arg{init}, $serialiser)
306 ->run ("$module\::run", sub { 460 ->run ("$module\::run", sub {
307 $fh = shift; 461 $fh = shift;
462
463 my ($id, $len);
308 $rw = AE::io $fh, 0, sub { 464 $rw = AE::io $fh, 0, sub {
309 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf; 465 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
310 my $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf; 466 $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
311 467
312 if ($len) { 468 if ($len) {
313 while (5 <= length $rbuf) { 469 while (8 <= length $rbuf) {
314 $len = unpack "L", $rbuf; 470 ($id, $len) = unpack "LL", $rbuf;
315 4 + $len <= length $rbuf 471 8 + $len <= length $rbuf
316 or last; 472 or last;
317 473
318 my @r = $t->(substr $rbuf, 4, $len); 474 my @r = $t->(substr $rbuf, 8, $len);
319 substr $rbuf, 0, $len + 4, ""; 475 substr $rbuf, 0, 8 + $len, "";
476
477 if ($id) {
478 if (@rcb) {
479 (shift @rcb)->(@r);
480 } elsif (my $cb = delete $rcb{$id}) {
481 $cb->(@r);
482 } else {
483 undef $rw; undef $ww;
484 $on_error->("unexpected data from child");
320 485 }
321 if (pop @r) { 486 } else {
322 $on_event->(@r); 487 $on_event->(@r);
323 } elsif (@rcb) {
324 (shift @rcb)->(@r);
325 } else {
326 undef $rw; undef $ww;
327 $on_error->("unexpected data from child");
328 } 488 }
329 } 489 }
330 } elsif (defined $len) { 490 } elsif (defined $len) {
331 undef $rw; undef $ww; # it ends here 491 undef $rw; undef $ww; # it ends here
332 492
333 if (@rcb) { 493 if (@rcb || %rcb) {
494 use Data::Dump;ddx[\@rcb,\%rcb];#d#
334 $on_error->("unexpected eof"); 495 $on_error->("unexpected eof");
335 } else { 496 } else {
336 $on_destroy->(); 497 $on_destroy->();
337 } 498 }
338 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) { 499 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
347 my $guard = Guard::guard { 508 my $guard = Guard::guard {
348 $shutdown = 1; 509 $shutdown = 1;
349 $ww ||= $fh && AE::io $fh, 1, $wcb; 510 $ww ||= $fh && AE::io $fh, 1, $wcb;
350 }; 511 };
351 512
513 my $id;
514
515 $arg{async}
352 sub { 516 ? sub {
353 push @rcb, pop; 517 $id = ($id == 0xffffffff ? 0 : $id) + 1;
518 $id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops
354 519
520 $rcb{$id} = pop;
521
355 $guard; # keep it alive 522 $guard; # keep it alive
356 523
357 $wbuf .= pack "L/a*", &$f; 524 $wbuf .= pack "LL/a*", $id, &$f;
358 $ww ||= $fh && AE::io $fh, 1, $wcb; 525 $ww ||= $fh && AE::io $fh, 1, $wcb;
359 } 526 }
527 : sub {
528 push @rcb, pop;
529
530 $guard; # keep it alive
531
532 $wbuf .= pack "L/a*", &$f;
533 $ww ||= $fh && AE::io $fh, 1, $wcb;
534 }
360} 535}
361 536
362=item $rpc->(..., $cb->(...)) 537=item $rpc->(..., $cb->(...))
363 538
364The RPC object returned by C<AnyEvent::Fork::RPC::run> is actually a code 539The RPC object returned by C<AnyEvent::Fork::RPC::run> is actually a code
379 554
380The other thing that can be done with the RPC object is to destroy it. In 555The other thing that can be done with the RPC object is to destroy it. In
381this case, the child process will execute all remaining RPC calls, report 556this case, the child process will execute all remaining RPC calls, report
382their results, and then exit. 557their results, and then exit.
383 558
559See the examples section earlier in this document for some actual
560examples.
561
384=back 562=back
385 563
386=head1 CHILD PROCESS USAGE 564=head1 CHILD PROCESS USAGE
387 565
388The following function is not available in this module. They are only 566The following function is not available in this module. They are only
396 574
397Send an event to the parent. Events are a bit like RPC calls made by the 575Send an event to the parent. Events are a bit like RPC calls made by the
398child process to the parent, except that there is no notion of return 576child process to the parent, except that there is no notion of return
399values. 577values.
400 578
579See the examples section earlier in this document for some actual
580examples.
581
401=back 582=back
583
584=head1 ADVANCED TOPICS
585
586=head2 Choosing a backend
587
588So how do you decide which backend to use? Well, that's your problem to
589solve, but here are some thoughts on the matter:
590
591=over 4
592
593=item Synchronous
594
595The synchronous backend does not rely on any external modules (well,
596except L<common::sense>, which works around a bug in how perl's warning
597system works). This keeps the process very small, for example, on my
598system, an empty perl interpreter uses 1492kB RSS, which becomes 2020kB
599after C<use warnings; use strict> (for people who grew up with C64s around
600them this is probably shocking every single time they see it). The worker
601process in the first example in this document uses 1792kB.
602
603Since the calls are done synchronously, slow jobs will keep newer jobs
604from executing.
605
606The synchronous backend also has no overhead due to running an event loop
607- reading requests is therefore very efficient, while writing responses is
608less so, as every response results in a write syscall.
609
610If the parent process is busy and a bit slow reading responses, the child
611waits instead of processing further requests. This also limits the amount
612of memory needed for buffering, as never more than one response has to be
613buffered.
614
615The API in the child is simple - you just have to define a function that
616does something and returns something.
617
618It's hard to use modules or code that relies on an event loop, as the
619child cannot execute anything while it waits for more input.
620
621=item Asynchronous
622
623The asynchronous backend relies on L<AnyEvent>, which tries to be small,
624but still comes at a price: On my system, the worker from example 1a uses
6253420kB RSS (for L<AnyEvent>, which loads L<EV>, which needs L<XSLoader>
626which in turn loads a lot of other modules such as L<warnings>, L<strict>,
627L<vars>, L<Exporter>...).
628
629It batches requests and responses reasonably efficiently, doing only as
630few reads and writes as needed, but needs to poll for events via the event
631loop.
632
633Responses are queued when the parent process is busy. This means the child
634can continue to execute any queued requests. It also means that a child
635might queue a lot of responses in memory when it generates them and the
636parent process is slow accepting them.
637
638The API is not a straightforward RPC pattern - you have to call a
639"done" callback to pass return values and signal completion. Also, more
640importantly, the API starts jobs as fast as possible - when 1000 jobs
641are queued and the jobs are slow, they will all run concurrently. The
642child must implement some queueing/limiting mechanism if this causes
643problems. Alternatively, the parent could limit the amount of rpc calls
644that are outstanding.
645
646Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is
647easy.
648
649=back
650
651=head2 Passing file descriptors
652
653Unlike L<AnyEvent::Fork>, this module has no in-built file handle or file
654descriptor passing abilities.
655
656The reason is that passing file descriptors is extraordinary tricky
657business, and conflicts with efficient batching of messages.
658
659There still is a method you can use: Create a
660C<AnyEvent::Util::portable_socketpair> and C<send_fh> one half of it to
661the process before you pass control to C<AnyEvent::Fork::RPC::run>.
662
663Whenever you want to pass a file descriptor, send an rpc request to the
664child process (so it expects the descriptor), then send it over the other
665half of the socketpair. The child should fetch the descriptor from the
666half it has passed earlier.
667
668Here is some (untested) pseudocode to that effect:
669
670 use AnyEvent::Util;
671 use AnyEvent::Fork::RPC;
672 use IO::FDPass;
673
674 my ($s1, $s2) = AnyEvent::Util::portable_socketpair;
675
676 my $rpc = AnyEvent::Fork
677 ->new
678 ->send_fh ($s2)
679 ->require ("MyWorker")
680 ->AnyEvent::Fork::RPC::run ("MyWorker::run"
681 init => "MyWorker::init",
682 );
683
684 undef $s2; # no need to keep it around
685
686 # pass an fd
687 $rpc->("i'll send some fd now, please expect it!", my $cv = AE::cv);
688
689 IO::FDPass fileno $s1, fileno $handle_to_pass;
690
691 $cv->recv;
692
693The MyWorker module could look like this:
694
695 package MyWorker;
696
697 use IO::FDPass;
698
699 my $s2;
700
701 sub init {
702 $s2 = $_[0];
703 }
704
705 sub run {
706 if ($_[0] eq "i'll send some fd now, please expect it!") {
707 my $fd = IO::FDPass::recv fileno $s2;
708 ...
709 }
710 }
711
712Of course, this might be blocking if you pass a lot of file descriptors,
713so you might want to look into L<AnyEvent::FDpasser> which can handle the
714gory details.
402 715
403=head1 SEE ALSO 716=head1 SEE ALSO
404 717
405L<AnyEvent::Fork> (to create the processes in the first place), 718L<AnyEvent::Fork> (to create the processes in the first place),
406L<AnyEvent::Fork::Pool> (to manage whole pools of processes). 719L<AnyEvent::Fork::Pool> (to manage whole pools of processes).

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines