File Coverage

blib/lib/MCE/Relay.pm

Criterion	Covered	Total	%
statement	153	186	82.2
branch	62	112	55.3
condition	7	12	58.3
subroutine	13	13	100.0
pod	0	3	0.0
total	235	326	72.0

###############################################################################
## ----------------------------------------------------------------------------
## Extends Many-Core Engine with relay capabilities.
##
###############################################################################

package MCE::Relay;

use strict;
use warnings;

no warnings qw( threads recursion uninitialized numeric );

our $VERSION = '1.889';

## no critic (Subroutines::ProhibitSubroutinePrototypes)

use constant {
   OUTPUT_W_RLA => 'W~RLA',  # Worker has relayed
   OUTPUT_R_NFY => 'R~NFY',  # Relay notification
};

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

my $LF = "\012";  Internals::SvREADONLY($LF, 1);
my $_imported;

sub import {

   return if ($_imported++);

   if ($INC{'MCE.pm'}) {
      _mce_m_init();
   }
   else {
      $\ = undef; require Carp;
      Carp::croak(
         "MCE::Relay cannot be used directly. Please consult the MCE::Relay\n".
         "documentation for more information.\n\n"
      );
   }

   return;
}

###############################################################################
## ----------------------------------------------------------------------------
## Output routines for the manager process.
##
###############################################################################

{
   my ($_MCE, $_DAU_R_SOCK_REF, $_DAU_R_SOCK, $_rla_nextid, $_max_workers);

   my %_output_function = (

      OUTPUT_W_RLA.$LF => sub {                   # Worker has relayed

         $_rla_nextid = 0 if ( ++$_rla_nextid == $_max_workers );

         return;
      },

      OUTPUT_R_NFY.$LF => sub {                   # Relay notification

         $_MCE->{_relayed}++;

         return;
      },

   );

   sub _mce_m_loop_begin {

      ($_MCE, $_DAU_R_SOCK_REF) = @_;

      my $_caller = $_MCE->{_caller};

      $_max_workers = (exists $_MCE->{user_tasks})
         ? $_MCE->{user_tasks}[0]{max_workers}
         : $_MCE->{max_workers};

      ## Write initial relay data.
      if (defined $_MCE->{init_relay}) {
         my $_ref = ref $_MCE->{init_relay};

         MCE::_croak("MCE::Relay: (init_relay) is not valid")
            if ($_ref ne '' && $_ref ne 'HASH' && $_ref ne 'ARRAY');

         my $_RLA_W_SOCK = $_MCE->{_rla_w_sock}->[0];
         my $_init_relay;

         $_MCE->{_relayed} = 0;

         if (ref $_MCE->{init_relay} eq '') {
            $_init_relay = $_MCE->{freeze}(\$_MCE->{init_relay}) . '0';
         }
         elsif (ref $_MCE->{init_relay} eq 'HASH') {
            $_init_relay = $_MCE->{freeze}($_MCE->{init_relay}) . '1';
         }
         elsif (ref $_MCE->{init_relay} eq 'ARRAY') {
            $_init_relay = $_MCE->{freeze}($_MCE->{init_relay}) . '2';
         }

         print {$_RLA_W_SOCK} length($_init_relay) . $LF . $_init_relay;

         $_rla_nextid = 0;
      }

      delete $MCE::RLA->{$_caller};

      return;
   }

   sub _mce_m_loop_end {

      ## Obtain final relay data.
      if (defined $_MCE->{init_relay}) {
         my $_RLA_R_SOCK = $_MCE->{_rla_r_sock}->[$_rla_nextid];
         my ($_caller, $_len, $_ret) = ($_MCE->{_caller});

         delete $_MCE->{_relayed};

         MCE::Util::_sock_ready($_RLA_R_SOCK, -1) if $^O eq 'MSWin32';
         chomp($_len = <$_RLA_R_SOCK>);
         read $_RLA_R_SOCK, $_ret, $_len;

         if (chop $_ret) {
            $MCE::RLA->{$_caller} = $_MCE->{thaw}($_ret);
         } else {
            $MCE::RLA->{$_caller} = ${ $_MCE->{thaw}($_ret) };
         }
      }

      ## Clear variables.
      $_MCE = $_DAU_R_SOCK_REF = $_DAU_R_SOCK = undef;
      $_rla_nextid = $_max_workers = undef;

      return;
   }

   sub _mce_m_init {

      MCE::_attach_plugin(
         \%_output_function, \&_mce_m_loop_begin, \&_mce_m_loop_end
      );

      return;
   }
}

