AnyEvent-Fork-RPC/RPC.pm

=head1 NAME

AnyEvent::Fork::RPC - simple RPC extension for AnyEvent::Fork

=head1 SYNOPSIS

   use AnyEvent::Fork::RPC;
   # use AnyEvent::Fork is not needed

   my $rpc = AnyEvent::Fork
      ->new
      ->require ("MyModule")
      ->AnyEvent::Fork::RPC::run (
         "MyModule::server",
      );

   my $cv = AE::cv;

   $rpc->(1, 2, 3, sub {
      print "MyModule::server returned @_\n";
      $cv->send;
   });

   $cv->recv;

=head1 DESCRIPTION

This module implements a simple RPC protocol and backend for processes
created via L<AnyEvent::Fork>, allowing you to call a function in the
child process and receive its return values (up to 4GB serialised).

It implements two different backends: a synchronous one that works like a
normal function call, and an asynchronous one that can run multiple jobs
concurrently in the child, using AnyEvent.

It also implements an asynchronous event mechanism from the child to the
parent, that could be used for progress indications or other information.

Loading this module also always loads L<AnyEvent::Fork>, so you can make a
separate C<use AnyEvent::Fork> if you wish, but you don't have to.

=head1 EXAMPLES

=head2 Example 1: Synchronous Backend

Here is a simple example that implements a backend that executes C<unlink>
and C<rmdir> calls, and reports their status back. It also reports the
number of requests it has processed every three requests, which is clearly
silly, but illustrates the use of events.

First the parent process:

   use AnyEvent;
   use AnyEvent::Fork::RPC;

   my $done = AE::cv;

   my $rpc = AnyEvent::Fork
      ->new
      ->require ("MyWorker")
      ->AnyEvent::Fork::RPC::run ("MyWorker::run",
         on_error   => sub { warn "FATAL: $_[0]"; exit 1 },
         on_event   => sub { warn "$_[0] requests handled\n" },
         on_destroy => $done,
      );

   for my $id (1..6) {
      $rpc->(rmdir => "/tmp/somepath/$id", sub {
         $_[0]
            or warn "/tmp/somepath/$id: $_[1]\n";
      });
   }

   undef $rpc;

   $done->recv;

The parent creates the process, queues a few rmdir's. It then forgets
about the C<$rpc> object, so that the child exits after it has handled the
requests, and then it waits till the requests have been handled.

The child is implemented using a separate module, C<MyWorker>, shown here:

   package MyWorker;

   my $count;

   sub run {
      my ($cmd, $path) = @_;

      AnyEvent::Fork::RPC::event ($count)
         unless ++$count % 3;

      my $status = $cmd eq "rmdir"  ? rmdir  $path
                 : $cmd eq "unlink" ? unlink $path
                 : die "fatal error, illegal command '$cmd'";

      $status or (0, "$!")
   }

   1

The C<run> function first sends a "progress" event every three calls, and
then executes C<rmdir> or C<unlink>, depending on the first parameter (or
dies with a fatal error - obviously, you must never let this happen :).

Eventually it returns the status value true if the command was successful,
or the status value 0 and the stringified error message.

On my system, running the first code fragment with the given
F<MyWorker.pm> in the current directory yields:

   /tmp/somepath/1: No such file or directory
   /tmp/somepath/2: No such file or directory
   3  requests handled
   /tmp/somepath/3: No such file or directory
   /tmp/somepath/4: No such file or directory
   /tmp/somepath/5: No such file or directory
   6  requests handled
   /tmp/somepath/6: No such file or directory

Obviously, none of the directories I am trying to delete even exist. Also,
the events and responses are processed in exactly the same order as
they were created in the child, which is true for both synchronous and
asynchronous backends.

Note that the parentheses in the call to C<AnyEvent::Fork::RPC::event> are
not optional. That is because the function isn't defined when the code is
compiled. You can make sure it is visible by pre-loading the correct
backend module in the call to C<require>:

      ->require ("AnyEvent::Fork::RPC::Sync", "MyWorker")

Since the backend module declares the C<event> function, loading it first
ensures that perl will correctly interpret calls to it.

And as a final remark, there is a fine module on CPAN that can
asynchronously C<rmdir> and C<unlink> and a lot more, and more efficiently
than this example, namely L<IO::AIO>.

