DEV: Allow validity of lock to be customizable for `DistributedMutex`.
- Allows a user to override the default lock validity of 60 seconds. - Also clean up test which was leaking a redis key
This commit is contained in:
parent
8cd4ceba49
commit
4f9e5e19c8
|
@ -1,8 +1,9 @@
|
||||||
# Cross-process locking using Redis.
|
# Cross-process locking using Redis.
|
||||||
class DistributedMutex
|
class DistributedMutex
|
||||||
|
DEFAULT_VALIDITY = 60
|
||||||
|
|
||||||
def self.synchronize(key, redis = nil, &blk)
|
def self.synchronize(key, redis: nil, validity: DEFAULT_VALIDITY, &blk)
|
||||||
self.new(key, redis).synchronize(&blk)
|
self.new(key, redis).synchronize(validity: DEFAULT_VALIDITY, &blk)
|
||||||
end
|
end
|
||||||
|
|
||||||
def initialize(key, redis = nil)
|
def initialize(key, redis = nil)
|
||||||
|
@ -15,16 +16,16 @@ class DistributedMutex
|
||||||
CHECK_READONLY_ATTEMPT ||= 10
|
CHECK_READONLY_ATTEMPT ||= 10
|
||||||
|
|
||||||
# NOTE wrapped in mutex to maintain its semantics
|
# NOTE wrapped in mutex to maintain its semantics
|
||||||
def synchronize
|
def synchronize(validity: DEFAULT_VALIDITY)
|
||||||
|
|
||||||
@mutex.lock
|
@mutex.lock
|
||||||
attempts = 0
|
attempts = 0
|
||||||
|
|
||||||
while !try_to_get_lock
|
while !try_to_get_lock(validity)
|
||||||
sleep 0.001
|
sleep 0.001
|
||||||
# in readonly we will never be able to get a lock
|
# in readonly we will never be able to get a lock
|
||||||
if @using_global_redis && Discourse.recently_readonly?
|
if @using_global_redis && Discourse.recently_readonly?
|
||||||
attempts += 1
|
attempts += 1
|
||||||
|
|
||||||
if attempts > CHECK_READONLY_ATTEMPT
|
if attempts > CHECK_READONLY_ATTEMPT
|
||||||
raise Discourse::ReadOnly
|
raise Discourse::ReadOnly
|
||||||
end
|
end
|
||||||
|
@ -40,18 +41,20 @@ class DistributedMutex
|
||||||
|
|
||||||
private
|
private
|
||||||
|
|
||||||
def try_to_get_lock
|
def try_to_get_lock(validity)
|
||||||
got_lock = false
|
got_lock = false
|
||||||
if @redis.setnx @key, Time.now.to_i + 60
|
|
||||||
@redis.expire @key, 60
|
if @redis.setnx @key, Time.now.to_i + validity
|
||||||
|
@redis.expire @key, validity
|
||||||
got_lock = true
|
got_lock = true
|
||||||
else
|
else
|
||||||
begin
|
begin
|
||||||
@redis.watch @key
|
@redis.watch @key
|
||||||
time = @redis.get @key
|
time = @redis.get @key
|
||||||
|
|
||||||
if time && time.to_i < Time.now.to_i
|
if time && time.to_i < Time.now.to_i
|
||||||
got_lock = @redis.multi do
|
got_lock = @redis.multi do
|
||||||
@redis.set @key, Time.now.to_i + 60
|
@redis.set @key, Time.now.to_i + validity
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
ensure
|
ensure
|
||||||
|
|
|
@ -2,9 +2,15 @@ require 'rails_helper'
|
||||||
require_dependency 'distributed_mutex'
|
require_dependency 'distributed_mutex'
|
||||||
|
|
||||||
describe DistributedMutex do
|
describe DistributedMutex do
|
||||||
|
let(:key) { "test_mutex_key" }
|
||||||
|
|
||||||
|
after do
|
||||||
|
$redis.del(key)
|
||||||
|
end
|
||||||
|
|
||||||
it "allows only one mutex object to have the lock at a time" do
|
it "allows only one mutex object to have the lock at a time" do
|
||||||
mutexes = (1..10).map do
|
mutexes = (1..10).map do
|
||||||
DistributedMutex.new("test_mutex_key")
|
DistributedMutex.new(key)
|
||||||
end
|
end
|
||||||
|
|
||||||
x = 0
|
x = 0
|
||||||
|
@ -22,9 +28,9 @@ describe DistributedMutex do
|
||||||
end
|
end
|
||||||
|
|
||||||
it "handles auto cleanup correctly" do
|
it "handles auto cleanup correctly" do
|
||||||
m = DistributedMutex.new("test_mutex_key")
|
m = DistributedMutex.new(key)
|
||||||
|
|
||||||
$redis.setnx "test_mutex_key", Time.now.to_i - 1
|
$redis.setnx key, Time.now.to_i - 1
|
||||||
|
|
||||||
start = Time.now.to_i
|
start = Time.now.to_i
|
||||||
m.synchronize do
|
m.synchronize do
|
||||||
|
@ -35,8 +41,26 @@ describe DistributedMutex do
|
||||||
expect(Time.now.to_i).to be <= start + 1
|
expect(Time.now.to_i).to be <= start + 1
|
||||||
end
|
end
|
||||||
|
|
||||||
|
it 'allows the validity of the lock to be configured' do
|
||||||
|
freeze_time
|
||||||
|
|
||||||
|
mutex = DistributedMutex.new(key)
|
||||||
|
|
||||||
|
mutex.synchronize(validity: 2) do
|
||||||
|
expect($redis.ttl(key)).to eq(2)
|
||||||
|
expect($redis.get(key).to_i).to eq(Time.now.to_i + 2)
|
||||||
|
end
|
||||||
|
|
||||||
|
mutex.synchronize do
|
||||||
|
expect($redis.ttl(key)).to eq(DistributedMutex::DEFAULT_VALIDITY)
|
||||||
|
|
||||||
|
expect($redis.get(key).to_i)
|
||||||
|
.to eq(Time.now.to_i + DistributedMutex::DEFAULT_VALIDITY)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
it "maintains mutex semantics" do
|
it "maintains mutex semantics" do
|
||||||
m = DistributedMutex.new("test_mutex_key")
|
m = DistributedMutex.new(key)
|
||||||
|
|
||||||
expect {
|
expect {
|
||||||
m.synchronize do
|
m.synchronize do
|
||||||
|
@ -55,7 +79,7 @@ describe DistributedMutex do
|
||||||
end
|
end
|
||||||
|
|
||||||
it "works even if redis is in readonly" do
|
it "works even if redis is in readonly" do
|
||||||
m = DistributedMutex.new("test_readonly")
|
m = DistributedMutex.new(key)
|
||||||
start = Time.now
|
start = Time.now
|
||||||
done = false
|
done = false
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue