119 lines
2.3 KiB
Ruby
119 lines
2.3 KiB
Ruby
# frozen_string_literal: true
|
|
|
|
# Cross-process locking using Redis.
|
|
class DistributedMutex
|
|
DEFAULT_VALIDITY ||= 60
|
|
|
|
def self.synchronize(key, redis: nil, validity: DEFAULT_VALIDITY, &blk)
|
|
self.new(
|
|
key,
|
|
redis: redis,
|
|
validity: validity
|
|
).synchronize(&blk)
|
|
end
|
|
|
|
def initialize(key, redis: nil, validity: DEFAULT_VALIDITY)
|
|
@key = key
|
|
@using_global_redis = true if !redis
|
|
@redis = redis || $redis
|
|
@mutex = Mutex.new
|
|
@validity = validity
|
|
end
|
|
|
|
CHECK_READONLY_ATTEMPT ||= 10
|
|
|
|
# NOTE wrapped in mutex to maintain its semantics
|
|
def synchronize
|
|
@mutex.synchronize do
|
|
expire_time = get_lock
|
|
|
|
begin
|
|
yield
|
|
ensure
|
|
current_time = redis.time[0]
|
|
unless current_time < expire_time
|
|
warn("held for too long")
|
|
end
|
|
|
|
unless unlock(expire_time)
|
|
warn("didn't unlock cleanly")
|
|
end
|
|
end
|
|
end
|
|
end
|
|
|
|
private
|
|
|
|
attr_reader :key
|
|
attr_reader :redis
|
|
attr_reader :validity
|
|
|
|
def warn(msg)
|
|
Rails.logger.warn("DistributedMutex(#{key.inspect}): #{msg}")
|
|
end
|
|
|
|
def get_lock
|
|
attempts = 0
|
|
|
|
while true
|
|
got_lock, expire_time = try_to_get_lock
|
|
if got_lock
|
|
return expire_time
|
|
end
|
|
|
|
sleep 0.001
|
|
# in readonly we will never be able to get a lock
|
|
if @using_global_redis && Discourse.recently_readonly?
|
|
attempts += 1
|
|
|
|
if attempts > CHECK_READONLY_ATTEMPT
|
|
raise Discourse::ReadOnly
|
|
end
|
|
end
|
|
end
|
|
end
|
|
|
|
def try_to_get_lock
|
|
got_lock = false
|
|
|
|
now = redis.time[0]
|
|
expire_time = now + validity
|
|
|
|
redis.watch key
|
|
|
|
current_expire_time = redis.get key
|
|
|
|
if current_expire_time.present? && current_expire_time.to_i > now
|
|
redis.unwatch
|
|
|
|
got_lock = false
|
|
else
|
|
result =
|
|
redis.multi do
|
|
redis.set key, expire_time.to_s
|
|
redis.expire key, validity
|
|
end
|
|
|
|
got_lock = result.present?
|
|
end
|
|
|
|
[got_lock, expire_time]
|
|
end
|
|
|
|
def unlock(expire_time)
|
|
redis.watch key
|
|
current_expire_time = redis.get key
|
|
|
|
if current_expire_time == expire_time.to_s
|
|
result =
|
|
redis.multi do
|
|
redis.del key
|
|
end
|
|
return result.present?
|
|
else
|
|
redis.unwatch
|
|
return false
|
|
end
|
|
end
|
|
end
|