[ViewVC] Diff of: cvs/AnyEvent-Fork-RPC/RPC.pm

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents):
Revision 1.25 by root, Sun Apr 28 14:17:59 2013 UTC vs.
Revision 1.46 by root, Sun Sep 15 20:18:14 2019 UTC

 =head1 NAME
 AnyEvent::Fork::RPC - simple RPC extension for AnyEvent::Fork
-THE API IS NOT FINISHED, CONSIDER THIS A BETA RELEASE
 =head1 SYNOPSIS
+   use AnyEvent::Fork;
    use AnyEvent::Fork::RPC;
-   # use AnyEvent::Fork is not needed
    my $rpc = AnyEvent::Fork
       ->new
       ->require ("MyModule")
       ->AnyEvent::Fork::RPC::run (
    $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
+created via L<AnyEvent::Fork> or L<AnyEvent::Fork::Remote>, allowing you
-child process and receive its return values (up to 4GB serialised).
+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
 silly, but illustrates the use of events.
 First the parent process:
    use AnyEvent;
+   use AnyEvent::Fork;
    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_error   => sub { warn "ERROR: $_[0]"; exit 1 },
          on_event   => sub { warn "$_[0] requests handled\n" },
          on_destroy => $done,
       );
    for my $id (1..6) {
 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
+L<AnyEvent> timers. While this is a bit silly (one could use timers in the
 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__>
 so silly anymore.
 Without further ado, here is the code:
    use AnyEvent;
+   use AnyEvent::Fork;
    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_error   => sub { warn "ERROR: $_[0]"; exit 1 },
          on_event   => sub { print $_[0] },
          on_destroy => $done,
       );
    for my $count (3, 2, 1) {
 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.
+=head2 Example 3: Asynchronous backend with Coro
+With L<Coro> you can create a nice asynchronous backend implementation by
+defining an rpc server function that creates a new Coro thread for every
+request that calls a function "normally", i.e. the parameters from the
+parent process are passed to it, and any return values are returned to the
+parent process, e.g.:
+   package My::Arith;
+   sub add {
+      return $_[0] + $_[1];
+   }
+   sub mul {
+      return $_[0] * $_[1];
+   }
+   sub run {
+      my ($done, $func, @arg) = @_;
+      Coro::async_pool {
+         $done->($func->(@arg));
+      };
+   }
+The C<run> function creates a new thread for every invocation, using the
+first argument as function name, and calls the C<$done> callback on it's
+return values. This makes it quite natural to define the C<add> and C<mul>
+functions to add or multiply two numbers and return the result.
+Since this is the asynchronous backend, it's quite possible to define RPC
+function that do I/O or wait for external events - their execution will
+overlap as needed.
+The above could be used like this:
+   my $rpc = AnyEvent::Fork
+      ->new
+      ->require ("MyWorker")
+      ->AnyEvent::Fork::RPC::run ("My::Arith::run",
+         on_error => ..., on_event => ..., on_destroy => ...,
+      );
+   $rpc->(add => 1, 3, Coro::rouse_cb); say Coro::rouse_wait;
+   $rpc->(mul => 3, 2, Coro::rouse_cb); say Coro::rouse_wait;
+The C<say>'s will print C<4> and C<6>.
+=head2 Example 4: Forward AnyEvent::Log messages using C<on_event>
+This partial example shows how to use the C<event> function to forward
+L<AnyEvent::Log> messages to the parent.
+For this, the parent needs to provide a suitable C<on_event>:
+   ->AnyEvent::Fork::RPC::run (
+      on_event => sub {
+         if ($_[0] eq "ae_log") {
+            my (undef, $level, $message) = @_;
+            AE::log $level, $message;
+         } else {
+            # other event types
+         }
+      },
+   )
+In the child, as early as possible, the following code should reconfigure
+L<AnyEvent::Log> to log via C<AnyEvent::Fork::RPC::event>:
+   $AnyEvent::Log::LOG->log_cb (sub {
+      my ($timestamp, $orig_ctx, $level, $message) = @{+shift};
+      if (defined &AnyEvent::Fork::RPC::event) {
+         AnyEvent::Fork::RPC::event (ae_log => $level, $message);
+      } else {
+         warn "[$$ before init] $message\n";
+      }
+   });
+There is an important twist - the C<AnyEvent::Fork::RPC::event> function
+is only defined when the child is fully initialised. If you redirect the
+log messages in your C<init> function for example, then the C<event>
+function might not yet be available. This is why the log callback checks
+whether the function is there using C<defined>, and only then uses it to
+log the message.
 =head1 PARENT PROCESS USAGE
 This module exports nothing, and only implements a single function:
 =over 4
 use Errno ();
 use Guard ();
 use AnyEvent;
-# explicit version on next line, as some cpan-testers test with the 0.1 version,
-# ignoring dependencies, and this line will at least give a clear indication of that.
-use AnyEvent::Fork 0.6; # we don't actually depend on it, this is for convenience
-our $VERSION = 1.1;
+our $VERSION = '2.0';
 =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:
 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
+If neither handler is provided, then the error is reported with loglevel
-start failing badly.
+C<error> via C<AE::log>.
 =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.
 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)
+=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 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 done => $function (default: C<CORE::exit>)
+The function to call when the asynchronous backend detects an end of file
+condition when reading from the communications socket I<and> there are no
+outstanding requests. It is ignored by the synchronous backend.
+By overriding this you can prolong the life of a RPC process after e.g.
+the parent has exited by running the event loop in the provided function
+(or simply calling it, for example, when your child process uses L<EV> you
+could provide L<EV::run> as C<done> function).
+Of course, in that case you are responsible for exiting at the appropriate
+time and not returning from
-=item async => $boolean (default: 0)
+=item async => $boolean (default: C<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
 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: $AnyEvent::Fork::RPC::STRING_SERIALISER)
+=item serialiser => $string (default: C<$AnyEvent::Fork::RPC::STRING_SERIALISER>)
 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
+By default, only octet strings can be passed between the processes,
-is reasonably fast and efficient and requires no extra modules.
+which is reasonably fast and efficient and requires no extra modules
+(the C<AnyEvent::Fork::RPC> distribution does not provide these extra
+serialiser modules).
 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
 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.
-Here are some examples - some of them are also available as global
+Here are some examples - all of them are also available as global
 variables that make them easier to use.
 =over 4
-=item octet strings - C<$AnyEvent::Fork::RPC::STRING_SERIALISER>
+=item C<$AnyEvent::Fork::RPC::STRING_SERIALISER> - octet strings only
-This serialiser concatenates length-prefixes octet strings, and is the
+This serialiser (currently the default) concatenates length-prefixes octet
-default.
+strings, and is the default. That means you can only pass (and return)
+strings containing character codes 0-255.
+The main advantages of this serialiser are the high speed and that it
+doesn't need another module. The main disadvantage is that you are very
+limited in what you can pass - only octet strings.
 Implementation:
    (
       sub { pack   "(w/a*)*", @_ },
       sub { unpack "(w/a*)*", shift }
    )
-=item json - C<$AnyEvent::Fork::RPC::JSON_SERIALISER>
+=item C<$AnyEvent::Fork::RPC::CBOR_XS_SERIALISER> - uses L<CBOR::XS>
+This serialiser creates CBOR::XS arrays - you have to make sure the
+L<CBOR::XS> module is installed for this serialiser to work. It can be
+beneficial for sharing when you preload the L<CBOR::XS> module in a template
+process.
+L<CBOR::XS> is about as fast as the octet string serialiser, but supports
+complex data structures (similar to JSON) and is faster than any of the
+other serialisers. If you have the L<CBOR::XS> module available, it's the
+best choice.
+The encoder enables C<allow_sharing> (so this serialisation method can
+encode cyclic and self-referencing data structures).
+Implementation:
+   use CBOR::XS ();
+   (
+      sub {    CBOR::XS::encode_cbor_sharing \@_ },
+      sub { @{ CBOR::XS::decode_cbor shift } }
+   )
+=item C<$AnyEvent::Fork::RPC::JSON_SERIALISER> - uses L<JSON::XS> or L<JSON>
 This serialiser creates JSON arrays - you have to make sure the L<JSON>
 module is installed for this serialiser to work. It can be beneficial for
 sharing when you preload the L<JSON> module in a template process.
    (
       sub {    JSON::encode_json \@_ },
       sub { @{ JSON::decode_json shift } }
    )
-=item storable - C<$AnyEvent::Fork::RPC::STORABLE_SERIALISER>
+=item C<$AnyEvent::Fork::RPC::STORABLE_SERIALISER> - L<Storable>
 This serialiser uses L<Storable>, which means it has high chance of
 serialising just about anything you throw at it, at the cost of having
-very high overhead per operation. It also comes with perl.
+very high overhead per operation. It also comes with perl. It should be
+used when you need to serialise complex data structures.
 Implementation:
    use Storable ();
    (
       sub {    Storable::freeze \@_ },
       sub { @{ Storable::thaw shift } }
    )
+=item C<$AnyEvent::Fork::RPC::NSTORABLE_SERIALISER> - portable Storable
+This serialiser also uses L<Storable>, but uses it's "network" format
+to serialise data, which makes it possible to talk to different
+perl binaries (for example, when talking to a process created with
+L<AnyEvent::Fork::Remote>).
+Implementation:
+   use Storable ();
+   (
+      sub {    Storable::nfreeze \@_ },
+      sub { @{ Storable::thaw shift } }
+   )
 =back
+=item buflen => $bytes (default: C<512 - 16>)
+The starting size of the read buffer for request and response data.
+C<AnyEvent::Fork::RPC> ensures that the buffer for reeading request and
+response data is large enough for at leats aingle request or response, and
+will dynamically enlarge the buffer if needed.
+While this ensures that memory is not overly wasted, it typically leads
+to having to do one syscall per request, which can be inefficient in some
+cases. In such cases, it can be beneficient to increase the buffer size to
+hold more than one request.
+=item buflen_req => $bytes (default: same as C<buflen>)
+Overrides C<buflen> for request data (as read by the forked process).
+=item buflen_res => $bytes (default: same as C<buflen>)
+Overrides C<buflen> for response data (replies read by the parent process).
 =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 })';