###############################################################################
## ----------------------------------------------------------------------------
## Relay methods.
##
###############################################################################

## Items below are folded into MCE.

package # hide from rpm
   MCE;

no warnings qw( threads recursion uninitialized redefine );

use Scalar::Util qw( weaken );

sub relay_final {

   my $x = shift; my $self = ref($x) ? $x : $MCE::MCE;

   _croak('MCE::relay_final: method is not allowed by the worker process')
      if ($self->{_wid});

   my $_caller = caller;

   if (exists $MCE::RLA->{$_caller}) {
      if (ref $MCE::RLA->{$_caller} eq '') {
         return delete $MCE::RLA->{$_caller};
      }
      elsif (ref $MCE::RLA->{$_caller} eq 'HASH') {
         return %{ delete $MCE::RLA->{$_caller} };
      }
      elsif (ref $MCE::RLA->{$_caller} eq 'ARRAY') {
         return @{ delete $MCE::RLA->{$_caller} };
      }

      # should not reach the following line
      delete $MCE::RLA->{$_caller};
   }

   return;
}

sub relay_recv {

   my $x = shift; my $self = ref($x) ? $x : $MCE::MCE;

   _croak('MCE::relay_recv: (init_relay) is not defined')
      unless (defined $self->{init_relay});
   _croak('MCE::relay_recv: method is not allowed by the manager process')
      unless ($self->{_wid});
   _croak('MCE::relay_recv: method is not allowed by task_id > 0')
      if ($self->{_task_id} > 0);

   my ($_chn, $_nxt, $_rdr, $_len, $_ref); local $_;

   local $\ = undef if (defined $\);
   local $/ = $LF   if ($/ ne $LF );

   $_chn = $self->{_chunk_id} || $self->{_wid};
   $_chn = ($_chn - 1) % $self->{max_workers};
   $_nxt = $_chn + 1;
   $_nxt = 0 if ($_nxt == $self->{max_workers});
   $_rdr = $self->{_rla_r_sock}->[$_chn];

   print {$self->{_dat_w_sock}->[0]} OUTPUT_W_RLA.$LF . '0'.$LF;

   MCE::Util::_sock_ready($_rdr, -1) if $^O eq 'MSWin32';
   chomp($_len = <$_rdr>);
   read $_rdr, $_, $_len;
   $_ref = chop $_;

   if ($_ref == 0) {                                 ## scalar value
      $self->{_rla_data} = ${ $self->{thaw}($_) };
      return unless defined wantarray;
      return $self->{_rla_data};
   }
   elsif ($_ref == 1) {                              ## hash reference
      $self->{_rla_data} = $self->{thaw}($_);
      return unless defined wantarray;
      return %{ $self->{_rla_data} };
   }
   elsif ($_ref == 2) {                              ## array reference
      $self->{_rla_data} = $self->{thaw}($_);
      return unless defined wantarray;
      return @{ $self->{_rla_data} };
   }

   return;
}