=head3 Example 1a: the same with the asynchronous backend

This example only shows what needs to be changed to use the async backend
instead. Doing this is not very useful, the purpose of this example is
to show the minimum amount of change that is required to go from the
synchronous to the asynchronous backend.

To use the async backend in the previous example, you need to add the
C<async> parameter to the C<AnyEvent::Fork::RPC::run> call:

      ->AnyEvent::Fork::RPC::run ("MyWorker::run",
         async      => 1,
         ...

And since the function call protocol is now changed, you need to adopt
C<MyWorker::run> to the async API.

First, you need to accept the extra initial C<$done> callback:

   sub run {
      my ($done, $cmd, $path) = @_;

And since a response is now generated when C<$done> is called, as opposed
to when the function returns, we need to call the C<$done> function with
the status:

      $done->($status or (0, "$!"));

A few remarks are in order. First, it's quite pointless to use the async
backend for this example - but it I<is> possible. Second, you can call
C<$done> before or after returning from the function. Third, having both
returned from the function and having called the C<$done> callback, the
child process may exit at any time, so you should call C<$done> only when
you really I<are> done.

=head2 Example 2: Asynchronous Backend

This example implements multiple count-downs in the child, using
L<AnyEvent> timers. While this is a bit silly (one could use timers in te
parent just as well), it illustrates the ability to use AnyEvent in the
child and the fact that responses can arrive in a different order then the
requests.

It also shows how to embed the actual child code into a C<__DATA__>
section, so it doesn't need any external files at all.

And when your parent process is often busy, and you have stricter timing
requirements, then running timers in a child process suddenly doesn't look
so silly anymore.

Without further ado, here is the code:

   use AnyEvent;
   use AnyEvent::Fork::RPC;

   my $done = AE::cv;

   my $rpc = AnyEvent::Fork
      ->new
      ->require ("AnyEvent::Fork::RPC::Async")
      ->eval (do { local $/; <DATA> })
      ->AnyEvent::Fork::RPC::run ("run",
         async      => 1,
         on_error   => sub { warn "FATAL: $_[0]"; exit 1 },
         on_event   => sub { print $_[0] },
         on_destroy => $done,
      );

   for my $count (3, 2, 1) {
      $rpc->($count, sub {
         warn "job $count finished\n";
      });
   }

   undef $rpc;

   $done->recv;

   __DATA__

   # this ends up in main, as we don't use a package declaration

   use AnyEvent;

   sub run {
      my ($done, $count) = @_;

      my $n;

      AnyEvent::Fork::RPC::event "starting to count up to $count\n";

      my $w; $w = AE::timer 1, 1, sub {
         ++$n;

         AnyEvent::Fork::RPC::event "count $n of $count\n";

         if ($n == $count) {
            undef $w;
            $done->();
         }
      };
   }

The parent part (the one before the C<__DATA__> section) isn't very
different from the earlier examples. It sets async mode, preloads
the backend module (so the C<AnyEvent::Fork::RPC::event> function is
declared), uses a slightly different C<on_event> handler (which we use
simply for logging purposes) and then, instead of loading a module with
the actual worker code, it C<eval>'s the code from the data section in the
child process.

It then starts three countdowns, from 3 to 1 seconds downwards, destroys
the rpc object so the example finishes eventually, and then just waits for
the stuff to trickle in.

The worker code uses the event function to log some progress messages, but
mostly just creates a recurring one-second timer.

The timer callback increments a counter, logs a message, and eventually,
when the count has been reached, calls the finish callback.

On my system, this results in the following output. Since all timers fire
at roughly the same time, the actual order isn't guaranteed, but the order
shown is very likely what you would get, too.

   starting to count up to 3
   starting to count up to 2
   starting to count up to 1
   count 1 of 3
   count 1 of 2
   count 1 of 1
   job 1 finished
   count 2 of 2
   job 2 finished
   count 2 of 3
   count 3 of 3
   job 3 finished

While the overall ordering isn't guaranteed, the async backend still
guarantees that events and responses are delivered to the parent process
in the exact same ordering as they were generated in the child process.

And unless your system is I<very> busy, it should clearly show that the
job started last will finish first, as it has the lowest count.

This concludes the async example. Since L<AnyEvent::Fork> does not
actually fork, you are free to use about any module in the child, not just
L<AnyEvent>, but also L<IO::AIO>, or L<Tk> for example.

=head1 PARENT PROCESS USAGE

This module exports nothing, and only implements a single function:

=over 4

=cut

package AnyEvent::Fork::RPC;

use common::sense;

use Errno ();
use Guard ();

use AnyEvent;
use AnyEvent::Fork; # we don't actually depend on it, this is for convenience

our $VERSION = 0.1;

=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]

The traditional way to call it. But it is way cooler to call it in the
following way:

=item my $rpc = $fork->AnyEvent::Fork::RPC::run ($function, [key => value...])

This C<run> function/method can be used in place of the
L<AnyEvent::Fork::run> method. Just like that method, it takes over
the L<AnyEvent::Fork> process, but instead of calling the specified
C<$function> directly, it runs a server that accepts RPC calls and handles
responses.

It returns a function reference that can be used to call the function in
the child process, handling serialisation and data transfers.

The following key/value pairs are allowed. It is recommended to have at
least an C<on_error> or C<on_event> handler set.

=over 4

=item on_error => $cb->($msg)

Called on (fatal) errors, with a descriptive (hopefully) message. If
this callback is not provided, but C<on_event> is, then the C<on_event>
callback is called with the first argument being the string C<error>,
followed by the error message.

If neither handler is provided it prints the error to STDERR and will
start failing badly.

=item on_event => $cb->(...)

Called for every call to the C<AnyEvent::Fork::RPC::event> function in the
child, with the arguments of that function passed to the callback.

Also called on errors when no C<on_error> handler is provided.

=item on_destroy => $cb->()

Called when the C<$rpc> object has been destroyed and all requests have
been successfully handled. This is useful when you queue some requests and
want the child to go away after it has handled them. The problem is that
the parent must not exit either until all requests have been handled, and
this can be accomplished by waiting for this callback.

=item init => $function (default none)

When specified (by name), this function is called in the child as the very
first thing when taking over the process, with all the arguments normally
passed to the C<AnyEvent::Fork::run> function, except the communications
socket.

It can be used to do one-time things in the child such as storing passed
parameters or opening database connections.

It is called very early - before the serialisers are created or the
C<$function> name is resolved into a function reference, so it could be
used to load any modules that provide the serialiser or function. It can
not, however, create events.

=item async => $boolean (default: 0)

The default server used in the child does all I/O blockingly, and only
allows a single RPC call to execute concurrently.

Setting C<async> to a true value switches to another implementation that
uses L<AnyEvent> in the child and allows multiple concurrent RPC calls.

The actual API in the child is documented in the section that describes
the calling semantics of the returned C<$rpc> function.

If you want to pre-load the actual back-end modules to enable memory
sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for
synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.

If you use a template process and want to fork both sync and async
children, then it is permissible to load both modules.

=item serialiser => $string (default: '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })')

All arguments, result data and event data have to be serialised to be
transferred between the processes. For this, they have to be frozen and
thawed in both parent and child processes.

By default, only octet strings can be passed between the processes, which
is reasonably fast and efficient.

For more complicated use cases, you can provide your own freeze and thaw
functions, by specifying a string with perl source code. It's supposed to
return two code references when evaluated: the first receives a list of
perl values and must return an octet string. The second receives the octet
string and must return the original list of values.

If you need an external module for serialisation, then you can either
pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>
or C<require> statement into the serialiser string. Or both.

=back

See the examples section earlier in this document for some actual
examples.

=cut

our $STRING_SERIALISER = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })';

