File Coverage

blib/lib/MCE/Shared.pm

Criterion	Covered	Total	%
statement	167	258	64.7
branch	84	192	43.7
condition	36	101	35.6
subroutine	23	27	85.1
pod	2	2	100.0
total	312	580	53.7

###############################################################################
## ----------------------------------------------------------------------------
## MCE extension for sharing data supporting threads and processes.
##
###############################################################################

package MCE::Shared;

use strict;
use warnings;

use 5.010001;

no warnings qw( threads recursion uninitialized once );

our $VERSION = '1.881';

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

use Carp ();

$Carp::Internal{ (__PACKAGE__) }++;

no overloading;

use MCE::Mutex ();
use MCE::Shared::Server ();
use Scalar::Util qw( blessed );

our @CARP_NOT = qw(
   MCE::Shared::Array    MCE::Shared::Hash     MCE::Shared::Queue
   MCE::Shared::Cache    MCE::Shared::Minidb   MCE::Shared::Scalar
   MCE::Shared::Condvar  MCE::Shared::Object   MCE::Shared::Sequence
   MCE::Shared::Handle   MCE::Shared::Ordhash  MCE::Shared::Server
);

sub import {
   no strict 'refs'; no warnings 'redefine';
   *{ caller().'::mce_open' } = \&open;

   return;
}

my $_share_deeply = 0;

###############################################################################
## ----------------------------------------------------------------------------
## Share function.
##
###############################################################################

sub share {
   shift if (defined $_[0] && $_[0] eq 'MCE::Shared');

   # construction via module option
   if ( ref $_[0] eq 'HASH' && $_[0]->{module} ) {
      my $_params = shift;
      my $_class  = $_params->{module};

      return MCE::Shared->condvar(@_) if ( $_class eq 'MCE::Shared::Condvar' );
      return MCE::Shared->handle(@_)  if ( $_class eq 'MCE::Shared::Handle' );
      return MCE::Shared->queue(@_)   if ( $_class eq 'MCE::Shared::Queue' );

      $_params->{class} = ':construct_module:';

      my $_obj = MCE::Shared::Server::_new(
         $_params, [ @_, delete $_params->{new} || 'new' ]
      );

      $_obj->[6] = MCE::Mutex->new( impl => 'Channel' ) unless (
         caller->isa('MCE::Hobo::_hash') || exists( $_params->{_DEEPLY_} )
      );

      return $_obj;
   }

   my $_params = ref $_[0] eq 'HASH' && ref $_[1] ? shift : {};
   my $_class  = blessed($_[0]);
   my $_obj;

   # class construction failed: e.g. share( class->new(...) )
   return '' if @_ && !$_[0] && $!;

   $_share_deeply = 1 if $_params->{_DEEPLY_};

   # blessed object, \@array, \%hash, or \$scalar
   if ( $_class ) {
      _incr_count($_[0]), return $_[0] if $_[0]->can('SHARED_ID');
      $_params->{'class'} = $_class;

      $_obj = MCE::Shared::Server::_new($_params, $_[0]);

      $_obj->[6] = MCE::Mutex->new( impl => 'Channel' )
         unless ( exists $_params->{_DEEPLY_} );
   }
   elsif ( ref $_[0] eq 'ARRAY' ) {
      if ( tied(@{ $_[0] }) && tied(@{ $_[0] })->can('SHARED_ID') ) {
         _incr_count(tied(@{ $_[0] })), return tied(@{ $_[0] });
      }
      $_obj = MCE::Shared->array($_params, @{ $_[0] });
      @{ $_[0] } = ();  tie @{ $_[0] }, 'MCE::Shared::Object', $_obj;
   }
   elsif ( ref $_[0] eq 'HASH' ) {
      if ( tied(%{ $_[0] }) && tied(%{ $_[0] })->can('SHARED_ID') ) {
         _incr_count(tied(%{ $_[0] })), return tied(%{ $_[0] });
      }
      $_obj = MCE::Shared->hash($_params, %{ $_[0] });
      %{ $_[0] } = ();  tie %{ $_[0] }, 'MCE::Shared::Object', $_obj;
   }
   elsif ( ref $_[0] eq 'SCALAR' && !ref ${ $_[0] } ) {
      if ( tied(${ $_[0] }) && tied(${ $_[0] })->can('SHARED_ID') ) {
         _incr_count(tied(${ $_[0] })), return tied(${ $_[0] });
      }
      $_obj = MCE::Shared->scalar($_params, ${ $_[0] });
      undef ${ $_[0] }; tie ${ $_[0] }, 'MCE::Shared::Object', $_obj;
   }

   # synopsis
   elsif ( ref $_[0] eq 'REF' ) {
      _croak('A "REF" type is not supported');
   }
   else {
      if ( ref $_[0] eq 'GLOB' ) {
         _incr_count(tied(*{ $_[0] })), return $_[0] if (
            tied(*{ $_[0] }) && tied(*{ $_[0] })->can('SHARED_ID')
         );
      }
      _croak('Synopsis: blessed object, \@array, \%hash, or \$scalar');
   }

   return $_obj;
}

