view release on metacpan or  search on metacpan

lib/MCE.pm  view on Meta::CPAN

###############################################################################
## ----------------------------------------------------------------------------
## MCE - Many-Core Engine for Perl providing parallel processing capabilities.
##
###############################################################################

package MCE;

use strict;
use warnings;

no warnings qw( threads recursion uninitialized );

our $VERSION = '1.904';

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

use Carp ();

my ($_has_threads, $_freeze, $_thaw, $_tid, $_oid);

BEGIN {
   local $@;

   if ( $^O eq 'MSWin32' && ! $INC{'threads.pm'} ) {
      eval 'use threads; use threads::shared;';
   }
   elsif ( $INC{'threads.pm'} && ! $INC{'threads/shared.pm'} ) {
      eval 'use threads::shared;';
   }

   $_has_threads = $INC{'threads.pm'} ? 1 : 0;
   $_tid = $_has_threads ? threads->tid() : 0;
   $_oid = "$$.$_tid";

   if ( $] ge '5.008008' && ! $INC{'PDL.pm'} ) {
      eval 'use Sereal::Encoder 3.015; use Sereal::Decoder 3.015;';
      if ( ! $@ ) {
         my $_encoder_ver = int( Sereal::Encoder->VERSION() );
         my $_decoder_ver = int( Sereal::Decoder->VERSION() );
         if ( $_encoder_ver - $_decoder_ver == 0 ) {
            $_freeze = \&Sereal::Encoder::encode_sereal;
            $_thaw   = \&Sereal::Decoder::decode_sereal;
         }
      }
   }

   if ( ! defined $_freeze ) {
      require Storable;
      $_freeze = \&Storable::freeze;
      $_thaw   = \&Storable::thaw;
   }
}

use IO::Handle ();
use Scalar::Util qw( looks_like_number refaddr reftype weaken );
use Socket qw( SOL_SOCKET SO_RCVBUF );
use Time::HiRes qw( sleep time );

use MCE::Util qw( $LF );
use MCE::Signal ();
use MCE::Mutex ();

our ($MCE, $RLA, $_que_template, $_que_read_size);
our (%_valid_fields_new);

my  ($TOP_HDLR, $_is_MSWin32, $_is_winenv, $_prev_mce);
my  (%_valid_fields_task, %_params_allowed_args);

BEGIN {
   ## Configure pack/unpack template for writing to and from the queue.
   ## Each entry contains 2 positive numbers: chunk_id & msg_id.
   ## Check for >= 64-bit, otherwize fall back to machine's word length.

   $_que_template  = ( ( log(~0+1) / log(2) ) >= 64 ) ? 'Q2' : 'I2';
   $_que_read_size = length pack($_que_template, 0, 0);

   ## Attributes used internally.
   ## _abort_msg _caller _chn _com_lock _dat_lock _mgr_live _rla_data _seed
   ## _chunk_id _pids _run_mode _single_dim _thrs _tids _task_wid _wid _wuf
   ## _exiting _exit_pid _last_sref _total_exited _total_running _total_workers
   ## _send_cnt _sess_dir _spawned _state _status _task _task_id _wrk_status
   ## _init_pid _init_total_workers _pids_t _pids_w _pids_c _relayed
   ##
   ## _bsb_r_sock _bsb_w_sock _com_r_sock _com_w_sock _dat_r_sock _dat_w_sock
   ## _que_r_sock _que_w_sock _rla_r_sock _rla_w_sock _data_channels
   ## _lock_chn   _mutex_n

   %_valid_fields_new = map { $_ => 1 } qw(
      max_workers tmp_dir use_threads user_tasks task_end task_name freeze thaw
      chunk_size input_data sequence job_delay spawn_delay submit_delay RS
      flush_file flush_stderr flush_stdout stderr_file stdout_file use_slurpio
      interval user_args user_begin user_end user_func user_error user_output
      bounds_only gather init_relay on_post_exit on_post_run parallel_io
      loop_timeout max_retries progress posix_exit
   );

lib/MCE.pm  view on Meta::CPAN

   }
   for my $_p (qw( user_begin user_func user_end )) {
      if (defined $_params_ref->{$_p}) {
         $self->{$_p} = delete $_params_ref->{$_p};
         $_requires_shutdown = 1;
      }
   }
   for my $_p (keys %{ $_params_ref }) {
      _croak("MCE::_sync_params: ($_p) is not a valid params argument")
         unless (exists $_params_allowed_args{$_p});

      $self->{$_p} = $_params_ref->{$_p};
   }

   return ($self->{_spawned}) ? $_requires_shutdown : 0;
}

