view release on metacpan or  search on metacpan

lib/Analizo/Batch/Runner/Parallel.pm  view on Meta::CPAN

package Analizo::Batch::Runner::Parallel;
use strict;
use warnings;
use ZMQ::FFI qw(ZMQ_PUSH ZMQ_PULL ZMQ_REQ ZMQ_REP);
use YAML::XS;

use parent qw( Analizo::Batch::Runner );

$YAML::XS::LoadBlessed = 1;
$YAML::XS::LoadCode = 1;
$YAML::XS::DumpCode = 1;

sub new {
  my ($class, $parallelism) = @_;
  $parallelism ||= 2;
  $class->SUPER::new(parallelism => $parallelism);
}

sub parallelism {
  my ($self) = @_;
  return $self->{parallelism};
}

sub actually_run {
  my ($self, $batch, $output) = @_;
  $self->start_workers();
  $self->coordinate_workers($batch, $output);
  $self->wait_for_workers();
}

sub _socket_spec {
  my ($name, $ppid) = @_;
  return "ipc:///tmp/.analizo-$name-$ppid";
}

sub start_workers {
  my ($self) = @_;
  $self->{workers} = [];
  my $n = $self->parallelism();
  my $ppid = $$;
  for my $i (1..$n) {
    my $pid = fork();
    if ($pid) {
      # on parent
      push(@{$self->{workers}}, $pid);
    } else {
      # on child
      $0 = '[analizo worker]';
      worker($ppid);
      exit();
    }
  }
  my $distributor_pid = fork();
  if ($distributor_pid) {
    push(@{$self->{workers}}, $distributor_pid);
  } else {
    $0 = '[analizo queue]';
    distributor($ppid, $n);
    exit();
  }
}

sub wait_for_workers {
  my ($self) = @_;
  for my $pid (@{$self->{workers}}) {
    waitpid($pid, 0);
  }
}

sub coordinate_workers {
  my ($self, $batch, $output) = @_;

  my $context = ZMQ::FFI->new();

  my $queue = $context->socket(ZMQ_PUSH);
  $queue->bind(_socket_spec('queue', $$));