###############################################################################
## ----------------------------------------------------------------------------
## Public functions.
##
###############################################################################

our $AUTOLOAD; # MCE::Shared::

sub AUTOLOAD {
   my $_fcn = $AUTOLOAD;  substr($_fcn, 0, rindex($_fcn,':') + 1, '');

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

   return MCE::Shared::Object::_init(@_) if $_fcn eq 'init';
   return MCE::Shared::Server::_start() if $_fcn eq 'start';
   return MCE::Shared::Server::_stop() if $_fcn eq 'stop';
   return MCE::Shared::Server::_pid() if $_fcn eq 'pid';

   if ( $_fcn eq 'array' || $_fcn eq 'hash' ) {
      _use( 'MCE::Shared::'.ucfirst($_fcn) );
      my $_params = ref $_[0] eq 'HASH' ? shift : {};

      $_params->{module} = ( $_fcn eq 'array' )
         ? 'MCE::Shared::Array' : 'MCE::Shared::Hash';

      my $_obj = &share($_params);
      delete $_params->{module};

      if ( scalar @_ ) {
         if ( $_share_deeply ) {
            $_params->{_DEEPLY_} = 1;
            if ( $_fcn eq 'array' ) {
               for ( my $i = 0; $i <= $#_; $i += 1 ) {
                  &_share($_params, $_obj, $_[$i]) if ref($_[$i]);
               }
            } else {
               for ( my $i = 1; $i <= $#_; $i += 2 ) {
                  &_share($_params, $_obj, $_[$i]) if ref($_[$i]);
               }
            }
         }
         $_obj->assign(@_);
      }

      $_share_deeply = 0;

      return $_obj;
   }
   elsif ( $_fcn eq 'handle' ) {
      require MCE::Shared::Handle unless $INC{'MCE/Shared/Handle.pm'};

      my $_obj = &share( MCE::Shared::Handle->new([]) );
      my $_fh  = \do { no warnings 'once'; local *FH };

      tie *{ $_fh }, 'MCE::Shared::Object', $_obj;
      if ( @_ ) { $_obj->OPEN(@_) or return ''; }

      return $_fh;
   }
   elsif ( $_fcn eq 'pdl' || $_fcn =~ /^pdl_(s?byte|u?short|u?long|indx|u?longlong|float|l?double|sequence|zeroe?s|ones|g?random)$/ ) {

      $_fcn = $1 if ( $_fcn ne 'pdl' );
      push @_, $_fcn; _use('PDL') or _croak($@);

      my $_obj = MCE::Shared::Server::_new(
         { 'class' => ':construct_pdl:' }, [ @_ ]
      );

      $_obj->[6] = MCE::Mutex->new( impl => 'Channel' );

      return $_obj;
   }

   # cache, condvar, minidb, ordhash, queue, scalar, sequence, et cetera
   $_fcn = 'sequence'  if $_fcn eq 'num_sequence';
   my $_pkg = ucfirst( lc $_fcn ); local $@;

   if ( $INC{"MCE/Shared/$_pkg.pm"} || eval "use MCE::Shared::$_pkg (); 1" ) {
      $_pkg = "MCE::Shared::$_pkg";

      return &share({}, $_pkg->new(@_)) if ( $_fcn =~ /^(?:condvar|queue)$/ );
      return &share({ module => $_pkg }, @_);
   }

   _croak("Can't locate object method \"$_fcn\" via package \"MCE::Shared\"");
}