sub relay (;&) {

   my ($self, $_code);

   if (ref $_[0] eq 'CODE') {
      ($self, $_code) = ($MCE::MCE, shift);
   } else {
      my $x = shift; $self = ref($x) ? $x : $MCE::MCE;
      $_code = shift;
   }

   _croak('MCE::relay: (init_relay) is not defined')
      unless (defined $self->{init_relay});
   _croak('MCE::relay: method is not allowed by the manager process')
      unless ($self->{_wid});
   _croak('MCE::relay: method is not allowed by task_id > 0')
      if ($self->{_task_id} > 0);

   if (ref $_code ne 'CODE') {
      _croak('MCE::relay: argument is not a code block') if (defined $_code);
   } else {
      weaken $_code;
   }

   my ($_chn, $_cid, $_nxt, $_rdr, $_wtr);

   local $\ = undef if (defined $\);
   local $/ = $LF   if ($/ ne $LF );

   $_chn = $_cid = $self->{_chunk_id} || $self->{_wid};
   $_chn = ($_chn - 1) % $self->{max_workers};
   $_nxt = $_chn + 1;
   $_nxt = 0 if ($_nxt == $self->{max_workers});
   $_rdr = $self->{_rla_r_sock}->[$_chn];
   $_wtr = $self->{_rla_w_sock}->[$_nxt];

   if (exists $self->{_rla_data}) {
      my $_tmp; local $_ = delete $self->{_rla_data};
      $_code->() if (ref $_code eq 'CODE');

      if (ref $_ eq '') {                         ## scalar value
         $_tmp = $self->{freeze}(\$_) . '0';
      }
      elsif (ref $_ eq 'HASH') {                  ## hash reference
         $_tmp = $self->{freeze}($_) . '1';
      }
      elsif (ref $_ eq 'ARRAY') {                 ## array reference
         $_tmp = $self->{freeze}($_) . '2';
      }

      print {$_wtr} length($_tmp) . $LF . $_tmp;
      print {$self->{_dat_w_sock}->[0]} OUTPUT_R_NFY.$LF . '0'.$LF;
      $self->{_relayed} = $_cid;
   }
   else {
      my ($_len, $_ref); local $_;
      print {$self->{_dat_w_sock}->[0]} OUTPUT_W_RLA.$LF . '0'.$LF;

      MCE::Util::_sock_ready($_rdr, -1) if $^O eq 'MSWin32';
      chomp($_len = <$_rdr>);
      read $_rdr, $_, $_len;
      $_ref = chop $_;

      if ($_ref == 0) {                              ## scalar value
         my $_ret = ${ $self->{thaw}($_) };
         local $_ = $_ret;      $_code->() if (ref $_code eq 'CODE');
         my $_tmp = $self->{freeze}(\$_) . '0';

         print {$_wtr} length($_tmp) . $LF . $_tmp;
         print {$self->{_dat_w_sock}->[0]} OUTPUT_R_NFY.$LF . '0'.$LF;
         $self->{_relayed} = $_cid;

         return unless defined wantarray;
         return $_ret;
      }
      elsif ($_ref == 1) {                           ## hash reference
         my %_ret = %{ $self->{thaw}($_) };
         local $_ = { %_ret };  $_code->() if (ref $_code eq 'CODE');
         my $_tmp = $self->{freeze}($_) . '1';

         print {$_wtr} length($_tmp) . $LF . $_tmp;
         print {$self->{_dat_w_sock}->[0]} OUTPUT_R_NFY.$LF . '0'.$LF;
         $self->{_relayed} = $_cid;

         return unless defined wantarray;
         return %_ret;
      }
      elsif ($_ref == 2) {                           ## array reference
         my @_ret = @{ $self->{thaw}($_) };
         local $_ = [ @_ret ];  $_code->() if (ref $_code eq 'CODE');
         my $_tmp = $self->{freeze}($_) . '2';

         print {$_wtr} length($_tmp) . $LF . $_tmp;
         print {$self->{_dat_w_sock}->[0]} OUTPUT_R_NFY.$LF . '0'.$LF;
         $self->{_relayed} = $_cid;

         return unless defined wantarray;
         return @_ret;
      }
   }

   return;
}

## Aliases.

*relay_lock   = \&relay_recv;
*relay_unlock = \&relay;

1;

__END__