sub run {
   my ($self, $function, %arg) = @_;

   my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER;
   my $on_event   = delete $arg{on_event};
   my $on_error   = delete $arg{on_error};
   my $on_destroy = delete $arg{on_destroy};
   
   # default for on_error is to on_event, if specified
   $on_error ||= $on_event
               ? sub { $on_event->(error => shift) }
               : sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" };

   # default for on_event is to raise an error
   $on_event ||= sub { $on_error->("event received, but no on_event handler") };

   my ($f, $t) = eval $serialiser; die $@ if $@;

   my (@rcb, %rcb, $fh, $shutdown, $wbuf, $ww);
   my ($rlen, $rbuf, $rw) = 512 - 16;

   my $wcb = sub {
      my $len = syswrite $fh, $wbuf;

      unless (defined $len) {
         if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
            undef $rw; undef $ww; # it ends here
            $on_error->("$!");
         }
      }

      substr $wbuf, 0, $len, "";

      unless (length $wbuf) {
         undef $ww;
         $shutdown and shutdown $fh, 1;
      }
   };

   my $module = "AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync");

   $self->require ($module)
        ->send_arg ($function, $arg{init}, $serialiser)
        ->run ("$module\::run", sub {
      $fh = shift;

      my ($id, $len);
      $rw = AE::io $fh, 0, sub {
         $rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
         $len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;

         if ($len) {
            while (8 <= length $rbuf) {
               ($id, $len) = unpack "LL", $rbuf;
               8 + $len <= length $rbuf
                  or last;

               my @r = $t->(substr $rbuf, 8, $len);
               substr $rbuf, 0, 8 + $len, "";

               if ($id) {
                  if (@rcb) {
                     (shift @rcb)->(@r);
                  } elsif (my $cb = delete $rcb{$id}) {
                     $cb->(@r);
                  } else {
                     undef $rw; undef $ww;
                     $on_error->("unexpected data from child");
                  }
               } else {
                  $on_event->(@r);
               }
            }
         } elsif (defined $len) {
            undef $rw; undef $ww; # it ends here

            if (@rcb || %rcb) {
               use Data::Dump;ddx[\@rcb,\%rcb];#d#
               $on_error->("unexpected eof");
            } else {
               $on_destroy->();
            }
         } elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
            undef $rw; undef $ww; # it ends here
            $on_error->("read: $!");
         }
      };

      $ww ||= AE::io $fh, 1, $wcb;
   });

   my $guard = Guard::guard {
      $shutdown = 1;
      $ww ||= $fh && AE::io $fh, 1, $wcb;
   };

   my $id;

   $arg{async}
      ? sub {
           $id = ($id == 0xffffffff ? 0 : $id) + 1;
           $id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops

           $rcb{$id} = pop;

           $guard; # keep it alive

           $wbuf .= pack "LL/a*", $id, &$f;
           $ww ||= $fh && AE::io $fh, 1, $wcb;
        }
      : sub {
           push @rcb, pop;

           $guard; # keep it alive

           $wbuf .= pack "L/a*", &$f;
           $ww ||= $fh && AE::io $fh, 1, $wcb;
        }
}

