File Coverage

blib/lib/MCE/Step.pm

Criterion	Covered	Total	%
statement	256	348	73.5
branch	116	238	48.7
condition	45	97	46.3
subroutine	21	26	80.7
pod	5	9	55.5
total	443	718	61.7

###############################################################################
## ----------------------------------------------------------------------------
## Parallel step model for building creative steps.
##
###############################################################################

package MCE::Step;

use strict;
use warnings;

no warnings qw( threads recursion uninitialized );

our $VERSION = '1.888';

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

use Scalar::Util qw( looks_like_number );

use MCE;
use MCE::Queue;

our @CARP_NOT = qw( MCE );

my $_tid = $INC{'threads.pm'} ? threads->tid() : 0;

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

###############################################################################
## ----------------------------------------------------------------------------
## Import routine.
##
###############################################################################

my ($_MCE, $_def, $_params, $_tag) = ({}, {}, {}, 'MCE::Step');
my ($_prev_c, $_prev_n, $_prev_t, $_prev_w) = ({}, {}, {}, {});
my ($_user_tasks, $_queue, $_last_task_id, $_lkup) = ({}, {}, {}, {});

sub import {
   my ($_class, $_pkg) = (shift, caller);

   my $_p = $_def->{$_pkg} = {
      MAX_WORKERS => 'auto',
      CHUNK_SIZE  => 'auto',
   };

   ## Import functions.
   if ($_pkg !~ /^MCE::/) {
      no strict 'refs'; no warnings 'redefine';
      *{ $_pkg.'::mce_step_f' } = \&run_file;
      *{ $_pkg.'::mce_step_s' } = \&run_seq;
      *{ $_pkg.'::mce_step'   } = \&run;
   }

   ## Process module arguments.
   while ( my $_argument = shift ) {
      my $_arg = lc $_argument;

      $_p->{MAX_WORKERS} = shift, next if ( $_arg eq 'max_workers' );
      $_p->{CHUNK_SIZE}  = shift, next if ( $_arg eq 'chunk_size' );
      $_p->{TMP_DIR}     = shift, next if ( $_arg eq 'tmp_dir' );
      $_p->{FREEZE}      = shift, next if ( $_arg eq 'freeze' );
      $_p->{THAW}        = shift, next if ( $_arg eq 'thaw' );
      $_p->{INIT_RELAY}  = shift, next if ( $_arg eq 'init_relay' );
      $_p->{USE_THREADS} = shift, next if ( $_arg eq 'use_threads' );

                           shift, next if ( $_arg eq 'fast' ); # ignored

      ## Sereal 3.015+, if available, is used automatically by MCE 1.8+.
      if ( $_arg eq 'sereal' ) {
         if ( shift eq '0' ) {
            require Storable;
            $_p->{FREEZE} = \&Storable::freeze;
            $_p->{THAW}   = \&Storable::thaw;
         }
         next;
      }

      _croak("Error: ($_argument) invalid module option");
   }

   $_p->{MAX_WORKERS} = MCE::_parse_max_workers($_p->{MAX_WORKERS});

   MCE::_validate_number($_p->{MAX_WORKERS}, 'MAX_WORKERS', $_tag);
   MCE::_validate_number($_p->{CHUNK_SIZE}, 'CHUNK_SIZE', $_tag)
      unless ($_p->{CHUNK_SIZE} eq 'auto');

   return;
}

###############################################################################
## ----------------------------------------------------------------------------
## The task end callback for when a task completes.
##
###############################################################################

sub _task_end {

   my ($_mce, $_task_id, $_task_name) = @_;
   my $_pid = $_mce->{_init_pid}.'.'.$_mce->{_caller};

   if (defined $_mce->{user_tasks}->[$_task_id + 1]) {
      my $n_workers = $_mce->{user_tasks}->[$_task_id + 1]->{max_workers};
      $_queue->{$_pid}[$_task_id]->enqueue((undef) x $n_workers);
   }

   $_params->{task_end}->($_mce, $_task_id, $_task_name)
      if (exists $_params->{task_end} && ref $_params->{task_end} eq 'CODE');

   return;
}

