File Coverage

blib/lib/Parallel/ForkManager.pm

Criterion	Covered	Total	%
statement	129	152	84.8
branch	42	62	67.7
condition	18	33	54.5
subroutine	29	35	82.8
pod	14	21	66.6
total	232	303	76.5

package Parallel::ForkManager;
our $AUTHORITY = 'cpan:DLUX';
# ABSTRACT:  A simple parallel processing fork manager
$Parallel::ForkManager::VERSION = '2.01'; # TRIAL
use POSIX ":sys_wait_h";
use Storable ();
use File::Spec;
use File::Temp ();
use File::Path ();
use Carp;

use Parallel::ForkManager::Child;

use strict;

use Moo;

has max_proc => (
    is => 'ro',
    required => 1,
    writer => 'set_max_procs',
);

has processes => (
    is => 'ro',
    default => sub { {} },
);

has parent_pid => (
    is => 'ro',
    default => sub { $$ },
);

has auto_cleanup => (
    is => 'rw',
    default => sub { 1 },
);

has waitpid_blocking_sleep => (
    is =>'ro',
    writer => 'set_waitpid_blocking_sleep',
    default => sub { 1 },
);

has tempdir => (
    is => 'ro',
    default => sub {
        File::Temp::tempdir(CLEANUP => 0);
    },
    trigger => sub {
        my( $self, $dir ) = @_;

        die qq|Temporary directory "$dir" doesn't exist or is not a directory.|
            unless -d $dir;

        $self->auto_cleanup(0);
    },
);

has child_role => (
    is => 'ro',
    default => 'Parallel::ForkManager::Child',
);

sub is_child  { 0 }
sub is_parent { 1 }


sub BUILDARGS {
    my ( undef, @args ) = @_;
    my %args;

    if ( $args[0] =~ /^\d+$/ ) {
        $args{max_proc} = shift @args;
        $args{tempdir} = shift @args if @args;
    }
    else {
        %args = @args;
    }

    return \%args;
}

sub start {
  my ($s,$identification)=@_;

  while ($s->{max_proc} && ( keys %{ $s->{processes} } ) >= $s->{max_proc}) {
    $s->on_wait;
    $s->wait_one_child(defined $s->{on_wait_period} ? &WNOHANG : undef);
  };
  $s->wait_children;
  if ($s->{max_proc}) {
    my $pid=fork();
    die "Cannot fork: $!" if !defined $pid;
    if ($pid) {  # in parent
      $s->{processes}->{$pid}=$identification;
      $s->on_start($pid,$identification);
    } else {
      Role::Tiny->apply_roles_to_object( $s, $s->child_role );
    }
    return $pid;
  }

  # non-forking mode
  $s->{processes}->{$$}=$identification;
  $s->on_start($$,$identification);
  return 0; # Simulating the child which returns 0
}

sub start_child {
    my $self = shift;
    my $sub = pop;
    my $identification = shift;

    $self->start( $identification ) # in the parent
            # ... or the child
        or $self->finish( 0, $sub->() );
}


sub finish {
  my ($s, $x, $r)=@_;

  if ($s->{max_proc} == 0) { # nofork
    $s->on_finish($$, $x ,$s->{processes}->{$$}, 0, 0, $r);
    delete $s->{processes}->{$$};
  }

  return 0;
}

sub wait_children {
  my ($s)=@_;

  return if !keys %{$s->{processes}};
  my $kid;
  do {
    $kid = $s->wait_one_child(&WNOHANG);
  } while defined $kid and ( $kid > 0 or $kid < -1 ); # AS 5.6/Win32 returns negative PIDs
};

*wait_childs=*wait_children; # compatibility
*reap_finished_children=*wait_children; # behavioral synonym for clarity

