File Coverage

blib/lib/AnyEvent/ProcessPool.pm

Criterion	Covered	Total	%
statement	72	73	98.6
branch	13	18	72.2
condition	5	6	83.3
subroutine	13	13	100.0
pod	2	4	50.0
total	105	114	92.1

package AnyEvent::ProcessPool;
# ABSTRACT: Asynchronously runs code concurrently in a pool of perl processes
$AnyEvent::ProcessPool::VERSION = '0.06_001'; # TRIAL

$AnyEvent::ProcessPool::VERSION = '0.06001';use common::sense;
use Carp;
use AnyEvent;
use AnyEvent::Util;
use AnyEvent::ProcessPool::Process;
use AnyEvent::ProcessPool::Task;
use AnyEvent::ProcessPool::Util qw(next_id cpu_count);

sub new {
  my ($class, %param) = @_;

  my $self = bless {
    workers  => $param{workers} || cpu_count,
    limit    => $param{limit},
    include  => $param{include},
    pid      => $$,
    pool     => [], # array of AE::PP::Process objects
    queue    => [], # array of [id, code] tasks
    complete => {}, # task_id => condvar: signals result to caller
    pending  => {}, # task_id => condvar: signals result internally
  }, $class;

  # Initialize workers but do not yet wait for them to be started
  if ($self->{limit}) {
    $AnyEvent::Util::MAX_FORKS = $self->{limit};
  }

  foreach (1 .. $self->{workers}) {
    my $worker = AnyEvent::ProcessPool::Process->new(limit => $self->{limit});
    push @{$self->{pool}}, $worker;
  }

  return $self;
}

sub join {
  my $self = shift;
  foreach my $task_id (keys %{$self->{complete}}) {
    if (my $cv = $self->{complete}{$task_id}) {
      $cv->recv;
    }
  }
}

sub DESTROY {
  my ($self, $global) = @_;
  return unless $self;
  return unless $self->{pid} == $$;

  # Unblock watchers for any remaining pending tasks
  if (ref $self->{pending}) {
    foreach my $cv (values %{$self->{pending}}) {
      $cv->croak('AnyEvent::ProcessPool destroyed with pending tasks remaining');
    }
  }

  # Terminate any workers still alive
  if (ref $self->{pool}) {
    foreach my $worker (@{$self->{pool}}) {
      $worker->stop if $worker;
    }
  }
}

sub async {
  my $self = shift;
  my $code = shift;
  my $id   = next_id;
  my $task = AnyEvent::ProcessPool::Task->new($code, [@_]);
  $self->{complete}{$id} = AE::cv;
  push @{$self->{queue}}, [$id, $task];
  $self->process_queue;
  return $self->{complete}{$id};
}

sub process_queue {
  my $self  = shift;
  my $queue = $self->{queue};
  my $pool  = $self->{pool};

  if (@$queue && @$pool) {
    my ($id, $task) = @{shift @$queue};
    my $worker = shift @$pool;

    $self->{pending}{$id} = $worker->run($task);

    $self->{pending}{$id}->cb(sub{
      my $task = shift->recv;

      if ($task->failed) {
        $self->{complete}{$id}->croak($task->result);
      } else {
        $self->{complete}{$id}->send($task->result);
      }

      delete $self->{pending}{$id};
      delete $self->{complete}{$id};

      push @{$self->{pool}}, $worker;
      $self->process_queue;
    });
  }
}

1;

__END__