File Coverage

blib/lib/Redis/DistLock.pm

Criterion	Covered	Total	%
statement	63	84	75.0
branch	20	38	52.6
condition	11	31	35.4
subroutine	13	15	86.6
pod	3	5	60.0
total	110	173	63.5

package Redis::DistLock;

use strict;
use warnings;

our $VERSION = '0.07';

use Digest::SHA qw( sha1_hex );
use MIME::Base64 qw( encode_base64 );
use Redis;
use Time::HiRes qw( time );

sub VERSION_CHECK()     { 1 }
sub RETRY_COUNT()       { 3 }
sub RETRY_DELAY()       { 0.2 }
sub DRIFT_FACTOR()      { 0.01 }
sub RELEASE_SCRIPT()    { '
if redis.call( "get", KEYS[1] ) == ARGV[1] then
    return redis.call( "del", KEYS[1] )
else
    return 0
end
' }
sub RELEASE_SHA1()      { sha1_hex( RELEASE_SCRIPT ) }
sub EXTEND_SCRIPT()     { '
if redis.call( "set", KEYS[1], ARGV[1], "XX", "PX", ARGV[2] ) then
    return "OK"
else
    return redis.call( "set", KEYS[1], ARGV[1], "NX", "PX", ARGV[2] )
end
' }
sub EXTEND_SHA1()       { sha1_hex( EXTEND_SCRIPT ) }

sub DESTROY {
    my $self = shift;

    # only has locks when auto release is enabled
    return if @{ $self->{locks} || [] } == 0;

    $self->release( $_ )
        for @{ $self->{locks} };
}

sub new {
    my $class = shift;
    my %args = @_ == 1 && ref( $_[0] )
             ? %{ $_[0] }
             : @_
    ;

    my $version_check = exists( $args{version_check} )
                      ?         $args{version_check}
                      :               VERSION_CHECK
    ;

    my $logger = exists( $args{logger} )
                      ?  $args{logger}
                      :  sub { warn @_ }
    ;

    my $quorum = int( @{ $args{servers} } / 2 + 1 );
    my @servers;

    for my $server ( @{ $args{servers} } ) {
        # connect might fail
        my $redis = ref( $server )
                       ? $server
                       : eval { Redis->new( server => $server, encoding => undef ) }
        ;
        unless ( $redis ) {
            $logger->( $@ );
            next;
        }
        push( @servers, $redis );

        if ( $version_check ) {
            my $info = $redis->info();

            die( "FATAL: cannot find the right redis version (needs at least 2.6.12 -- $1, $2, $3)" )
                unless $info &&
                       $info->{redis_version} &&
                       $info->{redis_version} =~ m!\A ([0-9]+) \. ([0-9]+) \. ([0-9]+) \z!x &&
                       (
                         ( $1 >  2 ) ||
                         ( $1 == 2 && $2 >  6 ) ||
                         ( $1 == 2 && $2 == 6 && $3 >= 12 )
                       )
            ;
        }

        # load script on all servers
        my $sha1 = $redis->script_load( RELEASE_SCRIPT );

        # ensure the script is everywhere the same
        if ( $sha1 ne RELEASE_SHA1 ) {
            die( "FATAL: script load results in different checksum!" );
        }

        $sha1 = $redis->script_load( EXTEND_SCRIPT );

        if ( $sha1 ne EXTEND_SHA1 ) {
            die( "FATAL: failed to load extend script: sha1 mismatch!" );
        }
    }

    if ( @servers < $quorum ) {
        die( "FATAL: could not establish enough connections (" . int( @servers ) . " < $quorum)" );
    }

    my $self = bless( {
        servers        => \@servers,
        quorum         => $quorum,
        retry_count    => $args{retry_count} || RETRY_COUNT,
        retry_delay    => $args{retry_delay} || RETRY_DELAY,
        locks          => [],
        logger         => $logger || sub {},
        auto_release   => $args{auto_release} || 0,
    }, $class );

    return $self;
}

sub _get_random_id {
    encode_base64( join( "", map chr( int( rand() * 256 ) ), 1 .. 24 ), "" );
}

sub lock {
    my ( $self, $resource, $ttl, $value, $extend ) = @_;
    my $retry_count = $self->{retry_count};

    $value = _get_random_id()
        unless defined( $value );

    while ( $retry_count-- > 0 ) {
        my $start = time();
        my $ok = 0;

        for my $redis ( @{ $self->{servers} } ) {
            # count successful locks, response only needs to be true
            $ok += eval {
                ! $extend
                ? $redis->set( $resource, $value, "NX", "PX", $ttl * 1000 )
                : $redis->evalsha( EXTEND_SHA1, 1, $resource, $value, $ttl * 1000 )
            } ? 1 : 0;

            $self->{logger}->( $@ )
                if $@;
        }

        my $drift = $ttl * DRIFT_FACTOR + 0.002;
        my $validity = $ttl - ( time() - $start ) - $drift;

        if ( $ok >= $self->{quorum} && $validity > 0 ) {
            my $lock = {
                validity    => $validity,
                resource    => $resource,
                value       => $value,
            };

            # track lock on demand only
            push( @{ $self->{locks} }, $lock )
                if $self->{auto_release};

            return $lock;
        }

        select( undef, undef, undef, rand( $self->{retry_delay} ) );
    }

    return undef;
}

sub release {
    my $self = shift;
    my ( $resource, $value ) = @_ == 1 && ref( $_[0] )
                             ? @{ $_[0] }{ qw{ resource value } }
                             : @_
    ;

    defined or $_ = ""
        for $resource, $value;

    for my $redis ( @{ $self->{servers} } ) {
        $redis->evalsha( RELEASE_SHA1, 1, $resource, $value );
    }
}

1;

__END__