###############################################################################
## ----------------------------------------------------------------------------
## Methods for MCE; step, enq, enqp, await.
##
###############################################################################

{
   no warnings 'redefine';

   sub MCE::step {

      my $x = shift; my $self = ref($x) ? $x : $MCE::MCE;
      my $_pid = $self->{_init_pid}.'.'.$self->{_caller};

      _croak('MCE::step: method is not allowed by the manager process')
         unless ($self->{_wid});

      my $_task_id = $self->{_task_id};

      if ($_task_id < $_last_task_id->{$_pid}) {
         $_queue->{$_pid}[$_task_id]->enqueue($self->freeze([ @_ ]));
      }
      else {
         _croak('MCE::step: method is not allowed by the last task');
      }

      return;
   }

   ############################################################################

   sub MCE::enq {

      my $x = shift; my $self = ref($x) ? $x : $MCE::MCE;
      my $_pid = $self->{_init_pid}.'.'.$self->{_caller};
      my $_name = shift;

      _croak('MCE::enq: method is not allowed by the manager process')
         unless ($self->{_wid});
      _croak('MCE::enq: (task_name) is not specified or valid')
         if (!defined $_name || !exists $_lkup->{$_pid}{$_name});
      _croak('MCE::enq: stepping to same task or backwards is not allowed')
         if ($_lkup->{$_pid}{$_name} <= $self->{_task_id});

      my $_task_id = $_lkup->{$_pid}{$_name} - 1;

      if ($_task_id < $_last_task_id->{$_pid}) {
         if (scalar @_ > 1) {
            my @_items = map { $self->freeze([ $_ ]) } @_;
            $_queue->{$_pid}[$_task_id]->enqueue(@_items);
         }
         else {
            $_queue->{$_pid}[$_task_id]->enqueue($self->freeze([ @_ ]));
         }
      }
      else {
         _croak('MCE::enq: method is not allowed by the last task');
      }

      return;
   }

   ############################################################################

   sub MCE::enqp {

      my $x = shift; my $self = ref($x) ? $x : $MCE::MCE;
      my $_pid = $self->{_init_pid}.'.'.$self->{_caller};
      my ($_name, $_p) = (shift, shift);

      _croak('MCE::enqp: method is not allowed by the manager process')
         unless ($self->{_wid});
      _croak('MCE::enqp: (task_name) is not specified or valid')
         if (!defined $_name || !exists $_lkup->{$_pid}{$_name});
      _croak('MCE::enqp: stepping to same task or backwards is not allowed')
         if ($_lkup->{$_pid}{$_name} <= $self->{_task_id});
      _croak('MCE::enqp: (priority) is not an integer')
         if (!looks_like_number($_p) || int($_p) != $_p);

      my $_task_id = $_lkup->{$_pid}{$_name} - 1;

      if ($_task_id < $_last_task_id->{$_pid}) {
         if (scalar @_ > 1) {
            my @_items = map { $self->freeze([ $_ ]) } @_;
            $_queue->{$_pid}[$_task_id]->enqueuep($_p, @_items);
         }
         else {
            $_queue->{$_pid}[$_task_id]->enqueuep($_p, $self->freeze([ @_ ]));
         }
      }
      else {
         _croak('MCE::enqp: method is not allowed by the last task');
      }

      return;
   }

   ############################################################################

   sub MCE::await {

      my $x = shift; my $self = ref($x) ? $x : $MCE::MCE;
      my $_pid = $self->{_init_pid}.'.'.$self->{_caller};
      my $_name = shift;

      _croak('MCE::await: method is not allowed by the manager process')
         unless ($self->{_wid});
      _croak('MCE::await: (task_name) is not specified or valid')
         if (!defined $_name || !exists $_lkup->{$_pid}{$_name});
      _croak('MCE::await: awaiting from same task or backwards is not allowed')
         if ($_lkup->{$_pid}{$_name} <= $self->{_task_id});

      my $_task_id = $_lkup->{$_pid}{$_name} - 1;  my $_t = shift || 0;

      _croak('MCE::await: (threshold) is not an integer')
         if (!looks_like_number($_t) || int($_t) != $_t);

      if ($_task_id < $_last_task_id->{$_pid}) {
         $_queue->{$_pid}[$_task_id]->await($_t);
      } else {
         _croak('MCE::await: method is not allowed by the last task');
      }

      return;
   }

}

