File Coverage

blib/lib/IPC/ConcurrencyLimit/Lock/Redis.pm

Criterion	Covered	Total	%
statement	22	24	91.6
branch			n/a
condition			n/a
subroutine	8	8	100.0
pod			n/a
total	30	32	93.7

package IPC::ConcurrencyLimit::Lock::Redis;
use 5.008001;
use strict;
use warnings;

our $VERSION = '0.01';

use Carp qw(croak);
use Redis;
use Redis::ScriptCache;
use Digest::SHA1 ();
use Data::UUID::MT ();
use Time::HiRes ();

our $UUIDGenerator = Data::UUID::MT->new(version => "4s");

use IPC::ConcurrencyLimit::Lock;
our @ISA = qw(IPC::ConcurrencyLimit::Lock);

use Class::XSAccessor
  getters => [qw(
    redis_conn
    max_procs
    key_name
    proc_info
    script_cache
    uuid
  )];

our $LuaScript_GetLock = q{
  local key       = KEYS[1]
  local max_procs = ARGV[1]
  local proc_info = ARGV[2]
  local i

  for i = 1, max_procs, 1 do
    local x = redis.call('hexists', key, i)
    if x == 0 then
      redis.call('hset', key, i, proc_info)
      return i
    end
  end

  return 0
};
our $LuaScriptHash_GetLock = Digest::SHA1::sha1_hex($LuaScript_GetLock);

# FIXME should this also check the uuid/etc like ClearOldLock?
our $LuaScript_ReleaseLock = q{
  local key    = KEYS[1]
  local lockno = ARGV[1]
  redis.call('hdel', key, lockno)
  return 1
};
our $LuaScriptHash_ReleaseLock = Digest::SHA1::sha1_hex($LuaScript_ReleaseLock);


# FIXME use this for heartbeat: Generate new data with update time
our $LuaScript_UpdateUUID = q{
  local key     = KEYS[1]
  local lockno  = ARGV[1]
  local olddata = ARGV[2]
  local newdata = ARGV[3]
  if redis.call('hget', key, lockno) == olddata then
    redis.call('hset', key, lockno, newdata)
    return 1
  end
  return 0
};
our $LuaScriptHash_UpdateUUID = Digest::SHA1::sha1_hex($LuaScript_UpdateUUID);

our $LuaScript_ClearOldLock = q{
  local key       = KEYS[1]
  local id        = ARGV[1]
  local proc_info = ARGV[2]

  local cleared = 0

  local x = redis.call('hget', key, id)
  if x == proc_info then
    redis.call('hdel', key, id)
    cleared = 1
  end

  return cleared
};
our $LuaScriptHash_ClearOldLock = Digest::SHA1::sha1_hex($LuaScript_ClearOldLock);


sub new {
  my $class = shift;
  my $opt = shift;

  my $max_procs = $opt->{max_procs}
    or croak("Need a 'max_procs' parameter");
  my $redis_conn = $opt->{redis_conn}
    or croak("Need a 'redis_conn' parameter");
  my $key_name = $opt->{key_name}
    or croak("Need a 'key_name' parameter");

  my $sc = Redis::ScriptCache->new(redis_conn => $redis_conn);
  $sc->register_script(\$LuaScript_GetLock, $LuaScriptHash_GetLock);
  $sc->register_script(\$LuaScript_ReleaseLock, $LuaScriptHash_ReleaseLock);
  $sc->register_script(\$LuaScript_UpdateUUID, $LuaScriptHash_UpdateUUID);

  my $proc_info = $opt->{proc_info};
  $proc_info = '' if not defined $proc_info;
  my $uuid = $UUIDGenerator->create;
  my $self = bless {
    max_procs    => $max_procs,
    redis_conn   => $redis_conn,
    key_name     => $key_name,
    id           => undef,
    script_cache => $sc,
    proc_info    => $proc_info,
    uuid         => $uuid,
  } => $class;

  $self->_get_lock($key_name, $max_procs, $sc, $uuid . "-" . $proc_info)
    or return undef;

  return $self;
}

sub _get_lock {
  my ($self, $key, $max_procs, $script_cache, $uuid_proc_info) = @_;

  my ($rv) = $script_cache->run_script(
    $LuaScriptHash_GetLock, [1, $key, $max_procs, $uuid_proc_info]
  );

  if (defined $rv and $rv > 0) {
    $self->{id} = $rv;
    return 1;
  }

  return();
}

sub _release_lock {
  my $self = shift;
  my $id = $self->id;
  return if not $id;

  $self->script_cache->run_script(
    $LuaScriptHash_ReleaseLock, [1, $self->key_name, $id]
  );

  $self->{id} = undef;
}

sub _updated_uuid {
  my ($self) = @_;
  my $old_uuid = $self->uuid;
  my $new_uuid = $UUIDGenerator->create;
  substr($new_uuid, 8, 8) = substr($old_uuid, 8, 8);
  vec($new_uuid, 15, 4) = vec($old_uuid, 15, 4);
  return $new_uuid;
}

sub heartbeat {
  my $self = shift;
  my $conn = $self->redis_conn;
  return() if not $conn;

  my $proc_info = $self->proc_info;
  my $new_uuid = $self->_updated_uuid;
  my $olddata = $self->uuid . "-" . $proc_info;
  my $newdata = $new_uuid . "-" . $proc_info;

  my $ok;
  eval {
    $ok = $self->script_cache->run_script(
      $LuaScriptHash_UpdateUUID,
      [ 1, $self->key_name, $self->id, $olddata, $newdata ]
    );
    1
  } or return(); # server gone away?

  if (not $ok) {
    return(); # lock was acquired by somebody else?
  }

  $self->{uuid} = $new_uuid;
  return 1; # probably all fine
}

sub DESTROY {
  local $@;
  my $self = shift;
  $self->_release_lock();
}


# This is so ugly because we compile slightly different code depending on whether
# we're running on a perl that can do big-endian-forced-quads or not.
# FIXME Will work on 64bit perls only. Implementation for 32bit integers welcome.
# FIXME is this worth it or should it just do a run-time perl version check like heartbeat()?
eval(<<'PRE' . ($] ge '5.010' ? <<'NEW_PERL' : <<'OLD_PERL') . <<'POST')
sub clear_old_locks {
  my ($class, $redis_conn, $key_name, $cutoff) = @_;

  my %hash = $redis_conn->hgetall($key_name);
  return if not keys(%hash);
  my $ncleared = 0;
  foreach my $lockid (keys %hash) {
PRE
    my ($quad) = unpack("Q>", $hash{$lockid});
    $quad -= $quad % 16; # 60 bit only
NEW_PERL
    my ($x, $y) = unpack("N2", $hash{$lockid});
    $y -= $y % 16; # 60 bit only
    my $quad = $x*2**32 + $y;
OLD_PERL
    if ($quad/1e7 < $cutoff) {
      $ncleared += $redis_conn->eval($LuaScript_ClearOldLock, 1, $key_name, $lockid, $hash{$lockid});
    }
  }
  return $ncleared;
}
1
POST
or do {
  my $err = $@ || 'Zombie error';
  die "Failed to compile clear_old_locks code: $err";
};

1;

__END__