# TODO document the method
sub retrieve {
    my( $self, $kid ) = @_;

    my $retrieved = undef;

    my $storable_tempfile = File::Spec->catfile($self->{tempdir}, 'Parallel-ForkManager-' . $self->{parent_pid} . '-' . $kid . '.txt');

    if (-e $storable_tempfile) {  # child has option of not storing anything, so we need to see if it did or not
      $retrieved = eval { Storable::retrieve($storable_tempfile) };

      # handle Storables errors
      if (not $retrieved or $@) {
        warn(qq|The storable module was unable to retrieve the child's data structure from the temporary file "$storable_tempfile":  | . join(', ', $@));
      }

      # clean up after ourselves
      unlink $storable_tempfile;
    }

    return $retrieved;
}

sub store {
    my( $self, $data ) = @_;

    return unless defined $data;

      my $storable_tempfile = File::Spec->catfile($self->{tempdir}, 'Parallel-ForkManager-' . $self->{parent_pid} . '-' . $$ . '.txt');
      my $stored = eval { return Storable::store($data, $storable_tempfile); };

      # handle Storables errors, IE logcarp or carp returning undef, or die (via logcroak or croak)
      if (not $stored or $@) {
        warn(qq|The storable module was unable to store the child's data structure to the temp file "$storable_tempfile":  | . join(', ', $@));
      }
}



sub wait_one_child {
  my ($s,$flag)=@_;

  my $kid;
  while (1) {
    $kid = $s->_waitpid($flag||=0);

    last unless defined $kid;

    last if $kid == 0 || $kid == -1; # AS 5.6/Win32 returns negative PIDs
    redo if !exists $s->{processes}->{$kid};
    my $id = delete $s->{processes}->{$kid};

    # retrieve child data structure, if any
    my $retrieved = $s->retrieve($kid);

    $s->on_finish( $kid, $? >> 8 , $id, $? & 0x7f, $? & 0x80 ? 1 : 0, $retrieved);
    last;
  }
  $kid;
};

sub wait_all_children {
  my ($s)=@_;

  while (keys %{ $s->{processes} }) {
    $s->on_wait;
    $s->wait_one_child(defined $s->{on_wait_period} ? &WNOHANG : undef);
  };
}

*wait_all_childs=*wait_all_children; # compatibility;

sub max_procs { $_[0]->max_proc }

sub running_procs {
    my $self = shift;

    my @pids = keys %{ $self->{processes} };
    return @pids;
}

sub wait_for_available_procs {
    my( $self, $nbr ) = @_;
    $nbr ||= 1;

    croak "number of processes '$nbr' higher than the max number of processes (@{[ $self->max_procs ]})"
        if $nbr > $self->max_procs;

    $self->wait_one_child until $self->max_procs - $self->running_procs >= $nbr;
}

sub run_on_finish {
  my ($s,$code,$pid)=@_;

  $s->{on_finish}->{$pid || 0}=$code;
}

sub on_finish {
  my ($s,$pid,@par)=@_;

  my $code=$s->{on_finish}->{$pid} || $s->{on_finish}->{0} or return 0;
  $code->($pid,@par);
};

sub run_on_wait {
  my ($s,$code, $period)=@_;

  $s->{on_wait}=$code;
  $s->{on_wait_period} = $period;
}

sub on_wait {
  my ($s)=@_;

  if(ref($s->{on_wait}) eq 'CODE') {
    $s->{on_wait}->();
    if (defined $s->{on_wait_period}) {
        local $SIG{CHLD} = sub { } if ! defined $SIG{CHLD};
        select undef, undef, undef, $s->{on_wait_period}
    };
  };
};

sub run_on_start {
  my ($s,$code)=@_;

  $s->{on_start}=$code;
}

sub on_start {
  my ($s,@par)=@_;

  $s->{on_start}->(@par) if ref($s->{on_start}) eq 'CODE';
};

sub _waitpid { # Call waitpid() in the standard Unix fashion.
    my( $self, $flag ) = @_;

    return $flag ? $self->_waitpid_non_blocking : $self->_waitpid_blocking;
}

sub _waitpid_non_blocking {
    my $self = shift;

    for my $pid ( $self->running_procs ) {
        my $p = waitpid $pid, &WNOHANG or next;

        return $pid if $p != -1;

        warn "child process '$pid' disappeared. A call to `waitpid` outside of Parallel::ForkManager might have reaped it.\n";
        # it's gone. let's clean the process entry
        delete $self->{processes}{$pid};
    }

    return;
}

sub _waitpid_blocking {
    my $self = shift;

    # pseudo-blocking
    if( my $sleep_period = $self->{waitpid_blocking_sleep} ) {
        while() {
            my $pid = $self->_waitpid_non_blocking;

            return $pid if defined $pid;

            return unless $self->running_procs;

            select undef, undef, undef, $sleep_period;
        }
    }

    return waitpid -1, 0;
}

sub DEMOLISH {
    my $self = shift;

    no warnings 'uninitialized';

    File::Path::remove_tree($self->{tempdir})
        if $self->{auto_cleanup} and $self->{parent_pid} == $$ and -d $self->{tempdir};
}

1;

__END__