###############################################################################
## ----------------------------------------------------------------------------
## Init and finish routines.
##
###############################################################################

sub init (@) {

   shift if (defined $_[0] && $_[0] eq 'MCE::Step');
   my $_pkg = "$$.$_tid.".caller();

   $_params->{$_pkg} = (ref $_[0] eq 'HASH') ? shift : { @_ };

   @_ = ();

   return;
}

sub finish (@) {

   shift if (defined $_[0] && $_[0] eq 'MCE::Step');
   my $_pkg = (defined $_[0]) ? shift : "$$.$_tid.".caller();

   if ( $_pkg eq 'MCE' ) {
      for my $_k ( keys %{ $_MCE } ) { MCE::Step->finish($_k, 1); }
   }
   elsif ( $_MCE->{$_pkg} && $_MCE->{$_pkg}{_init_pid} eq "$$.$_tid" ) {
      $_MCE->{$_pkg}->shutdown(@_) if $_MCE->{$_pkg}{_spawned};

      delete $_lkup->{$_pkg};
      delete $_last_task_id->{$_pkg};

      delete $_user_tasks->{$_pkg};
      delete $_prev_c->{$_pkg};
      delete $_prev_n->{$_pkg};
      delete $_prev_t->{$_pkg};
      delete $_prev_w->{$_pkg};
      delete $_MCE->{$_pkg};

      if (defined $_queue->{$_pkg}) {
         local $_;
         $_->DESTROY() for (@{ $_queue->{$_pkg} });
         delete $_queue->{$_pkg};
      }
   }

   @_ = ();

   return;
}

###############################################################################
## ----------------------------------------------------------------------------
## Parallel step with MCE -- file.
##
###############################################################################

sub run_file (@) {

   shift if (defined $_[0] && $_[0] eq 'MCE::Step');

   my ($_file, $_pos); my $_start_pos = (ref $_[0] eq 'HASH') ? 2 : 1;
   my $_pid = "$$.$_tid.".caller();

   if (defined (my $_p = $_params->{$_pid})) {
      delete $_p->{input_data} if (exists $_p->{input_data});
      delete $_p->{sequence}   if (exists $_p->{sequence});
   }
   else {
      $_params->{$_pid} = {};
   }

   for my $_i ($_start_pos .. @_ - 1) {
      my $_r = ref $_[$_i];
      if ($_r eq '' || $_r eq 'SCALAR' || $_r =~ /^(?:GLOB|FileHandle|IO::)/) {
         $_file = $_[$_i]; $_pos = $_i;
         last;
      }
   }

   if (defined $_file && ref $_file eq '' && $_file ne '') {
      _croak("$_tag: ($_file) does not exist")      unless (-e $_file);
      _croak("$_tag: ($_file) is not readable")     unless (-r $_file);
      _croak("$_tag: ($_file) is not a plain file") unless (-f $_file);
      $_params->{$_pid}{_file} = $_file;
   }
   elsif (ref $_file eq 'SCALAR' || ref($_file) =~ /^(?:GLOB|FileHandle|IO::)/) {
      $_params->{$_pid}{_file} = $_file;
   }
   else {
      _croak("$_tag: (file) is not specified or valid");
   }

   if (defined $_pos) {
      pop @_ for ($_pos .. @_ - 1);
   }

   return run(@_);
}

###############################################################################
## ----------------------------------------------------------------------------
## Parallel step with MCE -- sequence.
##
###############################################################################

