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, 2 months ago) by root
Branch: MAIN
Changes since 1.7: +8 -0 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::RPC;
8 # use AnyEvent::Fork is not needed
9
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 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 =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 on_error => sub { warn "FATAL: $_[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 =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 use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
159
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 =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 this can be accomplished by waiting for this callback.
207
208 =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 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 =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 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 If you use a template process and want to fork both sync and async
239 children, then it is permissible to load both modules.
240
241 =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 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 =back
261
262 See the examples section earlier in this document for some actual examples.
263
264 =cut
265
266 our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })';
267
268 sub run {
269 my ($self, $function, %arg) = @_;
270
271 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER;
272 my $on_event = delete $arg{on_event};
273 my $on_error = delete $arg{on_error};
274 my $on_destroy = delete $arg{on_destroy};
275
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 my (@rcb, $fh, $shutdown, $wbuf, $ww, $rw);
287 my ($rlen, $rbuf) = 512 - 16;
288
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 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
315 my $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
316
317 if ($len) {
318 while (4 <= length $rbuf) {
319 $len = unpack "L", $rbuf;
320 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 }
334 }
335 } elsif (defined $len) {
336 undef $rw; undef $ww; # it ends here
337
338 if (@rcb) {
339 $on_error->("unexpected eof");
340 } else {
341 $on_destroy->();
342 }
343 } 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 =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 See the examples section earlier in this document for some actual
390 examples.
391
392 =back
393
394 =head1 CHILD PROCESS USAGE
395
396 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
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 See the examples section earlier in this document for some actual
410 examples.
411
412 =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