无法释放 Redis 信号量锁

Question

我正在使用redis-semaphore gem，版本0.3.1。

出于某种原因，我偶尔无法释放过时的 Redis 锁。从我的分析来看，如果我的 Docker 进程在创建锁后崩溃，这似乎就会发生。

我在下面描述了我的调试过程，想知道是否有人可以建议如何进一步调试。

假设我们要创建一个具有此名称的 redis 锁：

name = "test"

我们将此变量插入两个不同的终端窗口中。首先，我们运行：

def lock_for_15_secs(name)
job = Redis::Semaphore.new(name.to_sym, redis: NonBlockingRedis.new(), custom_blpop: true, :stale_client_timeout => 15)
if job.lock(-1) == "0"
puts "Locked and starting"
sleep(15)
puts "Now it's stale, try to release in another process"
sleep(15)
puts "Now trying to unlock"
unlock = job.unlock
puts unlock == false ? "Wuhuu, already unlocked" : "Hm, should have been unlocked by another process, but wasn't"
end
end
lock_for_15_secs(name)

在第二个我们运行：

def release_and_lock(name)
job     = Redis::Semaphore.new(name.to_sym, redis: NonBlockingRedis.new(), custom_blpop: true, :stale_client_timeout => 15)
release = job.release_stale_locks!
count   = job.available_count
puts "Release reponse is #{release.inspect} and available count is #{count}"
if job.lock(-1) == "0"
puts "Wuhuu, we can lock it"
job.unlock
else
puts "Hmm, we can't lock it"
end
end
release_and_lock(name)

这通常按预期进行。在 15 秒内，第二个终端无法松开锁，但在 15 秒后运行时，它会松开。以下是release_and_lock(name)的输出。

在 15 秒过去之前：

irb(main):1:0> release_and_lock(name)
Release reponse is {"0"=>"1580292557.321834"} and available count is 0
Hmm, we can't lock it
=> nil

15 秒后：

irb(main):2:0> release_and_lock(name)
Release reponse is {"0"=>"1580292557.321834"} and available count is 1
Wuhuu, we can lock it
=> 1
irb(main):3:0> release_and_lock(name)
Release reponse is {} and available count is 1
Wuhuu, we can lock it

但是每当我看到陈旧的锁没有释放，并且我运行release_and_lock(name)进行诊断时，就会返回：

irb(main):4:0> release_and_lock(name)
Release reponse is {} and available count is 0
Hmm, we can't lock it

在这一点上，我唯一的选择是刷新 redis：

require 'non_blocking_redis'
non_blocking_redis = NonBlockingRedis.new()
non_blocking_redis.flushall

附言我的NonBlockingRedis继承自Redis：

class Redis
class Semaphore
def initialize(name, opts = {})
@custom_opts          = opts
@name                 = name
@resource_count       = opts.delete(:resources) || 1
@stale_client_timeout = opts.delete(:stale_client_timeout)
@redis                = opts.delete(:redis) || Redis.new(opts)
@use_local_time       = opts.delete(:use_local_time)
@custom_blpop         = opts.delete(:custom_blpop) # false=queue, true=cancel
@tokens               = []
end
def lock(timeout = 0)
exists_or_create!
release_stale_locks! if check_staleness?
token_pair    = @redis.blpop(available_key, timeout, @custom_blpop)
return false if token_pair.nil?
current_token = token_pair[1]
@tokens.push(current_token)
@redis.hset(grabbed_key, current_token, current_time.to_f)
if block_given?
begin
yield current_token
ensure
signal(current_token)
end
end
current_token
end
alias_method :wait, :lock
end
end
class NonBlockingRedis < Redis
def initialize(options = {})
if options.empty?
options = {
url: Rails.application.secrets.redis_url,
db:  Rails.application.secrets.redis_sidekiq_db,
driver: :hiredis,
network_timeout: 5
}
end
super(options)
end
def blpop(key, timeout, custom_blpop)
if custom_blpop
if timeout == -1
result = lpop(key)
return result if result.nil?
return [key, result]
else
super(key, timeout)
end
else
super
end
end
def lock(timeout = 0)
exists_or_create!
release_stale_locks! if check_staleness?
token_pair = @redis.blpop(available_key, timeout, @custom_blpop)
return false if token_pair.nil?
current_token = token_pair[1]
@tokens.push(current_token)
@redis.hset(grabbed_key, current_token, current_time.to_f)
if block_given?
begin
yield current_token
ensure
signal(current_token)
end
end
current_token
end
alias_method :wait, :lock
end
require 'non_blocking_redis'

Answer 1

😜一个很棒的错误 👏

错误

我认为如果你在它确实在SEMAPHORE:test:AVAILABLE上弹出时杀死进程，就会发生这种情况

很可能在这里 https://github.com/dv/redis-semaphore/blob/v0.3.1/lib/redis/semaphore.rb#L67

复制它

NonBlockingRedis.new.flushall
release_and_lock('test');
NonBlockingRedis.new.lpop('SEMAPHORE:test:AVAILABLE')

现在，您最初拥有：

SEMAPHORE:test:AVAILABLE                0
SEMAPHORE:test:VERSION          1
SEMAPHORE:test:EXISTS           1

在上面的代码之后，你会得到：

SEMAPHORE:test:VERSION          1
SEMAPHORE:test:EXISTS           1

代码检查SEMAPHORE:test:EXISTS，然后期望有SEMAPHORE:test:AVAILABLE/SEMAPHORE:test:GRABBED

溶液

从我的简短检查来看，我认为如果不进行修改就不可能使宝石工作。我尝试添加一个expiration:但不知何故，它设法禁用了SEMAPHORE:test:EXISTS的过期

NonBlockingRedis.new.ttl('SEMAPHORE:test:EXISTS') # => -1 and it should have been e.g. 20 seconds and going down

所以..也许修复将是

class Redis
class Semaphore
def exists_or_create!
token = @redis.getset(exists_key, EXISTS_TOKEN)
if token.nil? || all_tokens.empty?
create!
else
# Previous versions of redis-semaphore did not set `version_key`.
# Make sure it's set now, so we can use it in future versions.
if token == API_VERSION && @redis.get(version_key).nil?
@redis.set(version_key, API_VERSION)
end
true
end
end
end
end

all_tokens是 https://github.com/dv/redis-semaphore/blob/v0.3.1/lib/redis/semaphore.rb#L120

我很快就会打开宝石的 PR - 也许> https://github.com/dv/redis-semaphore/pull/66 🤷 ♂️

注1

不确定如何使用NonBlockingRedis，但它未在Redis::Semaphore中使用。你做lock(-1)在代码lpop中做。此外，代码从不调用您的lock。

随机

这是转储密钥的助手

class Test
def self.all
r = NonBlockingRedis.new
puts r.keys('*').map { |k|
[
k,
((r.hgetall(k) rescue r.get(k)) rescue r.lrange(k, 0, -1).join(' | '))
].join("tt")
}
end
end
> Test.all
SEMAPHORE:test:AVAILABLE                0
SEMAPHORE:test:VERSION          1
SEMAPHORE:test:EXISTS           1

为了完整起见，这是您抓住锁时的外观

SEMAPHORE:test:VERSION          1
SEMAPHORE:test:EXISTS           1
SEMAPHORE:test:GRABBED          {"0"=>"1583672948.7168388"}