ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-Fork-RPC/RPC.pm
Revision: 1.29
Committed: Sun Aug 25 21:52:15 2013 UTC (10 years, 9 months ago) by root
Branch: MAIN
Changes since 1.28: +61 -10 lines
Log Message:
*** empty log message ***

File Contents

# Content
1 =head1 NAME
2
3 AnyEvent::Fork::RPC - simple RPC extension for AnyEvent::Fork
4
5 =head1 SYNOPSIS
6
7 use AnyEvent::Fork;
8 use AnyEvent::Fork::RPC;
9
10 my $rpc = AnyEvent::Fork
11 ->new
12 ->require ("MyModule")
13 ->AnyEvent::Fork::RPC::run (
14 "MyModule::server",
15 );
16
17 use AnyEvent;
18
19 my $cv = AE::cv;
20
21 $rpc->(1, 2, 3, sub {
22 print "MyModule::server returned @_\n";
23 $cv->send;
24 });
25
26 $cv->recv;
27
28 =head1 DESCRIPTION
29
30 This module implements a simple RPC protocol and backend for processes
31 created via L<AnyEvent::Fork> or L<AnyEvent::Fork::Remote>, allowing you
32 to call a function in the child process and receive its return values (up
33 to 4GB serialised).
34
35 It implements two different backends: a synchronous one that works like a
36 normal function call, and an asynchronous one that can run multiple jobs
37 concurrently in the child, using AnyEvent.
38
39 It also implements an asynchronous event mechanism from the child to the
40 parent, that could be used for progress indications or other information.
41
42 =head1 EXAMPLES
43
44 =head2 Example 1: Synchronous Backend
45
46 Here is a simple example that implements a backend that executes C<unlink>
47 and C<rmdir> calls, and reports their status back. It also reports the
48 number of requests it has processed every three requests, which is clearly
49 silly, but illustrates the use of events.
50
51 First the parent process:
52
53 use AnyEvent;
54 use AnyEvent::Fork;
55 use AnyEvent::Fork::RPC;
56
57 my $done = AE::cv;
58
59 my $rpc = AnyEvent::Fork
60 ->new
61 ->require ("MyWorker")
62 ->AnyEvent::Fork::RPC::run ("MyWorker::run",
63 on_error => sub { warn "ERROR: $_[0]"; exit 1 },
64 on_event => sub { warn "$_[0] requests handled\n" },
65 on_destroy => $done,
66 );
67
68 for my $id (1..6) {
69 $rpc->(rmdir => "/tmp/somepath/$id", sub {
70 $_[0]
71 or warn "/tmp/somepath/$id: $_[1]\n";
72 });
73 }
74
75 undef $rpc;
76
77 $done->recv;
78
79 The parent creates the process, queues a few rmdir's. It then forgets
80 about the C<$rpc> object, so that the child exits after it has handled the
81 requests, and then it waits till the requests have been handled.
82
83 The child is implemented using a separate module, C<MyWorker>, shown here:
84
85 package MyWorker;
86
87 my $count;
88
89 sub run {
90 my ($cmd, $path) = @_;
91
92 AnyEvent::Fork::RPC::event ($count)
93 unless ++$count % 3;
94
95 my $status = $cmd eq "rmdir" ? rmdir $path
96 : $cmd eq "unlink" ? unlink $path
97 : die "fatal error, illegal command '$cmd'";
98
99 $status or (0, "$!")
100 }
101
102 1
103
104 The C<run> function first sends a "progress" event every three calls, and
105 then executes C<rmdir> or C<unlink>, depending on the first parameter (or
106 dies with a fatal error - obviously, you must never let this happen :).
107
108 Eventually it returns the status value true if the command was successful,
109 or the status value 0 and the stringified error message.
110
111 On my system, running the first code fragment with the given
112 F<MyWorker.pm> in the current directory yields:
113
114 /tmp/somepath/1: No such file or directory
115 /tmp/somepath/2: No such file or directory
116 3 requests handled
117 /tmp/somepath/3: No such file or directory
118 /tmp/somepath/4: No such file or directory
119 /tmp/somepath/5: No such file or directory
120 6 requests handled
121 /tmp/somepath/6: No such file or directory
122
123 Obviously, none of the directories I am trying to delete even exist. Also,
124 the events and responses are processed in exactly the same order as
125 they were created in the child, which is true for both synchronous and
126 asynchronous backends.
127
128 Note that the parentheses in the call to C<AnyEvent::Fork::RPC::event> are
129 not optional. That is because the function isn't defined when the code is
130 compiled. You can make sure it is visible by pre-loading the correct
131 backend module in the call to C<require>:
132
133 ->require ("AnyEvent::Fork::RPC::Sync", "MyWorker")
134
135 Since the backend module declares the C<event> function, loading it first
136 ensures that perl will correctly interpret calls to it.
137
138 And as a final remark, there is a fine module on CPAN that can
139 asynchronously C<rmdir> and C<unlink> and a lot more, and more efficiently
140 than this example, namely L<IO::AIO>.
141
142 =head3 Example 1a: the same with the asynchronous backend
143
144 This example only shows what needs to be changed to use the async backend
145 instead. Doing this is not very useful, the purpose of this example is
146 to show the minimum amount of change that is required to go from the
147 synchronous to the asynchronous backend.
148
149 To use the async backend in the previous example, you need to add the
150 C<async> parameter to the C<AnyEvent::Fork::RPC::run> call:
151
152 ->AnyEvent::Fork::RPC::run ("MyWorker::run",
153 async => 1,
154 ...
155
156 And since the function call protocol is now changed, you need to adopt
157 C<MyWorker::run> to the async API.
158
159 First, you need to accept the extra initial C<$done> callback:
160
161 sub run {
162 my ($done, $cmd, $path) = @_;
163
164 And since a response is now generated when C<$done> is called, as opposed
165 to when the function returns, we need to call the C<$done> function with
166 the status:
167
168 $done->($status or (0, "$!"));
169
170 A few remarks are in order. First, it's quite pointless to use the async
171 backend for this example - but it I<is> possible. Second, you can call
172 C<$done> before or after returning from the function. Third, having both
173 returned from the function and having called the C<$done> callback, the
174 child process may exit at any time, so you should call C<$done> only when
175 you really I<are> done.
176
177 =head2 Example 2: Asynchronous Backend
178
179 This example implements multiple count-downs in the child, using
180 L<AnyEvent> timers. While this is a bit silly (one could use timers in te
181 parent just as well), it illustrates the ability to use AnyEvent in the
182 child and the fact that responses can arrive in a different order then the
183 requests.
184
185 It also shows how to embed the actual child code into a C<__DATA__>
186 section, so it doesn't need any external files at all.
187
188 And when your parent process is often busy, and you have stricter timing
189 requirements, then running timers in a child process suddenly doesn't look
190 so silly anymore.
191
192 Without further ado, here is the code:
193
194 use AnyEvent;
195 use AnyEvent::Fork;
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 "ERROR: $_[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
246 The parent part (the one before the C<__DATA__> section) isn't very
247 different from the earlier examples. It sets async mode, preloads
248 the backend module (so the C<AnyEvent::Fork::RPC::event> function is
249 declared), uses a slightly different C<on_event> handler (which we use
250 simply for logging purposes) and then, instead of loading a module with
251 the actual worker code, it C<eval>'s the code from the data section in the
252 child process.
253
254 It then starts three countdowns, from 3 to 1 seconds downwards, destroys
255 the rpc object so the example finishes eventually, and then just waits for
256 the stuff to trickle in.
257
258 The worker code uses the event function to log some progress messages, but
259 mostly just creates a recurring one-second timer.
260
261 The timer callback increments a counter, logs a message, and eventually,
262 when the count has been reached, calls the finish callback.
263
264 On my system, this results in the following output. Since all timers fire
265 at roughly the same time, the actual order isn't guaranteed, but the order
266 shown 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
281 While the overall ordering isn't guaranteed, the async backend still
282 guarantees that events and responses are delivered to the parent process
283 in the exact same ordering as they were generated in the child process.
284
285 And unless your system is I<very> busy, it should clearly show that the
286 job started last will finish first, as it has the lowest count.
287
288 This concludes the async example. Since L<AnyEvent::Fork> does not
289 actually fork, you are free to use about any module in the child, not just
290 L<AnyEvent>, but also L<IO::AIO>, or L<Tk> for example.
291
292 =head2 Example 3: Asynchronous backend with Coro
293
294 With L<Coro> you can create a nice asynchronous backend implementation by
295 defining an rpc server function that creates a new Coro thread for every
296 request that calls a function "normally", i.e. the parameters from the
297 parent process are passed to it, and any return values are returned to the
298 parent process, e.g.:
299
300 package My::Arith;
301
302 sub add {
303 return $_[0] + $_[1];
304 }
305
306 sub mul {
307 return $_[0] * $_[1];
308 }
309
310 sub run {
311 my ($done, $func, @arg) = @_;
312
313 Coro::async_pool {
314 $done->($func->(@arg));
315 };
316 }
317
318 The C<run> function creates a new thread for every invocation, using the
319 first argument as function name, and calls the C<$done> callback on it's
320 return values. This makes it quite natural to define the C<add> and C<mul>
321 functions to add or multiply two numbers and return the result.
322
323 Since this is the asynchronous backend, it's quite possible to define RPC
324 function that do I/O or wait for external events - their execution will
325 overlap as needed.
326
327 The above could be used like this:
328
329 my $rpc = AnyEvent::Fork
330 ->new
331 ->require ("MyWorker")
332 ->AnyEvent::Fork::RPC::run ("My::Arith::run",
333 on_error => ..., on_event => ..., on_destroy => ...,
334 );
335
336 $rpc->(add => 1, 3, Coro::rouse_cb); say Coro::rouse_wait;
337 $rpc->(mul => 3, 2, Coro::rouse_cb); say Coro::rouse_wait;
338
339 The C<say>'s will print C<4> and C<6>.
340
341 =head1 PARENT PROCESS USAGE
342
343 This module exports nothing, and only implements a single function:
344
345 =over 4
346
347 =cut
348
349 package AnyEvent::Fork::RPC;
350
351 use common::sense;
352
353 use Errno ();
354 use Guard ();
355
356 use AnyEvent;
357
358 our $VERSION = 1.1;
359
360 =item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
361
362 The traditional way to call it. But it is way cooler to call it in the
363 following way:
364
365 =item my $rpc = $fork->AnyEvent::Fork::RPC::run ($function, [key => value...])
366
367 This C<run> function/method can be used in place of the
368 L<AnyEvent::Fork::run> method. Just like that method, it takes over
369 the L<AnyEvent::Fork> process, but instead of calling the specified
370 C<$function> directly, it runs a server that accepts RPC calls and handles
371 responses.
372
373 It returns a function reference that can be used to call the function in
374 the child process, handling serialisation and data transfers.
375
376 The following key/value pairs are allowed. It is recommended to have at
377 least an C<on_error> or C<on_event> handler set.
378
379 =over 4
380
381 =item on_error => $cb->($msg)
382
383 Called on (fatal) errors, with a descriptive (hopefully) message. If
384 this callback is not provided, but C<on_event> is, then the C<on_event>
385 callback is called with the first argument being the string C<error>,
386 followed by the error message.
387
388 If neither handler is provided, then the error is reported with loglevel
389 C<error> via C<AE::log>.
390
391 =item on_event => $cb->(...)
392
393 Called for every call to the C<AnyEvent::Fork::RPC::event> function in the
394 child, with the arguments of that function passed to the callback.
395
396 Also called on errors when no C<on_error> handler is provided.
397
398 =item on_destroy => $cb->()
399
400 Called when the C<$rpc> object has been destroyed and all requests have
401 been successfully handled. This is useful when you queue some requests and
402 want the child to go away after it has handled them. The problem is that
403 the parent must not exit either until all requests have been handled, and
404 this can be accomplished by waiting for this callback.
405
406 =item init => $function (default none)
407
408 When specified (by name), this function is called in the child as the very
409 first thing when taking over the process, with all the arguments normally
410 passed to the C<AnyEvent::Fork::run> function, except the communications
411 socket.
412
413 It can be used to do one-time things in the child such as storing passed
414 parameters or opening database connections.
415
416 It is called very early - before the serialisers are created or the
417 C<$function> name is resolved into a function reference, so it could be
418 used to load any modules that provide the serialiser or function. It can
419 not, however, create events.
420
421 =item async => $boolean (default: 0)
422
423 The default server used in the child does all I/O blockingly, and only
424 allows a single RPC call to execute concurrently.
425
426 Setting C<async> to a true value switches to another implementation that
427 uses L<AnyEvent> in the child and allows multiple concurrent RPC calls (it
428 does not support recursion in the event loop however, blocking condvar
429 calls will fail).
430
431 The actual API in the child is documented in the section that describes
432 the calling semantics of the returned C<$rpc> function.
433
434 If you want to pre-load the actual back-end modules to enable memory
435 sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for
436 synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
437
438 If you use a template process and want to fork both sync and async
439 children, then it is permissible to load both modules.
440
441 =item serialiser => $string (default: $AnyEvent::Fork::RPC::STRING_SERIALISER)
442
443 All arguments, result data and event data have to be serialised to be
444 transferred between the processes. For this, they have to be frozen and
445 thawed in both parent and child processes.
446
447 By default, only octet strings can be passed between the processes, which
448 is reasonably fast and efficient and requires no extra modules.
449
450 For more complicated use cases, you can provide your own freeze and thaw
451 functions, by specifying a string with perl source code. It's supposed to
452 return two code references when evaluated: the first receives a list of
453 perl values and must return an octet string. The second receives the octet
454 string and must return the original list of values.
455
456 If you need an external module for serialisation, then you can either
457 pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>
458 or C<require> statement into the serialiser string. Or both.
459
460 Here are some examples - some of them are also available as global
461 variables that make them easier to use.
462
463 =over 4
464
465 =item octet strings - C<$AnyEvent::Fork::RPC::STRING_SERIALISER>
466
467 This serialiser concatenates length-prefixes octet strings, and is the
468 default. That means you can only pass (and return) strings containing
469 character codes 0-255.
470
471 Implementation:
472
473 (
474 sub { pack "(w/a*)*", @_ },
475 sub { unpack "(w/a*)*", shift }
476 )
477
478 =item json - C<$AnyEvent::Fork::RPC::JSON_SERIALISER>
479
480 This serialiser creates JSON arrays - you have to make sure the L<JSON>
481 module is installed for this serialiser to work. It can be beneficial for
482 sharing when you preload the L<JSON> module in a template process.
483
484 L<JSON> (with L<JSON::XS> installed) is slower than the octet string
485 serialiser, but usually much faster than L<Storable>, unless big chunks of
486 binary data need to be transferred.
487
488 Implementation:
489
490 use JSON ();
491 (
492 sub { JSON::encode_json \@_ },
493 sub { @{ JSON::decode_json shift } }
494 )
495
496 =item storable - C<$AnyEvent::Fork::RPC::STORABLE_SERIALISER>
497
498 This serialiser uses L<Storable>, which means it has high chance of
499 serialising just about anything you throw at it, at the cost of having
500 very high overhead per operation. It also comes with perl. It should be
501 used when you need to serialise complex data structures.
502
503 Implementation:
504
505 use Storable ();
506 (
507 sub { Storable::freeze \@_ },
508 sub { @{ Storable::thaw shift } }
509 )
510
511 =item portable storable - C<$AnyEvent::Fork::RPC::NSTORABLE_SERIALISER>
512
513 This serialiser also uses L<Storable>, but uses it's "network" format
514 to serialise data, which makes it possible to talk to different
515 perl binaries (for example, when talking to a process created with
516 L<AnyEvent::Fork::Remote>).
517
518 Implementation:
519
520 use Storable ();
521 (
522 sub { Storable::nfreeze \@_ },
523 sub { @{ Storable::thaw shift } }
524 )
525
526 =back
527
528 =back
529
530 See the examples section earlier in this document for some actual
531 examples.
532
533 =cut
534
535 our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })';
536 our $JSON_SERIALISER = 'use JSON (); (sub { JSON::encode_json \@_ }, sub { @{ JSON::decode_json shift } })';
537 our $STORABLE_SERIALISER = 'use Storable (); (sub { Storable::freeze \@_ }, sub { @{ Storable::thaw shift } })';
538 our $NSTORABLE_SERIALISER = 'use Storable (); (sub { Storable::nfreeze \@_ }, sub { @{ Storable::thaw shift } })';
539
540 sub run {
541 my ($self, $function, %arg) = @_;
542
543 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER;
544 my $on_event = delete $arg{on_event};
545 my $on_error = delete $arg{on_error};
546 my $on_destroy = delete $arg{on_destroy};
547
548 # default for on_error is to on_event, if specified
549 $on_error ||= $on_event
550 ? sub { $on_event->(error => shift) }
551 : sub { AE::log die => "AnyEvent::Fork::RPC: uncaught error: $_[0]." };
552
553 # default for on_event is to raise an error
554 $on_event ||= sub { $on_error->("event received, but no on_event handler") };
555
556 my ($f, $t) = eval $serialiser; die $@ if $@;
557
558 my (@rcb, %rcb, $fh, $shutdown, $wbuf, $ww);
559 my ($rlen, $rbuf, $rw) = 512 - 16;
560
561 my $wcb = sub {
562 my $len = syswrite $fh, $wbuf;
563
564 unless (defined $len) {
565 if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
566 undef $rw; undef $ww; # it ends here
567 $on_error->("$!");
568 }
569 }
570
571 substr $wbuf, 0, $len, "";
572
573 unless (length $wbuf) {
574 undef $ww;
575 $shutdown and shutdown $fh, 1;
576 }
577 };
578
579 my $module = "AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync");
580
581 $self->require ($module)
582 ->send_arg ($function, $arg{init}, $serialiser)
583 ->run ("$module\::run", sub {
584 $fh = shift;
585
586 my ($id, $len);
587 $rw = AE::io $fh, 0, sub {
588 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
589 $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
590
591 if ($len) {
592 while (8 <= length $rbuf) {
593 ($id, $len) = unpack "NN", $rbuf;
594 8 + $len <= length $rbuf
595 or last;
596
597 my @r = $t->(substr $rbuf, 8, $len);
598 substr $rbuf, 0, 8 + $len, "";
599
600 if ($id) {
601 if (@rcb) {
602 (shift @rcb)->(@r);
603 } elsif (my $cb = delete $rcb{$id}) {
604 $cb->(@r);
605 } else {
606 undef $rw; undef $ww;
607 $on_error->("unexpected data from child");
608 }
609 } else {
610 $on_event->(@r);
611 }
612 }
613 } elsif (defined $len) {
614 undef $rw; undef $ww; # it ends here
615
616 if (@rcb || %rcb) {
617 $on_error->("unexpected eof");
618 } else {
619 $on_destroy->()
620 if $on_destroy;
621 }
622 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
623 undef $rw; undef $ww; # it ends here
624 $on_error->("read: $!");
625 }
626 };
627
628 $ww ||= AE::io $fh, 1, $wcb;
629 });
630
631 my $guard = Guard::guard {
632 $shutdown = 1;
633
634 shutdown $fh, 1 if $fh && !$ww;
635 };
636
637 my $id;
638
639 $arg{async}
640 ? sub {
641 $id = ($id == 0xffffffff ? 0 : $id) + 1;
642 $id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops
643
644 $rcb{$id} = pop;
645
646 $guard if 0; # keep it alive
647
648 $wbuf .= pack "NN/a*", $id, &$f;
649 $ww ||= $fh && AE::io $fh, 1, $wcb;
650 }
651 : sub {
652 push @rcb, pop;
653
654 $guard; # keep it alive
655
656 $wbuf .= pack "N/a*", &$f;
657 $ww ||= $fh && AE::io $fh, 1, $wcb;
658 }
659 }
660
661 =item $rpc->(..., $cb->(...))
662
663 The RPC object returned by C<AnyEvent::Fork::RPC::run> is actually a code
664 reference. There are two things you can do with it: call it, and let it go
665 out of scope (let it get destroyed).
666
667 If C<async> was false when C<$rpc> was created (the default), then, if you
668 call C<$rpc>, the C<$function> is invoked with all arguments passed to
669 C<$rpc> except the last one (the callback). When the function returns, the
670 callback will be invoked with all the return values.
671
672 If C<async> was true, then the C<$function> receives an additional
673 initial argument, the result callback. In this case, returning from
674 C<$function> does nothing - the function only counts as "done" when the
675 result callback is called, and any arguments passed to it are considered
676 the return values. This makes it possible to "return" from event handlers
677 or e.g. Coro threads.
678
679 The other thing that can be done with the RPC object is to destroy it. In
680 this case, the child process will execute all remaining RPC calls, report
681 their results, and then exit.
682
683 See the examples section earlier in this document for some actual
684 examples.
685
686 =back
687
688 =head1 CHILD PROCESS USAGE
689
690 The following function is not available in this module. They are only
691 available in the namespace of this module when the child is running,
692 without having to load any extra modules. They are part of the child-side
693 API of L<AnyEvent::Fork::RPC>.
694
695 =over 4
696
697 =item AnyEvent::Fork::RPC::event ...
698
699 Send an event to the parent. Events are a bit like RPC calls made by the
700 child process to the parent, except that there is no notion of return
701 values.
702
703 See the examples section earlier in this document for some actual
704 examples.
705
706 =back
707
708 =head1 ADVANCED TOPICS
709
710 =head2 Choosing a backend
711
712 So how do you decide which backend to use? Well, that's your problem to
713 solve, but here are some thoughts on the matter:
714
715 =over 4
716
717 =item Synchronous
718
719 The synchronous backend does not rely on any external modules (well,
720 except L<common::sense>, which works around a bug in how perl's warning
721 system works). This keeps the process very small, for example, on my
722 system, an empty perl interpreter uses 1492kB RSS, which becomes 2020kB
723 after C<use warnings; use strict> (for people who grew up with C64s around
724 them this is probably shocking every single time they see it). The worker
725 process in the first example in this document uses 1792kB.
726
727 Since the calls are done synchronously, slow jobs will keep newer jobs
728 from executing.
729
730 The synchronous backend also has no overhead due to running an event loop
731 - reading requests is therefore very efficient, while writing responses is
732 less so, as every response results in a write syscall.
733
734 If the parent process is busy and a bit slow reading responses, the child
735 waits instead of processing further requests. This also limits the amount
736 of memory needed for buffering, as never more than one response has to be
737 buffered.
738
739 The API in the child is simple - you just have to define a function that
740 does something and returns something.
741
742 It's hard to use modules or code that relies on an event loop, as the
743 child cannot execute anything while it waits for more input.
744
745 =item Asynchronous
746
747 The asynchronous backend relies on L<AnyEvent>, which tries to be small,
748 but still comes at a price: On my system, the worker from example 1a uses
749 3420kB RSS (for L<AnyEvent>, which loads L<EV>, which needs L<XSLoader>
750 which in turn loads a lot of other modules such as L<warnings>, L<strict>,
751 L<vars>, L<Exporter>...).
752
753 It batches requests and responses reasonably efficiently, doing only as
754 few reads and writes as needed, but needs to poll for events via the event
755 loop.
756
757 Responses are queued when the parent process is busy. This means the child
758 can continue to execute any queued requests. It also means that a child
759 might queue a lot of responses in memory when it generates them and the
760 parent process is slow accepting them.
761
762 The API is not a straightforward RPC pattern - you have to call a
763 "done" callback to pass return values and signal completion. Also, more
764 importantly, the API starts jobs as fast as possible - when 1000 jobs
765 are queued and the jobs are slow, they will all run concurrently. The
766 child must implement some queueing/limiting mechanism if this causes
767 problems. Alternatively, the parent could limit the amount of rpc calls
768 that are outstanding.
769
770 Blocking use of condvars is not supported.
771
772 Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is
773 easy.
774
775 =back
776
777 =head2 Passing file descriptors
778
779 Unlike L<AnyEvent::Fork>, this module has no in-built file handle or file
780 descriptor passing abilities.
781
782 The reason is that passing file descriptors is extraordinary tricky
783 business, and conflicts with efficient batching of messages.
784
785 There still is a method you can use: Create a
786 C<AnyEvent::Util::portable_socketpair> and C<send_fh> one half of it to
787 the process before you pass control to C<AnyEvent::Fork::RPC::run>.
788
789 Whenever you want to pass a file descriptor, send an rpc request to the
790 child process (so it expects the descriptor), then send it over the other
791 half of the socketpair. The child should fetch the descriptor from the
792 half it has passed earlier.
793
794 Here is some (untested) pseudocode to that effect:
795
796 use AnyEvent::Util;
797 use AnyEvent::Fork;
798 use AnyEvent::Fork::RPC;
799 use IO::FDPass;
800
801 my ($s1, $s2) = AnyEvent::Util::portable_socketpair;
802
803 my $rpc = AnyEvent::Fork
804 ->new
805 ->send_fh ($s2)
806 ->require ("MyWorker")
807 ->AnyEvent::Fork::RPC::run ("MyWorker::run"
808 init => "MyWorker::init",
809 );
810
811 undef $s2; # no need to keep it around
812
813 # pass an fd
814 $rpc->("i'll send some fd now, please expect it!", my $cv = AE::cv);
815
816 IO::FDPass fileno $s1, fileno $handle_to_pass;
817
818 $cv->recv;
819
820 The MyWorker module could look like this:
821
822 package MyWorker;
823
824 use IO::FDPass;
825
826 my $s2;
827
828 sub init {
829 $s2 = $_[0];
830 }
831
832 sub run {
833 if ($_[0] eq "i'll send some fd now, please expect it!") {
834 my $fd = IO::FDPass::recv fileno $s2;
835 ...
836 }
837 }
838
839 Of course, this might be blocking if you pass a lot of file descriptors,
840 so you might want to look into L<AnyEvent::FDpasser> which can handle the
841 gory details.
842
843 =head1 EXCEPTIONS
844
845 There are no provisions whatsoever for catching exceptions at this time -
846 in the child, exeptions might kill the process, causing calls to be lost
847 and the parent encountering a fatal error. In the parent, exceptions in
848 the result callback will not be caught and cause undefined behaviour.
849
850 =head1 SEE ALSO
851
852 L<AnyEvent::Fork>, to create the processes in the first place.
853
854 L<AnyEvent::Fork::Remote>, likewise, but helpful for remote processes.
855
856 L<AnyEvent::Fork::Pool>, to manage whole pools of processes.
857
858 =head1 AUTHOR AND CONTACT INFORMATION
859
860 Marc Lehmann <schmorp@schmorp.de>
861 http://software.schmorp.de/pkg/AnyEvent-Fork-RPC
862
863 =cut
864
865 1
866