File Coverage

blib/lib/Proc/Forkmap.pm

Criterion	Covered	Total	%
statement	66	118	55.9
branch	14	56	25.0
condition	5	15	33.3
subroutine	7	10	70.0
pod	2	2	100.0
total	94	201	46.7

package Proc::Forkmap;
use IO::Select;
use POSIX ":sys_wait_h";
use File::Temp qw(tempfile);
use strict;
use warnings;
use v5.10;
require Exporter;

our @ISA = qw(Exporter);
our @EXPORT = qw/
  forkmap_settings
  forkmap
/;

our $VERSION = '0.2305';

my $MAX_PROCS = 4;
my $TIMEOUT = 0;
my $IPC = 1;
my $TEMPFILE_DIR;

sub forkmap_settings {
  my %opts = @_;
  $MAX_PROCS = $opts{MAX_PROCS} if exists $opts{MAX_PROCS};
  $TIMEOUT = $opts{TIMEOUT} if exists $opts{TIMEOUT};
  $IPC = $opts{IPC} if exists $opts{IPC};
  $TEMPFILE_DIR = $opts{TEMPFILE_DIR} if exists $opts{TEMPFILE_DIR};
}

sub forkmap (&@) {
  my $fn = shift;
  my $cb = ref $_[-1] eq 'CODE' ? pop : undef;
  my @args = @_;

  unless ($IPC) {
    my $procs = 0;
    my $i = 0;

    while ($i <= $#args || $procs > 0) {
      while ($procs < $MAX_PROCS && $i <= $#args) {
        my $idx = $i;
        my $pid = fork();
        die "bad fork: $!" unless defined $pid;
        if ($pid == 0) {
          if ($TIMEOUT) {
            local $SIG{ALRM} = sub { die "timeout: pid [$$]" };
            alarm $TIMEOUT;
            eval {
              local $_ = $args[$idx];
              $fn->($_);
              alarm 0;
            };
            if ($@) {
              die "$@";
            }
          } else {
            local $_ = $args[$idx];
            $fn->($_);
          }
          $cb->($idx) if $cb;
          exit 0;
        }
        $procs++;
        $i++;
      }

      while (waitpid(-1, WNOHANG) > 0) {
        $procs--;
      }
      select undef, undef, undef, 0.1 if $procs >= $MAX_PROCS || $i > $#args;
    }
    return;
  }

  my $sel = IO::Select->new();
  my @res;
  my $procs = 0;
  my $i = 0;

  while ($i <= $#args || $procs > 0) {
    while ($procs < $MAX_PROCS && $i <= $#args) {
      pipe(my $r, my $w) or die "bad pipe: $!";
      my $idx = $i;
      my $pid = fork();
      die "bad fork: $!" unless defined $pid;
      if ($pid == 0) {
        close $r;
        my ($fh, $fname) = tempfile(DIR => $TEMPFILE_DIR, UNLINK => 0);
        if ($TIMEOUT) {
          local $SIG{ALRM} = sub { die "timeout: pid [$$] index [$idx]" };
          alarm $TIMEOUT;
          eval {
            local $_ = $args[$idx];
            my $t = $fn->($_);
            print $fh $t if defined $t;
            alarm 0;
          };
          if ($@) {
            close $fh if defined fileno $fh;
            unlink $fname if defined $fname;
            die "$@";
          }
        } else {
          eval {
            local $_ = $args[$idx];
            my $t = $fn->($_);
            print $fh $t if defined $t;
          };
          if ($@) {
            close $fh if defined fileno $fh;
            unlink $fname if defined $fname;
            die "$@";
          }
        }
        close $fh;
        print $w "$idx:$fname\n";
        close $w;
        exit 0;
      }
      close $w;
      $sel->add($r);
      $procs++;
      $i++;
    }

    for my $fh ($sel->can_read(0.1)) {
      my $line = <$fh>;
      if (defined $line) {
        chomp $line;
        my ($idx, $fname) = split /:/, $line, 2;
        open my $f, "<", $fname or do {
          warn "bad file [$fname]: $!";
          $res[$idx] = undef;
          next;
        };
        my $d = do {local $/; <$f>};
        close $f;
        unlink $fname;

        if ($cb) {
          $cb->($idx, $d);
        } else {
          $res[$idx] = $d;
        }
      } else {
        $sel->remove($fh);
        close $fh;
      }
    }
    while (waitpid(-1, WNOHANG) > 0) {
      $procs--;
    }
  }

  return $cb ? undef : @res;
}

1;

__END__