Class: RedisLocks::Semaphore

Inherits:
Object
  • Object
show all
Defined in:
lib/redis_locks/semaphore.rb

Constant Summary collapse

NAMESPACE = 
"semaphore-lua"
SETUP_SCRIPT =

Removes stale locks, then ensures that all resources which aren’t locked are marked as available.

"  local avail_key = KEYS[1]\n  local grabbed_key = KEYS[2]\n\n  local expected_resources = tonumber(ARGV[1])\n  local stale_before = tonumber(ARGV[2])\n\n  redis.call('zremrangebyscore', grabbed_key, -1, stale_before)\n\n  local found_resources = redis.call('llen', avail_key) + redis.call('zcard', grabbed_key)\n  if found_resources < expected_resources then\n    for i=1,(expected_resources - found_resources) do\n      redis.call('rpush', avail_key, 1)\n    end\n  end\n"
SETUP_DIGEST = 
Digest::SHA1.hexdigest(SETUP_SCRIPT)

Instance Method Summary collapse

Constructor Details

#initialize(key, redis:, resources: 1, stale_client_timeout: 86400) ⇒ Semaphore

‘resources` is the number of clients allowed to lock the semaphore concurrently.

‘stale_client_timeout` is the threshold of time before we assume that something has gone terribly wrong with a client and we invalidate its lock.

Raises:

  • (ArgumentError) 


65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/redis_locks/semaphore.rb', line 65

def initialize(key, redis:, resources: 1, stale_client_timeout: 86400)
  @key = key
  @resource_count = resources.to_i
  @stale_client_timeout = stale_client_timeout.to_f
  @redis = redis
  @tokens = []

  raise ArgumentError.new("Lock key is required") if @key.nil? || @key.empty?
  raise ArgumentError.new("resources must be > 0") unless @resource_count > 0
  raise ArgumentError.new("stale_client_timeout must be > 0") unless @stale_client_timeout > 0
end

Instance Method Details

#delete! ⇒ Object

Forcefully clear the lock. Be careful!



78
79
80
81
82
 
# File 'lib/redis_locks/semaphore.rb', line 78

def delete!
  @redis.del(available_key)
  @redis.del(grabbed_key)
  @tokens = []
end

#lock(timeout: nil, &block) ⇒ Object

Acquire a resource from the semaphore, if available. Returns false if no resources are available.

‘timeout` is how long to wait, blocking, until a resource is available. The default is nil, meaning don’t block. A timeout of zero means block forever. (This is a bit weird, but corresponds to how blpop uses timeout values.)

If passed a block, if a resource is available, runs the block and then unlocks.

If called without a block, if a resource is available, returns a token. Caller is then responsible for unlocking the token.

This isn’t atomic–if the process dies, we could remove something from the available queue without adding it to the grabbed set–but that’s ok, the semaphore will recover just as if this was a stale client that left its token in the grabbed set forever.



101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/redis_locks/semaphore.rb', line 101

def lock(timeout: nil, &block)
  ensure_exists_and_release_stale_locks!

  success =
    if timeout
      !@redis.blpop(available_key, timeout.to_i).nil?
    else
      !@redis.lpop(available_key).nil?
    end

  return false unless success

  token = SecureRandom.hex(16)
  @tokens.push(token)
  @redis.zadd(grabbed_key, epoch_f, token)

  return_or_yield(token, &block)
end

#lock!(timeout: nil, &block) ⇒ Object

Raises:



124
125
126
127
128
 
# File 'lib/redis_locks/semaphore.rb', line 124

def lock!(timeout: nil, &block)
  token = lock(timeout: timeout)
  raise SemaphoreUnavailable.new(@key, @resource_count) unless token
  return_or_yield(token, &block)
end

#unlock(token = @tokens.pop) ⇒ Object Also known as: signal

Release a resource back to the semaphore. Should normally be called with an explicit token.

This isn’t atomic–if the process dies, we could remove something from the blocked set without adding it to the available queue–but that’s ok, the semaphore will recover just as if this was a stale client that left its token in the grabbed set forever.



141
142
143
144
145
146
147
148
149
150
 
# File 'lib/redis_locks/semaphore.rb', line 141

def unlock(token = @tokens.pop)
  return unless token

  removed = @redis.zrem grabbed_key, token
  if removed
    @redis.lpush available_key, 1
  end

  removed
end

#wait(timeout: 0, &block) ⇒ Object 



120
121
122
 
# File 'lib/redis_locks/semaphore.rb', line 120

def wait(timeout: 0, &block)
  lock(timeout: timeout, &block)
end

#wait!(timeout: 0, &block) ⇒ Object 



130
131
132
 
# File 'lib/redis_locks/semaphore.rb', line 130

def wait!(timeout: 0, &block)
  lock!(timeout: timeout, &block)
end