sub open (@) {
   shift if ( defined $_[0] && $_[0] eq 'MCE::Shared' );
   require MCE::Shared::Handle unless $INC{'MCE/Shared/Handle.pm'};

   my $_obj;
   if ( ref $_[0] eq 'GLOB' && tied *{ $_[0] } &&
        ref tied(*{ $_[0] }) eq 'MCE::Shared::Object' ) {

      $_obj = tied *{ $_[0] };
   }
   elsif ( @_ ) {
      if ( ref $_[0] eq 'GLOB' && tied *{ $_[0] } ) {
         close $_[0] if defined ( fileno $_[0] );
      }
      $_obj = &share( MCE::Shared::Handle->new([]) );
      $_[0] = \do { no warnings 'once'; local *FH };
      tie *{ $_[0] }, 'MCE::Shared::Object', $_obj;
   }

   shift; _croak("Not enough arguments for open") unless @_;

   if ( !defined wantarray ) {
      $_obj->OPEN(@_) or _croak("open error: $!");
   } else {
      $_obj->OPEN(@_);
   }
}

###############################################################################
## ----------------------------------------------------------------------------
## TIE support.
##
###############################################################################

sub TIEARRAY {
   shift; $_share_deeply = 1;

   ( ref($_[0]) eq 'HASH' && exists $_[0]->{'module'} )
      ? _tie('TIEARRAY', @_) : MCE::Shared->array(@_);
}

sub TIEHANDLE {
   shift; require MCE::Shared::Handle unless $INC{'MCE/Shared/Handle.pm'};

   # Tie *FH, 'MCE::Shared', { module => 'MCE::Shared::Handle' }, '>>', \*STDOUT
   # doesn't work on the Windows platform. We'd let OPEN handle the ref instead.

   shift if ref($_[0]) eq 'HASH' && $_[0]->{'module'} eq 'MCE::Shared::Handle';

   if ( ref($_[0]) eq 'HASH' && exists $_[0]->{'module'} ) {
      if ( @_ == 3 && ref $_[2] && defined( my $_fd = fileno($_[2]) ) ) {
         _tie('TIEHANDLE', $_[0], $_[1]."&=$_fd");
      } else {
         _tie('TIEHANDLE', @_);
      }
   }
   else {
      my $_obj = &share( MCE::Shared::Handle->new([]) );
      if ( @_ ) { $_obj->OPEN(@_) or return ''; }

      $_obj;
   }
}

sub TIEHASH {
   shift; $_share_deeply = 1;

   return _tie('TIEHASH', @_) if (
      ref($_[0]) eq 'HASH' && exists $_[0]->{'module'}
   );

   my ($_cache, $_ordered);

   if ( ref $_[0] eq 'HASH' ) {
      if ( $_[0]->{'ordered'} || $_[0]->{'ordhash'} ) {
         $_ordered = 1; shift;
      } elsif ( exists $_[0]->{'max_age'} || exists $_[0]->{'max_keys'} ) {
         $_cache = 1;
      }
   }
   else {
      if ( @_ < 3 && ( $_[0] eq 'ordered' || $_[0] eq 'ordhash' ) ) {
         $_ordered = $_[1]; splice(@_, 0, 2);
      } elsif ( @_ < 5 && ( $_[0] eq 'max_age' || $_[0] eq 'max_keys' ) ) {
         $_cache = 1;
      }
   }

   return MCE::Shared->cache(@_)   if $_cache;
   return MCE::Shared->ordhash(@_) if $_ordered;
   return MCE::Shared->hash(@_);
}

sub TIESCALAR {
   shift;

   ( ref($_[0]) eq 'HASH' && exists $_[0]->{'module'} )
      ? _tie('TIESCALAR', @_) : MCE::Shared->scalar(@_);
}

###############################################################################
## ----------------------------------------------------------------------------
## Private functions.
##
###############################################################################

sub _croak {
   if ( $INC{'MCE.pm'} ) {
      goto &MCE::_croak;
   } else {
      require MCE::Shared::Base unless $INC{'MCE/Shared/Base.pm'};
      goto &MCE::Shared::Base::_croak;
   }
}

