File Coverage

blib/lib/IPC/ConcurrencyLimit/WithLatestStandby.pm

Criterion	Covered	Total	%
statement	76	96	79.1
branch	25	54	46.3
condition	10	24	41.6
subroutine	11	18	61.1
pod	0	6	0.0
total	122	198	61.6

package IPC::ConcurrencyLimit::WithLatestStandby;
use 5.008001;
use strict;
use warnings;

our $VERSION = '0.16';

use Carp qw(croak);
use Time::HiRes qw(sleep time);
use IPC::ConcurrencyLimit;

sub new {
    my $class  = shift;
    my %params = @_;
    my $type   = delete $params{type};
    $type = 'Flock' if not defined $type;
    croak( __PACKAGE__ . " only supports 'Flock' for now")
        if $type ne 'Flock';

    if (defined $params{max_procs} and $params{max_procs}!=1) {
        croak( __PACKAGE__ . " does not support max_procs!=1, use multiple objects instead.");
    }

    my $process_name_change= $params{process_name_change} || 0;
    my $path=        $params{path}          || die __PACKAGE__ . '->new: missing mandatory parameter `path`';
    my $file_prefix= $params{file_prefix}   || "";
    my $poll_time=   $params{poll_time}     || $params{interval} || 1; # seconds to poll (may be fraction)
    my $retries=     $params{retries}       || undef;
    my $timeout=     $params{timeout}       || undef;
    my $debug=       $params{debug}         || 0; # show debug?
    my $debug_sub=   $params{debug_sub}     || undef;

    my $retry_sub= (ref $retries)     ? $retries :
                   (defined $retries &&
                    defined $timeout) ? sub { $_[0] <= $retries && $_[2] <= $timeout } :
                   (defined $retries) ? sub { $_[0] <= $retries } :
                   (defined $timeout) ? sub { $_[2] <= $timeout } :
                                        sub { 1 };

    # primary is replace by standby1, is replaced by
    # standby2, is replaced by standby3. However, standby1
    # will exit when standby2's lock is held by another process.
    my @names= map { ($file_prefix ? "$file_prefix.$_" : $_) }
                     qw(primary standby1 standby2 standby3);
    my @lockers= map {
        IPC::ConcurrencyLimit->new(
            type => $type,
            max_procs => 1,
            # future proofing
            $type eq "Flock" ? (
                file_prefix => $_,
                path => $path,
            ) : (),
        )
    } @names;

    return bless {
        poll_time   => $poll_time,
        timeout     => $timeout,    # FYI
        retries     => $retries,    # FYI
        lock_name   => \@names,
        locker      => \@lockers,
        debug       => $debug,
        debug_sub   => $debug_sub || sub { warn @_,"\n" },
        retry_sub   => $retry_sub,
        process_name_change => $process_name_change,
    }, $class;
}

sub _diag {
    my ($self, $fmt, @args)= @_;
    if (!@args) {
        $self->{debug_sub}->($fmt);
    } else {
        $self->{debug_sub}->(sprintf $fmt, @args);
    }
}


sub get_lock {
    my ($self) = @_;

    my $locker= $self->{locker};
    my $names= $self->{lock_name};

    my $old_oh= $0;

    $0 = "$old_oh - acquire"
        if $self->{process_name_change};

    # try to get the rightmost lock (standby3) if we don't get it
    # then we exit out. this shouldn't really happen if other things
    # are sane, for instance when $poll_time is much smaller than
    # the rate we allocate new workers.
    my $locker_id= $#$locker;
    if ( $locker->[$locker_id]->get_lock() ) {
        $self->_diag( "Got a $names->[$locker_id] lock");
    } else {
        $self->_diag( "Failed to get a $names->[$locker_id] lock, entry lock is held by another process" )
            if $self->{debug};
        $0 = "$old_oh - no-lock-acquired"
            if $self->{process_name_change};
        return;
    }

    # Each worker tries to acquire the lock to its left. If it does
    # then it abandons its old lock. If that means the worker ends up
    # on locker_id 0 then they are done, and can do work.
    # The first standby worker also looks to its right to see if there
    # is a replacement process for it, if there is it exits, leaving
    # a gap and letting the replacements shuffle left.
    my $tries= 0;
    my $lock_tries= 0;
    my $standby_start= time();
    my $lock_start= time();
    my $poll_time= $self->{poll_time};

    while ( $locker_id > 0 ) {
        $0 = "$old_oh - $names->[$locker_id]"
            if $self->{process_name_change};

        # can we shuffle our lock left?
        if ( $locker->[$locker_id - 1]->get_lock() ) {
            $self->_diag( "Got a $names->[$locker_id -1] lock, dropping old $names->[$locker_id] lock");
            # yep, we got the lock to the left, so drop our old lock,
            # and move the pointer left at the same time.
            $locker->[ $locker_id-- ]->release_lock();
            $lock_tries= 0;
            $lock_start= time();
            next;
        }

        unless ($self->{retry_sub}->(++$tries, ++$lock_tries, time - $standby_start, time - $lock_start)) {
            $0 = "$old_oh - no-lock-timeout"
                if $self->{process_name_change};
            return;
        }

        # check if we are the first standby worker.
        if ( $locker_id == 1 ) {
            # yep - we are the first standby worker,
            # so check if the lock to our right is being held:
            if ( $locker->[$locker_id + 1]->get_lock() ) {
                # we got the lock, which means nothing else
                # holds it. so we release the lock and move on.
                $locker->[$locker_id + 1]->release_lock();
            } else {
                $self->_diag(
                    "A newer worker is holding the $names->[$locker_id+1] lock, will exit to let it take over"
                ) if $self->{debug};
                # we failed to get the lock, which means there is a newer
                # process that can replace us so return/exit - this frees up
                # our lock and lets the newer process to move into our position.
                $0 = "$old_oh - no-lock-retired"
                    if $self->{process_name_change};
                return;
            }
        }

        # nope - the lock to our left is being held so sleep a while before
        # we try again. We use the rand and the formula so that items to the
        # right poll faster than items to the left, and to reduce the chance
        # that lock holder 1 and lock holder 3 poll lock 2 at the same time
        # forever. The formula guarantees that items to the left poll faster,
        # and the rand ensures there is jitter.
        sleep rand(($poll_time / $locker_id)*2);
    }

    # assert that $locker_id is 0 at this point.
    die "panic: We should not reach this point with \$locker_id larger than 0, got $locker_id"
        if $locker_id;

    $self->_diag("Got $names->[$locker_id] lock, we are allowed to do work.")
        if $self->{debug};

    # at this point we should be $locker_id == 0 and we can do work.
    if ($self->{process_name_change}) {
        if ($self->{process_name_change} > 1) {
            $0 = $old_oh;
        } else {
            $0 = "$old_oh - $names->[$locker_id]"
        }
    }
    return 1;
}


sub is_locked {
    my $self = shift;
    return $self->{locker}[0]->is_locked(@_);
}

sub release_lock {
    my $self = shift;
    return $self->{locker}[0]->release_lock(@_);
}

sub lock_id {
    my $self = shift;
    return $self->{locker}[0]->lock_id(@_);
}

sub heartbeat {
    my $self = shift;
    return $self->{locker}[0]->heartbeat;
}


1;

__END__