File Coverage

blib/lib/Proc/Queue.pm

Criterion	Covered	Total	%
statement	161	227	70.9
branch	76	192	39.5
condition	27	66	40.9
subroutine	28	35	80.0
pod	15	19	78.9
total	307	539	56.9

package Proc::Queue;

require 5.006;

our $VERSION = '1.23';

use strict;
# use warnings;
require Exporter;
use Carp;
use POSIX ":sys_wait_h";
use Errno qw(ENOENT EPERM);

our @ISA = qw(Exporter);
our %EXPORT_TAGS = ( all => [ qw( fork_now
				  weight
				  waitpids
				  run_back
				  run_back_now
				  system_back
				  system_back_now
				  all_exit_ok
				  running_now ) ] );

our @EXPORT_OK = ( @{ $EXPORT_TAGS{'all'} } );
our @EXPORT = qw();

# are we running on Windows?
use constant _win => $^O=~/win32/i;

# parameters
my $queue_size=4; # max number of councurrent processes running.
my $debug=0; # shows debug info.
my $trace=0; # shows function calls.
my $delay=0; # delay between fork calls.
my $ignore_children=0; # similar to $SIG{CHILD}='IGNORE';

my $weight=1;
my $allow_excess=1;

my $last=0; # last time fork was called.

# module status
my $queue_now=0;
my %process;
my @captured;

# set STDERR as unbuffered so all the carp calls work as expected
{ my $oldfh=select STDERR; $|=1; select $oldfh }

# extended import to support parameter configuration from use statment
sub import {
  my ($pkg,@opts)=@_;
  my $i;
  for ($i=0; $i<@opts; $i++) {
    my $o=$opts[$i];
    if( $o eq 'size'
        or $o eq 'debug'
        or $o eq 'trace'
	or $o eq 'delay'
	or $o eq 'weight'
        or $o eq 'ignore_childs'
	or $o eq 'ignore_children'
        or $o eq 'allow_excess') {
      $#opts>$i or croak "option '$o' needs a value";
      my $value=$opts[$i+1];
      { no strict qw( subs refs );
	&$o($value) }
      splice @opts,$i--,2;
    }
  }
  carp "Exporting '".join("', '",@opts)."' symbols from Proc::Queue" if $debug;
  @_=($pkg,@opts);
  goto &Exporter::import;
}

sub delay {
  return $delay unless @_;
  my $old_delay=$delay;
  $delay=$_[0];
  carp "Proc queue delay set to ${delay}s, it was $old_delay" if $debug;
  $old_delay;
}

sub size {
  return $queue_size unless @_;
  my $size=shift;
  my $old_size=$queue_size;
  croak "invalid value for Proc::Queue size ($size), min value is 1"
    unless $size >= 1;
  $queue_size=$size;
  carp "Proc queue size set to $size, it was $old_size" if $debug;
  $old_size;
}

sub weight {
  return $weight unless @_;
  my $old_weight=$weight;
  croak "invalid value for Proc::Queue weight ($_[0]), min value is 1"
    unless int($_[0])>=1;
  $weight=int($_[0]);
  carp "Proc weight set to $weight, it was $old_weight" if $debug;
  $old_weight;
}

sub debug {
  return $debug unless @_;
  my $old_debug=$debug;
  if ($_[0]) {
    $debug=1;
    carp "debug mode ON";
  }
  else {
    $debug=0;
    carp "debug mode OFF" if $old_debug;
  }
  return $old_debug;
}

sub trace {
  return $trace unless @_;
  my $old_trace=$trace;
  if ($_[0]) {
    $trace=1;
    carp "trace mode ON" if $debug;
  }
  else {
    $trace=0;
    carp "trace mode OFF" if $debug;
  }
  return $old_trace;
}

sub  allow_excess {
  return $allow_excess unless @_;
  my $old_allow_excess=$allow_excess;
  if ($_[0]) {
    $allow_excess=1;
    carp "allow_excess mode ON" if $debug;
  }
  else {
    $allow_excess=0;
    carp "allow_excess mode OFF" if $debug;
  }
  $old_allow_excess;
}

