File Coverage

blib/lib/Proc/Forkmap.pm

Criterion	Covered	Total	%
statement	20	124	16.1
branch	0	30	0.0
condition	0	16	0.0
subroutine	7	18	38.8
pod	7	7	100.0
total	34	195	17.4

package Proc::Forkmap;
use POSIX qw(:sys_wait_h);
use Proc::Fork;
use Carp;
use IO::Pipe;
use strict;
use warnings;
use 5.010;
require Exporter;
our @ISA = qw(Exporter);
our @EXPORT_OK = qw(forkmap);

our $MAX_PROC = 4;
our $VERSION = '0.2008';
our $TIMEOUT = 0;

sub new {
  my $class = shift;
  my $self = bless {@_}, $class;
  $self->_init;
  return $self;
}

sub _init {
  my $self = shift;
  $self->{max_kids} //= 4;
  $self->{ipc} //= 0;
  $self->{non_blocking} //= 1;
  $self->{timeout} //= 0;
}

sub max_kids {
  my ($self, $n) = @_;
  $n // return $self->{max_kids};
  $self->{max_kids} = $n;
}

sub non_blocking {
  my ($self, $n) = @_;
  $n // return $self->{non_blocking};
  $self->{non_blocking} = $n;
}

sub ipc {
  my ($self, $n) = @_;
  $n // return $self->{ipc};
  $self->{ipc} = $n;
}

sub timeout {
  my ($self, $n) = @_;
  $n // return $self->{timeout};
  $self->{timeout} = $n;
}

sub fmap {
  my ($self, $code) = (shift, shift);
  my %pids = ();
  my @rs = ();  # result set of child return values
  my $max = $self->max_kids;
  my $ipc = $self->ipc;
  my $timeout = $self->timeout;
  my $non_blocking = $self->non_blocking;
  for my $proc (@_) {
    my $pipe = $ipc ? IO::Pipe->new : {};
    # max kids?
    while ($max == keys %pids) {
      # free a spot in queue when a process completes
      for my $pid (keys %pids) {
        if (my $kid = waitpid($pid, WNOHANG)) {
          delete $pids{$kid};
          last;
        }
      }
    }
    my $fn = sub {
      my $rs = shift;
      if ($ipc) {
        $pipe->writer();
        $pipe->autoflush;
        print $pipe $rs if defined $rs;
      }
      return 1;
    };
    run_fork {  # processes fork here
      parent {
        my $kid = shift;
        $pids{$kid}++;
        if ($ipc) {
          $pipe->reader();
          if ($non_blocking) {
            $pipe->blocking(0);
          } else {
            $pipe->blocking(1);
          }
          while(<$pipe>) {
            push @rs, $_;
          }
        }
      }
      child {
        if (!$timeout) {
          my $rs = $code->($proc);
          $fn->($rs);
          exit;
        }
        eval {
          local $SIG{ALRM} = sub {
            die "alarm\n"
          };
          alarm $timeout;
          my $rs = $code->($proc);
          $fn->($rs);
          alarm 0;
        };
        if ($@) {
          if ($@ eq "alarm\n") {
            print STDERR "error: timeout $$\n";
          }
          die unless $@ eq "alarm\n";
        }
        exit;
      }
      error {
        die "error: couldn't fork";
      }
    };
  }
  1 while (wait() != -1);  # wait for the stragglers to finish
  return @rs;
}

sub forkmap (&@) {
  my ($code, @args) = @_;
  my %pids = ();
  my $max = $MAX_PROC;
  for my $proc (@args) {
    while ($max == keys %pids) {
      for my $pid (keys %pids) {
        if (my $kid = waitpid($pid, WNOHANG)) {
          delete $pids{$kid};
          last;
        }
      }
    }
    run_fork {
      parent {
        my $kid = shift;
        $pids{$kid}++;
      }
      child {
        if (!$TIMEOUT) {
          local $_ = $proc;
          $code->();
          exit;
        }
        eval {
          local $SIG{ALRM} = sub {
            die "alarm\n"
          };
          alarm $TIMEOUT;
          local $_ = $proc;
          $code->();
          alarm 0;
        };
        if ($@) {
          if ($@ eq "alarm\n") {
            print STDERR "error: timeout $$\n";
          }
          die unless $@ eq "alarm\n";
        }
        exit;
      }
      error {
        die "error: couldn't fork";
      }
    };
  }
  1 while (wait() != -1);
  return 1;
}

1;

__END__