File Coverage

blib/lib/MCE/Shared/Condvar.pm

Criterion	Covered	Total	%
statement	101	126	80.1
branch	23	52	44.2
condition	4	21	19.0
subroutine	23	43	53.4
pod	18	18	100.0
total	169	260	65.0

###############################################################################
## ----------------------------------------------------------------------------
## Condvar helper class.
##
###############################################################################

package MCE::Shared::Condvar;

use strict;
use warnings;

use 5.010001;

no warnings qw( threads recursion uninitialized numeric );

our $VERSION = '1.881';

use MCE::Shared::Base ();
use MCE::Util ();
use MCE::Mutex ();

use overload (
   q("")    => \&MCE::Shared::Base::_stringify,
   q(0+)    => \&MCE::Shared::Base::_numify,
   fallback => 1
);

my $LF = "\012"; Internals::SvREADONLY($LF, 1);
my $_tid = $INC{'threads.pm'} ? threads->tid() : 0;
my $_reset_flg = 1;

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

sub DESTROY {
   my ($_cv) = @_;
   my $_pid  = $_tid ? $$ .'.'. $_tid : $$;

   if ($_cv->{_init_pid} eq $_pid) {
      MCE::Util::_destroy_socks($_cv, qw(_cw_sock _cr_sock));
   }

   return;
}

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

sub new {
   my ($_class, $_cv) = (shift, {});

   $_cv->{_init_pid} = $_tid ? $$ .'.'. $_tid : $$;
   $_cv->{_value}    = shift || 0;
   $_cv->{_count}    = 0;

   MCE::Util::_sock_pair($_cv, qw(_cr_sock _cw_sock), undef, 1);

   MCE::Shared::Object::_reset(), $_reset_flg = ''
      if $_reset_flg && $INC{'MCE/Shared/Server.pm'};

   bless $_cv, $_class;
}

sub get { $_[0]->{_value} }
sub set { $_[0]->{_value} = $_[1] }

# The following methods applies to shared-context only and handled by
# MCE::Shared::Object.

sub lock      { }
sub unlock    { }

sub broadcast { }
sub signal    { }
sub timedwait { }
sub wait      { }

###############################################################################
## ----------------------------------------------------------------------------
## Sugar API, mostly resembles https://redis.io/commands#string primitives.
##
###############################################################################

# append ( string )