sub ignore_children {
  return $ignore_children unless @_;
  my $old_ignore_children=$ignore_children;
  if ($_[0]) {
    $ignore_children=1;
    carp "ignore_children mode ON" if $debug;
  }
  else {
    $ignore_children=0;
    carp "ignore_children mode OFF" if $debug;
  }
  return $old_ignore_children;
}

*ignore_childs = \&ignore_children;

# sub to store internally captured processes
sub _push_captured {
  push @captured, shift,$? unless $ignore_children;
  croak "captured stack is corrupted" if (@captured & 1)
}

# do the real wait and housekeeping
sub _wait () {
  carp "Proc::Queue::_wait private function called" if $debug && $trace;
  carp "Waiting for child processes to exit" if $debug;
  my $w=CORE::wait;
  if ($w != -1) {
    if(exists $process{$w}) {
      $queue_now -= delete($process{$w});
      carp "Process $w has exited, $queue_now processes running now" if $debug;
    }
    else {
      carp "Unknow process $w has exited, ignoring it" if $debug;
    }
  }
  else {
    carp "No child processes left, continuing" if $debug;
  }
  return $w;
}

sub new_wait () {
  carp "Proc::Queue::wait called" if $trace;
  if(@captured) {
    my $w=shift @captured;
    $?=shift @captured;
    carp "Wait returning old child $w captured in wait" if $debug;
    return $w;
  }
  return _wait;
}

sub _waitpid ($$) {
  my ($pid,$flags)=@_;
  carp "Proc::Queue::_waitpid($pid,$flags) private function called" if $debug && $trace;
  carp "Waiting for child process $pid to exit" if $debug;
  my $w=CORE::waitpid($pid,$flags);
  if ($w != -1) {
    if(exists $process{$w}) {
      $queue_now -= delete($process{$w});
      carp "Process $w has exited, $queue_now processes running now" if $debug;
    }
    else {
      carp "Unknow process $w has exited, ignoring it" if $debug;
    }
  }
  else {
    carp "No child processes left, continuing" if $debug;
  }
  return $w;
}

sub _clean() {
  my $pid;
  while (1) {
    $pid=_waitpid(-1,WNOHANG);
    return unless ((_win && $pid < -1) || $pid >0);
    _push_captured $pid
  }
}

