view release on metacpan or  search on metacpan

lib/Coro/ProcessPool.pm  view on Meta::CPAN

=head2 join

Cedes control to the event loop until the pool has completed all remaining
tasks and woken up any threads watching them.

=head2 defer

Queues a task to be processed by the pool. Tasks may specified in either of two
forms, as a code ref or the fully qualified name of a perl class which
implements two methods, C<new> and C<run>. Any remaining arguments to C<defer>
are passed unchanged to the code ref or the C<new> method of the task class.

C<defer> will immediately return an L<AnyEvent/condvar> that will wait for and
return the result of the task (or croak if the task generated an error).

  # Using a code ref
  my $cv = $pool->defer(\&func, $arg1, $arg2, $arg3);
  my $result = $cv->recv;

  # With a task class
  my $cv = $pool->defer('Some::Task::Class', $arg1, $arg2, $arg3);
  my $result = $cv->recv;

=head2 process

Calls defer and immediately calls C<recv> on the returned condvar, returning
the result. This is useful if your workflow includes multiple threads which
share the same pool. All arguments are passed unchanged to C<defer>.

=head2 map

Like perl's C<map>, applies a code ref to a list of arguments. This method will
cede until all results have been returned by the pool, returning the result as
a list. The order of arguments and results is preserved as expected.

  my @results = $pool->map(\&func, $arg1, $arg2, $arg3);

=head2 pipeline

Returns a L<Coro::ProcessPool::Pipeline> object which can be used to pipe
requests through to the process pool. Results then come out the other end of
the pipe, not necessarily in the order in which they were queued. It is up to
the calling code to perform task accounting (for example, by passing an id in
as one of the arguments to the task class).

  my $pipe = $pool->pipeline;

  my $producer = async {
    foreach my $args (@tasks) {
      $pipe->queue('Some::Class', $args);
    }

    $pipe->shutdown;
  };

  while (my $result = $pipe->next) {
    ...
  }

All arguments to C<pipeline()> are passed transparently to the constructor of
L<Coro::ProcessPool::Pipeline>. There is no limit to the number of pipelines
which may be created for a pool.

=head1 A NOTE ABOUT IMPORTS AND CLOSURES

Code refs are serialized using L<Data::Dump::Streamer>, allowing closed over
variables to be available to the code being called in the sub-process. Mutated
variables are I<not> updated when the result is returned.

See L<Data::Dump::Streamer/Caveats-Dumping-Closures-(CODE-Refs)> for important
notes regarding closures.

=head2 Use versus require

The C<use> pragma is run at compile time, whereas C<require> is evaluated at
runtime. Because of this, the use of C<use> in code passed directly to the
C<process> method can fail in the worker process because the C<use> statement
has already been evaluated in the parent process when the calling code was
compiled.

This will not work:

  $pool->process(sub {
    use Foo;
    my $foo = Foo->new();
  });

This will work:

  $pool->process(sub {
    require Foo;
    my $foo = Foo->new();
  });

If C<use> is necessary (for example, to import a method or transform the
calling code via import), it is recommended to move the code into its own
module (or to expliticly call require and import in the subroutine), which can
then be called in the anonymous routine:

  package Bar;

  use Foo;

  sub dostuff {
    ...
  }

Then, in your caller:

  $pool->process(sub {
    require Bar;
    Bar::dostuff();
  });

Alternately, a task class may be used if dependency management is causing a
headaches:

  my $result = $pool->process('Task::Class', @args);

=head1 COMPATIBILITY