+our $STRING_SERIALISER    = '(sub { pack "(w/a*)*", @_ }, sub { unpack "(w/a*)*", shift })';
+our $CBOR_XS_SERIALISER   = 'use CBOR::XS (); (sub { CBOR::XS::encode_cbor_sharing \@_ }, sub { @{ CBOR::XS::decode_cbor shift } })';
-our $JSON_SERIALISER     = 'use JSON (); (sub { JSON::encode_json \@_ }, sub { @{ JSON::decode_json shift } })';
+our $JSON_SERIALISER      = 'use JSON     (); (sub { JSON::encode_json             \@_ }, sub { @{ JSON::decode_json     shift } })';
-our $STORABLE_SERIALISER = 'use Storable (); (sub { Storable::freeze \@_ }, sub { @{ Storable::thaw shift } })';
+our $STORABLE_SERIALISER  = 'use Storable (); (sub { Storable::freeze  \@_ }, sub { @{ Storable::thaw shift } })';
+our $NSTORABLE_SERIALISER = 'use Storable (); (sub { Storable::nfreeze \@_ }, sub { @{ Storable::thaw shift } })';
 sub run {
    my ($self, $function, %arg) = @_;
    my $serialiser = delete $arg{serialiser} || $STRING_SERIALISER;
    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" };
+               : sub { AE::log die => "AnyEvent::Fork::RPC: uncaught error: $_[0]." };
    # 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 ($rlen, $rbuf, $rw) = $arg{buflen_res} || $arg{buflen} || 512 - 16;
    my $wcb = sub {
       my $len = syswrite $fh, $wbuf;
       unless (defined $len) {
       }
    };
    my $module = "AnyEvent::Fork::RPC::" . ($arg{async} ? "Async" : "Sync");
-   $self->require ($module)
+   $self->eval ("use $module 2 ()")
-        ->send_arg ($function, $arg{init}, $serialiser)
+        ->send_arg (
+             function   => $function,
+             init       => $arg{init},
+             serialiser => $serialiser,
+             done       => $arg{done} || "$module\::do_exit",
+             rlen       => $arg{buflen_req} || $arg{buflen} || 512 - 16,
+             -10 # the above are 10 arguments
+          )
         ->run ("$module\::run", sub {
-      $fh = shift;
+      $fh = shift
+         or return $on_error->("connection failed");
       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;
 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>.
+Note that these functions are typically not yet declared when code is
+compiled into the child, because the backend module is only loaded when
+you call C<run>, which is typically the last method you call on the fork
+object.
+Therefore, you either have to explicitly pre-load the right backend module
+or mark calls to these functions as function calls, e.g.:
+   AnyEvent::Fork::RPC::event (0 => "five");
+   AnyEvent::Fork::RPC::event->(0 => "five");
+   &AnyEvent::Fork::RPC::flush;
 =over 4
-=item AnyEvent::Fork::RPC::event ...
+=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.
+Note: the event data, like any data send to the parent, might not be sent
+immediatelly but queued for later sending, so there is no guarantee that
+the event has been sent to the parent when the call returns - when you
+e.g. exit directly after calling this function, the parent might never
+receive the event. See the next function for a remedy.
+=item $success = AnyEvent::Fork::RPC::flush ()
+Synchronously wait and flush the reply data to the parent. Returns true on
+success and false otherwise (i.e. when the reply data cannot be written at
+all). Ignoring the success status is a common and healthy behaviour.
+Only the "async" backend does something on C<flush> - the "sync" backend
+is not buffering reply data and always returns true from this function.
+Normally, reply data might or might not be written to the parent
+immediatelly but is buffered. This can greatly improve performance and
+efficiency, but sometimes can get in your way: for example. when you want
+to send an error message just before exiting, or when you want to ensure
+replies timely reach the parent before starting a long blocking operation.
+In these cases, you can call this function to flush any outstanding reply
+data to the parent. This is done blockingly, so no requests will be
+handled and no event callbacks will be called.
+For example, you could wrap your request function in a C<eval> block and
+report the exception string back to the caller just before exiting:
+   sub req {
+      ...
+      eval {
+         ...
+      };
+      if ($@) {
+         AnyEvent::RPC::event (throw => "$@");
+         AnyEvent::RPC::flush ();
+         exit;
+      }
+      ...
+   }
+=back
+=head2 PROCESS EXIT
+If and when the child process exits depends on the backend and
+configuration. Apart from explicit exits (e.g. by calling C<exit>) or
+runtime conditions (uncaught exceptions, signals etc.), the backends exit
+under these conditions:
+=over 4
+=item Synchronous Backend
+The synchronous backend is very simple: when the process waits for another
+request to arrive and the writing side (usually in the parent) is closed,
+it will exit normally, i.e. as if your main program reached the end of the
+file.
+That means that if your parent process exits, the RPC process will usually
+exit as well, either because it is idle anyway, or because it executes a
+request. In the latter case, you will likely get an error when the RPc
+process tries to send the results to the parent (because agruably, you
+shouldn't exit your parent while there are still outstanding requests).
+The process is usually quiescent when it happens, so it should rarely be a
+problem, and C<END> handlers can be used to clean up.
+=item Asynchronous Backend
+For the asynchronous backend, things are more complicated: Whenever it
+listens for another request by the parent, it might detect that the socket
+was closed (e.g. because the parent exited). It will sotp listening for
+new requests and instead try to write out any remaining data (if any) or
+simply check whether the socket can be written to. After this, the RPC
+process is effectively done - no new requests are incoming, no outstanding
+request data can be written back.
+Since chances are high that there are event watchers that the RPC server
+knows nothing about (why else would one use the async backend if not for
+the ability to register watchers?), the event loop would often happily
+continue.
+This is why the asynchronous backend explicitly calls C<CORE::exit> when
+it is done (under other circumstances, such as when there is an I/O error
+and there is outstanding data to write, it will log a fatal message via
+L<AnyEvent::Log>, also causing the program to exit).
+You can override this by specifying a function name to call via the C<done>
+parameter instead.
 =back
 =head1 ADVANCED TOPICS
 are queued and the jobs are slow, they will all run concurrently. The
 child must implement some queueing/limiting mechanism if this causes
 problems. Alternatively, the parent could limit the amount of rpc calls
 that are outstanding.
-Blocking use of condvars is not supported.
+Blocking use of condvars is not supported (in the main thread, outside of
+e.g. L<Coro> threads).
 Using event-based modules such as L<IO::AIO>, L<Gtk2>, L<Tk> and so on is
 easy.
 =back
 half it has passed earlier.
 Here is some (untested) pseudocode to that effect:
    use AnyEvent::Util;
+   use AnyEvent::Fork;
    use AnyEvent::Fork::RPC;
    use IO::FDPass;
    my ($s1, $s2) = AnyEvent::Util::portable_socketpair;
 gory details.
 =head1 EXCEPTIONS
 There are no provisions whatsoever for catching exceptions at this time -
-in the child, exeptions might kill the process, causing calls to be lost
+in the child, exceptions might kill the process, causing calls to be lost
 and the parent encountering a fatal error. In the parent, exceptions in
 the result callback will not be caught and cause undefined behaviour.
 =head1 SEE ALSO
 L<AnyEvent::Fork>, to create the processes in the first place.
+L<AnyEvent::Fork::Remote>, likewise, but helpful for remote processes.
 L<AnyEvent::Fork::Pool>, to manage whole pools of processes.
 =head1 AUTHOR AND CONTACT INFORMATION
  Marc Lehmann <schmorp@schmorp.de>

Diff Legend

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

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents): Revision 1.25 by root, Sun Apr 28 14:17:59 2013 UTC vs. Revision 1.46 by root, Sun Sep 15 20:18:14 2019 UTC

Diff Legend

Comparing AnyEvent-Fork-RPC/RPC.pm (file contents):
Revision 1.25 by root, Sun Apr 28 14:17:59 2013 UTC vs.
Revision 1.46 by root, Sun Sep 15 20:18:14 2019 UTC