sub append {
   length( $_[0]->{_value} .= $_[1] // '' );
}

# decr
# decrby ( number )
# incr
# incrby ( number )
# getdecr
# getincr

sub decr    { --$_[0]->{_value}               }
sub decrby  {   $_[0]->{_value} -= $_[1] || 0 }
sub incr    { ++$_[0]->{_value}               }
sub incrby  {   $_[0]->{_value} += $_[1] || 0 }
sub getdecr {   $_[0]->{_value}--        // 0 }
sub getincr {   $_[0]->{_value}++        // 0 }

# getset ( value )

sub getset {
   my $old = $_[0]->{_value};
   $_[0]->{_value} = $_[1];

   $old;
}

# len ( )

sub len {
   length $_[0]->{_value};
}

###############################################################################
## ----------------------------------------------------------------------------
## Server functions.
##
###############################################################################

{
   use bytes;

   use constant {
      SHR_O_CVB => 'O~CVB',  # Condvar broadcast
      SHR_O_CVS => 'O~CVS',  # Condvar signal
      SHR_O_CVT => 'O~CVT',  # Condvar timedwait
      SHR_O_CVW => 'O~CVW',  # Condvar wait
   };

   my ( $_DAU_R_SOCK_REF, $_DAU_R_SOCK, $_obj, $_id );

   my %_output_function = (

      SHR_O_CVB.$LF => sub {                      # Condvar broadcast
         $_DAU_R_SOCK = ${ $_DAU_R_SOCK_REF };
         chomp($_id = <$_DAU_R_SOCK>);

         my $_var = $_obj->{ $_id } || do {
            print {$_DAU_R_SOCK} $LF;
         };
         for my $_i (1 .. $_var->{_count}) {
            syswrite($_var->{_cw_sock}, $LF);
         }

         $_var->{_count} = 0;
         print {$_DAU_R_SOCK} $LF;

         return;
      },

      SHR_O_CVS.$LF => sub {                      # Condvar signal
         $_DAU_R_SOCK = ${ $_DAU_R_SOCK_REF };
         chomp($_id = <$_DAU_R_SOCK>);

         my $_var = $_obj->{ $_id } || do {
            print {$_DAU_R_SOCK} $LF;
         };
         if ( $_var->{_count} >= 0 ) {
            syswrite($_var->{_cw_sock}, $LF);
            $_var->{_count} -= 1;
         }

         print {$_DAU_R_SOCK} $LF;

         return;
      },

      SHR_O_CVT.$LF => sub {                      # Condvar timedwait
         $_DAU_R_SOCK = ${ $_DAU_R_SOCK_REF };
         chomp($_id = <$_DAU_R_SOCK>);

         my $_var = $_obj->{ $_id } || do {
            print {$_DAU_R_SOCK} $LF;
         };

         $_var->{_count} -= 1;
         print {$_DAU_R_SOCK} $LF;

         return;
      },

      SHR_O_CVW.$LF => sub {                      # Condvar wait
         $_DAU_R_SOCK = ${ $_DAU_R_SOCK_REF };
         chomp($_id = <$_DAU_R_SOCK>);

         my $_var = $_obj->{ $_id } || do {
            print {$_DAU_R_SOCK} $LF;
         };

         $_var->{_count} += 1;
         print {$_DAU_R_SOCK} $LF;

         return;
      },

   );

   sub _init_mgr {
      my $_function;
      ( $_DAU_R_SOCK_REF, $_obj, $_function ) = @_;

      for my $key ( keys %_output_function ) {
         last if exists($_function->{$key});
         $_function->{$key} = $_output_function{$key};
      }

      return;
   }
}

###############################################################################
## ----------------------------------------------------------------------------
## Object package.
##
###############################################################################

## Items below are folded into MCE::Shared::Object.

package # hide from rpm
   MCE::Shared::Object;

use strict;
use warnings;

no warnings qw( threads recursion uninitialized numeric once );

use Time::HiRes qw( alarm sleep );
use bytes;

no overloading;

my $_is_MSWin32 = ($^O eq 'MSWin32') ? 1 : 0;

my ($_DAT_LOCK, $_DAT_W_SOCK, $_DAU_W_SOCK, $_dat_ex, $_dat_un, $_chn, $_obj);

sub _init_condvar {
   ($_DAT_LOCK, $_DAT_W_SOCK, $_DAU_W_SOCK, $_dat_ex, $_dat_un, $_chn, $_obj) = @_;

   return;
}

# broadcast ( floating_seconds )
# broadcast ( )

sub broadcast {
   my $_id = $_[0]->[0];
   return unless ( my $_CV = $_obj->{ $_id } );
   return unless ( exists $_CV->{_cr_sock} );

   sleep($_[1]) if defined $_[1];

   _req1('O~CVB', $_id.$LF);
   $_[0]->[6]->unlock();

   sleep(0);
}

# signal ( floating_seconds )
# signal ( )

sub signal {
   my $_id = $_[0]->[0];
   return unless ( my $_CV = $_obj->{ $_id } );
   return unless ( exists $_CV->{_cr_sock} );

   sleep($_[1]) if defined $_[1];

   _req1('O~CVS', $_id.$LF);
   $_[0]->[6]->unlock();

   sleep(0);
}

# timedwait ( floating_seconds )

sub timedwait {
   my $_id = $_[0]->[0];
   my $_timeout = $_[1];

   return unless ( my $_CV = $_obj->{ $_id } );
   return unless ( exists $_CV->{_cr_sock} );
   return $_[0]->wait() unless $_timeout;

   _croak('Condvar: timedwait (timeout) is not valid')
      if (!looks_like_number($_timeout) || $_timeout < 0);

   _req1('O~CVW', $_id.$LF);
   $_[0]->[6]->unlock();

   $_timeout = 0.0003 if $_timeout < 0.0003;

   local $@; eval {
      local $SIG{ALRM} = sub { alarm 0; die "alarm clock restart\n" };
      alarm $_timeout unless $_is_MSWin32;

      die "alarm clock restart\n"
         if $_is_MSWin32 && MCE::Util::_sock_ready($_CV->{_cr_sock}, $_timeout);

      (!$_is_MSWin32)
         ? (MCE::Util::_sysread($_CV->{_cr_sock}, my($_b1), 1), alarm(0))
         : (MCE::Util::_sysread($_CV->{_cr_sock}, my($_b2), 1));
   };

   alarm 0 unless $_is_MSWin32;

   if ($@) {
      chomp($@), _croak($@) unless $@ eq "alarm clock restart\n";
      _req1('O~CVT', $_id.$LF);

      return '';
   }

   return 1;
}

# wait ( )

sub wait {
   my $_id = $_[0]->[0];
   return unless ( my $_CV = $_obj->{ $_id } );
   return unless ( exists $_CV->{_cr_sock} );

   _req1('O~CVW', $_id.$LF);
   $_[0]->[6]->unlock();

   MCE::Util::_sock_ready($_CV->{_cr_sock}) if $_is_MSWin32;
   MCE::Util::_sysread($_CV->{_cr_sock}, my($_b), 1);

   return 1;
}

1;

__END__