File Coverage

blib/lib/Thread/Barrier.pm

Criterion	Covered	Total	%
statement	15	90	16.6
branch	0	38	0.0
condition	0	14	0.0
subroutine	5	15	33.3
pod	5	6	83.3
total	25	163	15.3

###########################################################################
#
# See the README file included with the
# distribution for license information.
#
###########################################################################

package Thread::Barrier;

use 5.008;
use strict;
use warnings;

use threads::shared;

our $VERSION = '0.300_02';
$VERSION = eval $VERSION;

###########################################################################
# Public Methods
###########################################################################

#
# new - creates a new Thread::Barrier object
#
# Arguments:
#
# threshold
#   Specifies the required number of threads that
#   must block on the barrier before it is released.
# opt => val ...
#   Optional arguments to new()
#
# Returns a Thread::Barrier object on success, dies on failure.
#
sub new {
    my ($class, $threshold, %opts) = @_;
    $opts{RaiseError} = 1 unless exists $opts{RaiseError};

    # threads::shared 1.43 (perl 5.18.0) does not yet support shared
    # CODE refs, which would be obvious/ideal for our 'Action' support.
    # So, our Thread::Barrier object isn't itself shared, but one of
    # its members is.
    #
    # Object structure (ARRAY ref):
    #
    #   [   {barrier implementation},    <-- shared
    #       \&optional_coderef        ]  <-- non-shared
    #

    my $self = bless [], $class;

    $self->[0] = &share({});
    %{$self->[0]} = (
        threshold           => 0,    # threads required to release barrier
        count               => 0,    # number of threads blocking on barrier
        generation          => 0,    # incremented when barrier is released
       #broken              => 0,    # true if broken
       #first_released_$gen => 1,    # if present, $gen was just released
    );

    $self->set_threshold($threshold) if $threshold; # may die
    while (my ($opt, $val) = each(%opts)) {
      if ($opt eq 'Action' and defined($val)) {
        _confess("Invalid Action parameter to $class->new")
        unless ref($val) eq 'CODE';
        $self->[1] = $val;
        next;
      }
      if ($opt eq 'RaiseError') {
        $self->[0]->{RaiseError} = !!$val;
        next;
      }
      _confess("Unrecognized parameter '$opt' to $class->new");
    }

    return $self;
}


#
# init - set the threshold value for the barrier
#
# *** DEPRECATED ***
#
# Arguments:
#
# threshold
#   Specifies the required number of threads that
#   must block on the barrier before it is released.
#
# Returns the passed argument.
#
sub init {
    my($self, $threshold) = @_;
    $self->set_threshold($threshold);
    return $threshold;
}


#
# wait - block until a sufficient number of threads have reached the barrier
#
# Arguments:
#
# none
#
# Returns true to one of threads released upon barrier reset, false to
# all others.
#
sub wait {
    my ($self, $timeo) = @_;
    my ($bar, $act) = @$self; # Unwrap our actual barrier and Action (if any)
    my ($gen, $i);

    $timeo = $self->_normalize_timeout($timeo)
      if defined($timeo);

    lock $bar;

    $gen = $bar->{generation};
    $i   = $bar->{count}++;

    if (! $self->_try_release) {
        unless (defined $timeo) {
          # block
          while ($bar->{generation} == $gen and not $bar->{broken}) {
            cond_wait($bar);
          }
        } else {
          while ($bar->{generation} == $gen and not $bar->{broken}) {
            last if !cond_timedwait($bar, $timeo);
          }
          $bar->{broken} = 1 if $bar->{generation} == $gen;
        }
    }

    # Are we the first awake from our generation?  Run Action if any
    if (delete $bar->{"first_released_${gen}"} and $act) {
      my $ok = eval { $act->(); 1; };
      if (! $ok) {
        $bar->{broken} = 1;
        die($@ || "Barrier action failed");
      }
    } elsif ($bar->{broken}) {
      _croak("Barrier broken") if $bar->{RaiseError};
      return undef;
    }

    # In our implementation, the first one to arrive gets the serial
    # indicator
    return ($i == 0);
}


#
# set_threshold - adjust the barrier's threshold, possibly releasing it
#                  if enough threads are blocking.
#
# Arguments:
#
# threshold
#   Specifies the required number of threads that
#   must block on the barrier before it is released.
#
# Returns true if barrier is released, false otherwise.
#
sub set_threshold {
    my($self, $threshold) = @_;
    my $err;

    # validate threshold
    for ($threshold) {
        $err = "no argument supplied", last unless defined $_;
        $err = "invalid argument supplied", last if /[^0-9]/;
    }
    if ($err) {
        no warnings 'once';
        local $Carp::CarpLevel = 1;
        _confess($err);
    }

    # apply new threshold, possibly releasing barrier
    lock $self->[0];
    $self->[0]->{threshold} = $threshold;

    # check for release condition
    $self->_try_release;
}


#
# threshold - accessor for debugging purposes
#
sub threshold {
    my $bar = shift->[0];
    lock $bar;
    return $bar->{threshold};
}


#
# count - accessor for debugging purposes
#
sub count {
    my $bar = shift->[0];
    lock $bar;
    return $bar->{count};
}


###########################################################################
# Private Methods
###########################################################################

sub _confess {
  require Carp;
  goto &Carp::confess;
}

sub _croak {
  require Carp;
  goto &Carp::croak;
}

#
# _try_release - release the barrier if a sufficient number of threads
#                have reached the barrier.
#                N.B.:  Assumes the barrier is locked
#
# Arguments:
#
#   none
#
# Returns true if barrier is released, false otherwise.
#
sub _try_release {
    my $bar = shift->[0];

    return undef if $bar->{count} < $bar->{threshold};

    # reset barrier and release
    my $gen = $bar->{generation}++;
    $bar->{"first_released_${gen}"} = 1;
    $bar->{count} = 0;

    cond_broadcast($bar);
    return 1;
}

sub _normalize_timeout {
  my ($self, $timeo) = @_;
  $timeo =~ /^\d+(?:\.\d+)*$/
    or _croak("Invalid timeout specification ($timeo)");
  $timeo += time();
}

1;
__END__