Forks::Super::Job - object representing a background task

    use Forks::Super;

    $pid = Forks::Super::fork( \%options );  # see Forks::Super
    $job = Forks::Super::Job::get($pid);
    $job = Forks::Super::Job::getByName($name);

    print "Current job state is $job->{state}\n";
    print "Job was created at ", scalar localtime($job->{created}), "\n";

    use Forks::Super 'overload';
    $job = Forks::Super::fork( \%options );
    print "Process id of new job is $job\n";
    print "Current state is ", $job->state, "\n";
    waitpid $job, 0;
    print "Exit status was ", $job->status, "\n";

    my ($longest_job, $longest_time) = (-1, -1);
    foreach $job (@Forks::Super::ALL_JOBS) {
        if ($job->is_complete) {
            $job_time = $job->{end} - $job->{start};
            if ($job_time > $longest_time) {
                ($longest_job, $longest_time) = ($job, $job_time);
            }
        }
    }
    print STDERR "The job that took the longest was $job: ${job_time}s\n";

    $job_or_pid = fork { %options };

    if ($job_or_pid != $another_pid) { ... }
    kill 'TERM', $job_or_pid;

    $job_or_pid->{real_pid}
    $job_or_pid->suspend

    my $job = fork { cmd => $command };
    while (<$job>) {
        print "Output from $job: $_\n";
    }

#
# Forks::Super::Job - object representing a task to perform in
#                     a background process
# See the subpackages for some implementation details
#

package Forks::Super::Job;
use Forks::Super::Debug qw(debug);
use Forks::Super::Util qw(is_number qualify_sub_name IS_WIN32 is_pipe);
use Forks::Super::Config qw(:all);
use Forks::Super::Job::Ipc;   # does windows prefer to load Ipc before Timeout?
use Forks::Super::Job::Timeout;
use Forks::Super::Queue qw(queue_job);
use Forks::Super::Job::OS;
use Forks::Super::Job::Callback qw(run_callback);
use Signals::XSIG;
use Exporter;
use POSIX ':sys_wait_h';
use Carp;
use IO::Handle;
use strict;
use warnings;

our @ISA = qw(Exporter);
our @EXPORT = qw(@ALL_JOBS %ALL_JOBS);
our $VERSION = '0.52';

our (@ALL_JOBS, %ALL_JOBS, $WIN32_PROC, $WIN32_PROC_PID);
our $OVERLOAD_ENABLED = 0;
our $INSIDE_END_QUEUE = 0;
our $_RETRY_PAUSE;

my $use_overload = $ENV{FORKS_SUPER_JOB_OVERLOAD};
if (!defined($use_overload)) {
  $use_overload = 1;
}
if ($use_overload) {
  enable_overload();
} else {
  enable_overload();
  disable_overload();
}

#############################################################################
# Object methods (meant to be called as $job->xxx(@args))

sub new {
  my ($class, $opts) = @_;
  my $this = {};
  if (ref $opts eq 'HASH') {
    $this->{$_} = $opts->{$_} foreach keys %$opts;
  }

  $this->{__opts__} = $opts;

  $this->{created} = Time::HiRes::time();
  $this->{state} = 'NEW';
  $this->{ppid} = $$;
  if (!defined $this->{_is_bg}) {
    $this->{_is_bg} = 0;
  }
  if (!defined $this->{debug}) {
    $this->{debug} = $Forks::Super::Debug::DEBUG;
  }
  # 0.41: fix overload bug here by putting  bless  before  push @ALL_JOBS
  bless $this, 'Forks::Super::Job';
  push @ALL_JOBS, $this;
  if ($this->{debug}) {
    debug("New job created: ", $this->toString());
  }
  return $this;
}