sub run_seq (@) {

   shift if (defined $_[0] && $_[0] eq 'MCE::Step');

   my ($_begin, $_end, $_pos); my $_start_pos = (ref $_[0] eq 'HASH') ? 2 : 1;
   my $_pid = "$$.$_tid.".caller();

   if (defined (my $_p = $_params->{$_pid})) {
      delete $_p->{sequence}   if (exists $_p->{sequence});
      delete $_p->{input_data} if (exists $_p->{input_data});
      delete $_p->{_file}      if (exists $_p->{_file});
   }
   else {
      $_params->{$_pid} = {};
   }

   for my $_i ($_start_pos .. @_ - 1) {
      my $_r = ref $_[$_i];

      if ($_r eq '' || $_r =~ /^Math::/ || $_r eq 'HASH' || $_r eq 'ARRAY') {
         $_pos = $_i;

         if ($_r eq '' || $_r =~ /^Math::/) {
            $_begin = $_[$_pos], $_end = $_[$_pos + 1];
            $_params->{$_pid}{sequence} = [
               $_[$_pos], $_[$_pos + 1], $_[$_pos + 2], $_[$_pos + 3]
            ];
         }
         elsif ($_r eq 'HASH') {
            $_begin = $_[$_pos]->{begin}, $_end = $_[$_pos]->{end};
            $_params->{$_pid}{sequence} = $_[$_pos];
         }
         elsif ($_r eq 'ARRAY') {
            $_begin = $_[$_pos]->[0], $_end = $_[$_pos]->[1];
            $_params->{$_pid}{sequence} = $_[$_pos];
         }

         last;
      }
   }

   _croak("$_tag: (sequence) is not specified or valid")
      unless (exists $_params->{$_pid}{sequence});
   _croak("$_tag: (begin) is not specified for sequence")
      unless (defined $_begin);
   _croak("$_tag: (end) is not specified for sequence")
      unless (defined $_end);

   $_params->{$_pid}{sequence_run} = undef;

   if (defined $_pos) {
      pop @_ for ($_pos .. @_ - 1);
   }

   return run(@_);
}

###############################################################################
## ----------------------------------------------------------------------------
## Parallel step with MCE.
##
###############################################################################

