/ 
IO-Async-0.10
River stage three • 121 direct dependents • 154 total dependents
/ IO::Async::DetachedCode

NAME

IO::Async::DetachedCode - a class that allows a block of code to execute asynchronously in detached child processes

SYNOPSIS

Usually this object would be constructed indirectly, via an IO::Async::Set:

use IO::Async::Set::...;
my $set = IO::Async::Set::...

$set->enable_childmanager;

my $code = $set->detach_code(
   code => sub {
      my ( $number ) = @_;
      return is_prime( $number );
   }
);

$code->call(
   args => [ 123454321 ],
   on_return => sub {
      my $isprime = shift;
      print "123454321 " . ( $isprime ? "is" : "is not" ) . " a prime number\n";
   },
   on_error => sub {
      print STDERR "Cannot determine if it's prime - $_[0]\n";
   },
);

$set->loop_forever;

It can also be used directly. In this case, extra effort must be taken to pass an IO::Async::Set object:

my $set = IO::Async::Set::...

my $code = IO::Async::DetachedCode->new(
   set => $set,
   code => sub { ... },
);

DESCRIPTION

This module provides a class that allows a block of code to "detach" from the main process, and execute independently in its own child processes. The object itself acts as a proxy to this code block, allowing arguments to be passed to it each time it is called, and returning results back to a callback function in the main process.

The object represents the code block itself, rather than one specific invocation of it. It can be called multiple times, by the call() method. Multiple outstanding invocations can be called; they will be executed in the order they were queued. If only one worker process is used then results will be returned in the order they were called. If multiple are used, then each request will be sent in the order called, but timing differences between each worker may mean results are returned in a different order.

The default marshalling code can only cope with plain scalars or undef values; no references, objects, or IO handles may be passed to the function each time it is called. If references are required then code based on Storable may be used instead to pass these. See the documentation on the marshaller parameter of new() method.

The IO::Async framework generally provides mechanisms for multiplexing IO tasks between different handles, so there aren't many occasions when such detached code is necessary. Two cases where this does become useful are:

  1. When a large amount of computationally-intensive work needs to be performed (for example, the is_prime() test in the example in the SYNOPSIS).

  2. When a blocking OS syscall or library-level function needs to be called, and no nonblocking or asynchronous version is supplied.

CONSTRUCTOR

$code = IO::Async::DetachedCode->new( %params )

This function returns a new instance of a IO::Async::DetachedCode object. The %params hash takes the following keys:

set => IO::Async::Set

A reference to an IO::Async::Set object. The set must have the child manager enabled.

code => CODE

A block of code to call in the child process. It will be invoked in list context each time the call() method is is called, passing in the arguments given. The result will be given to the on_result or on_return callback provided to the call() method.

stream => STRING: socket or pipe

Optional string, specifies which sort of stream will be used to attach to each worker. socket uses only one file descriptor per worker in the parent process, but not all systems may be able to use it. If the system does not allow PF_UNIX socket pairs, then pipe can be used instead. This will use two file descriptors per worker in the parent process, however.

If not supplied, the socket method is used.

marshaller => STRING: flat or storable

Optional string, specifies the way that call arguments and return values are marshalled over the stream that connects the worker and parent processes. The flat method is small, simple and fast, but can only cope with strings or undef; cannot cope with any references. The storable method uses the Storable module to marshall arbitrary reference structures.

If not supplied, the flat method is used.

workers => INT

Optional integer, specifies the number of parallel workers to create.

If not supplied, 1 is used.

exit_on_die => BOOL

Optional boolean, controls what happens after the code throws an exception. If missing or false, the worker will continue running to process more requests. If true, the worker will be shut down. A new worker might be constructed by the call method to replace it, if necessary.

Since the code block will be called multiple times within the same child process, it must take care not to modify any of its state that might affect subsequent calls. Since it executes in a child process, it cannot make any modifications to the state of the parent program. Therefore, all the data required to perform its task must be represented in the call arguments, and all of the result must be represented in the return values.

METHODS

$code->call( %params )

This method causes one invocation of the code block to be executed in a free worker. If there are no free workers available at the time this method is called, the request will be queued, to be sent to the first worker that later becomes available. The request will already have been serialised by the marshaller, so it will be safe to modify any referenced data structures in the arguments after this call returns.

If the number of available workers is less than the number supplied to the constructor (perhaps because some of them were shut down because of exit_on_die) and they are all busy, then a new one will be created to perform this request.

The %params hash takes the following keys:

args => ARRAY

A reference to the array of arguments to pass to the code.

on_result => CODE

A callback that is invoked when the code has been executed. If the code returned normally, it is called as:

$on_result->( 'return', @values )

If the code threw an exception, or some other error occured such as a closed connection or the process died, it is called as:

$on_result->( 'error', $exception_name )

or

on_return => CODE and on_error => CODE

Two callbacks to use in either of the circumstances given above. They will be called directly, without the leading 'return' or 'error' value.

The args key must always be supplied. Either the on_result or both the on_return and on_error keys must also be supplied.

$code->shutdown

This method requests that the detached worker processes stop running. All pending calls to the code are finished with a 'shutdown' error, and the worker processes exit.

$n_workers = $code->workers

This method in scalar context returns the number of workers currently running.

@worker_pids = $code->workers

This method in list context returns a list of the PID numbers of all the currently running worker processes.

TODO

  • Allow other argument/return value marshalling code - perhaps an arbitrary object.

  • Dynamic pooling of multiple worker processes, with min/max watermarks.

  • Fall back on a pipe pair if socketpair doesn't work.

NOTES

For the record, 123454321 is 11111 * 11111, a square number, and therefore not prime.

AUTHOR

Paul Evans <leonerd@leonerd.org.uk>