###############################################################################
## ----------------------------------------------------------------------------
## Dispatch methods.
##
###############################################################################

sub _dispatch {
   my @_args = @_; my $_is_thread = shift @_args;
   my $self = $MCE = $_args[0];

   ## To avoid (Scalars leaked: N) messages; fixed in Perl 5.12.x
   @_ = ();

   $ENV{'PERL_MCE_IPC'} = 'win32' if ( $_is_MSWin32 && (
      defined($self->{max_retries}) ||
      $INC{'MCE/Child.pm'} ||
      $INC{'MCE/Hobo.pm'}
   ));

   delete $self->{_relayed};

   $self->{_is_thread} = $_is_thread;
   $self->{_pid}       = $_is_thread ? $$ .'.'. threads->tid() : $$;

   if (!$self->{use_threads}) {
      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 (non-thread workers only). Ditto for Math::Prime::Util,
   # Math::Random, Math::Random::MT::Auto, and PDL.
   #
   # MCE 1.892, 2024-06-08: Enable predictability running threads.
   # Output matches non-threads for CORE, Math::Prime::Util, and
   # Math::Random::MT::Auto. https://perlmonks.org/?node_id=11159834

   {
      my $_wid  = $_args[1];
      my $_seed = abs($self->{_seed} - ($_wid * 100000)) % 2147483560;

      CORE::srand($_seed) if (!$self->{use_threads} || $] ge '5.020000'); # drand48
      Math::Prime::Util::srand($_seed) if $INC{'Math/Prime/Util.pm'};

      # [etj] identified a race condition in PDL running threads
      # https://perlmonks.org/?node_id=11159841
      if (!$self->{use_threads}) {
         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+
      }
   }

   if (!$self->{use_threads} && $INC{'Math/Random.pm'}) {
      my ($_wid, $_cur_seed) = ($_args[1], Math::Random::random_get_seed());

      my $_new_seed = ($_cur_seed < 1073741781)
         ? $_cur_seed + (($_wid * 100000) % 1073741780)
         : $_cur_seed - (($_wid * 100000) % 1073741780);

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

   if ($INC{'Math/Random/MT/Auto.pm'}) {
      my ($_wid, $_cur_seed) = (
         $_args[1], Math::Random::MT::Auto::get_seed()->[0]
      );
      my $_new_seed = ($_cur_seed < 1073741781)
         ? $_cur_seed + (($_wid * 100000) % 1073741780)
         : $_cur_seed - (($_wid * 100000) % 1073741780);

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

   ## Run.

   _worker_main(@_args, \@_plugin_worker_init);

   _exit($self);
}

sub _dispatch_thread {
   my ($self, $_wid, $_task, $_task_id, $_task_wid, $_params) = @_;

   @_ = (); local $_;

   my $_thr = threads->create( \&_dispatch,
      1, $self, $_wid, $_task, $_task_id, $_task_wid, $_params
   );

   _croak("MCE::_dispatch_thread: Failed to spawn worker $_wid: $!")
      if (!defined $_thr);

   ## Store into an available slot (restart), otherwise append to arrays.
   if (defined $_params) { for my $_i (0 .. @{ $self->{_tids} } - 1) {
      unless (defined $self->{_tids}->[$_i]) {
         $self->{_thrs}->[$_i] = $_thr;
         $self->{_tids}->[$_i] = $_thr->tid();
         return;
      }
   }}

   push @{ $self->{_thrs} }, $_thr;