=item $rpc->(..., $cb->(...))

The RPC object returned by C<AnyEvent::Fork::RPC::run> is actually a code
reference. There are two things you can do with it: call it, and let it go
out of scope (let it get destroyed).

If C<async> was false when C<$rpc> was created (the default), then, if you
call C<$rpc>, the C<$function> is invoked with all arguments passed to
C<$rpc> except the last one (the callback). When the function returns, the
callback will be invoked with all the return values.

If C<async> was true, then the C<$function> receives an additional
initial argument, the result callback. In this case, returning from
C<$function> does nothing - the function only counts as "done" when the
result callback is called, and any arguments passed to it are considered
the return values. This makes it possible to "return" from event handlers
or e.g. Coro threads.

The other thing that can be done with the RPC object is to destroy it. In
this case, the child process will execute all remaining RPC calls, report
their results, and then exit.

See the examples section earlier in this document for some actual
examples.

=back

=head1 CHILD PROCESS USAGE

The following function is not available in this module. They are only
available in the namespace of this module when the child is running,
without having to load any extra modules. They are part of the child-side
API of L<AnyEvent::Fork::RPC>.

=over 4

=item AnyEvent::Fork::RPC::event ...

Send an event to the parent. Events are a bit like RPC calls made by the
child process to the parent, except that there is no notion of return
values.

See the examples section earlier in this document for some actual
examples.

=back

=head1 SEE ALSO

L<AnyEvent::Fork> (to create the processes in the first place),
L<AnyEvent::Fork::Pool> (to manage whole pools of processes).