sub _incr_count {
   # increments counter for safety during destroy
   MCE::Shared::Server::_incr_count($_[0]->SHARED_ID);
}

sub _share {
   $_[2] = &share($_[0], $_[2]);

   MCE::Shared::Object::_req2(
      'M~DEE', $_[1]->SHARED_ID()."\n", $_[2]->SHARED_ID()."\n"
   );
}

sub _tie {
   my ( $_fcn, $_params ) = ( shift, shift );

   _use( my $_module = $_params->{'module'} ) or _croak("$@\n");

   _croak("Can't locate object method \"$_fcn\" via package \"$_module\"")
      unless eval qq{ $_module->can('$_fcn') };

   $_params->{class} = ':construct_module:';
   $_params->{tied } = 1;

   my $_obj;

   if ( $_params->{'module'}->isa('MCE::Shared::Array') ) {
      $_obj = MCE::Shared::Server::_new($_params, [ (), $_fcn ]);
      if ( @_ ) {
         $_params->{_DEEPLY_} = 1; delete $_params->{module};
         for ( my $i = 0; $i <= $#_; $i += 1 ) {
            &_share($_params, $_obj, $_[$i]) if ref($_[$i]);
         }
         $_obj->assign(@_);
      }
   }
   elsif ( $_params->{'module'}->isa('MCE::Shared::Hash') ) {
      $_obj = MCE::Shared::Server::_new($_params, [ (), $_fcn ]);
      if ( @_ ) {
         $_params->{_DEEPLY_} = 1; delete $_params->{module};
         for ( my $i = 1; $i <= $#_; $i += 2 ) {
            &_share($_params, $_obj, $_[$i]) if ref($_[$i]);
         }
         $_obj->assign(@_);
      }
   }
   else {
      $_obj = MCE::Shared::Server::_new($_params, [ @_, $_fcn ]);
   }

   if ( $_obj && $_obj->[2] ) {
      ##
      # Set encoder/decoder automatically for supported DB modules.
      # - AnyDBM_File, DB_File, GDBM_File, NDBM_File, ODBM_File, SDBM_File,
      # - CDB_File, SQLite_File, Tie::Array::DBD, Tie::Hash::DBD,
      # - BerkeleyDB::*, KyotoCabinet::DB, TokyoCabinet::*
      ##
      $_obj->[2] = MCE::Shared::Server::_get_freeze(),
      $_obj->[3] = MCE::Shared::Server::_get_thaw();
   }

   $_obj->[6] = MCE::Mutex->new( impl => 'Channel' );

   $_share_deeply = 0;

   return $_obj;
}

sub _use {
   my $_class = $_[0];

   return 1 if $_class eq 'main';

   if ( $_class =~ /(.*)::_/ ) {
      # e.g. MCE::Hobo::_hash
      eval "require $1" unless $INC{ join('/',split(/::/,$1)).'.pm' };
   }
   elsif ( $_class =~ /^(BerkeleyDB)::(?:Btree|Hash|Queue|Recno)$/ ) {
      eval "require $1" unless $INC{"$1.pm"};
   }
   elsif ( $_class =~ /^(TokyoCabinet|KyotoCabinet)::[ABH]?DB$/ ) {
      eval "require $1" unless $INC{"$1.pm"};
   }
   elsif ( $_class =~ /^Tie::(?:Std|Extra)Hash$/ ) {
      eval "require Tie::Hash" unless $INC{'Tie/Hash.pm'};
   }
   elsif ( $_class eq 'Tie::StdArray' ) {
      eval "require Tie::Array" unless $INC{'Tie/Array.pm'};
   }
   elsif ( $_class eq 'Tie::StdScalar' ) {
      eval "require Tie::Scalar" unless $INC{'Tie/Scalar.pm'};
   }

   return 1 if eval q{
      $_class->can('new') ||
      $_class->can('TIEARRAY') || $_class->can('TIEHANDLE') ||
      $_class->can('TIEHASH')  || $_class->can('TIESCALAR')
   };

   if ( !exists $INC{ join('/',split(/::/,$_class)).'.pm' } ) {
      # remove tainted'ness from $_class
      ($_class) = $_class =~ /(.*)/;

      eval "use $_class (); 1" or return '';
   }

   return 1;
}

1;

__END__