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.1 by root, Wed Apr 17 15:55:59 2013 UTC vs.
Revision 1.8 by root, Wed Apr 17 20:24:36 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
42=head1 EXAMPLES
43
44=head2 Synchronous Backend
45
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
48number of requests it has processed every three requests, which is clearly
49silly, but illustrates the use of events.
50
51First 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
79The parent creates the process, queues a few rmdir's. It then forgets
80about the C<$rpc> object, so that the child exits after it has handled the
81requests, and then it waits till the requests have been handled.
82
83The 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
104The C<run> function first sends a "progress" event every three calls, and
105then executes C<rmdir> or C<unlink>, depending on the first parameter (or
106dies with a fatal error - obviously, you must never let this happen :).
107
108Eventually it returns the status value true if the command was successful,
109or the status value 0 and the stringified error message.
110
111On my system, running the first code fragment with the given
112F<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
123Obviously, none of the directories I am trying to delete even exist. Also,
124the events and responses are processed in exactly the same order as
125they were created in the child, which is true for both synchronous and
126asynchronous backends.
127
128Note that the parentheses in the call to C<AnyEvent::Fork::RPC::event> are
129not optional. That is because the function isn't defined when the code is
130compiled. You can make sure it is visible by pre-loading the correct
131backend module in the call to C<require>:
132
133 ->require ("AnyEvent::Fork::RPC::Sync", "MyWorker")
134
135Since the backend module declares the C<event> function, loading it first
136ensures that perl will correctly interpret calls to it.
137
138And 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
140than this example, namely L<IO::AIO>.
141
39=head1 PARENT PROCESS USAGE 142=head1 PARENT PROCESS USAGE
40 143
41This module exports nothing, and only implements a single function: 144This module exports nothing, and only implements a single function:
42 145
43=over 4 146=over 4
… …
50 153
51use Errno (); 154use Errno ();
52use Guard (); 155use Guard ();
53 156
54use AnyEvent; 157use AnyEvent;
55#use AnyEvent::Fork; 158use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
56 159
57our $VERSION = 0.1; 160our $VERSION = 0.1;
58 161
59=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...] 162=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
60 163
… …
92Called for every call to the C<AnyEvent::Fork::RPC::event> function in the 195Called for every call to the C<AnyEvent::Fork::RPC::event> function in the
93child, with the arguments of that function passed to the callback. 196child, with the arguments of that function passed to the callback.
94 197
95Also called on errors when no C<on_error> handler is provided. 198Also called on errors when no C<on_error> handler is provided.
96 199
200=item on_destroy => $cb->()
201
202Called when the C<$rpc> object has been destroyed and all requests have
203been successfully handled. This is useful when you queue some requests and
204want the child to go away after it has handled them. The problem is that
205the parent must not exit either until all requests have been handled, and
206this can be accomplished by waiting for this callback.
207
97=item init => $function (default none) 208=item init => $function (default none)
98 209
99When specified (by name), this function is called in the child as the very 210When specified (by name), this function is called in the child as the very
100first thing when taking over the process, with all the arguments normally 211first thing when taking over the process, with all the arguments normally
101passed to the C<AnyEvent::Fork::run> function, except the communications 212passed to the C<AnyEvent::Fork::run> function, except the communications
102socket. 213socket.
103 214
104It can be used to do one-time things in the child such as storing passed 215It can be used to do one-time things in the child such as storing passed
105parameters or opening database connections. 216parameters or opening database connections.
106 217
218It is called very early - before the serialisers are created or the
219C<$function> name is resolved into a function reference, so it could be
220used to load any modules that provide the serialiser or function. It can
221not, however, create events.
222
107=item async => $boolean (default: 0) 223=item async => $boolean (default: 0)
108 224
109The default server used in the child does all I/O blockingly, and only 225The default server used in the child does all I/O blockingly, and only
110allows a single RPC call to execute concurrently. 226allows a single RPC call to execute concurrently.
111 227
112Setting C<async> to a true value switches to another implementation that 228Setting C<async> to a true value switches to another implementation that
113uses L<AnyEvent> in the child and allows multiple concurrent RPC calls. 229uses L<AnyEvent> in the child and allows multiple concurrent RPC calls.
114 230
115The actual API in the child is documented in the section that describes 231The actual API in the child is documented in the section that describes
116the calling semantics of the returned C<$rpc> function. 232the calling semantics of the returned C<$rpc> function.
233
234If you want to pre-load the actual back-end modules to enable memory
235sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for
236synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
237
238If you use a template process and want to fork both sync and async
239children, then it is permissible to load both modules.
117 240
118=item serialiser => $string (default: '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })') 241=item serialiser => $string (default: '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })')
119 242
120All arguments, result data and event data have to be serialised to be 243All arguments, result data and event data have to be serialised to be
121transferred between the processes. For this, they have to be frozen and 244transferred between the processes. For this, they have to be frozen and
… …
128functions, by specifying a string with perl source code. It's supposed to 251functions, by specifying a string with perl source code. It's supposed to
129return two code references when evaluated: the first receives a list of 252return two code references when evaluated: the first receives a list of
130perl values and must return an octet string. The second receives the octet 253perl values and must return an octet string. The second receives the octet
131string and must return the original list of values. 254string and must return the original list of values.
132 255
256If 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>
258or C<require> statement into the serialiser string. Or both.
259
133=back 260=back
134 261
262See the examples section earlier in this document for some actual examples.
263
135=cut 264=cut
136 265
137our $SERIALISE_STRINGS = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })'; 266our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })';
138 267
139sub run { 268sub run {
140 my ($self, $function, %arg) = @_; 269 my ($self, $function, %arg) = @_;
141 270
142 my $serialiser = delete $arg{serialiser} || $SERIALISE_STRINGS; 271 my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER;
143 my $on_event = delete $arg{on_event}; 272 my $on_event = delete $arg{on_event};
144 my $on_error = delete $arg{on_error}; 273 my $on_error = delete $arg{on_error};
274 my $on_destroy = delete $arg{on_destroy};
145 275
146 # default for on_error is to on_event, if specified 276 # default for on_error is to on_event, if specified
147 $on_error ||= $on_event 277 $on_error ||= $on_event
148 ? sub { $on_event->(error => shift) } 278 ? sub { $on_event->(error => shift) }
149 : sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" }; 279 : sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" };
… …
151 # default for on_event is to raise an error 281 # default for on_event is to raise an error
152 $on_event ||= sub { $on_error->("event received, but no on_event handler") }; 282 $on_event ||= sub { $on_error->("event received, but no on_event handler") };
153 283
154 my ($f, $t) = eval $serialiser; die $@ if $@; 284 my ($f, $t) = eval $serialiser; die $@ if $@;
155 285
156 my (@rcb, $fh, $shutdown, $wbuf, $ww, $rbuf, $rw); 286 my (@rcb, $fh, $shutdown, $wbuf, $ww, $rw);
287 my ($rlen, $rbuf) = 512 - 16;
157 288
158 my $wcb = sub { 289 my $wcb = sub {
159 my $len = syswrite $fh, $wbuf; 290 my $len = syswrite $fh, $wbuf;
160 291
161 if (!defined $len) { 292 if (!defined $len) {
… …
178 $self->require ($module) 309 $self->require ($module)
179 ->send_arg ($function, $arg{init}, $serialiser) 310 ->send_arg ($function, $arg{init}, $serialiser)
180 ->run ("$module\::run", sub { 311 ->run ("$module\::run", sub {
181 $fh = shift; 312 $fh = shift;
182 $rw = AE::io $fh, 0, sub { 313 $rw = AE::io $fh, 0, sub {
314 $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
183 my $len = sysread $fh, $rbuf, 512 + length $rbuf, length $rbuf; 315 my $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
184 316
185 if ($len) { 317 if ($len) {
186 while (5 <= length $rbuf) { 318 while (4 <= length $rbuf) {
187 $len = unpack "L", $rbuf; 319 $len = unpack "L", $rbuf;
188 if (4 + $len <= length $rbuf) { 320 4 + $len <= length $rbuf
321 or last;
322
189 my @r = $t->(substr $rbuf, 4, $len); 323 my @r = $t->(substr $rbuf, 4, $len);
190 substr $rbuf, 0, $len + 4, ""; 324 substr $rbuf, 0, $len + 4, "";
191 325
192 if (pop @r) { 326 if (pop @r) {
193 $on_event->(@r); 327 $on_event->(@r);
194 } elsif (@rcb) { 328 } elsif (@rcb) {
195 (shift @rcb)->(@r); 329 (shift @rcb)->(@r);
196 } else { 330 } else {
197 undef $rw; undef $ww; 331 undef $rw; undef $ww;
198 $on_error->("unexpected data from child"); 332 $on_error->("unexpected data from child");
199 }
200 } 333 }
201 } 334 }
202 } elsif (defined $len) { 335 } elsif (defined $len) {
203 undef $rw; undef $ww; # it ends here 336 undef $rw; undef $ww; # it ends here
337
338 if (@rcb) {
204 $on_error->("unexpected eof") 339 $on_error->("unexpected eof");
205 if @rcb; 340 } else {
341 $on_destroy->();
342 }
206 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) { 343 } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
207 undef $rw; undef $ww; # it ends here 344 undef $rw; undef $ww; # it ends here
208 $on_error->("read: $!"); 345 $on_error->("read: $!");
209 } 346 }
210 }; 347 };
… …
225 $wbuf .= pack "L/a*", &$f; 362 $wbuf .= pack "L/a*", &$f;
226 $ww ||= $fh && AE::io $fh, 1, $wcb; 363 $ww ||= $fh && AE::io $fh, 1, $wcb;
227 } 364 }
228} 365}
229 366
367=item $rpc->(..., $cb->(...))
368
369The RPC object returned by C<AnyEvent::Fork::RPC::run> is actually a code
370reference. There are two things you can do with it: call it, and let it go
371out of scope (let it get destroyed).
372
373If C<async> was false when C<$rpc> was created (the default), then, if you
374call C<$rpc>, the C<$function> is invoked with all arguments passed to
375C<$rpc> except the last one (the callback). When the function returns, the
376callback will be invoked with all the return values.
377
378If C<async> was true, then the C<$function> receives an additional
379initial argument, the result callback. In this case, returning from
380C<$function> does nothing - the function only counts as "done" when the
381result callback is called, and any arguments passed to it are considered
382the return values. This makes it possible to "return" from event handlers
383or e.g. Coro threads.
384
385The other thing that can be done with the RPC object is to destroy it. In
386this case, the child process will execute all remaining RPC calls, report
387their results, and then exit.
388
389See the examples section earlier in this document for some actual
390examples.
391
230=back 392=back
231 393
232=head1 CHILD PROCESS USAGE 394=head1 CHILD PROCESS USAGE
233 395
234These functions are not available in this module. They are only available 396The following function is not available in this module. They are only
235in the namespace of this module when the child is running, without 397available in the namespace of this module when the child is running,
236having to load any extra module. They are part of the child-side API of 398without having to load any extra modules. They are part of the child-side
237L<AnyEvent::Fork::RPC>. 399API of L<AnyEvent::Fork::RPC>.
238 400
239=over 4 401=over 4
240
241=item AnyEvent::Fork::RPC::quit
242
243This function can be called to gracefully stop the child process when it
244is idle.
245
246After this function is called, the process stops handling incoming RPC
247requests, but outstanding events and function return values will be sent
248to the parent. When all data has been sent, the process calls C<exit>.
249
250Since the parent might not expect the child to exit at random points in
251time, it is often better to signal the parent by sending an C<event> and
252letting the parent close down the child process.
253 402
254=item AnyEvent::Fork::RPC::event ... 403=item AnyEvent::Fork::RPC::event ...
255 404
256Send an event to the parent. Events are a bit like RPC calls made by the 405Send an event to the parent. Events are a bit like RPC calls made by the
257child process to the parent, except that there is no notion of return 406child process to the parent, except that there is no notion of return
258values. 407values.
259 408
409See the examples section earlier in this document for some actual
410examples.
411
260=back 412=back
261 413
262=head1 SEE ALSO 414=head1 SEE ALSO
263 415
264L<AnyEvent::Fork> (to create the processes in the first place), 416L<AnyEvent::Fork> (to create the processes in the first place),

Diff Legend

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