sub new_waitpid ($$) {
  my ($pid,$flags)=@_;
  carp "Proc::Queue::waitpid called" if $trace;
  foreach my $i (0..$#captured) {
    next if $i&1;
    my $r;
    if ($pid==$captured[$i] or $pid==-1) {
      croak "corrupted captured stack" unless ($#captured & 1);
      ($r,$?)=splice @captured,$i,2;
      return $r;
    }
  }
  return _waitpid($pid,$flags);
}

sub new_exit (;$ ) {
  my $e=@_?shift:0;
  carp "Proc::Queue::exit(".(defined($e)?$e:'undef').") called" if $trace;
  carp "Process $$ exiting with value ".(defined($e)?$e:'undef') if $debug;
  return CORE::exit($e);
}

# use Time::Hires::time if available;
BEGIN { eval "use Time::HiRes 'time'" }

sub _fork () {
  carp "Proc::Queue::_fork called" if $trace && $debug;
  if ($delay>0) {
    my $wait=$last+$delay - time;
    if ($wait>0) {
      carp "Delaying $wait seconds before forking" if $debug;
      select (undef,undef,undef,$wait);
    }
    $last=time;
  }
  my $f=CORE::fork;
  if (defined($f)) {
    if($f == 0) {
      carp "Process $$ now running" if $debug;
      # reset queue internal vars in child proccess;
      $queue_size=1;
      $queue_now=0;
      %process=();
      @captured=();
    }
    else {
      $process{$f}=$weight;
      $queue_now+=$weight;
      carp "Child forked (pid=$f), $queue_now processes running now" if $debug;
    }
  }
  else {
    carp "Fork failed: $!" if $debug;
  }
  return $f;
}

sub new_fork () {
  carp "Proc::Queue::fork called" if $trace;
  while($queue_now and
        $queue_now + ($allow_excess ? 1 : $weight) > $queue_size) {
    carp "Waiting for some process to finish" if $debug;
    my $nw;
    if (($nw=_wait) != -1) {
      _push_captured $nw;
    }
    else {
      carp "Proc queue seems to be corrupted, $queue_now children lost";
      last;
    }
  }
  return _fork;
}

sub fork_now () {
  carp "Proc::Queue::fork_now called" if $trace;
  return _fork;
}

sub waitpids {
  carp "Proc::Queue::waitpids(".join(", ",@_).")" if $trace;
  my @result;
  foreach my $pid (@_) {
    if (defined($pid)) {
      carp "Waiting for child $pid to exit" if $debug;
      my $r=new_waitpid($pid,0);
      if ((_win && $r < -1) || $r > 0) {
	carp "Child $r return $?" if $debug;
	push @result,$r,$?;
      }
      else {
	carp "No such child returned while waiting for $pid" if $debug;
	push @result,$r,undef;
      }
    }
    else {
      carp "Undef arg found";
      push @result,undef,undef
    };
  }
  return @result;
}


sub _run_back {
  my ($now,$code)=@_;
  carp "Proc::Queue::_run_back($now,$code) called" if $trace and $debug;
  my $f=$now ? fork_now : new_fork;
  if(defined($f) and $f==0) {
    carp "Running code $code in forked child $$" if $debug;
    $?=0;
    eval {
      &$code();
    };
    if ($@) {
      carp "Uncaught exception $@" if $debug;
      new_exit(255)
    }
    else {
      carp "Code $code in child $$ returns '$?'" if $? && $debug;
    }
    new_exit($?)
  }
  return $f;
}

sub run_back (&) {
  my $code=shift;
  carp "Proc::Queue::run_back($code) called" if $trace;
  return _run_back(0,$code)
}

sub run_back_now (&) {
  my $code=shift;
  carp "Proc::Queue::run_back_now($code) called" if $trace;
  return _run_back(1,$code)
}

sub _system_back {
  my $now=shift;
  carp "Proc::Queue::_system_back($now, ".join(", ",@_).") called" if $trace and $debug;
  if (0 and @_ > 1) { # TODO, search command on the PATH
      unless (-e $_[0]) {
          carp "command '$_[0]' not found" if $debug;
          $! = ENOENT;
          return undef;
      }
      unless (-f _ and -x _) {
          carp "permission to execute command '$_[0]' denied" if $debug;
          $! = EPERM;
          return undef;
      }
  }
  my $f=$now ? fork_now : new_fork;
  if(defined($f) and $f==0) {
    carp "Running exec(".join(", ",@_).") in forked child $$" if $debug;
    { exec(@_) }
    carp "exec(".join(", ",@_).") failed" if $debug;
    require POSIX;
    POSIX::_exit(255);
  }
  return $f;
}

sub system_back {
  carp "Proc::Queue::system_back(".join(", ",@_).") called" if $trace;
  return _system_back(0,@_);
}

sub system_back_now {
  carp "Proc::Queue::system_back_now(".join(", ",@_).") called" if $trace;
  return _system_back(1,@_);
}

sub all_exit_ok {
  carp "Proc::Queue::all_exit_ok(".join(", ",@_).")" if $trace;
  my @result=waitpids(@_);
  my $i;
  for($i=0; $i<@result; $i++) {
    next unless $i&1;
    if (!defined($result[$i]) or $result[$i]) {
      carp "Child ".$_[$i>>1]." fail with code $result[$i], waitpid return $result[$i-1]" if $debug;
      return 0;
    }
  }
  carp "All children run ok" if $debug;
  return 1;
}

# this function is mostly for testing pourposes:
sub running_now () {
  _clean;
  return $queue_now;
}

*CORE::GLOBAL::wait = \&new_wait;
*CORE::GLOBAL::waitpid = \&new_waitpid;
*CORE::GLOBAL::exit = \&new_exit;
*CORE::GLOBAL::fork = \&new_fork;


1;

__END__