2019-05-03 06:17:27 +08:00
|
|
|
# frozen_string_literal: true
|
|
|
|
|
2014-04-11 13:43:33 +08:00
|
|
|
# Cross-process locking using Redis.
|
2019-08-14 16:56:43 +08:00
|
|
|
# Expiration happens when the current time is greater than the expire time
|
2014-04-11 13:43:33 +08:00
|
|
|
class DistributedMutex
|
2022-07-11 20:16:37 +08:00
|
|
|
DEFAULT_VALIDITY = 60
|
|
|
|
CHECK_READONLY_ATTEMPTS = 10
|
|
|
|
|
|
|
|
LOCK_SCRIPT = DiscourseRedis::EvalHelper.new <<~LUA
|
|
|
|
local now = redis.call("time")[1]
|
|
|
|
local expire_time = now + ARGV[1]
|
|
|
|
local current_expire_time = redis.call("get", KEYS[1])
|
|
|
|
|
|
|
|
if current_expire_time and tonumber(now) <= tonumber(current_expire_time) then
|
|
|
|
return nil
|
|
|
|
else
|
|
|
|
local result = redis.call("setex", KEYS[1], ARGV[1] + 1, tostring(expire_time))
|
|
|
|
return expire_time
|
|
|
|
end
|
|
|
|
LUA
|
|
|
|
|
|
|
|
UNLOCK_SCRIPT = DiscourseRedis::EvalHelper.new <<~LUA
|
|
|
|
local current_expire_time = redis.call("get", KEYS[1])
|
|
|
|
|
|
|
|
if current_expire_time == ARGV[1] then
|
|
|
|
local result = redis.call("del", KEYS[1])
|
|
|
|
return result ~= nil
|
|
|
|
else
|
|
|
|
return false
|
|
|
|
end
|
|
|
|
LUA
|
2014-04-11 13:43:33 +08:00
|
|
|
|
2019-02-20 09:23:42 +08:00
|
|
|
def self.synchronize(key, redis: nil, validity: DEFAULT_VALIDITY, &blk)
|
2019-02-20 09:28:10 +08:00
|
|
|
self.new(
|
|
|
|
key,
|
|
|
|
redis: redis,
|
|
|
|
validity: validity
|
|
|
|
).synchronize(&blk)
|
2014-07-30 12:04:27 +08:00
|
|
|
end
|
|
|
|
|
2019-02-20 09:28:10 +08:00
|
|
|
def initialize(key, redis: nil, validity: DEFAULT_VALIDITY)
|
2014-04-11 13:43:33 +08:00
|
|
|
@key = key
|
2018-09-19 13:49:18 +08:00
|
|
|
@using_global_redis = true if !redis
|
2019-12-03 17:05:53 +08:00
|
|
|
@redis = redis || Discourse.redis
|
2014-04-14 08:51:46 +08:00
|
|
|
@mutex = Mutex.new
|
2019-02-20 09:28:10 +08:00
|
|
|
@validity = validity
|
2014-04-11 13:43:33 +08:00
|
|
|
end
|
|
|
|
|
2014-04-14 08:51:46 +08:00
|
|
|
# NOTE wrapped in mutex to maintain its semantics
|
2019-02-20 09:28:10 +08:00
|
|
|
def synchronize
|
2022-07-11 20:16:37 +08:00
|
|
|
result = nil
|
|
|
|
|
2019-08-01 16:12:05 +08:00
|
|
|
@mutex.synchronize do
|
|
|
|
expire_time = get_lock
|
|
|
|
|
|
|
|
begin
|
2022-07-11 20:16:37 +08:00
|
|
|
result = yield
|
2019-08-01 16:12:05 +08:00
|
|
|
ensure
|
|
|
|
current_time = redis.time[0]
|
2019-08-05 09:57:35 +08:00
|
|
|
if current_time > expire_time
|
|
|
|
warn("held for too long, expected max: #{@validity} secs, took an extra #{current_time - expire_time} secs")
|
2019-08-01 16:12:05 +08:00
|
|
|
end
|
|
|
|
|
2022-07-11 20:16:37 +08:00
|
|
|
unlocked = UNLOCK_SCRIPT.eval(redis, [prefixed_key], [expire_time.to_s])
|
|
|
|
if !unlocked && current_time <= expire_time
|
2019-08-14 16:56:43 +08:00
|
|
|
warn("the redis key appears to have been tampered with before expiration")
|
2019-08-01 16:12:05 +08:00
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
2022-07-11 20:16:37 +08:00
|
|
|
|
|
|
|
result
|
2019-08-01 16:12:05 +08:00
|
|
|
end
|
|
|
|
|
|
|
|
private
|
|
|
|
|
|
|
|
attr_reader :key
|
|
|
|
attr_reader :redis
|
|
|
|
attr_reader :validity
|
|
|
|
|
|
|
|
def get_lock
|
2018-09-19 13:49:18 +08:00
|
|
|
attempts = 0
|
|
|
|
|
2019-08-01 16:12:05 +08:00
|
|
|
while true
|
2022-07-11 20:16:37 +08:00
|
|
|
expire_time = LOCK_SCRIPT.eval(redis, [prefixed_key], [validity])
|
|
|
|
|
|
|
|
return expire_time if expire_time
|
2019-08-01 16:12:05 +08:00
|
|
|
|
2014-04-14 08:51:46 +08:00
|
|
|
sleep 0.001
|
2022-07-11 20:16:37 +08:00
|
|
|
|
2018-09-19 13:49:18 +08:00
|
|
|
# in readonly we will never be able to get a lock
|
2018-09-20 13:15:29 +08:00
|
|
|
if @using_global_redis && Discourse.recently_readonly?
|
2018-09-19 14:07:29 +08:00
|
|
|
attempts += 1
|
2019-02-20 09:23:42 +08:00
|
|
|
|
2022-07-11 20:16:37 +08:00
|
|
|
if attempts > CHECK_READONLY_ATTEMPTS
|
2018-09-19 14:07:29 +08:00
|
|
|
raise Discourse::ReadOnly
|
|
|
|
end
|
2018-09-19 13:49:18 +08:00
|
|
|
end
|
2014-04-14 08:51:46 +08:00
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2022-07-11 20:16:37 +08:00
|
|
|
def prefixed_key
|
|
|
|
@prefixed_key ||= redis.respond_to?(:namespace_key) ? redis.namespace_key(key) : key
|
2014-04-11 13:43:33 +08:00
|
|
|
end
|
|
|
|
|
2022-07-11 20:16:37 +08:00
|
|
|
def warn(msg)
|
|
|
|
Rails.logger.warn("DistributedMutex(#{key.inspect}): #{msg}")
|
2019-08-01 16:12:05 +08:00
|
|
|
end
|
2014-04-11 13:43:33 +08:00
|
|
|
end
|