ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent-Fork-RPC/RPC.pm
Revision: 1.8
Committed: Wed Apr 17 20:24:36 2013 UTC (11 years, 1 month ago) by root
Branch: MAIN
Changes since 1.7: +8 -0 lines
Log Message:
*** empty log message ***

File Contents

# User Rev Content
1 root 1.1 =head1 NAME
2    
3     AnyEvent::Fork::RPC - simple RPC extension for AnyEvent::Fork
4    
5     =head1 SYNOPSIS
6    
7     use AnyEvent::Fork::RPC;
8 root 1.7 # use AnyEvent::Fork is not needed
9 root 1.1
10     my $rpc = AnyEvent::Fork
11     ->new
12     ->require ("MyModule")
13     ->AnyEvent::Fork::RPC::run (
14     "MyModule::server",
15     );
16    
17     my $cv = AE::cv;
18    
19     $rpc->(1, 2, 3, sub {
20     print "MyModule::server returned @_\n";
21     $cv->send;
22     });
23    
24     $cv->recv;
25    
26     =head1 DESCRIPTION
27    
28     This module implements a simple RPC protocol and backend for processes
29     created via L<AnyEvent::Fork>, allowing you to call a function in the
30     child process and receive its return values (up to 4GB serialised).
31    
32     It implements two different backends: a synchronous one that works like a
33     normal function call, and an asynchronous one that can run multiple jobs
34     concurrently in the child, using AnyEvent.
35    
36     It also implements an asynchronous event mechanism from the child to the
37     parent, that could be used for progress indications or other information.
38    
39 root 1.7 Loading this module also always loads L<AnyEvent::Fork>, so you can make a
40     separate C<use AnyEvent::Fork> if you wish, but you don't have to.
41    
42 root 1.4 =head1 EXAMPLES
43    
44     =head2 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 root 1.5 on_error => sub { warn "FATAL: $_[0]"; exit 1 },
64 root 1.4 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 root 1.6 On my system, running the first code fragment with the given
112 root 1.4 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 root 1.1 =head1 PARENT PROCESS USAGE
143    
144     This module exports nothing, and only implements a single function:
145    
146     =over 4
147    
148     =cut
149    
150     package AnyEvent::Fork::RPC;
151    
152     use common::sense;
153    
154     use Errno ();
155     use Guard ();
156    
157     use AnyEvent;
158 root 1.7 use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
159 root 1.1
160     our $VERSION = 0.1;
161    
162     =item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
163    
164     The traditional way to call it. But it is way cooler to call it in the
165     following way:
166    
167     =item my $rpc = $fork->AnyEvent::Fork::RPC::run ($function, [key => value...])
168    
169     This C<run> function/method can be used in place of the
170     L<AnyEvent::Fork::run> method. Just like that method, it takes over
171     the L<AnyEvent::Fork> process, but instead of calling the specified
172     C<$function> directly, it runs a server that accepts RPC calls and handles
173     responses.
174    
175     It returns a function reference that can be used to call the function in
176     the child process, handling serialisation and data transfers.
177    
178     The following key/value pairs are allowed. It is recommended to have at
179     least an C<on_error> or C<on_event> handler set.
180    
181     =over 4
182    
183     =item on_error => $cb->($msg)
184    
185     Called on (fatal) errors, with a descriptive (hopefully) message. If
186     this callback is not provided, but C<on_event> is, then the C<on_event>
187     callback is called with the first argument being the string C<error>,
188     followed by the error message.
189    
190     If neither handler is provided it prints the error to STDERR and will
191     start failing badly.
192    
193     =item on_event => $cb->(...)
194    
195     Called for every call to the C<AnyEvent::Fork::RPC::event> function in the
196     child, with the arguments of that function passed to the callback.
197    
198     Also called on errors when no C<on_error> handler is provided.
199    
200 root 1.4 =item on_destroy => $cb->()
201    
202     Called when the C<$rpc> object has been destroyed and all requests have
203     been successfully handled. This is useful when you queue some requests and
204     want the child to go away after it has handled them. The problem is that
205     the parent must not exit either until all requests have been handled, and
206 root 1.6 this can be accomplished by waiting for this callback.
207 root 1.4
208 root 1.1 =item init => $function (default none)
209    
210     When specified (by name), this function is called in the child as the very
211     first thing when taking over the process, with all the arguments normally
212     passed to the C<AnyEvent::Fork::run> function, except the communications
213     socket.
214    
215     It can be used to do one-time things in the child such as storing passed
216     parameters or opening database connections.
217    
218 root 1.4 It is called very early - before the serialisers are created or the
219     C<$function> name is resolved into a function reference, so it could be
220     used to load any modules that provide the serialiser or function. It can
221     not, however, create events.
222    
223 root 1.1 =item async => $boolean (default: 0)
224    
225     The default server used in the child does all I/O blockingly, and only
226     allows a single RPC call to execute concurrently.
227    
228     Setting C<async> to a true value switches to another implementation that
229     uses L<AnyEvent> in the child and allows multiple concurrent RPC calls.
230    
231     The actual API in the child is documented in the section that describes
232     the calling semantics of the returned C<$rpc> function.
233    
234 root 1.2 If you want to pre-load the actual back-end modules to enable memory
235     sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for
236     synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
237    
238 root 1.4 If you use a template process and want to fork both sync and async
239 root 1.6 children, then it is permissible to load both modules.
240 root 1.4
241 root 1.1 =item serialiser => $string (default: '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })')
242    
243     All arguments, result data and event data have to be serialised to be
244     transferred between the processes. For this, they have to be frozen and
245     thawed in both parent and child processes.
246    
247     By default, only octet strings can be passed between the processes, which
248     is reasonably fast and efficient.
249    
250     For more complicated use cases, you can provide your own freeze and thaw
251     functions, by specifying a string with perl source code. It's supposed to
252     return two code references when evaluated: the first receives a list of
253     perl values and must return an octet string. The second receives the octet
254     string and must return the original list of values.
255    
256 root 1.2 If you need an external module for serialisation, then you can either
257     pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>
258     or C<require> statement into the serialiser string. Or both.
259    
260 root 1.1 =back
261    
262 root 1.8 See the examples section earlier in this document for some actual examples.
263    
264 root 1.1 =cut
265    
266 root 1.2 our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })';
267    
268 root 1.1 sub run {
269     my ($self, $function, %arg) = @_;
270    
271 root 1.2 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER;
272 root 1.1 my $on_event = delete $arg{on_event};
273     my $on_error = delete $arg{on_error};
274 root 1.4 my $on_destroy = delete $arg{on_destroy};
275 root 1.1
276     # default for on_error is to on_event, if specified
277     $on_error ||= $on_event
278     ? sub { $on_event->(error => shift) }
279     : sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" };
280    
281     # default for on_event is to raise an error
282     $on_event ||= sub { $on_error->("event received, but no on_event handler") };
283    
284     my ($f, $t) = eval $serialiser; die $@ if $@;
285    
286 root 1.4 my (@rcb, $fh, $shutdown, $wbuf, $ww, $rw);
287     my ($rlen, $rbuf) = 512 - 16;
288 root 1.1
289     my $wcb = sub {
290     my $len = syswrite $fh, $wbuf;
291    
292     if (!defined $len) {
293     if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
294     undef $rw; undef $ww; # it ends here
295     $on_error->("$!");
296     }
297     }
298    
299     substr $wbuf, 0, $len, "";
300    
301     unless (length $wbuf) {
302     undef $ww;
303     $shutdown and shutdown $fh, 1;
304     }
305     };
306    
307     my $module = "AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync");
308    
309     $self->require ($module)
310     ->send_arg ($function, $arg{init}, $serialiser)
311     ->run ("$module\::run", sub {
312     $fh = shift;
313     $rw = AE::io $fh, 0, sub {
314 root 1.4 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
315     my $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
316 root 1.1
317     if ($len) {
318 root 1.7 while (4 <= length $rbuf) {
319 root 1.1 $len = unpack "L", $rbuf;
320 root 1.2 4 + $len <= length $rbuf
321     or last;
322    
323     my @r = $t->(substr $rbuf, 4, $len);
324     substr $rbuf, 0, $len + 4, "";
325    
326     if (pop @r) {
327     $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 root 1.1 }
334     }
335     } elsif (defined $len) {
336     undef $rw; undef $ww; # it ends here
337 root 1.4
338     if (@rcb) {
339     $on_error->("unexpected eof");
340     } else {
341     $on_destroy->();
342     }
343 root 1.1 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
344     undef $rw; undef $ww; # it ends here
345     $on_error->("read: $!");
346     }
347     };
348    
349     $ww ||= AE::io $fh, 1, $wcb;
350     });
351    
352     my $guard = Guard::guard {
353     $shutdown = 1;
354     $ww ||= $fh && AE::io $fh, 1, $wcb;
355     };
356    
357     sub {
358     push @rcb, pop;
359    
360     $guard; # keep it alive
361    
362     $wbuf .= pack "L/a*", &$f;
363     $ww ||= $fh && AE::io $fh, 1, $wcb;
364     }
365     }
366    
367 root 1.4 =item $rpc->(..., $cb->(...))
368    
369     The RPC object returned by C<AnyEvent::Fork::RPC::run> is actually a code
370     reference. There are two things you can do with it: call it, and let it go
371     out of scope (let it get destroyed).
372    
373     If C<async> was false when C<$rpc> was created (the default), then, if you
374     call C<$rpc>, the C<$function> is invoked with all arguments passed to
375     C<$rpc> except the last one (the callback). When the function returns, the
376     callback will be invoked with all the return values.
377    
378     If C<async> was true, then the C<$function> receives an additional
379     initial argument, the result callback. In this case, returning from
380     C<$function> does nothing - the function only counts as "done" when the
381     result callback is called, and any arguments passed to it are considered
382     the return values. This makes it possible to "return" from event handlers
383     or e.g. Coro threads.
384    
385     The other thing that can be done with the RPC object is to destroy it. In
386     this case, the child process will execute all remaining RPC calls, report
387     their results, and then exit.
388    
389 root 1.8 See the examples section earlier in this document for some actual
390     examples.
391    
392 root 1.1 =back
393    
394     =head1 CHILD PROCESS USAGE
395    
396 root 1.4 The following function is not available in this module. They are only
397     available in the namespace of this module when the child is running,
398     without having to load any extra modules. They are part of the child-side
399     API of L<AnyEvent::Fork::RPC>.
400 root 1.1
401     =over 4
402    
403     =item AnyEvent::Fork::RPC::event ...
404    
405     Send an event to the parent. Events are a bit like RPC calls made by the
406     child process to the parent, except that there is no notion of return
407     values.
408    
409 root 1.8 See the examples section earlier in this document for some actual
410     examples.
411    
412 root 1.1 =back
413    
414     =head1 SEE ALSO
415    
416     L<AnyEvent::Fork> (to create the processes in the first place),
417     L<AnyEvent::Fork::Pool> (to manage whole pools of processes).
418    
419     =head1 AUTHOR AND CONTACT INFORMATION
420    
421     Marc Lehmann <schmorp@schmorp.de>
422     http://software.schmorp.de/pkg/AnyEvent-Fork-RPC
423    
424     =cut
425    
426     1
427