=head1 AUTHOR AND CONTACT INFORMATION

 Marc Lehmann <schmorp@schmorp.de>
 http://software.schmorp.de/pkg/AnyEvent-Fork-RPC

=cut

1

Revision:	1.11
Committed:	Thu Apr 18 07:59:46 2013 UTC (11 years, 1 month ago) by root
Branch:	MAIN
Changes since 1.10:	+111 -2 lines
Log Message:	* empty log message *
#	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 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::RPC;
55
56	my $done = AE::cv;
57
58	my $rpc = AnyEvent::Fork
59	->new
60	->require ("MyWorker")
61	->AnyEvent::Fork::RPC::run ("MyWorker::run",
62	on_error => sub { warn "FATAL: $_[0]"; exit 1 },
63	on_event => sub { warn "$_[0] requests handled\n" },
64	on_destroy => $done,
65	);
66
67	for my $id (1..6) {
68	$rpc->(rmdir => "/tmp/somepath/$id", sub {
69	$_[0]
70	or warn "/tmp/somepath/$id: $_[1]\n";
71	});
72	}
73
74	undef $rpc;
75
76	$done->recv;
77
78	The parent creates the process, queues a few rmdir's. It then forgets
79	about the C<$rpc> object, so that the child exits after it has handled the
80	requests, and then it waits till the requests have been handled.
81
82	The child is implemented using a separate module, C<MyWorker>, shown here:
83
84	package MyWorker;
85
86	my $count;
87
88	sub run {
89	my ($cmd, $path) = @_;
90
91	AnyEvent::Fork::RPC::event ($count)
92	unless ++$count % 3;
93
94	my $status = $cmd eq "rmdir" ? rmdir $path
95	: $cmd eq "unlink" ? unlink $path
96	: die "fatal error, illegal command '$cmd'";
97
98	$status or (0, "$!")
99	}
100
101	1
102
103	The C<run> function first sends a "progress" event every three calls, and
104	then executes C<rmdir> or C<unlink>, depending on the first parameter (or
105	dies with a fatal error - obviously, you must never let this happen :).
106
107	Eventually it returns the status value true if the command was successful,
108	or the status value 0 and the stringified error message.
109
110	On my system, running the first code fragment with the given
111	F<MyWorker.pm> in the current directory yields:
112
113	/tmp/somepath/1: No such file or directory
114	/tmp/somepath/2: No such file or directory
115	3 requests handled
116	/tmp/somepath/3: No such file or directory
117	/tmp/somepath/4: No such file or directory
118	/tmp/somepath/5: No such file or directory
119	6 requests handled
120	/tmp/somepath/6: No such file or directory
121
122	Obviously, none of the directories I am trying to delete even exist. Also,
123	the events and responses are processed in exactly the same order as
124	they were created in the child, which is true for both synchronous and
125	asynchronous backends.
126
127	Note that the parentheses in the call to C<AnyEvent::Fork::RPC::event> are
128	not optional. That is because the function isn't defined when the code is
129	compiled. You can make sure it is visible by pre-loading the correct
130	backend module in the call to C<require>:
131
132	->require ("AnyEvent::Fork::RPC::Sync", "MyWorker")
133
134	Since the backend module declares the C<event> function, loading it first
135	ensures that perl will correctly interpret calls to it.
136
137	And as a final remark, there is a fine module on CPAN that can
138	asynchronously C<rmdir> and C<unlink> and a lot more, and more efficiently
139	than this example, namely L<IO::AIO>.
140
141	=head3 Example 1a: the same with the asynchronous backend
142
143	This example only shows what needs to be changed to use the async backend
144	instead. Doing this is not very useful, the purpose of this example is
145	to show the minimum amount of change that is required to go from the
146	synchronous to the asynchronous backend.
147
148	To use the async backend in the previous example, you need to add the
149	C<async> parameter to the C<AnyEvent::Fork::RPC::run> call:
150
151	->AnyEvent::Fork::RPC::run ("MyWorker::run",
152	async => 1,
153	...
154
155	And since the function call protocol is now changed, you need to adopt
156	C<MyWorker::run> to the async API.
157
158	First, you need to accept the extra initial C<$done> callback:
159
160	sub run {
161	my ($done, $cmd, $path) = @_;
162
163	And since a response is now generated when C<$done> is called, as opposed
164	to when the function returns, we need to call the C<$done> function with
165	the status:
166
167	$done->($status or (0, "$!"));
168
169	A few remarks are in order. First, it's quite pointless to use the async
170	backend for this example - but it I<is> possible. Second, you can call
171	C<$done> before or after returning from the function. Third, having both
172	returned from the function and having called the C<$done> callback, the
173	child process may exit at any time, so you should call C<$done> only when
174	you really I<are> done.
175
176	=head2 Example 2: Asynchronous Backend
177
178	This example implements multiple count-downs in the child, using
179	L<AnyEvent> timers. While this is a bit silly (one could use timers in te
180	parent just as well), it illustrates the ability to use AnyEvent in the
181	child and the fact that responses can arrive in a different order then the
182	requests.
183
184	It also shows how to embed the actual child code into a C<__DATA__>
185	section, so it doesn't need any external files at all.
186
187	And when your parent process is often busy, and you have stricter timing
188	requirements, then running timers in a child process suddenly doesn't look
189	so silly anymore.
190
191	Without 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
244	The parent part (the one before the C<__DATA__> section) isn't very
245	different from the earlier examples. It sets async mode, preloads
246	the backend module (so the C<AnyEvent::Fork::RPC::event> function is
247	declared), uses a slightly different C<on_event> handler (which we use
248	simply for logging purposes) and then, instead of loading a module with
249	the actual worker code, it C<eval>'s the code from the data section in the
250	child process.
251
252	It then starts three countdowns, from 3 to 1 seconds downwards, destroys
253	the rpc object so the example finishes eventually, and then just waits for
254	the stuff to trickle in.
255
256	The worker code uses the event function to log some progress messages, but
257	mostly just creates a recurring one-second timer.
258
259	The timer callback increments a counter, logs a message, and eventually,
260	when the count has been reached, calls the finish callback.
261
262	On my system, this results in the following output. Since all timers fire
263	at roughly the same time, the actual order isn't guaranteed, but the order
264	shown 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
279	While the overall ordering isn't guaranteed, the async backend still
280	guarantees that events and responses are delivered to the parent process
281	in the exact same ordering as they were generated in the child process.
282
283	And unless your system is I<very> busy, it should clearly show that the
284	job started last will finish first, as it has the lowest count.
285
286	This concludes the async example. Since L<AnyEvent::Fork> does not
287	actually fork, you are free to use about any module in the child, not just
288	L<AnyEvent>, but also L<IO::AIO>, or L<Tk> for example.
289
290	=head1 PARENT PROCESS USAGE
291
292	This module exports nothing, and only implements a single function:
293
294	=over 4
295
296	=cut
297
298	package AnyEvent::Fork::RPC;
299
300	use common::sense;
301
302	use Errno ();
303	use Guard ();
304
305	use AnyEvent;
306	use AnyEvent::Fork; # we don't actually depend on it, this is for convenience
307
308	our $VERSION = 0.1;
309
310	=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
311
312	The traditional way to call it. But it is way cooler to call it in the
313	following way:
314
315	=item my $rpc = $fork->AnyEvent::Fork::RPC::run ($function, [key => value...])
316
317	This C<run> function/method can be used in place of the
318	L<AnyEvent::Fork::run> method. Just like that method, it takes over
319	the L<AnyEvent::Fork> process, but instead of calling the specified
320	C<$function> directly, it runs a server that accepts RPC calls and handles
321	responses.
322
323	It returns a function reference that can be used to call the function in
324	the child process, handling serialisation and data transfers.
325
326	The following key/value pairs are allowed. It is recommended to have at
327	least an C<on_error> or C<on_event> handler set.
328
329	=over 4
330
331	=item on_error => $cb->($msg)
332
333	Called on (fatal) errors, with a descriptive (hopefully) message. If
334	this callback is not provided, but C<on_event> is, then the C<on_event>
335	callback is called with the first argument being the string C<error>,
336	followed by the error message.
337
338	If neither handler is provided it prints the error to STDERR and will
339	start failing badly.
340
341	=item on_event => $cb->(...)
342
343	Called for every call to the C<AnyEvent::Fork::RPC::event> function in the
344	child, with the arguments of that function passed to the callback.
345
346	Also called on errors when no C<on_error> handler is provided.
347
348	=item on_destroy => $cb->()
349
350	Called when the C<$rpc> object has been destroyed and all requests have
351	been successfully handled. This is useful when you queue some requests and
352	want the child to go away after it has handled them. The problem is that
353	the parent must not exit either until all requests have been handled, and
354	this can be accomplished by waiting for this callback.
355
356	=item init => $function (default none)
357
358	When specified (by name), this function is called in the child as the very
359	first thing when taking over the process, with all the arguments normally
360	passed to the C<AnyEvent::Fork::run> function, except the communications
361	socket.
362
363	It can be used to do one-time things in the child such as storing passed
364	parameters or opening database connections.
365
366	It is called very early - before the serialisers are created or the
367	C<$function> name is resolved into a function reference, so it could be
368	used to load any modules that provide the serialiser or function. It can
369	not, however, create events.
370
371	=item async => $boolean (default: 0)
372
373	The default server used in the child does all I/O blockingly, and only
374	allows a single RPC call to execute concurrently.
375
376	Setting C<async> to a true value switches to another implementation that
377	uses L<AnyEvent> in the child and allows multiple concurrent RPC calls.
378
379	The actual API in the child is documented in the section that describes
380	the calling semantics of the returned C<$rpc> function.
381
382	If you want to pre-load the actual back-end modules to enable memory
383	sharing, then you should load C<AnyEvent::Fork::RPC::Sync> for
384	synchronous, and C<AnyEvent::Fork::RPC::Async> for asynchronous mode.
385
386	If you use a template process and want to fork both sync and async
387	children, then it is permissible to load both modules.
388
389	=item serialiser => $string (default: '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })')
390
391	All arguments, result data and event data have to be serialised to be
392	transferred between the processes. For this, they have to be frozen and
393	thawed in both parent and child processes.
394
395	By default, only octet strings can be passed between the processes, which
396	is reasonably fast and efficient.
397
398	For more complicated use cases, you can provide your own freeze and thaw
399	functions, by specifying a string with perl source code. It's supposed to
400	return two code references when evaluated: the first receives a list of
401	perl values and must return an octet string. The second receives the octet
402	string and must return the original list of values.
403
404	If you need an external module for serialisation, then you can either
405	pre-load it into your L<AnyEvent::Fork> process, or you can add a C<use>
406	or C<require> statement into the serialiser string. Or both.
407
408	=back
409
410	See the examples section earlier in this document for some actual
411	examples.
412
413	=cut
414
415	our $STRING_SERIALISER = '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })';
416
417	sub run {
418	my ($self, $function, %arg) = @_;
419
420	my $serialiser = delete $arg{serialiser} \|\| $STRING_SERIALISER;
421	my $on_event = delete $arg{on_event};
422	my $on_error = delete $arg{on_error};
423	my $on_destroy = delete $arg{on_destroy};
424
425	# default for on_error is to on_event, if specified
426	$on_error \|\|= $on_event
427	? sub { $on_event->(error => shift) }
428	: sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" };
429
430	# default for on_event is to raise an error
431	$on_event \|\|= sub { $on_error->("event received, but no on_event handler") };
432
433	my ($f, $t) = eval $serialiser; die $@ if $@;
434
435	my (@rcb, %rcb, $fh, $shutdown, $wbuf, $ww);
436	my ($rlen, $rbuf, $rw) = 512 - 16;
437
438	my $wcb = sub {
439	my $len = syswrite $fh, $wbuf;
440
441	unless (defined $len) {
442	if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
443	undef $rw; undef $ww; # it ends here
444	$on_error->("$!");
445	}
446	}
447
448	substr $wbuf, 0, $len, "";
449
450	unless (length $wbuf) {
451	undef $ww;
452	$shutdown and shutdown $fh, 1;
453	}
454	};
455
456	my $module = "AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync");
457
458	$self->require ($module)
459	->send_arg ($function, $arg{init}, $serialiser)
460	->run ("$module\::run", sub {
461	$fh = shift;
462
463	my ($id, $len);
464	$rw = AE::io $fh, 0, sub {
465	$rlen = $rlen * 2 + 16 if $rlen - 128 < length $rbuf;
466	$len = sysread $fh, $rbuf, $rlen - length $rbuf, length $rbuf;
467
468	if ($len) {
469	while (8 <= length $rbuf) {
470	($id, $len) = unpack "LL", $rbuf;
471	8 + $len <= length $rbuf
472	or last;
473
474	my @r = $t->(substr $rbuf, 8, $len);
475	substr $rbuf, 0, 8 + $len, "";
476
477	if ($id) {
478	if (@rcb) {
479	(shift @rcb)->(@r);
480	} elsif (my $cb = delete $rcb{$id}) {
481	$cb->(@r);
482	} else {
483	undef $rw; undef $ww;
484	$on_error->("unexpected data from child");
485	}
486	} else {
487	$on_event->(@r);
488	}
489	}
490	} elsif (defined $len) {
491	undef $rw; undef $ww; # it ends here
492
493	if (@rcb \|\| %rcb) {
494	use Data::Dump;ddx[\@rcb,\%rcb];#d#
495	$on_error->("unexpected eof");
496	} else {
497	$on_destroy->();
498	}
499	} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
500	undef $rw; undef $ww; # it ends here
501	$on_error->("read: $!");
502	}
503	};
504
505	$ww \|\|= AE::io $fh, 1, $wcb;
506	});
507
508	my $guard = Guard::guard {
509	$shutdown = 1;
510	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
511	};
512
513	my $id;
514
515	$arg{async}
516	? sub {
517	$id = ($id == 0xffffffff ? 0 : $id) + 1;
518	$id = ($id == 0xffffffff ? 0 : $id) + 1 while exists $rcb{$id}; # rarely loops
519
520	$rcb{$id} = pop;
521
522	$guard; # keep it alive
523
524	$wbuf .= pack "LL/a*", $id, &$f;
525	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
526	}
527	: sub {
528	push @rcb, pop;
529
530	$guard; # keep it alive
531
532	$wbuf .= pack "L/a*", &$f;
533	$ww \|\|= $fh && AE::io $fh, 1, $wcb;
534	}
535	}
536
537	=item $rpc->(..., $cb->(...))
538
539	The RPC object returned by C<AnyEvent::Fork::RPC::run> is actually a code
540	reference. There are two things you can do with it: call it, and let it go
541	out of scope (let it get destroyed).
542
543	If C<async> was false when C<$rpc> was created (the default), then, if you
544	call C<$rpc>, the C<$function> is invoked with all arguments passed to
545	C<$rpc> except the last one (the callback). When the function returns, the
546	callback will be invoked with all the return values.
547
548	If C<async> was true, then the C<$function> receives an additional
549	initial argument, the result callback. In this case, returning from
550	C<$function> does nothing - the function only counts as "done" when the
551	result callback is called, and any arguments passed to it are considered
552	the return values. This makes it possible to "return" from event handlers
553	or e.g. Coro threads.
554
555	The other thing that can be done with the RPC object is to destroy it. In
556	this case, the child process will execute all remaining RPC calls, report
557	their results, and then exit.
558
559	See the examples section earlier in this document for some actual
560	examples.
561
562	=back
563
564	=head1 CHILD PROCESS USAGE
565
566	The following function is not available in this module. They are only
567	available in the namespace of this module when the child is running,
568	without having to load any extra modules. They are part of the child-side
569	API of L<AnyEvent::Fork::RPC>.
570
571	=over 4
572
573	=item AnyEvent::Fork::RPC::event ...
574
575	Send an event to the parent. Events are a bit like RPC calls made by the
576	child process to the parent, except that there is no notion of return
577	values.
578
579	See the examples section earlier in this document for some actual
580	examples.
581
582	=back
583
584	=head1 SEE ALSO
585
586	L<AnyEvent::Fork> (to create the processes in the first place),
587	L<AnyEvent::Fork::Pool> (to manage whole pools of processes).
588
589	=head1 AUTHOR AND CONTACT INFORMATION
590
591	Marc Lehmann <schmorp@schmorp.de>
592	http://software.schmorp.de/pkg/AnyEvent-Fork-RPC
593
594	=cut
595
596	1
597