ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-Fork-RPC/RPC.pm
(Generate patch)

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents):
Revision 1.9 by root, Wed Apr 17 21:48:35 2013 UTC vs.
Revision 1.15 by root, Thu Apr 18 13:15:23 2013 UTC

… …
39Loading this module also always loads L<AnyEvent::Fork>, so you can make a 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. 40separate C<use AnyEvent::Fork> if you wish, but you don't have to.
41 41
42=head1 EXAMPLES 42=head1 EXAMPLES
43 43
44=head2 Synchronous Backend 44=head2 Example 1: Synchronous Backend
45 45
46Here 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>
47and 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
48number of requests it has processed every three requests, which is clearly 48number of requests it has processed every three requests, which is clearly
49silly, but illustrates the use of events. 49silly, but illustrates the use of events.
50 50
51First the parent process: 51First the parent process:
52 52
53 use AnyEvent; 53 use AnyEvent;
54 use AnyEvent::Fork;
55 use AnyEvent::Fork::RPC; 54 use AnyEvent::Fork::RPC;
56 55
57 my $done = AE::cv; 56 my $done = AE::cv;
58 57
59 my $rpc = AnyEvent::Fork 58 my $rpc = AnyEvent::Fork
… …
137 136
138And 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
139asynchronously 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
140than this example, namely L<IO::AIO>. 139than this example, namely L<IO::AIO>.
141 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
142=head1 PARENT PROCESS USAGE 290=head1 PARENT PROCESS USAGE
143 291
144This module exports nothing, and only implements a single function: 292This module exports nothing, and only implements a single function:
145 293
146=over 4 294=over 4
… …
224 372
225The default server used in the child does all I/O blockingly, and only 373The default server used in the child does all I/O blockingly, and only
226allows a single RPC call to execute concurrently. 374allows a single RPC call to execute concurrently.
227 375
228Setting C<async> to a true value switches to another implementation that 376Setting C<async> to a true value switches to another implementation that
229uses L<AnyEvent> in the child and allows multiple concurrent RPC calls. 377uses L<AnyEvent> in the child and allows multiple concurrent RPC calls (it
378does not support recursion in the event loop however, blocking condvar
379calls will fail).
230 380
231The actual API in the child is documented in the section that describes 381The actual API in the child is documented in the section that describes
232the calling semantics of the returned C<$rpc> function. 382the calling semantics of the returned C<$rpc> function.
233 383
234If you want to pre-load the actual back-end modules to enable memory 384If you want to pre-load the actual back-end modules to enable memory
… …
236synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode. 386synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
237 387
238If you use a template process and want to fork both sync and async 388If you use a template process and want to fork both sync and async
239children, then it is permissible to load both modules. 389children, then it is permissible to load both modules.
240 390
241=item serialiser => $string (default: '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })') 391=item serialiser => $string (default: $AnyEvent::Fork::RPC::STRING_SERIALISER)
242 392
243All arguments, result data and event data have to be serialised to be 393All arguments, result data and event data have to be serialised to be
244transferred between the processes. For this, they have to be frozen and 394transferred between the processes. For this, they have to be frozen and
245thawed in both parent and child processes. 395thawed in both parent and child processes.
246 396
247By default, only octet strings can be passed between the processes, which 397By default, only octet strings can be passed between the processes, which
248is reasonably fast and efficient. 398is reasonably fast and efficient and requires no extra modules.
249 399
250For more complicated use cases, you can provide your own freeze and thaw 400For 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 401functions, by specifying a string with perl source code. It's supposed to
252return two code references when evaluated: the first receives a list of 402return two code references when evaluated: the first receives a list of
253perl values and must return an octet string. The second receives the octet 403perl values and must return an octet string. The second receives the octet
… …
255 405
256If you need an external module for serialisation, then you can either 406If 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> 407pre-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. 408or C<require> statement into the serialiser string. Or both.
259 409
410Here are some examples - some of them are also available as global
411variables that make them easier to use.
412
413=over 4
414
415=item octet strings - C<$AnyEvent::Fork::RPC::STRING_SERIALISER>
416
417This serialiser concatenates length-prefixes octet strings, and is the
418default.
419
420Implementation:
421
422 (
423 sub { pack "(w/a*)*", @_ },
424 sub { unpack "(w/a*)*", shift }
425 )
426
427=item json - C<$AnyEvent::Fork::RPC::JSON_SERIALISER>
428
429This serialiser creates JSON arrays - you have to make sure the L<JSON>
430module is installed for this serialiser to work. It can be beneficial for
431sharing when you preload the L<JSON> module in a template process.
432
433L<JSON> (with L<JSON::XS> installed) is slower than the octet string
434serialiser, but usually much faster than L<Storable>, unless big chunks of
435binary data need to be transferred.
436
437Implementation:
438
439 use JSON ();
440 (
441 sub { JSON::encode_json \@_ },
442 sub { @{ JSON::decode_json shift } }
443 )
444
445=item storable - C<$AnyEvent::Fork::RPC::STORABLE_SERIALISER>
446
447This serialiser uses L<Storable>, which means it has high chance of
448serialising just about anything you throw at it, at the cost of having
449very high overhead per operation. It also comes with perl.
450
451Implementation:
452
453 use Storable ();
454 (
455 sub { Storable::freeze \@_ },
456 sub { @{ Storable::thaw shift } }
457 )
458
459=back
460
260=back 461=back
261 462
262See the examples section earlier in this document for some actual 463See the examples section earlier in this document for some actual
263examples. 464examples.
264 465
265=cut 466=cut
266 467
267our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })'; 468our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })';
469our $JSON_SERIALISER = 'use JSON (); (sub { JSON::encode_json \@_ }, sub { @{ JSON::decode_json shift } })';
470our $STORABLE_SERIALISER = 'use Storable (); (sub { Storable::freeze \@_ }, sub { @{ Storable::thaw shift } })';
268 471
269sub run { 472sub run {
270 my ($self, $function, %arg) = @_; 473 my ($self, $function, %arg) = @_;
271 474
272 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER; 475 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER;
… …
341 } 544 }
342 } elsif (defined $len) { 545 } elsif (defined $len) {
343 undef $rw; undef $ww; # it ends here 546 undef $rw; undef $ww; # it ends here
344 547
345 if (@rcb || %rcb) { 548 if (@rcb || %rcb) {
346 use Data::Dump;ddx[\@rcb,\%rcb];#d#
347 $on_error->("unexpected eof"); 549 $on_error->("unexpected eof");
348 } else { 550 } else {
349 $on_destroy->(); 551 $on_destroy->();
350 } 552 }
351 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) { 553 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
… …
431See the examples section earlier in this document for some actual 633See the examples section earlier in this document for some actual
432examples. 634examples.
433 635
434=back 636=back
435 637
638=head1 ADVANCED TOPICS
639
640=head2 Choosing a backend
641
642So how do you decide which backend to use? Well, that's your problem to
643solve, but here are some thoughts on the matter:
644
645=over 4
646
647=item Synchronous
648
649The synchronous backend does not rely on any external modules (well,
650except L<common::sense>, which works around a bug in how perl's warning
651system works). This keeps the process very small, for example, on my
652system, an empty perl interpreter uses 1492kB RSS, which becomes 2020kB
653after C<use warnings; use strict> (for people who grew up with C64s around
654them this is probably shocking every single time they see it). The worker
655process in the first example in this document uses 1792kB.
656
657Since the calls are done synchronously, slow jobs will keep newer jobs
658from executing.
659
660The synchronous backend also has no overhead due to running an event loop
661- reading requests is therefore very efficient, while writing responses is
662less so, as every response results in a write syscall.
663
664If the parent process is busy and a bit slow reading responses, the child
665waits instead of processing further requests. This also limits the amount
666of memory needed for buffering, as never more than one response has to be
667buffered.
668
669The API in the child is simple - you just have to define a function that
670does something and returns something.
671
672It's hard to use modules or code that relies on an event loop, as the
673child cannot execute anything while it waits for more input.
674
675=item Asynchronous
676
677The asynchronous backend relies on L<AnyEvent>, which tries to be small,
678but still comes at a price: On my system, the worker from example 1a uses
6793420kB RSS (for L<AnyEvent>, which loads L<EV>, which needs L<XSLoader>
680which in turn loads a lot of other modules such as L<warnings>, L<strict>,
681L<vars>, L<Exporter>...).
682
683It batches requests and responses reasonably efficiently, doing only as
684few reads and writes as needed, but needs to poll for events via the event
685loop.
686
687Responses are queued when the parent process is busy. This means the child
688can continue to execute any queued requests. It also means that a child
689might queue a lot of responses in memory when it generates them and the
690parent process is slow accepting them.
691
692The API is not a straightforward RPC pattern - you have to call a
693"done" callback to pass return values and signal completion. Also, more
694importantly, the API starts jobs as fast as possible - when 1000 jobs
695are queued and the jobs are slow, they will all run concurrently. The
696child must implement some queueing/limiting mechanism if this causes
697problems. Alternatively, the parent could limit the amount of rpc calls
698that are outstanding.
699
700Blocking use of condvars is not supported.
701
702Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is
703easy.
704
705=back
706
707=head2 Passing file descriptors
708
709Unlike L<AnyEvent::Fork>, this module has no in-built file handle or file
710descriptor passing abilities.
711
712The reason is that passing file descriptors is extraordinary tricky
713business, and conflicts with efficient batching of messages.
714
715There still is a method you can use: Create a
716C<AnyEvent::Util::portable_socketpair> and C<send_fh> one half of it to
717the process before you pass control to C<AnyEvent::Fork::RPC::run>.
718
719Whenever you want to pass a file descriptor, send an rpc request to the
720child process (so it expects the descriptor), then send it over the other
721half of the socketpair. The child should fetch the descriptor from the
722half it has passed earlier.
723
724Here is some (untested) pseudocode to that effect:
725
726 use AnyEvent::Util;
727 use AnyEvent::Fork::RPC;
728 use IO::FDPass;
729
730 my ($s1, $s2) = AnyEvent::Util::portable_socketpair;
731
732 my $rpc = AnyEvent::Fork
733 ->new
734 ->send_fh ($s2)
735 ->require ("MyWorker")
736 ->AnyEvent::Fork::RPC::run ("MyWorker::run"
737 init => "MyWorker::init",
738 );
739
740 undef $s2; # no need to keep it around
741
742 # pass an fd
743 $rpc->("i'll send some fd now, please expect it!", my $cv = AE::cv);
744
745 IO::FDPass fileno $s1, fileno $handle_to_pass;
746
747 $cv->recv;
748
749The MyWorker module could look like this:
750
751 package MyWorker;
752
753 use IO::FDPass;
754
755 my $s2;
756
757 sub init {
758 $s2 = $_[0];
759 }
760
761 sub run {
762 if ($_[0] eq "i'll send some fd now, please expect it!") {
763 my $fd = IO::FDPass::recv fileno $s2;
764 ...
765 }
766 }
767
768Of course, this might be blocking if you pass a lot of file descriptors,
769so you might want to look into L<AnyEvent::FDpasser> which can handle the
770gory details.
771
436=head1 SEE ALSO 772=head1 SEE ALSO
437 773
438L<AnyEvent::Fork> (to create the processes in the first place), 774L<AnyEvent::Fork> (to create the processes in the first place),
439L<AnyEvent::Fork::Pool> (to manage whole pools of processes). 775L<AnyEvent::Fork::Pool> (to manage whole pools of processes).
440 776

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines