File Coverage

blib/lib/MCE/Hobo.pm

Criterion	Covered	Total	%
statement	344	533	64.5
branch	164	430	38.1
condition	58	179	32.4
subroutine	56	78	71.7
pod	24	24	100.0
total	646	1244	51.9

###############################################################################
## ----------------------------------------------------------------------------
## A threads-like parallelization module.
##
###############################################################################

use strict;
use warnings;

use 5.010001;

no warnings qw( threads recursion uninitialized once redefine );

package MCE::Hobo;

our $VERSION = '1.893';

## no critic (BuiltinFunctions::ProhibitStringyEval)
## no critic (Subroutines::ProhibitExplicitReturnUndef)
## no critic (Subroutines::ProhibitSubroutinePrototypes)
## no critic (TestingAndDebugging::ProhibitNoStrict)

use MCE::Signal ();
use MCE::Mutex ();
use MCE::Channel ();
use Time::HiRes 'sleep';

use overload (
   q(==)    => \&equal,
   q(!=)    => sub { !equal(@_) },
   fallback => 1
);

sub import {
   if (caller !~ /^MCE::/) {
      no strict 'refs'; no warnings 'redefine';
      *{ caller().'::mce_async' } = \&mce_async;
   }
   return;
}

## The POSIX module has many symbols. Try not loading it simply
## to have WNOHANG. The following covers most platforms.

use constant {
   _WNOHANG => ( $INC{'POSIX.pm'} )
      ? &POSIX::WNOHANG : ( $^O eq 'solaris' ) ? 64 : 1
};

my ( $_MNGD, $_DATA, $_DELY, $_LIST ) = ( {}, {}, {}, {} );

my $_freeze = MCE::Channel::_get_freeze();
my $_thaw   = MCE::Channel::_get_thaw();

my $_is_MSWin32 = ( $^O eq 'MSWin32' ) ? 1 : 0;
my $_tid        = ( $INC{'threads.pm'} ) ? threads->tid() : 0;
my $_yield_secs = ( $^O =~ /mswin|mingw|msys|cygwin/i ) ? 0.015 : 0.008;

sub CLONE {
   $_tid = threads->tid(), &_clear() if $INC{'threads.pm'};
}

sub _clear {
   %{ $_LIST } = ();
}

sub _max_workers {
   my ( $cpus ) = @_;
   if ( $cpus eq 'auto' ) {
      $cpus = MCE::Util::get_ncpu();
   }
   elsif ( $cpus =~ /^([0-9.]+)%$/ ) {
      my ( $percent, $ncpu ) = ( $1 / 100, MCE::Util::get_ncpu() );
      $cpus = $ncpu * $percent + 0.5;
   }
   $cpus = 1 if $cpus !~ /^[\d\.]+$/ || $cpus < 1;
   return int($cpus);
}

###############################################################################
## ----------------------------------------------------------------------------
## Init routine.
##
###############################################################################

bless my $_SELF = { MGR_ID => "$$.$_tid", WRK_ID => $$ }, __PACKAGE__;

sub MCE::Hobo::_guard::DESTROY {
   my ($pkg, $id) = @{ $_[0] };

   if (defined $pkg && $id eq "$$.$_tid") {
      @{ $_[0] } = ();
      MCE::Hobo->finish($pkg);
   }

   return;
}

sub init {
   shift if ( defined $_[0] && $_[0] eq __PACKAGE__ );

   # -- options ----------------------------------------------------------
   # max_workers hobo_timeout posix_exit on_start on_finish void_context
   # ---------------------------------------------------------------------

   my $opt = ( ref $_[0] eq 'HASH' ) ? shift : { @_ };
   my $pkg = "$$.$_tid.".( delete $opt->{caller} || caller() );
   my $mngd = $_MNGD->{$pkg} = $opt;

   @_ = ();

   $mngd->{MGR_ID} = "$$.$_tid", $mngd->{PKG} = $pkg,
   $mngd->{WRK_ID} =  $$;

   &_force_reap($pkg), $_DATA->{$pkg}->clear() if ( defined $_LIST->{$pkg} );

   if ( !defined $_LIST->{$pkg} ) {
      $MCE::_GMUTEX->lock() if ( $_tid && $MCE::_GMUTEX );
      sleep 0.015 if $_tid;

      # Start the shared-manager process if not running.
      MCE::Shared->start() if $INC{'MCE/Shared.pm'};

      my $chnl = MCE::Channel->new( impl => 'Mutex' );
      $_LIST->{ $pkg } = MCE::Hobo::_ordhash->new();
      $_DELY->{ $pkg } = MCE::Hobo::_delay->new( $chnl );
      $_DATA->{ $pkg } = MCE::Hobo::_hash->new();
      $_DATA->{"$pkg:id"} = 0;

      $_DATA->{"$pkg:seed"} = int(CORE::rand() * 1e9);

      $MCE::_GMUTEX->unlock() if ( $_tid && $MCE::_GMUTEX );
   }

   if ( !exists $mngd->{posix_exit} ) {
      $mngd->{posix_exit} = 1 if (
         $^S || $_tid || $INC{'Mojo/IOLoop.pm'} ||
         $INC{'Coro.pm'} || $INC{'LWP/UserAgent.pm'} || $INC{'stfl.pm'} ||
         $INC{'Curses.pm'} || $INC{'CGI.pm'} || $INC{'FCGI.pm'} ||
         $INC{'Tk.pm'} || $INC{'Wx.pm'} || $INC{'Win32/GUI.pm'} ||
         $INC{'Gearman/Util.pm'} || $INC{'Gearman/XS.pm'}
      );
   }

   if ( defined $mngd->{max_workers} ) {
      $mngd->{max_workers} = _max_workers($mngd->{max_workers});
   }

   if ( $INC{'LWP/UserAgent.pm'} && !$INC{'Net/HTTP.pm'} ) {
      local $@; eval 'require Net::HTTP; require Net::HTTPS';
   }

   require POSIX
      if ( $mngd->{on_finish} && !$INC{'POSIX.pm'} && !$_is_MSWin32 );

   defined wantarray
      ? bless([$pkg, "$$.$_tid"], MCE::Hobo::_guard::)
      : ();
}

###############################################################################
## ----------------------------------------------------------------------------
## 'new', 'mce_async', and 'create' for threads-like similarity.
##
###############################################################################

## 'new' and 'tid' are aliases for 'create' and 'pid' respectively.

*new = \&create, *tid = \&pid;

## Use "goto" trick to avoid pad problems from 5.8.1 (fixed in 5.8.2)
## Tip found in threads::async.

sub mce_async (&;@) {
   goto &create;
}

sub create {
   my $caller = caller();

   my $mngd = $_MNGD->{ "$$.$_tid.$caller" } || do {
      # construct mngd internally on first use unless defined
      init( caller => $caller ); $_MNGD->{ "$$.$_tid.$caller" };
   };

   shift if ( $_[0] eq __PACKAGE__ );

   # ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~

   my $self = bless ref $_[0] eq 'HASH' ? { %{ shift() } } : { }, __PACKAGE__;

   $self->{IGNORE} = 1 if $SIG{CHLD} eq 'IGNORE';
   $self->{MGR_ID} = $mngd->{MGR_ID}, $self->{PKG} = $mngd->{PKG};
   $self->{ident } = shift if ( !ref $_[0] && ref $_[1] eq 'CODE' );

   my $func = shift; $func = $caller.'::'.$func
      if ( !ref $func && length $func && index($func,':') < 0 );

   if ( !defined $func ) {
      local $\; print {*STDERR} "code function is not specified or valid\n";
      return undef;
   }

   my ( $list, $max_workers, $pkg ) = (
      $_LIST->{ $mngd->{PKG} }, $mngd->{max_workers}, $mngd->{PKG}
   );

   $_DATA->{"$pkg:id"} = 10000 if ( ( my $id = ++$_DATA->{"$pkg:id"} ) >= 2e9 );

   # Reap completed hobo processes.
   if ( $self->{IGNORE} || ($max_workers && $list->len() >= $max_workers) ) {
      local ($SIG{CHLD}, $!, $?, $_);
      map {
         $_ = substr($_, 1); # strip leading 'R'
         my $hobo = $list->del($_);
         if ( ! $hobo->{REAPED} ) {
             waitpid($hobo->{WRK_ID}, 0);
             _reap_hobo($hobo, 0);
         }
         ();
      }
      $_DATA->{$pkg}->get_done();
   }

   # Wait for a slot if saturated.
   _wait_one($pkg) if ( $max_workers && $list->len() >= $max_workers );

   # ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~ ~~~

   $MCE::_GMUTEX->lock() if ( $_tid && $MCE::_GMUTEX );

   my @args = @_; @_ = ();  # To avoid (Scalars leaked: N) messages
   my ( $killed, $pid );

   {
      local $SIG{TERM} = local $SIG{INT} = sub { $killed = $_[0] }
         if ( !$_is_MSWin32 && $] ge '5.010001' );

      local $SIG{TTIN}, local $SIG{TTOU}, local $SIG{WINCH}
         if ( !$_is_MSWin32 );

      $pid = fork();

      if ( !defined $pid ) {                                # error
         local $\; print {*STDERR} "fork error: $!\n";
      }
      elsif ( $pid ) {                                      # parent
         $self->{WRK_ID} = $pid;
         $list->set($pid, $self);
         $mngd->{on_start}->($pid, $self->{ident}) if $mngd->{on_start};
      }
      else {                                                # child
         %{ $_LIST } = (), $_SELF = $self;

         local $SIG{TERM} = local $SIG{INT} = local $SIG{ABRT} = \&_trap,
         local $SIG{SEGV} = local $SIG{HUP} = \&_trap,
         local $SIG{QUIT} = \&_quit;
         local $SIG{CHLD};

         MCE::Shared::init() if $INC{'MCE/Shared.pm'};
         $_DATA->{ $_SELF->{PKG} }->set('S'.$$, '') unless $self->{IGNORE};
         CORE::kill($killed, $$) if $killed;

         MCE::Child->_clear() if $INC{'MCE/Child.pm'};
         MCE::Hobo->_clear() if $INC{'MCE/Hobo.pm'};

         # Set the seed of the base generator uniquely between workers.
         # The new seed is computed using the current seed and ID value.
         # One may set the seed at the application level for predictable
         # results. Ditto for PDL, Math::Prime::Util, Math::Random, and
         # Math::Random::MT::Auto.

         {
            my $seed = abs($_DATA->{"$pkg:seed"} - ($id * 100000)) % 2147483560;

            CORE::srand($seed);
            PDL::srand($seed) if $INC{'PDL.pm'} && PDL->can('srand'); # PDL 2.062 ~ 2.089
            PDL::srandom($seed) if $INC{'PDL.pm'} && PDL->can('srandom'); # PDL 2.089_01+
            Math::Prime::Util::srand($seed) if $INC{'Math/Prime/Util.pm'};
         }

         if ( $INC{'Math/Random.pm'} ) {
            my $cur_seed = Math::Random::random_get_seed();
            my $new_seed = ($cur_seed < 1073741781)
               ? $cur_seed + ((abs($id) * 10000) % 1073741780)
               : $cur_seed - ((abs($id) * 10000) % 1073741780);

            Math::Random::random_set_seed($new_seed, $new_seed);
         }

         if ( $INC{'Math/Random/MT/Auto.pm'} ) {
            my $cur_seed = Math::Random::MT::Auto::get_seed()->[0];
            my $new_seed = ($cur_seed < 1073741781)
               ? $cur_seed + ((abs($id) * 10000) % 1073741780)
               : $cur_seed - ((abs($id) * 10000) % 1073741780);

            Math::Random::MT::Auto::set_seed($new_seed);
         }

         _dispatch($mngd, $func, \@args);
      }
   }

   $MCE::_GMUTEX->unlock() if ( $_tid && $MCE::_GMUTEX );

   CORE::kill($killed, $$) if $killed;

   return $pid ? $self : undef;
}

###############################################################################
## ----------------------------------------------------------------------------
## Public methods.
##
###############################################################################

sub equal {
   return 0 unless ( ref $_[0] && ref $_[1] );
   $_[0]->{WRK_ID} == $_[1]->{WRK_ID} ? 1 : 0;
}

sub error {
   _croak('Usage: $hobo->error()') unless ref( my $self = $_[0] );
   $self->join() unless $self->{REAPED};
   $self->{ERROR} || undef;
}

sub exit {
   shift if ( defined $_[0] && $_[0] eq __PACKAGE__ );

   my ( $self ) = ( ref $_[0] ? shift : $_SELF );
   my ( $pkg, $wrk_id ) = ( $self->{PKG}, $self->{WRK_ID} );

   if ( $wrk_id == $$ && $self->{MGR_ID} eq "$$.$_tid" ) {
      MCE::Hobo->finish('MCE'); CORE::exit(@_);
   }
   elsif ( $wrk_id == $$ ) {
      alarm 0; my ( $exit_status, @res ) = @_; $? = $exit_status || 0;
      $_DATA->{$pkg}->set('R'.$wrk_id, @res ? $_freeze->(\@res) : '')
         unless $self->{IGNORE};
      die "Hobo exited ($?)\n";
      _exit($?); # not reached
   }

   return $self if $self->{REAPED};

   if ( defined $_DATA->{$pkg} ) {
      sleep $_yield_secs until $_DATA->{$pkg}->exists('S'.$wrk_id);
   } else {
      sleep 0.030;
   }

   if ($_is_MSWin32) {
      CORE::kill('KILL', $wrk_id) if CORE::kill('ZERO', $wrk_id);
   } else {
      CORE::kill('QUIT', $wrk_id) if CORE::kill('ZERO', $wrk_id);
   }

   $self;
}

sub finish {
   _croak('Usage: MCE::Hobo->finish()') if ref($_[0]);
   shift if ( defined $_[0] && $_[0] eq __PACKAGE__ );

   my $pkg = defined($_[0]) ? shift : "$$.$_tid.".caller();

   if ( $pkg eq 'MCE' ) {
      for my $key ( keys %{ $_LIST } ) { MCE::Hobo->finish($key); }
   }
   elsif ( defined $_LIST->{$pkg} ) {
      return if $MCE::Signal::KILLED;

      if ( exists $_DELY->{$pkg} ) {
         &_force_reap($pkg);
         delete($_DELY->{$pkg}), delete($_DATA->{"$pkg:seed"}),
         delete($_LIST->{$pkg}), delete($_DATA->{"$pkg:id"}),
         delete($_MNGD->{$pkg}), delete($_DATA->{ $pkg });
      }
   }

   @_ = ();

   return;
}

sub is_joinable {
   _croak('Usage: $hobo->is_joinable()') unless ref( my $self = $_[0] );
   my ( $wrk_id, $pkg ) = ( $self->{WRK_ID}, $self->{PKG} );

   if ( $wrk_id == $$ ) {
      '';
   }
   elsif ( $self->{MGR_ID} eq "$$.$_tid" ) {
      return '' if $self->{REAPED};
      local $!;
      ( waitpid($wrk_id, _WNOHANG) == 0 ) ? '' : do {
         _reap_hobo($self, 0) unless $self->{REAPED};
         1;
      };
   }
   else {
      _croak('Error: $hobo->is_joinable() not called by managed process')
         if ( $self->{IGNORE} );

      return '' if $self->{REAPED};
      $_DATA->{$pkg}->exists('R'.$wrk_id) ? 1 : '';
   }
}

sub is_running {
   _croak('Usage: $hobo->is_running()') unless ref( my $self = $_[0] );
   my ( $wrk_id, $pkg ) = ( $self->{WRK_ID}, $self->{PKG} );

   if ( $wrk_id == $$ ) {
      1;
   }
   elsif ( $self->{MGR_ID} eq "$$.$_tid" ) {
      return '' if $self->{REAPED};
      local $!;
      ( waitpid($wrk_id, _WNOHANG) == 0 ) ? 1 : do {
         _reap_hobo($self, 0) unless $self->{REAPED};
         '';
      };
   }
   else {
      _croak('Error: $hobo->is_running() not called by managed process')
         if ( $self->{IGNORE} );

      return '' if $self->{REAPED};
      $_DATA->{$pkg}->exists('R'.$wrk_id) ? '' : 1;
   }
}

sub join {
   _croak('Usage: $hobo->join()') unless ref( my $self = $_[0] );
   my ( $wrk_id, $pkg ) = ( $self->{WRK_ID}, $self->{PKG} );

   if ( $self->{REAPED} ) {
      _croak('Hobo already joined') unless exists( $self->{RESULT} );
      $_LIST->{$pkg}->del($wrk_id) if ( defined $_LIST->{$pkg} );

      return ( defined wantarray )
         ? wantarray ? @{ delete $self->{RESULT} } : delete( $self->{RESULT} )->[-1]
         : ();
   }

   if ( $wrk_id == $$ ) {
      _croak('Cannot join self');
   }
   elsif ( $self->{MGR_ID} eq "$$.$_tid" ) {
      # remove from list after reaping
      if ( $_tid ) {
         local $SIG{CHLD};
         _reap_hobo($self, 1);
         $_LIST->{$pkg}->del($wrk_id);
      }
      else {
         local ($SIG{CHLD}, $!);
         waitpid($wrk_id, 0);
         _reap_hobo($self, 0);
         $_LIST->{$pkg}->del($wrk_id);
      }
   }
   else {
      _croak('Error: $hobo->join() not called by managed process')
         if ( $self->{IGNORE} );

      sleep 0.3 until ( $_DATA->{$pkg}->exists('R'.$wrk_id) );
      _reap_hobo($self, 0);
   }

   return unless ( exists $self->{RESULT} );

   ( defined wantarray )
      ? wantarray ? @{ delete $self->{RESULT} } : delete( $self->{RESULT} )->[-1]
      : ();
}

sub kill {
   _croak('Usage: $hobo->kill()') unless ref( my $self = $_[0] );
   my ( $wrk_id, $pkg, $signal ) = ( $self->{WRK_ID}, $self->{PKG}, $_[1] );

   if ( $wrk_id == $$ ) {
      CORE::kill($signal || 'INT', $$);
      return $self;
   }
   if ( $self->{MGR_ID} eq "$$.$_tid" ) {
      return $self if $self->{REAPED};
      if ( defined $_DATA->{$pkg} ) {
         sleep $_yield_secs until $_DATA->{$pkg}->exists('S'.$wrk_id);
      } else {
         sleep 0.030;
      }
   }

   CORE::kill($signal || 'INT', $wrk_id) if CORE::kill('ZERO', $wrk_id);

   $self;
}

sub list {
   _croak('Usage: MCE::Hobo->list()') if ref($_[0]);
   my $pkg = "$$.$_tid.".caller();

   ( defined $_LIST->{$pkg} ) ? $_LIST->{$pkg}->vals() : ();
}

sub list_pids {
   _croak('Usage: MCE::Hobo->list_pids()') if ref($_[0]);
   my $pkg = "$$.$_tid.".caller(); local $_;

   ( defined $_LIST->{$pkg} ) ? map { $_->pid } $_LIST->{$pkg}->vals() : ();
}

sub list_joinable {
   _croak('Usage: MCE::Hobo->list_joinable()') if ref($_[0]);
   my $pkg = "$$.$_tid.".caller();

   return () unless ( my $list = $_LIST->{$pkg} );
   local ($!, $?, $_);

   map {
      ( waitpid($_->{WRK_ID}, _WNOHANG) == 0 ) ? () : do {
         _reap_hobo($_, 0) unless $_->{REAPED};
         $_;
      };
   }
   $list->vals();
}

sub list_running {
   _croak('Usage: MCE::Hobo->list_running()') if ref($_[0]);
   my $pkg = "$$.$_tid.".caller();

   return () unless ( my $list = $_LIST->{$pkg} );
   local ($!, $?, $_);

   map {
      ( waitpid($_->{WRK_ID}, _WNOHANG) == 0 ) ? $_ : do {
         _reap_hobo($_, 0) unless $_->{REAPED};
         ();
      };
   }
   $list->vals();
}

sub max_workers {
   _croak('Usage: MCE::Hobo->max_workers()') if ref($_[0]);
   my $mngd = $_MNGD->{ "$$.$_tid.".caller() } || do {
      # construct mngd internally on first use unless defined
      init(); $_MNGD->{ "$$.$_tid.".caller() };
   };
   shift if ( $_[0] eq __PACKAGE__ );

   $mngd->{max_workers} = _max_workers(shift) if @_;
   $mngd->{max_workers};
}

sub pending {
   _croak('Usage: MCE::Hobo->pending()') if ref($_[0]);
   my $pkg = "$$.$_tid.".caller();

   ( defined $_LIST->{$pkg} ) ? $_LIST->{$pkg}->len() : 0;
}

sub pid {
   ref($_[0]) ? $_[0]->{WRK_ID} : $_SELF->{WRK_ID};
}

sub result {
   _croak('Usage: $hobo->result()') unless ref( my $self = $_[0] );
   return $self->join() unless $self->{REAPED};

   _croak('Hobo already joined') unless exists( $self->{RESULT} );
   wantarray ? @{ delete $self->{RESULT} } : delete( $self->{RESULT} )->[-1];
}

sub seed {
   _croak('Usage: MCE::Hobo->seed()') if ref($_[0]);
   my $pkg = exists $_SELF->{PKG} ? $_SELF->{PKG} : "$$.$_tid.".caller();

   return $_DATA->{"$pkg:seed"};
}

sub self {
   ref($_[0]) ? $_[0] : $_SELF;
}

sub wait_all {
   _croak('Usage: MCE::Hobo->wait_all()') if ref($_[0]);
   my $pkg = "$$.$_tid.".caller();

   return wantarray ? () : 0
      if ( !defined $_LIST->{$pkg} || !$_LIST->{$pkg}->len() );

   local $_; ( wantarray )
      ? map { $_->join(); $_ } $_LIST->{$pkg}->vals()
      : map { $_->join(); () } $_LIST->{$pkg}->vals();
}

*waitall = \&wait_all; # compatibility

sub wait_one {
   _croak('Usage: MCE::Hobo->wait_one()') if ref($_[0]);
   my $pkg = "$$.$_tid.".caller();

   return undef
      if ( !defined $_LIST->{$pkg} || !$_LIST->{$pkg}->len() );

   _wait_one($pkg);
}

*waitone = \&wait_one; # compatibility

sub yield {
   _croak('Usage: MCE::Hobo->yield()') if ref($_[0]);
   shift if ( defined $_[0] && $_[0] eq __PACKAGE__ );

   my $pkg = $_SELF->{PKG} || do {
      my $mngd = $_MNGD->{ "$$.$_tid.".caller() } || do {
         # construct mngd internally on first use unless defined
         init(); $_MNGD->{ "$$.$_tid.".caller() };
      };
      $mngd->{PKG};
   };

   return unless $_DELY->{$pkg};
   my $seconds = $_DELY->{$pkg}->seconds(@_);

   MCE::Util::_sleep( $seconds );
}

###############################################################################
## ----------------------------------------------------------------------------
## Private methods.
##
###############################################################################

sub _croak {
   if ( $INC{'MCE.pm'} ) {
      goto &MCE::_croak;
   }
   else {
      $SIG{__DIE__}  = \&MCE::Signal::_die_handler;
      $SIG{__WARN__} = \&MCE::Signal::_warn_handler;

      $\ = undef; goto &Carp::croak;
   }
}

sub _dispatch {
   my ( $mngd, $func, $args ) = @_;

   $mngd->{WRK_ID} = $_SELF->{WRK_ID} = $$, $? = 0;
   $ENV{PERL_MCE_IPC} = 'win32' if $_is_MSWin32;

   {
      local $!;
      (*STDERR)->autoflush(1) if defined( fileno *STDERR );
      (*STDOUT)->autoflush(1) if defined( fileno *STDOUT );
   }

   # Run task.
   my $hobo_timeout = ( exists $_SELF->{hobo_timeout} )
      ? $_SELF->{hobo_timeout} : $mngd->{hobo_timeout};

   my $void_context = ( exists $_SELF->{void_context} )
      ? $_SELF->{void_context} : $mngd->{void_context};

   my @res; my $timed_out = 0;

   local $SIG{'ALRM'} = sub {
      alarm 0; $timed_out = 1; $SIG{__WARN__} = sub {};
      die "Hobo timed out\n";
   };

   if ( $void_context || $_SELF->{IGNORE} ) {
      no strict 'refs';
      eval { alarm($hobo_timeout || 0); $func->(@{ $args }) };
   }
   else {
      no strict 'refs';
      @res = eval { alarm($hobo_timeout || 0); $func->(@{ $args }) };
   }

   alarm 0;
   $@ = "Hobo timed out" if $timed_out;

   if ( $@ ) {
      _exit($?) if ( $@ =~ /^Hobo exited \(\S+\)$/ );
      my $err = $@; $? = 1; $err =~ s/, <__ANONIO__> line \d+//;

      if ( ! $_SELF->{IGNORE} ) {
         $_DATA->{ $_SELF->{PKG} }->set('S'.$$, $err),
         $_DATA->{ $_SELF->{PKG} }->set('R'.$$, '');
      }

      if ( !$timed_out && !$mngd->{on_finish} && !$INC{'MCE/Simple.pm'} ) {
         use bytes; warn "Hobo $$ terminated abnormally: reason $err\n";
      }
   }
   else {
      shift(@res) if ref($res[0]) =~ /^MCE::(?:Barrier|Semaphore)::_guard/s;
      $_DATA->{ $_SELF->{PKG} }->set('R'.$$, @res ? $_freeze->(\@res) : '')
         if ( ! $_SELF->{IGNORE} );
   }

   _exit($?);
}

sub _exit {
   my ( $exit_status ) = @_;

   # Check for nested workers not yet joined.
   MCE::Hobo->finish('MCE') if ( !$_SELF->{SIGNALED} && keys %{ $_LIST } );

   # Exit hobo process.
   $SIG{__DIE__}  = sub {} unless $_tid;
   $SIG{__WARN__} = sub {};

   threads->exit($exit_status) if ( $INC{'threads.pm'} && $_is_MSWin32 );
   CORE::kill('KILL', $$) if ( $_SELF->{SIGNALED} && !$_is_MSWin32 );

   my $posix_exit = ( exists $_SELF->{posix_exit} )
      ? $_SELF->{posix_exit} : $_MNGD->{ $_SELF->{PKG} }{posix_exit};

   if ( $posix_exit && !$_is_MSWin32 ) {
      eval { MCE::Mutex::Channel::_destroy() };
      POSIX::_exit($exit_status) if $INC{'POSIX.pm'};
      CORE::kill('KILL', $$);
   }

   CORE::exit($exit_status);
}

sub _force_reap {
   my ( $count, $pkg ) = ( 0, @_ );
   return unless ( defined $_LIST->{$pkg} && $_LIST->{$pkg}->len() );

   for my $hobo ( $_LIST->{$pkg}->vals() ) {
      next if $hobo->{IGNORE};

      if ( $hobo->is_running() ) {
         sleep($_yield_secs), CORE::kill('KILL', $hobo->pid())
            if CORE::kill('ZERO', $hobo->pid());
         $count++;
      }
   }

   $_LIST->{$pkg}->clear();

   warn "Finished with active hobo processes [$pkg] ($count)\n"
      if ( $count && !$_is_MSWin32 );

   return;
}

sub _quit {
   return MCE::Signal::defer($_[0]) if $MCE::Signal::IPC;

   alarm 0; my ( $name ) = @_;
   $_SELF->{SIGNALED} = 1, $name =~ s/^SIG//;

   $SIG{$name} = sub {}, CORE::kill($name, -$$)
      if ( exists $SIG{$name} );

   if ( ! $_SELF->{IGNORE} ) {
      my ( $pkg, $wrk_id ) = ( $_SELF->{PKG}, $_SELF->{WRK_ID} );
      $_DATA->{$pkg}->set('R'.$wrk_id, '');
   }

   _exit(0);
}

sub _reap_hobo {
   my ( $hobo, $wait_flag ) = @_;
   return if ( !$hobo || !defined $hobo->{PKG} );

   local @_ = $_DATA->{ $hobo->{PKG} }->get($hobo->{WRK_ID}, $wait_flag);

   ( $hobo->{ERROR}, $hobo->{RESULT}, $hobo->{REAPED} ) =
      ( pop || '', length $_[0] ? $_thaw->(pop) : [], 1 );

   return if $hobo->{IGNORE};

   my ( $exit, $err ) = ( $? || 0, $hobo->{ERROR} );
   my ( $code, $sig ) = ( $exit >> 8, $exit & 0x7f );

   if ( $code > 100 && !$err ) {
      $code = 2, $sig = 1,  $err = 'Hobo received SIGHUP'  if $code == 101;
      $code = 2, $sig = 2,  $err = 'Hobo received SIGINT'  if $code == 102;
      $code = 2, $sig = 6,  $err = 'Hobo received SIGABRT' if $code == 106;
      $code = 2, $sig = 11, $err = 'Hobo received SIGSEGV' if $code == 111;
      $code = 2, $sig = 15, $err = 'Hobo received SIGTERM' if $code == 115;

      $hobo->{ERROR} = $err;
   }

   if ( my $on_finish = $_MNGD->{ $hobo->{PKG} }{on_finish} ) {
      $on_finish->(
         $hobo->{WRK_ID}, $code, $hobo->{ident}, $sig, $err,
         @{ $hobo->{RESULT} }
      );
   }

   return;
}

sub _trap {
   return MCE::Signal::defer($_[0]) if $MCE::Signal::IPC;

   alarm 0; my ( $exit_status, $name ) = ( 2, @_ );
   $_SELF->{SIGNALED} = 1, $name =~ s/^SIG//;

   $SIG{$name} = sub {}, CORE::kill($name, -$$)
      if ( exists $SIG{$name} );

   if    ( $name eq 'HUP'  ) { $exit_status = 101 }
   elsif ( $name eq 'INT'  ) { $exit_status = 102 }
   elsif ( $name eq 'ABRT' ) { $exit_status = 106 }
   elsif ( $name eq 'SEGV' ) { $exit_status = 111 }
   elsif ( $name eq 'TERM' ) { $exit_status = 115 }

   if ( ! $_SELF->{IGNORE} ) {
      my ( $pkg, $wrk_id ) = ( $_SELF->{PKG}, $_SELF->{WRK_ID} );
      $_DATA->{$pkg}->set('R'.$wrk_id, '');
   }

   _exit($exit_status);
}

sub _wait_one {
   my ( $pkg ) = @_;
   my ( $list, $self, $wrk_id ) = ( $_LIST->{$pkg} ); local $!;

   while () {
      for my $hobo ( $list->vals() ) {
         $wrk_id = $hobo->{WRK_ID};
         return  $list->del($wrk_id) if $hobo->{REAPED};
         $self = $list->del($wrk_id), last if waitpid($wrk_id, _WNOHANG);
      }
      last if $self;
      sleep $_yield_secs;
   }

   _reap_hobo($self, 0);

   $self;
}

###############################################################################
## ----------------------------------------------------------------------------
## Delay implementation suited for MCE::Hobo.
##
###############################################################################

package # hide from rpm
   MCE::Hobo::_delay;

sub new {
   my ( $class, $chnl, $delay ) = @_;

   if ( !defined $delay ) {
      $delay = ($^O =~ /mswin|mingw|msys|cygwin/i) ? 0.015 : 0.008;
   }

   $chnl->send(undef);

   bless [ $delay, $chnl ], $class;
}

sub seconds {
   my ( $self, $how_long ) = @_;
   my $delay = defined($how_long) ? $how_long : $self->[0];
   my $lapse = $self->[1]->recv();
   my $time  = MCE::Util::_time();

   if ( !$delay || !defined $lapse ) {
      $lapse = $time;
   }
   elsif ( $lapse + $delay - $time < 0 ) {
      $lapse += int( abs($time - $lapse) / $delay + 0.5 ) * $delay;
   }

   $self->[1]->send( $lapse += $delay );

   return $lapse - $time;
}

###############################################################################
## ----------------------------------------------------------------------------
## Hash and ordhash implementations suited for MCE::Hobo.
##
###############################################################################

package # hide from rpm
   MCE::Hobo::_hash;

use MCE::Shared ();
use Time::HiRes 'sleep';

use constant {
   _WNOHANG => ( $INC{'POSIX.pm'} )
      ? &POSIX::WNOHANG : ( $^O eq 'solaris' ) ? 64 : 1
};

sub new {
   bless \ MCE::Shared->share({ module => 'MCE::Shared::Hash' }), shift;
}

sub clear  { ${ $_[0] }->clear(); }
sub exists { ${ $_[0] }->exists($_[1]); }
sub set    { ${ $_[0] }->set($_[1], $_[2]); }

sub get_done {
   my ( $self ) = @_;
   my @ret;

   for my $key (${ $self }->keys) {
      push @ret, $key if ( substr($key, 0, 1) eq 'R' );
   }

   return @ret;
}

sub get {
   my ( $self, $wrk_id, $wait_flag ) = @_;

   if ( $wait_flag ) {
      local $!;
      ( ${ $self }->exists('R'.$wrk_id) ) ? waitpid($wrk_id, 0) : do {
         while () {
            if ( ! ${ $self }->exists('R'.$wrk_id) ) {
               last if waitpid($wrk_id, _WNOHANG);
               sleep(0.030), next;
            }
            waitpid($wrk_id, 0), last;
         }
      };
   }

   ${ $self }->_get_hobo_data($wrk_id);
}

package # hide from rpm
   MCE::Hobo::_ordhash;

sub new    { bless [ {}, [], {}, 0 ], shift; }  # data, keys, indx, gcnt
sub exists { CORE::exists $_[0]->[0]{ $_[1] }; }
sub get    { $_[0]->[0]{ $_[1] }; }
sub len    { scalar keys %{ $_[0]->[0] }; }

sub clear {
   my ( $self ) = @_;
   %{ $self->[0] } = @{ $self->[1] } = %{ $self->[2] } = (), $self->[3] = 0;

   return;
}

sub del {
   my ( $self, $key ) = @_;
   return undef unless defined( my $off = delete $self->[2]{$key} );

   # tombstone
   $self->[1][$off] = undef;

   # GC keys and refresh index
   if ( ++$self->[3] > @{ $self->[1] } * 0.667 ) {
      my ( $keys, $indx ) = ( $self->[1], $self->[2] );
      my $i; $i = $self->[3] = 0;
      for my $k ( @{ $keys } ) {
         $keys->[$i] = $k, $indx->{$k} = $i++ if defined($k);
      }
      splice @{ $keys }, $i;
   }

   delete $self->[0]{$key};
}

sub set {
   my ( $self, $key ) = @_;
   $self->[0]{$key} = $_[2], return 1 if exists($self->[0]{$key});

   $self->[2]{$key} = @{ $self->[1] }; push @{ $self->[1] }, $key;
   $self->[0]{$key} = $_[2];

   return 1;
}

sub vals {
   my ( $self ) = @_;
   $self->[3]
      ? @{ $self->[0] }{ grep defined($_), @{ $self->[1] } }
      : @{ $self->[0] }{ @{ $self->[1] } };
}

1;

__END__