sub reuse {
  my ($job, $opts) = @_;
  if (ref $opts ne 'HASH') {
    $opts = { @_[1..$#_] };
  }
  my %opts;
  if (defined $job->{__opts__}) {
    %opts = %{$job->{__opts__}};
  }
  for (keys %$opts) {
    $opts{$_} = $opts->{$_};
  }

  return Forks::Super::fork( \%opts ) ;
}

sub is_complete {
  my $job = shift;
  return defined($job->{state}) &&
    ($job->{state} eq 'COMPLETE' || $job->{state} eq 'REAPED');
}

sub is_started {
  my $job = shift;
  return $job->is_complete || $job->is_active || 
    (defined($job->{state}) && $job->{state} eq 'SUSPENDED');
}

sub is_active {
  my $job = shift;
  return defined($job->{state}) && $job->{state} eq 'ACTIVE';
}

sub is_suspended {
  my $job = shift;
  return defined($job->{state}) && $job->{state} =~ /SUSPENDED/;
}

sub is_deferred {
  my $job = shift;
  return defined($job->{state}) && $job->{state} =~ /DEFERRED/;
}

sub waitpid {
  my ($job, $flags, $timeout) = @_;
  return Forks::Super::Wait::waitpid($job->{pid}, $flags, $timeout || 0);
}

sub wait {
  my ($job, $timeout) = @_;
  if (defined($timeout) && $timeout == 0) { # ZZZ
    return Forks::Super::Wait::waitpid($job->{pid}, &WNOHANG);
  }
  return Forks::Super::Wait::waitpid($job->{pid}, 0, $timeout || 0);
}

sub kill {
  my ($job, $signal) = @_;
  if (!defined($signal) || $signal eq '') {
    $signal = Forks::Super::Util::signal_number('INT') || 1;
  }
  return Forks::Super::kill($signal, $job);
}

sub state {
  my $job = shift;
  return $job->{state};
}

sub status {
  my $job = shift;
  return $job->{status};  # may be undefined
}

#
# Produces string representation of a Forks::Super::Job object.
#
sub toString {
  my $job = shift;
  my @to_display = qw(pid state create);
  foreach my $attr (qw(real_pid style cmd exec sub args dir start end reaped
		       status closure pgid child_fh queue_priority
		       timeout expiration)) {
    push @to_display, $attr if defined $job->{$attr};
  }
  my @output = ();
  foreach my $attr (@to_display) {
    next unless defined $job->{$attr};
    if (ref $job->{$attr} eq 'ARRAY') {
      push @output, "$attr=[" . join(q{,},@{$job->{$attr}}) . ']';
    } else {
      push @output, "$attr=" . $job->{$attr};
    }
  }
  return '{' . join ( ';' , @output), '}';
}

sub toFullString {
  my $job = shift;
  my @output = ();
  foreach my $attr (sort keys %$job) {
    next unless defined $job->{$attr};
    if (ref $job->{$attr} eq 'ARRAY') {
      push @output, "$attr=[" . join(',', @{$job->{$attr}}) . ']';
    } elsif (ref $job->{$attr} eq 'HASH') {
      push @output, "$attr={", 
	join(',', map {"$_=>$job->{$attr}{$_}"
		     } sort keys %{$job->{$attr}}), '}';
    } else {
      push @output, "$attr=$job->{$attr}";
    }
  }
  return '{' . join(';', @output), '}';
}

sub toShortString {
  my $job = shift;
  if (defined $job->{short_string}) {
    return $job->{short_string};
  }
  my @to_display = ();
  foreach my $attr (qw(pid state cmd exec sub args closure real_pid)) {
    push @to_display, $attr if defined $job->{$attr};
  }
  my @output;
  foreach my $attr (@to_display) {
    if (ref $job->{$attr} eq 'ARRAY') {
      push @output, "$attr=[" . join(",", @{$job->{$attr}}) . "]";
    } else {
      push @output, "$attr=" . $job->{$attr};
    }
  }
  return $job->{short_string} = "{" . join(";",@output) . "}";
}

sub _mark_complete {
  my $job = shift;
  $job->{end} = Time::HiRes::time();
  $job->{state} = 'COMPLETE';

  $job->run_callback('collect');
  $job->run_callback('finish');
  return;
}

sub _mark_reaped {
  my $job = shift;
  $job->{state} = 'REAPED';
  $job->{reaped} = Time::HiRes::time();
  $? = $job->{status};
  debug("Job $job->{pid} reaped") if $job->{debug};
  return;
}

#
# determine whether a job is eligible to start
#
sub can_launch {
  no strict 'refs';

  my $job = shift;
  $job->{last_check} = Time::HiRes::time();
  if (defined $job->{can_launch}) {
    if (ref $job->{can_launch} eq 'CODE') {
      return $job->{can_launch}->($job);
    } elsif (ref $job->{can_launch} eq '') {
      my $can_launch_sub = $job->{can_launch};
      return $can_launch_sub->($job);
    }
  } else {
    return $job->_can_launch;
  }
}

sub _can_launch_delayed_start_check {
  my $job = shift;
  return 1 if !defined($job->{start_after}) ||
    Time::HiRes::time() >= $job->{start_after};

  debug('Forks::Super::Job::_can_launch(): ',
	'start delay requested. launch fail') if $job->{debug};

  # delay option should normally be associated with queue on busy behavior.
  # any reason not to make this the default ?
  #  delay + fail   is pretty dumb
  #  delay + block  is like sleep + fork

  $job->{_on_busy} = 'QUEUE' if not defined $job->{on_busy};
  return 0;
}

sub _can_launch_dependency_check {
  my $job = shift;
  my @dep_on = defined($job->{depend_on}) ? @{$job->{depend_on}} : ();
  my @dep_start = defined($job->{depend_start}) ? @{$job->{depend_start}} : ();

  foreach my $dj (@dep_on) {
    my $j = $ALL_JOBS{$dj};
    if (not defined $j) {
      carp "Forks::Super::Job: ",
	"dependency $dj for job $job->{pid} is invalid. Ignoring.\n";
      next;
    }
    unless ($j->is_complete) {
      debug('Forks::Super::Job::_can_launch(): ',
	"job waiting for job $j->{pid} to finish. launch fail.")
	if $j->{debug};
      return 0;
    }
  }

  foreach my $dj (@dep_start) {
    my $j = $ALL_JOBS{$dj};
    if (not defined $j) {
      carp "Forks::Super::Job ",
	"start dependency $dj for job $job->{pid} is invalid. Ignoring.\n";
      next;
    }
    unless ($j->is_started) {
      debug('Forks::Super::Job::_can_launch(): ',
	"job waiting for job $j->{pid} to start. launch fail.")
	if $j->{debug};
      return 0;
    }
  }
  return 1;
}

#
# default function for determining whether the system
# is too busy to create a new child process or not
#
sub _can_launch {
  no warnings qw(once);

  my $job = shift;
  my $max_proc = defined($job->{max_proc})
    ? $job->{max_proc} : $Forks::Super::MAX_PROC;
  my $max_load = defined($job->{max_load})
    ? $job->{max_load} : $Forks::Super::MAX_LOAD;
  my $force = defined($job->{max_load}) && $job->{force};

  if ($force) {
    debug('Forks::Super::Job::_can_launch(): force attr set. launch ok')
      if $job->{debug};
    return 1;
  }

  return 0 if not $job->_can_launch_delayed_start_check;
  return 0 if not $job->_can_launch_dependency_check;

  if ($max_proc > 0) {
    my $num_active = count_active_processes();
    if ($num_active >= $max_proc) {
      debug('Forks::Super::Job::_can_launch(): ',
	"active jobs $num_active exceeds limit $max_proc. ",
	    'launch fail.') if $job->{debug};
      return 0;
    }
  }

  if ($max_load > 0) {
    my $load = get_cpu_load();
    if ($load > $max_load) {
      debug('Forks::Super::Job::_can_launch(): ',
	"cpu load $load exceeds limit $max_load. launch fail.")
	if $job->{debug};
      return 0;
    }
  }

  debug('Forks::Super::Job::_can_launch(): system not busy. launch ok.')
    if $job->{debug};
  return 1;
}

# Perl system fork() call. Encapsulated here so it can be overridden 
# and mocked for testing. See t/17-retries.t
sub _CORE_fork { return CORE::fork }

#
# make a system fork call and configure the job object
# in the parent and the child processes
#
sub launch {
  my $job = shift;
  if ($job->is_started) {
    Carp::confess "Forks::Super::Job::launch() ",
	"called on a job in state $job->{state}!\n";
  }

  if ($$ != $Forks::Super::MAIN_PID && $Forks::Super::CHILD_FORK_OK > 0) {
    $Forks::Super::MAIN_PID = $$;
    $Forks::Super::CHILD_FORK_OK--;
  }

  if ($$ != $Forks::Super::MAIN_PID && $Forks::Super::CHILD_FORK_OK < 1) {
    return _launch_from_child($job);
  }
  $job->_preconfig_fh;
  $job->_preconfig2;





  my $retries = $job->{retries} || 0;



  my $pid = _CORE_fork();
  while (!defined($pid) && $retries-- > 0) {
    warn "Forks::Super::launch: ",
      "system fork call returned undef. Retrying ...\n";
    $_RETRY_PAUSE ||= 1.0;
    my $delay = 1.0 + $_RETRY_PAUSE * (($job->{retries} || 1) - $retries);
    Forks::Super::Util::pause($delay);
    $pid = _CORE_fork();
  }







  if (!defined $pid) {
    debug('Forks::Super::Job::launch(): CORE::fork() returned undefined!')
      if $job->{debug};
    return;
  }


  if (Forks::Super::Util::isValidPid($pid)) {

    # parent
    return _postlaunch_parent($pid, $job);

  } elsif ($pid == 0) {

    _postlaunch_child($job);
    return 0;

  } else {

    Carp::confess "Forks::Super::launch(): ",
	"Somehow we got invalid pid=$pid from fork call.";
    return;

  }
}

sub _postlaunch_parent {
  my ($pid, $job) = @_;
  $ALL_JOBS{$pid} = $job;
  if (defined($job->{state}) &&
      $job->{state} ne 'NEW' &&
      $job->{state} ne 'LAUNCHING' &&
      $job->{state} ne 'DEFERRED') {
    warn "Forks::Super::Job::launch(): ",
      "job $pid already has state: $job->{state}\n";
  } else {
    $job->{state} = 'ACTIVE';

    #
    # it is possible that this child exited quickly and has already
    # been reaped in the SIGCHLD handler. In that case, the signal
    # handler should have made an entry in %Forks::Super::Sigchld::BASTARD_DATA
    # for this process.
    #
    Forks::Super::Sigchld::handle_bastards($pid);
  }
  $job->{real_pid} = $pid;
  $job->{pid} = $pid unless defined $job->{pid};
  $job->{start} = Time::HiRes::time();

  $job->_config_parent;
  $job->run_callback('start');
  Forks::Super::Sigchld::handle_CHLD(-1);
  if ($$ != $Forks::Super::MAIN_PID) {
    $XSIG{CHLD}[-1] = \&Forks::Super::Sigchld::handle_CHLD;
  }

  return $OVERLOAD_ENABLED ? $job : $pid;
}

sub _postlaunch_child {
  my $job = shift;
  Forks::Super::init_child() if defined &Forks::Super::init_child;
  $job->_config_child;

  local $ENV{_FORK_PPID} = $$;
  local $ENV{_FORK_PID} = $$;

  if ($job->{style} eq 'cmd' || $job->{style} eq 'exec') {

      if (defined($job->{fh_config}->{stdin})
	  && defined($job->{fh_config}->{sockets})) {

	  $job->_postlaunch_child_to_proc;

      } elsif ($job->{style} eq 'cmd') {

	  $job->_postlaunch_child_to_cmd;

      } else {

	  debug("Exec'ing [ @{$job->{exec}} ]") if $job->{debug};
	  exec( @{$job->{exec}} );

      }

  } elsif ($job->{style} eq 'sub') {

      $job->_postlaunch_child_to_sub;

  }
  return 0;
}

sub _postlaunch_child_to_proc {
    my $job = shift;
    my $proch = Forks::Super::Job::Ipc::_gensym();
    $job->{cmd} ||= $job->{exec};
    my $p1 = open $proch, '|-', @{$job->{cmd}};
    print $proch $job->{fh_config}->{stdin};
    close $proch;
    my $c1 = $?;
    debug("Exit code of $$ was $c1 ", $c1>>8) if $job->{debug};
    deinit_child();
    exit $c1 >> 8;
}

sub _postlaunch_child_to_cmd {
    my $job = shift;
    debug("Executing [ @{$job->{cmd}} ]") if $job->{debug};

    my $c1;
    if (&IS_WIN32) {
      # There are lots of ways to spawn a process in Windows
      if (Forks::Super::Config::CONFIG('Win32::Process')) {
	$c1 = Forks::Super::Job::OS::Win32::open_win32_process($job);
      } else {
	$c1 = Forks::Super::Job::OS::Win32::open3_win32_process($job);
      }
    } else {

      $c1 = system( @{$job->{cmd}} );
    }
    debug("Exit code of $$ was $c1 ", $c1>>8) if $job->{debug};
    deinit_child();
    exit $c1 >> 8;
}

sub _postlaunch_child_to_sub {
    my $job = shift;
    my $sub = $job->{sub};
    my @args = @{$job->{args} || []};

    my $error;
    eval {
      no strict 'refs';
      $job->{_cleanup_code} = \&deinit_child;
      $sub->(@args);
      delete $job->{_cleanup_code};
      1;
    } or do {
      $error = $@;
    };
 
    if ($job->{debug}) {
      if ($error) {
	debug("JOB $$ SUBROUTINE CALL HAD AN ERROR: $error");
      }
      debug("Job $$ subroutine call has completed");
    }
    deinit_child();
    if ($error) {
	die $error,"\n";
    }
    exit 0;
}

sub _launch_from_child {
  my $job = shift;
  if ($Forks::Super::CHILD_FORK_OK == 0) {
    carp 'Forks::Super::fork() not allowed\n',
      "in child process $$ while \$Forks::Super::CHILD_FORK_OK ",
	"is not set!\n";

    return;
  } elsif ($Forks::Super::CHILD_FORK_OK == -1) {
    carp "Forks::Super::fork() call not allowed\n",
	"in child process $$ while \$Forks::Super::CHILD_FORK_OK <= 0.\n",
	  "Will create child of child with CORE::fork()\n";

    my $pid = _CORE_fork();
    if (defined($pid) && $pid == 0) {
      # child of child
      if (defined &Forks::Super::init_child) {
	Forks::Super::init_child();
      } else {
	init_child();
      }
      return $pid;
    }
    return $pid;
  }
  return;
}

sub suspend {
  my $j = shift;
  $j = Forks::Super::Job::get($j) if ref $j ne 'Forks::Super::Job';
  my $pid = $j->{real_pid};
  if ($j->{state} eq 'ACTIVE') {
    local $! = 0;
    my $kill_result = Forks::Super::kill('STOP', $j);
    if ($kill_result > 0) {
      $j->{state} = 'SUSPENDED';
      return 1;
    }
    carp "'STOP' signal not received by $pid, job ", $j->toString(), "\n";
    return;
  }
  if ($j->{state} eq 'DEFERRED') {
    $j->{state} = 'SUSPENDED-DEFERRED';
    return -1;
  }
  if ($j->is_complete) {
    carp "Forks::Super::Job::suspend(): called on completed job ", 
      $j->{pid}, "\n";
    return;
  }
  if ($j->{state} eq 'SUSPENDED') {
    carp "Forks::Super::Job: suspend called on suspended job ", $j->{pid};
    return;
  }
  carp "Forks::Super::Job: suspend called on job ", $j->toString(), "\n";
  return;
}

sub resume {
  my $j = shift;
  $j = Forks::Super::Job::get($j) if ref $j ne 'Forks::Super::Job';
  my $pid = $j->{real_pid};
  if ($j->{state} eq 'SUSPENDED') {
    local $! = 0;
    my $kill_result = Forks::Super::kill('CONT', $j);
    if ($kill_result > 0) {
      $j->{state} = 'ACTIVE';
      return 1;
    }
    carp "'CONT' signal not received by $pid, job ", $j->toString(), "\n";
    return;
  }
  if ($j->{state} eq 'SUSPENDED-DEFERRED') {
    $j->{state} = 'DEFERRED';
    return -1;
  }
  if ($j->is_complete) {
    carp "Forks::Super::Job::resume(): called on a completed job ", 
      $j->{pid}, "\n";
    return;
  }
  carp "Forks::Super::Job::resume(): called on job in state ", 
    $j->{state}, "\n";
  return;
}

#
# do further initialization of a Forks::Super::Job object,
# mainly setting derived fields
#
sub _preconfig {
  my $job = shift;

  $job->_preconfig_style;
  $job->_preconfig_busy_action;
  $job->_preconfig_start_time;
  $job->_preconfig_dependencies;
  Forks::Super::Job::Callback::_preconfig_callbacks($job);
  Forks::Super::Job::OS::_preconfig_os($job);
  return;
}

# some final initialization just before launch
sub _preconfig2 {
  my $job = shift;
  if (!defined $job->{debug}) {
    $job->{debug} = $Forks::Super::Debug::DEBUG;
  }
  return;
}

sub _preconfig_style {
  my $job = shift;

  ###################
  # set up style.
  #

  if (0 && defined $job->{run}) {   # not enabled
    $job->_preconfig_style_run;
  }

  if (defined $job->{cmd}) {
    if (ref $job->{cmd} ne 'ARRAY') {
      $job->{cmd} = [ $job->{cmd} ];
    }
    $job->{style} = 'cmd';
  } elsif (defined $job->{exec}) {
    if (ref $job->{exec} ne 'ARRAY') {
      $job->{exec} = [ $job->{exec} ];
    }
    $job->{style} = 'exec';
  } elsif (defined $job->{sub}) {
    $job->{style} = 'sub';
    $job->{sub} = qualify_sub_name $job->{sub};
    if (defined $job->{args}) {
      if (ref $job->{args} ne 'ARRAY') {
	$job->{args} = [ $job->{args} ];
      }
    } else {
      $job->{args} = [];
    }
  } else {
    $job->{style} = 'natural';
  }
  return;
}

sub _preconfig_style_run {    ### for future use
  my $job = shift;
  if (ref $job->{run} ne 'ARRAY') {
    $job->{run} = [ $job->{run} ];
  }

  return;

  # How will we use or emulate the rich functionality
  # of IPC::Run?
  #
  # inputs are a "harness specification"
  # build a harness
  # on "launch", call $harness->start
  # when the job is reaped, call $harness->finish

  # one feature of IPC::Run harnesses is that they
  # may be reused!

}

sub _preconfig_busy_action {
  my $job = shift;

  ######################
  # what will we do if the job cannot launch?
  #
  if (defined $job->{on_busy}) {
    $job->{_on_busy} = $job->{on_busy};
  } else {
    no warnings 'once';
    $job->{_on_busy} = $Forks::Super::ON_BUSY || 'block';
  }
  $job->{_on_busy} = uc $job->{_on_busy};

  ########################
  # make a queue priority available if needed
  #
  if (not defined $job->{queue_priority}) {
    $job->{queue_priority} = Forks::Super::Queue::get_default_priority();
  }
  return;
}

sub _preconfig_start_time {
  my $job = shift;

  ###########################
  # configure a future start time
  my $start_after = 0;
  if (defined $job->{delay}) {
    $start_after
      = Time::HiRes::time() 
	+ Forks::Super::Job::Timeout::_time_from_natural_language(
		$job->{delay}, 1);
  }
  if (defined $job->{start_after}) {
    my $start_after2 
      = Forks::Super::Job::Timeout::_time_from_natural_language(
		$job->{start_after}, 0);
    if ($start_after < $start_after2) {
      $start_after = $start_after2 
    }
  }
  if ($start_after) {
    $job->{start_after} = $start_after;
    delete $job->{delay};
    debug('Forks::Super::Job::_can_launch(): start delay requested.')
      if $job->{debug};
  }
  return;
}

sub _preconfig_dependencies {
  my $job = shift;

  ##########################
  # assert dependencies are expressed as array refs
  # expand job names to pids
  #
  if (defined $job->{depend_on}) {
    if (ref $job->{depend_on} ne 'ARRAY') {
      $job->{depend_on} = [ $job->{depend_on} ];
    }
    $job->{depend_on} = _resolve_names($job, $job->{depend_on});
  }
  if (defined $job->{depend_start}) {
    if (ref $job->{depend_start} ne 'ARRAY') {
      $job->{depend_start} = [ $job->{depend_start} ];
    }
    $job->{depend_start} = _resolve_names($job, $job->{depend_start});
  }
  return;
}

# convert job names in an array to job ids, if necessary
sub _resolve_names {
  my $job = shift;
  my @in = @{$_[0]};
  my @out = ();
  foreach my $id (@in) {
    if (ref $id eq 'Forks::Super::Job') {
      push @out, $id;
    } elsif (is_number($id) && defined($ALL_JOBS{$id})) {
      push @out, $id;
    } else {
      my @j = Forks::Super::Job::getByName($id);
      if (@j > 0) {
	foreach my $j (@j) {
	  next if \$j eq \$job; 
	  # $j eq $job was not sufficient when $job is overloaded
	  # and $job->{pid} has not been set.

	  push @out, $j->{pid};
	}
      } else {
	carp "Forks::Super: Job ",
	  "dependency identifier \"$id\" is invaild. Ignoring\n";
      }
    }
  }
  return [ @out ];
}

#
# set some additional attributes of a Forks::Super::Job after the
# child is successfully launched.
#
sub _config_parent {
  my $job = shift;
  $job->_config_fh_parent;
  $job->_config_timeout_parent;
  if ($Forks::Super::SysInfo::CONFIG{'getpgrp'}) {
    # when  timeout =>   or   expiration =>  is used,
    # PGID of child will be set to child PID
    if (defined($job->{timeout}) || defined($job->{expiration})) {
      $job->{pgid} = $job->{real_pid};
    } else {
      if (not eval { $job->{pgid} = getpgrp($job->{real_pid}) }) {
	$Forks::Super::SysInfo::CONFIG{'getpgrp'} = 0;
	$job->{pgid} = $job->{real_pid};
      }
    }
  }
  return;
}

sub _config_child {
  my $job = shift;
  $Forks::Super::Job::self = $job;
  $job->_config_callback_child;
  $job->_config_debug_child;
  $job->_config_timeout_child;
  $job->_config_os_child;
  $job->_config_fh_child;
  $job->_config_dir;
  return;
}

sub _config_debug_child {
  my $job = shift;
  if ($job->{debug} && $job->{undebug}) {
    if ($Forks::Super::Config::IS_TEST) {
      debug("Disabling debugging in child $$");
    }
    $Forks::Super::Debug::DEBUG = 0;
    $job->{debug} = 0;
  }
  return;
}

sub _config_dir {
  my $job = shift;
  $job->{dir} ||= $job->{chdir};
  if (defined $job->{dir}) {
    if (!chdir $job->{dir}) {
      croak "Forks::Super::Job::launch(): ",
	"Invalid \"dir\" option: \"$job->{dir}\" $!\n";
    }
  }
  return;
}

END {
  no warnings 'internal';
  $INSIDE_END_QUEUE = 1;
  if ($$ == ($Forks::Super::MAIN_PID ||= $$)) {

    # disable SIGCHLD handler during cleanup. Hopefully this will fix
    # intermittent test failures where all subtests pass but the
    # test exits with non-zero exit status (e.g., t/42d-filehandles.t)

    untie %SIG;
    if ($] >= 5.007003) {
      delete $SIG{CHLD};
    } else {
      $SIG{CHLD} = 'IGNORE';
    }

    Forks::Super::Queue::_cleanup();
    Forks::Super::Job::Ipc::_cleanup();
  } else {
    if (defined($Forks::Super::Job::self)
       && defined($Forks::Super::Job::self->{_cleanup_code})) {
      no strict 'refs';
      $Forks::Super::Job::self->{_cleanup_code}->();
    }
    Forks::Super::Job::Timeout::_cleanup_child();
  }
}

#############################################################################
# Package methods (meant to be called as Forks::Super::Job::xxx(@args))

sub enable_overload {
  if (!$OVERLOAD_ENABLED) {
    $OVERLOAD_ENABLED = 1;

    if (!eval {
      use overload
	'""' => sub { $_[0]->{pid} },
	'+' => sub { $_[0]->{pid} + $_[1] },
        '*' => sub { $_[0]->{pid} * $_[1] },
        '&' => sub { $_[0]->{pid} & $_[1] },
        '|' => sub { $_[0]->{pid} | $_[1] },
        '^' => sub { $_[0]->{pid} ^ $_[1] },
        '~' => sub { ~$_[0]->{pid} },         # since 0.37
        '<=>' => sub { $_[2] ? $_[1] <=> $_[0]->{pid} 
			     : $_[0]->{pid} <=> $_[1] },
        'cmp' => sub { $_[2] ? $_[1] cmp $_[0]->{pid} 
			     : $_[0]->{pid} cmp $_[1] },
        '-'   => sub { $_[2] ? $_[1]  -  $_[0]->{pid} 
			     : $_[0]->{pid}  -  $_[1] },
        '/'   => sub { $_[2] ? $_[1]  /  $_[0]->{pid} 
			     : $_[0]->{pid}  /  $_[1] },
        '%'   => sub { $_[2] ? $_[1]  %  $_[0]->{pid} 
			     : $_[0]->{pid}  %  $_[1] },
        '**'  => sub { $_[2] ? $_[1]  ** $_[0]->{pid} 
			     : $_[0]->{pid}  ** $_[1] },
        '<<'  => sub { $_[2] ? $_[1]  << $_[0]->{pid} 
			     : $_[0]->{pid}  << $_[1] },
        '>>'  => sub { $_[2] ? $_[1]  >> $_[0]->{pid} 
			     : $_[0]->{pid}  >> $_[1] },
        'x'   => sub { $_[2] ? $_[1]  x  $_[0]->{pid} 
			     : $_[0]->{pid}  x  $_[1] },
        'cos'  => sub { cos $_[0]->{pid} },
        'sin'  => sub { sin $_[0]->{pid} },
        'exp'  => sub { exp $_[0]->{pid} },
        'log'  => sub { log $_[0]->{pid} },
        'sqrt' => sub { sqrt $_[0]->{pid} },
        'int'  => sub { int $_[0]->{pid} },
        'abs'  => sub { abs $_[0]->{pid} },
        'atan2' => sub { $_[2] ? atan2($_[1],$_[0]->{pid}) 
			       : atan2($_[0]->{pid},$_[1]) };

      # XXX - why doesn't it work when I include
      #       '<>' => sub { ... }
      #    in the  use overload  block?
      no strict 'refs';
      *{'Forks::Super::Job::(<>'} = sub {
	return $_[0]->read_stdout();
      };
      1 }            # end eval { use overload ... }
	) {
      carp "Error enabling overloading on Forks::Super::Job objects: $@\n";
    } elsif ($Forks::Super::Debug::DEBUG) {
        debug("Enabled overloading on Forks::Super::Job objects");
    }
  }
  return;
}

sub disable_overload {
  if ($OVERLOAD_ENABLED) {
    $OVERLOAD_ENABLED = 0;
    eval { no overload values %overload::ops; 1 }
        or Forks::Super::Debug::carp_once "Forks::Super::Job ",
    		"disable overload failed: $@";
  }
  return;
}

# returns a Forks::Super::Job object with the given identifier
sub get {
  my $id = shift;
  if (!defined $id) {
    Carp::cluck "undef value passed to Forks::Super::Job::get()";
  }
  if (ref $id eq 'Forks::Super::Job') {
    return $id;
  }
  if (defined $ALL_JOBS{$id}) {
    return $ALL_JOBS{$id};
  }
  return getByPid($id) || getByName($id);
}

sub getByPid {
  my $id = shift;
  if (is_number($id)) {
    my @j = grep { (defined($_->{pid}) && $_->{pid} == $id) ||
		   (defined($_->{real_pid}) && $_->{real_pid} == $id)
		 } @ALL_JOBS;
    return $j[0] if @j > 0;
  }
  return;
}

sub getByName {
  my $id = shift;
  my @j = grep { defined($_->{name}) && $_->{name} eq $id } @ALL_JOBS;
  if (@j > 0) {
    return wantarray ? @j : $j[0];
  }
  return;
}

# retrieve a job object for a pid or job name, if necessary
sub _resolve {
  if (ref $_[0] ne 'Forks::Super::Job') {
    my $job = get($_[0]);
    if (defined $job) {
      return $_[0] = $job;
    }
    return $job;
  }
  return $_[0];
}

#
# count the number of active processes
#
sub count_active_processes {
  my $optional_pgid = shift;
  if (defined $optional_pgid) {
    return scalar grep {
      $_->{state} eq 'ACTIVE'
	and $_->{pgid} == $optional_pgid } @ALL_JOBS;
  }
  return scalar grep { defined($_->{state})
			 && $_->{state} eq 'ACTIVE' } @ALL_JOBS;
}

sub count_alive_processes {
  my ($count_bg, $optional_pgid) = @_;
  my @alive = grep { $_->{state} eq 'ACTIVE' ||
		     $_->{state} eq 'COMPLETE' ||
		     $_->{state} eq 'DEFERRED' ||
		     $_->{state} eq 'LAUNCHING' || # rare
		     $_->{state} eq 'SUSPENDED' ||
		     $_->{state} eq 'SUSPENDED-DEFERRED' 
		   } @ALL_JOBS;
  if (!$count_bg) {
    @alive = grep { $_->{_is_bg} == 0 } @alive;
  }
  if (defined $optional_pgid) {
    @alive = grep { $_->{pgid} == $optional_pgid } @alive;
  }
  return scalar @alive;
}

#
# _reap should distinguish:
#
#    all alive jobs (ACTIVE+COMPLETE+SUSPENDED+DEFERRED+SUSPENDED-DEFERRED)
#    all active jobs (ACTIVE + COMPLETE + DEFERRED)
#    filtered alive jobs (by optional pgid)
#    filtered ACTIVE + COMPLETE + DEFERRED jobs
#
#    if  all_active==0  and  all_alive>0,  
#    then see Wait::WAIT_ACTION_ON_SUSPENDED_JOBS
#
sub count_processes {
  my ($count_bg, $optional_pgid) = @_;
  my @alive = grep { $_->{state} ne 'REAPED' && $_->{state} ne 'NEW' 
		   } @ALL_JOBS;
  if (!$count_bg) {
    @alive = grep { $_->{_is_bg} == 0 } @alive;
  }
  my @active = grep { $_->{state} !~ /SUSPENDED/ } @alive;
  my @filtered_active = @active;
  if (defined $optional_pgid) {
    @filtered_active = grep { $_->{pgid} == $optional_pgid } @filtered_active;
  }

  my @n = (scalar(@filtered_active), scalar(@alive), scalar(@active));

  if ($Forks::Super::Debug::DEBUG) {
    debug("count_processes(): @n");
    debug("count_processes(): Filtered active: ",
	  $filtered_active[0]->toString()) if $n[0];
    debug("count_processes(): Alive: ", $alive[0]->toShortString()) if $n[1];
    debug("count_processes(): Active: @active") if $n[2];
  }

  return @n;
}

sub init_child {
  Forks::Super::Job::Ipc::init_child();
  return;
}

sub deinit_child {
  Forks::Super::Job::Ipc::deinit_child();
  return;
}

#
# get the current CPU load. May not be possible
# to do on all operating systems.
#
sub get_cpu_load {
  return Forks::Super::Job::OS::get_cpu_load();
}

sub dispose {
  foreach my $job (@_) {

    my $pid = $job->{pid};
    my $real_pid = $job->{real_pid} || $pid;

    $job->close_fh('all');
    delete $Forks::Super::CHILD_STDIN{$pid};
    delete $Forks::Super::CHILD_STDIN{$real_pid};
    delete $Forks::Super::CHILD_STDOUT{$pid};
    delete $Forks::Super::CHILD_STDOUT{$real_pid};
    delete $Forks::Super::CHILD_STDERR{$pid};
    delete $Forks::Super::CHILD_STDERR{$real_pid};

    foreach my $attr ('f_in','f_out','f_err') {
      my $file = $job->{fh_config} && $job->{fh_config}->{$attr};
      if (defined($file) && -f $file) {
	$! = 0;
	if (unlink $file) {
	  delete $Forks::Super::Job::Ipc::IPC_FILES{$file};
	} elsif ($INSIDE_END_QUEUE) {
	  warn "unlink failed for \"$file\": $! $^E\n";
	  warn "@{$Forks::Super::Job::Ipc::IPC_FILES{$file}}\n";
	}
      }
    }

    # XXX - disposed jobs should go to %ARCHIVED_JOBS, @ARCHIVED_JOBS
    my @k = grep { $ALL_JOBS{$_} eq $job } keys %ALL_JOBS;
    delete $ALL_JOBS{$_} for @k;

    delete $job->{$_} for keys %$job;
    $job->{disposed} ||= time;
  }
  @ALL_JOBS = grep { !$_->{disposed} } @ALL_JOBS;
  return;
}

#
# Print information about all known jobs.
#
sub printAll {
  print "ALL JOBS\n";
  print "--------\n";
  foreach my $job
    (sort {$a->{pid} <=> $b->{pid} ||
	     $a->{created} <=> $b->{created}} @ALL_JOBS) {

      print $job->toString(), "\n";
      print "----------------------------\n";
    }
  return;
}

sub get_win32_proc { return $WIN32_PROC; }
sub get_win32_proc_pid { return $WIN32_PROC_PID; }

1;

__END__