AnyEvent-Fork-RPC/RPC.pm

=head1 NAME

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

=head1 SYNOPSIS

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

   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.

=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;

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 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.

=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.

=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.

=back

=cut

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

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

   my $serialiser = delete $arg{serialiser} || $SERIALISE_STRINGS;
   my $on_event   = delete $arg{on_event};
   my $on_error   = delete $arg{on_error};
   
   # 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, $fh, $shutdown, $wbuf, $ww, $rbuf, $rw);

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

      if (!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;
      $rw = AE::io $fh, 0, sub {
         my $len = sysread $fh, $rbuf, 512 + length $rbuf, length $rbuf;

         if ($len) {
            while (5 <= length $rbuf) {
               $len = unpack "L", $rbuf;
               if (4 + $len <= length $rbuf) {
                  my @r = $t->(substr $rbuf, 4, $len);
                  substr $rbuf, 0, $len + 4, "";
                  
                  if (pop @r) {
                     $on_event->(@r);
                  } elsif (@rcb) {
                     (shift @rcb)->(@r);
                  } else {
                     undef $rw; undef $ww;
                     $on_error->("unexpected data from child");
                  }
               }
            }
         } elsif (defined $len) {
            undef $rw; undef $ww; # it ends here
            $on_error->("unexpected eof")
               if @rcb;
         } 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;
   };

   sub {
      push @rcb, pop;

      $guard; # keep it alive

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

=back

=head1 CHILD PROCESS USAGE

These functions are 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 module. They are part of the child-side API of
L<AnyEvent::Fork::RPC>.

=over 4

=item AnyEvent::Fork::RPC::quit

This function can be called to gracefully stop the child process when it
is idle.

After this function is called, the process stops handling incoming RPC
requests, but outstanding events and function return values will be sent
to the parent. When all data has been sent, the process calls C<exit>.

Since the parent might not expect the child to exit at random points in
time, it is often better to signal the parent by sending an C<event> and
letting the parent close down the child process.

=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.

=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.1
Committed:	Wed Apr 17 15:55:59 2013 UTC (11 years, 1 month ago) by root
Branch:	MAIN
Log Message:	* empty log message *
#	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;
8			use AnyEvent::Fork::RPC;
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			=head1 PARENT PROCESS USAGE
40
41			This module exports nothing, and only implements a single function:
42
43			=over 4
44
45			=cut
46
47			package AnyEvent::Fork::RPC;
48
49			use common::sense;
50
51			use Errno ();
52			use Guard ();
53
54			use AnyEvent;
55			#use AnyEvent::Fork;
56
57			our $VERSION = 0.1;
58
59			=item my $rpc = AnyEvent::Fork::RPC::run $fork, $function, [key => value...]
60
61			The traditional way to call it. But it is way cooler to call it in the
62			following way:
63
64			=item my $rpc = $fork->AnyEvent::Fork::RPC::run ($function, [key => value...])
65
66			This C<run> function/method can be used in place of the
67			L<AnyEvent::Fork::run> method. Just like that method, it takes over
68			the L<AnyEvent::Fork> process, but instead of calling the specified
69			C<$function> directly, it runs a server that accepts RPC calls and handles
70			responses.
71
72			It returns a function reference that can be used to call the function in
73			the child process, handling serialisation and data transfers.
74
75			The following key/value pairs are allowed. It is recommended to have at
76			least an C<on_error> or C<on_event> handler set.
77
78			=over 4
79
80			=item on_error => $cb->($msg)
81
82			Called on (fatal) errors, with a descriptive (hopefully) message. If
83			this callback is not provided, but C<on_event> is, then the C<on_event>
84			callback is called with the first argument being the string C<error>,
85			followed by the error message.
86
87			If neither handler is provided it prints the error to STDERR and will
88			start failing badly.
89
90			=item on_event => $cb->(...)
91
92			Called for every call to the C<AnyEvent::Fork::RPC::event> function in the
93			child, with the arguments of that function passed to the callback.
94
95			Also called on errors when no C<on_error> handler is provided.
96
97			=item init => $function (default none)
98
99			When specified (by name), this function is called in the child as the very
100			first thing when taking over the process, with all the arguments normally
101			passed to the C<AnyEvent::Fork::run> function, except the communications
102			socket.
103
104			It can be used to do one-time things in the child such as storing passed
105			parameters or opening database connections.
106
107			=item async => $boolean (default: 0)
108
109			The default server used in the child does all I/O blockingly, and only
110			allows a single RPC call to execute concurrently.
111
112			Setting C<async> to a true value switches to another implementation that
113			uses L<AnyEvent> in the child and allows multiple concurrent RPC calls.
114
115			The actual API in the child is documented in the section that describes
116			the calling semantics of the returned C<$rpc> function.
117
118			=item serialiser => $string (default: '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })')
119
120			All arguments, result data and event data have to be serialised to be
121			transferred between the processes. For this, they have to be frozen and
122			thawed in both parent and child processes.
123
124			By default, only octet strings can be passed between the processes, which
125			is reasonably fast and efficient.
126
127			For more complicated use cases, you can provide your own freeze and thaw
128			functions, by specifying a string with perl source code. It's supposed to
129			return two code references when evaluated: the first receives a list of
130			perl values and must return an octet string. The second receives the octet
131			string and must return the original list of values.
132
133			=back
134
135			=cut
136
137			our $SERIALISE_STRINGS = '(sub { pack "(w/a)", @_ }, sub { unpack "(w/a)", shift })';
138
139			sub run {
140			my ($self, $function, %arg) = @_;
141
142			my $serialiser = delete $arg{serialiser} \|\| $SERIALISE_STRINGS;
143			my $on_event = delete $arg{on_event};
144			my $on_error = delete $arg{on_error};
145
146			# default for on_error is to on_event, if specified
147			$on_error \|\|= $on_event
148			? sub { $on_event->(error => shift) }
149			: sub { die "AnyEvent::Fork::RPC: uncaught error: $_[0].\n" };
150
151			# default for on_event is to raise an error
152			$on_event \|\|= sub { $on_error->("event received, but no on_event handler") };
153
154			my ($f, $t) = eval $serialiser; die $@ if $@;
155
156			my (@rcb, $fh, $shutdown, $wbuf, $ww, $rbuf, $rw);
157
158			my $wcb = sub {
159			my $len = syswrite $fh, $wbuf;
160
161			if (!defined $len) {
162			if ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
163			undef $rw; undef $ww; # it ends here
164			$on_error->("$!");
165			}
166			}
167
168			substr $wbuf, 0, $len, "";
169
170			unless (length $wbuf) {
171			undef $ww;
172			$shutdown and shutdown $fh, 1;
173			}
174			};
175
176			my $module = "AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync");
177
178			$self->require ($module)
179			->send_arg ($function, $arg{init}, $serialiser)
180			->run ("$module\::run", sub {
181			$fh = shift;
182			$rw = AE::io $fh, 0, sub {
183			my $len = sysread $fh, $rbuf, 512 + length $rbuf, length $rbuf;
184
185			if ($len) {
186			while (5 <= length $rbuf) {
187			$len = unpack "L", $rbuf;
188			if (4 + $len <= length $rbuf) {
189			my @r = $t->(substr $rbuf, 4, $len);
190			substr $rbuf, 0, $len + 4, "";
191
192			if (pop @r) {
193			$on_event->(@r);
194			} elsif (@rcb) {
195			(shift @rcb)->(@r);
196			} else {
197			undef $rw; undef $ww;
198			$on_error->("unexpected data from child");
199			}
200			}
201			}
202			} elsif (defined $len) {
203			undef $rw; undef $ww; # it ends here
204			$on_error->("unexpected eof")
205			if @rcb;
206			} elsif ($! != Errno::EAGAIN && $! != Errno::EWOULDBLOCK) {
207			undef $rw; undef $ww; # it ends here
208			$on_error->("read: $!");
209			}
210			};
211
212			$ww \|\|= AE::io $fh, 1, $wcb;
213			});
214
215			my $guard = Guard::guard {
216			$shutdown = 1;
217			$ww \|\|= $fh && AE::io $fh, 1, $wcb;
218			};
219
220			sub {
221			push @rcb, pop;
222
223			$guard; # keep it alive
224
225			$wbuf .= pack "L/a*", &$f;
226			$ww \|\|= $fh && AE::io $fh, 1, $wcb;
227			}
228			}
229
230			=back
231
232			=head1 CHILD PROCESS USAGE
233
234			These functions are not available in this module. They are only available
235			in the namespace of this module when the child is running, without
236			having to load any extra module. They are part of the child-side API of
237			L<AnyEvent::Fork::RPC>.
238
239			=over 4
240
241			=item AnyEvent::Fork::RPC::quit
242
243			This function can be called to gracefully stop the child process when it
244			is idle.
245
246			After this function is called, the process stops handling incoming RPC
247			requests, but outstanding events and function return values will be sent
248			to the parent. When all data has been sent, the process calls C<exit>.
249
250			Since the parent might not expect the child to exit at random points in
251			time, it is often better to signal the parent by sending an C<event> and
252			letting the parent close down the child process.
253
254			=item AnyEvent::Fork::RPC::event ...
255
256			Send an event to the parent. Events are a bit like RPC calls made by the
257			child process to the parent, except that there is no notion of return
258			values.
259
260			=back
261
262			=head1 SEE ALSO
263
264			L<AnyEvent::Fork> (to create the processes in the first place),
265			L<AnyEvent::Fork::Pool> (to manage whole pools of processes).
266
267			=head1 AUTHOR AND CONTACT INFORMATION
268
269			Marc Lehmann <schmorp@schmorp.de>
270			http://software.schmorp.de/pkg/AnyEvent-Fork-RPC
271
272			=cut
273
274			1
275