sub run (@) {

   shift if (defined $_[0] && $_[0] eq 'MCE::Step');

   my $_pkg = caller() eq 'MCE::Step' ? caller(1) : caller();
   my $_pid = "$$.$_tid.$_pkg";

   if (ref $_[0] eq 'HASH') {
      $_params->{$_pid} = {} unless defined $_params->{$_pid};
      for my $_p (keys %{ $_[0] }) {
         $_params->{$_pid}{$_p} = $_[0]->{$_p};
      }

      shift;
   }

   ## -------------------------------------------------------------------------

   my (@_code, @_name, @_thrs, @_wrks); my $_init_mce = 0; my $_pos = 0;

   %{ $_lkup->{$_pid} } = ();

   while (ref $_[0] eq 'CODE') {
      push @_code, $_[0];

      if (defined (my $_p = $_params->{$_pid})) {
         push @_name, (ref $_p->{task_name} eq 'ARRAY')
            ? $_p->{task_name}->[$_pos] : undef;
         push @_thrs, (ref $_p->{use_threads} eq 'ARRAY')
            ? $_p->{use_threads}->[$_pos] : undef;
         push @_wrks, (ref $_p->{max_workers} eq 'ARRAY')
            ? $_p->{max_workers}->[$_pos] : undef;
      }

      $_lkup->{$_pid}{ $_name[ $_pos ] } = $_pos if (defined $_name[ $_pos ]);

      $_init_mce = 1 if (
         !defined $_prev_c->{$_pid}[$_pos] ||
         $_prev_c->{$_pid}[$_pos] != $_code[$_pos]
      );

      $_init_mce = 1 if ($_prev_n->{$_pid}[$_pos] ne $_name[$_pos]);
      $_init_mce = 1 if ($_prev_t->{$_pid}[$_pos] ne $_thrs[$_pos]);
      $_init_mce = 1 if ($_prev_w->{$_pid}[$_pos] ne $_wrks[$_pos]);

      $_prev_c->{$_pid}[$_pos] = $_code[$_pos];
      $_prev_n->{$_pid}[$_pos] = $_name[$_pos];
      $_prev_t->{$_pid}[$_pos] = $_thrs[$_pos];
      $_prev_w->{$_pid}[$_pos] = $_wrks[$_pos];

      shift; $_pos++;
   }

   if (defined $_prev_c->{$_pid}[$_pos]) {
      pop @{ $_prev_c->{$_pid} } for ($_pos .. $#{ $_prev_c->{$_pid } });
      pop @{ $_prev_n->{$_pid} } for ($_pos .. $#{ $_prev_n->{$_pid } });
      pop @{ $_prev_t->{$_pid} } for ($_pos .. $#{ $_prev_t->{$_pid } });
      pop @{ $_prev_w->{$_pid} } for ($_pos .. $#{ $_prev_w->{$_pid } });

      $_init_mce = 1;
   }

   return unless (scalar @_code);

   ## -------------------------------------------------------------------------

   my $_input_data; my $_max_workers = $_def->{$_pkg}{MAX_WORKERS};
   my $_r = ref $_[0];

   if (@_ == 1 && $_r =~ /^(?:ARRAY|HASH|SCALAR|GLOB|FileHandle|IO::)/) {
      $_input_data = shift;
   }

   if (defined (my $_p = $_params->{$_pid})) {
      $_max_workers = MCE::_parse_max_workers($_p->{max_workers})
         if (exists $_p->{max_workers} && ref $_p->{max_workers} ne 'ARRAY');

      delete $_p->{sequence}   if (defined $_input_data || scalar @_);
      delete $_p->{user_func}  if (exists $_p->{user_func});
      delete $_p->{user_tasks} if (exists $_p->{user_tasks});
   }

   if (@_code > 1 && $_max_workers > 1) {
      $_max_workers = int($_max_workers / @_code + 0.5) + 1;
   }

   my $_chunk_size = MCE::_parse_chunk_size(
      $_def->{$_pkg}{CHUNK_SIZE}, $_max_workers, $_params->{$_pid},
      $_input_data, scalar @_
   );

   if (defined (my $_p = $_params->{$_pid})) {
      if (exists $_p->{_file}) {
         $_input_data = delete $_p->{_file};
      } else {
         $_input_data = $_p->{input_data} if exists $_p->{input_data};
      }
   }

   ## -------------------------------------------------------------------------

   MCE::_save_state($_MCE->{$_pid});

   if ($_init_mce || !exists $_queue->{$_pid}) {
      $_MCE->{$_pid}->shutdown() if (defined $_MCE->{$_pid});
      $_queue->{$_pid} = [] if (!defined $_queue->{$_pid});

      my $_Q = $_queue->{$_pid};
      pop(@{ $_Q })->DESTROY for (@_code .. @{ $_Q });

      push @{ $_Q }, MCE::Queue->new(await => 1)
         for (@{ $_Q } .. @_code - 2);

      $_last_task_id->{$_pid} = @_code - 1;

      ## must clear arrays for nested session to work with Perl < v5.14
      _gen_user_tasks($_pid,$_Q, [@_code],[@_name],[@_thrs],[@_wrks], $_chunk_size);

      @_code = @_name = @_thrs = @_wrks = ();

      my %_opts = (
         max_workers => $_max_workers, task_name => $_tag,
         user_tasks  => $_user_tasks->{$_pid}, task_end  => \&_task_end,
      );

      if (defined (my $_p = $_params->{$_pid})) {
         local $_;

         for (keys %{ $_p }) {
            next if ($_ eq 'max_workers' && ref $_p->{max_workers} eq 'ARRAY');
            next if ($_ eq 'task_name'   && ref $_p->{task_name}   eq 'ARRAY');
            next if ($_ eq 'use_threads' && ref $_p->{use_threads} eq 'ARRAY');

            next if ($_ eq 'chunk_size');
            next if ($_ eq 'input_data');
            next if ($_ eq 'sequence_run');
            next if ($_ eq 'task_end');

            _croak("$_tag: ($_) is not a valid constructor argument")
               unless (exists $MCE::_valid_fields_new{$_});

            $_opts{$_} = $_p->{$_};
         }
      }

      for my $_k (qw/ tmp_dir freeze thaw init_relay use_threads /) {
         $_opts{$_k} = $_def->{$_pkg}{uc($_k)}
            if (exists $_def->{$_pkg}{uc($_k)} && !exists $_opts{$_k});
      }

      $_MCE->{$_pid} = MCE->new(pkg => $_pkg, %_opts);
   }
   else {
      ## Workers may persist after running. Thus, updating the MCE instance.
      ## These options do not require respawning.
      if (defined (my $_p = $_params->{$_pid})) {
         for my $_k (qw(
            RS interval stderr_file stdout_file user_error user_output
            job_delay submit_delay on_post_exit on_post_run user_args
            flush_file flush_stderr flush_stdout gather max_retries
         )) {
            $_MCE->{$_pid}{$_k} = $_p->{$_k} if (exists $_p->{$_k});
         }
      }
   }

   ## -------------------------------------------------------------------------

   my @_a; my $_wa = wantarray; $_MCE->{$_pid}{gather} = \@_a if (defined $_wa);

   if (defined $_input_data) {
      @_ = ();
      $_MCE->{$_pid}->process({ chunk_size => $_chunk_size }, $_input_data);
      delete $_MCE->{$_pid}{input_data};
   }
   elsif (scalar @_) {
      $_MCE->{$_pid}->process({ chunk_size => $_chunk_size }, \@_);
      delete $_MCE->{$_pid}{input_data};
   }
   else {
      if (defined $_params->{$_pid} && exists $_params->{$_pid}{sequence}) {
         $_MCE->{$_pid}->run({
             chunk_size => $_chunk_size,
             sequence   => $_params->{$_pid}{sequence}
         }, 0);
         if (exists $_params->{$_pid}{sequence_run}) {
             delete $_params->{$_pid}{sequence_run};
             delete $_params->{$_pid}{sequence};
         }
         delete $_MCE->{$_pid}{sequence};
      }
      else {
         $_MCE->{$_pid}->run({ chunk_size => $_chunk_size }, 0);
      }
   }

   MCE::_restore_state();

   # destroy queue(s) if MCE::run requested workers to shutdown
   if (!$_MCE->{$_pid}{_spawned}) {
      $_->DESTROY() for @{ $_queue->{$_pid} };
      delete $_queue->{$_pid};
   }

   delete $_MCE->{$_pid}{gather} if (defined $_wa);

   return ((defined $_wa) ? @_a : ());
}

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

sub _croak {

   goto &MCE::_croak;
}

sub _gen_user_func {

   my ($_qref, $_cref, $_chunk_size, $_pos) = @_;

   my $_q_in = $_qref->[$_pos - 1];
   my $_code = $_cref->[$_pos];

   return sub {
      my ($_mce) = @_;

      $_mce->{_next_jmp} = sub { goto _MCE_STEP__NEXT; };
      $_mce->{_last_jmp} = sub { goto _MCE_STEP__LAST; };

      _MCE_STEP__NEXT:

      while (defined (local $_ = $_q_in->dequeue())) {
         my $_args = $_mce->thaw($_);  $_ = $_args->[0];
         $_code->($_mce, @{ $_args });
      }

      _MCE_STEP__LAST:

      return;
   };
}

sub _gen_user_tasks {

   my ($_pid, $_qref, $_cref, $_nref, $_tref, $_wref, $_chunk_size) = @_;

   @{ $_user_tasks->{$_pid} } = ();

   push @{ $_user_tasks->{$_pid} }, {
      task_name   => $_nref->[0],
      use_threads => $_tref->[0],
      max_workers => $_wref->[0],
      user_func   => sub { $_cref->[0]->(@_); return; }
   };

   for my $_pos (1 .. @{ $_cref } - 1) {
      push @{ $_user_tasks->{$_pid} }, {
         task_name   => $_nref->[$_pos],
         use_threads => $_tref->[$_pos],
         max_workers => $_wref->[$_pos],
         user_func   => _gen_user_func(
            $_qref, $_cref, $_chunk_size, $_pos
         )
      };
   }

   return;
}

1;

__END__