Class: CountingSemaphore::RedisSemaphore

Inherits:
Object
  • Object
show all
Includes:
WithLeaseSupport
Defined in:
lib/counting_semaphore/redis_semaphore.rb

Defined Under Namespace

Classes: NullPool

Constant Summary collapse

LEASE_EXPIRATION_SECONDS = 
5
GET_LEASE_SCRIPT =

Lua script for atomic lease acquisition Returns: [success, lease_key, current_usage] success: 1 if lease was acquired, 0 if no capacity lease_key: the key of the acquired lease (if successful) current_usage: current usage count after operation

<<~LUA
  local lease_key = KEYS[1]
  local lease_set_key = KEYS[2]
  local capacity = tonumber(ARGV[1])
  local permit_count = tonumber(ARGV[2])
  local expiration_seconds = tonumber(ARGV[3])
  
  -- Get all active leases from the set and calculate current usage
  local lease_keys = redis.call('SMEMBERS', lease_set_key)
  local current_usage = 0
  local valid_leases = {}
  
  for i, key in ipairs(lease_keys) do
    local permits = redis.call('GET', key)
    if permits then
      local permits_from_lease = tonumber(permits)
      if permits_from_lease then
        current_usage = current_usage + permits_from_lease
        table.insert(valid_leases, key)
      else
        -- Remove lease with invalid permit count
        redis.call('DEL', key)
        redis.call('SREM', lease_set_key, key)
      end
    else
      -- Lease key doesn't exist, remove from set
      redis.call('SREM', lease_set_key, key)
    end
  end
  
  -- Check if we have capacity
  local available = capacity - current_usage
  if available >= permit_count then
    -- Set lease with TTL (value is just the permit count)
    redis.call('SETEX', lease_key, expiration_seconds, permit_count)
    -- Add lease key to the set
    redis.call('SADD', lease_set_key, lease_key)
    -- Set TTL on the set (4x the lease TTL to ensure cleanup)
    redis.call('EXPIRE', lease_set_key, expiration_seconds * 4)
    
    return {1, lease_key, current_usage + permit_count}
  else
    return {0, '', current_usage}
  end
LUA
GET_USAGE_SCRIPT =

Lua script for getting current usage Returns: current_usage (integer)

<<~LUA
  local lease_set_key = KEYS[1]
  local expiration_seconds = tonumber(ARGV[1])
  
  -- Get all active leases from the set and calculate current usage
  local lease_keys = redis.call('SMEMBERS', lease_set_key)
  local current_usage = 0
  local has_valid_leases = false
  
  for i, lease_key in ipairs(lease_keys) do
    local permits = redis.call('GET', lease_key)
    if permits then
      local permits_from_lease = tonumber(permits)
      if permits_from_lease then
        current_usage = current_usage + permits_from_lease
        has_valid_leases = true
      else
        -- Remove lease with invalid permit count
        redis.call('DEL', lease_key)
        redis.call('SREM', lease_set_key, lease_key)
      end
    else
      -- Lease key doesn't exist, remove from set
      redis.call('SREM', lease_set_key, lease_key)
    end
  end
  
  -- Refresh TTL on the set if there are valid leases (4x the lease TTL)
  if has_valid_leases then
    redis.call('EXPIRE', lease_set_key, expiration_seconds * 4)
  end
  
  return current_usage
LUA
RELEASE_LEASE_SCRIPT =

Lua script for atomic lease release and signal Returns: 1 (success)

<<~LUA
  local lease_key = KEYS[1]
  local queue_key = KEYS[2]
  local lease_set_key = KEYS[3]
  local permit_count = tonumber(ARGV[1])
  local max_signals = tonumber(ARGV[2])
  
  -- Remove the lease
  redis.call('DEL', lease_key)
  -- Remove from the lease set
  redis.call('SREM', lease_set_key, lease_key)
  
  -- Signal waiting clients about the released permits
  redis.call('LPUSH', queue_key, 'permits:' .. permit_count)
  
  -- Trim queue to prevent indefinite growth (atomic)
  redis.call('LTRIM', queue_key, 0, max_signals - 1)
  
  return 1
LUA
GET_LEASE_SCRIPT_SHA =

Precomputed script SHAs

Digest::SHA1.hexdigest(GET_LEASE_SCRIPT)
GET_USAGE_SCRIPT_SHA = 
Digest::SHA1.hexdigest(GET_USAGE_SCRIPT)
RELEASE_LEASE_SCRIPT_SHA = 
Digest::SHA1.hexdigest(RELEASE_LEASE_SCRIPT)

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from WithLeaseSupport

#currently_leased, #with_lease

Constructor Details

#initialize(capacity, namespace, redis: nil, logger: CountingSemaphore::NullLogger, lease_expiration_seconds: LEASE_EXPIRATION_SECONDS) ⇒ RedisSemaphore

Initialize the semaphore with a maximum capacity and required namespace.

Parameters:

  • capacity (Integer) —

    Maximum number of concurrent operations allowed (also called permits)

  • namespace (String) —

    Required namespace for Redis keys

  • redis (Redis, ConnectionPool) (defaults to: nil) —

    Optional Redis client or connection pool (defaults to new Redis instance)

  • logger (Logger) (defaults to: CountingSemaphore::NullLogger) —

    the logger

Raises:

  • (ArgumentError) —

    if capacity is not positive



142
143
144
145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/counting_semaphore/redis_semaphore.rb', line 142

def initialize(capacity, namespace, redis: nil, logger: CountingSemaphore::NullLogger, lease_expiration_seconds: LEASE_EXPIRATION_SECONDS)
  raise ArgumentError, "Capacity must be positive, got #{capacity}" unless capacity > 0

  # Require Redis only when RedisSemaphore is used
  require "redis" unless defined?(Redis)

  @capacity = capacity
  @redis_connection_pool = wrap_redis_client_with_pool(redis || Redis.new)
  @namespace = namespace
  @lease_expiration_seconds = lease_expiration_seconds
  @logger = logger

  # Scripts are precomputed and will be loaded on-demand if needed
end

Instance Attribute Details

#capacity ⇒ Integer (readonly)

Returns:

  • (Integer) 


133
134
135
 
# File 'lib/counting_semaphore/redis_semaphore.rb', line 133

def capacity
  @capacity
end

Instance Method Details

#acquire(permits = 1) ⇒ CountingSemaphore::Lease

Acquires the given number of permits from this semaphore, blocking until all are available.

Parameters:

  • permits (Integer) (defaults to: 1) —

    Number of permits to acquire (default: 1)

Returns:

Raises:

  • (ArgumentError) —

    if permits is not an integer or is less than one



182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/counting_semaphore/redis_semaphore.rb', line 182

def acquire(permits = 1)
  raise ArgumentError, "Permits must be at least 1, got #{permits}" if permits < 1
  if permits > @capacity
    raise ArgumentError, "Cannot acquire #{permits} permits as capacity is only #{@capacity}"
  end

  lease_key = acquire_lease_internal(permits, timeout_seconds: nil)
  @logger.debug { "Acquired #{permits} permits with lease #{lease_key}" }

  CountingSemaphore::Lease.new(
    semaphore: self,
    id: lease_key,
    permits: permits
  )
end

#available_permits ⇒ Integer

Returns the current number of permits available in this semaphore.

Returns:

  • (Integer) —

    Number of available permits



244
245
246
247
 
# File 'lib/counting_semaphore/redis_semaphore.rb', line 244

def available_permits
  current_usage = get_current_usage
  @capacity - current_usage
end

#debug_info ⇒ Hash

Returns debugging information about the current state of the semaphore. Includes current usage, capacity, available permits, and details about active leases.

Examples:

info = semaphore.debug_info
puts "Usage: #{info[:usage]}/#{info[:capacity]}"
info[:active_leases].each { |lease| puts "Lease: #{lease[:key]} - #{lease[:permits]} permits" }

Returns:

  • (Hash) —

    A hash containing :usage, :capacity, :available, and :active_leases



269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
 
# File 'lib/counting_semaphore/redis_semaphore.rb', line 269

def debug_info
  usage = get_current_usage
  lease_set_key = "#{@namespace}:lease_set"
  lease_keys = with_redis { |redis| redis.smembers(lease_set_key) }
  active_leases = []

  lease_keys.each do |lease_key|
    permits = with_redis { |redis| redis.get(lease_key) }
    next unless permits

    active_leases << {
      key: lease_key,
      permits: permits.to_i
    }
  end

  {
    usage: usage,
    capacity: @capacity,
    available: @capacity - usage,
    active_leases: active_leases
  }
end

#drain_permits ⇒ CountingSemaphore::Lease?

Acquires and returns all permits that are immediately available. Note: For distributed semaphores, this may not be perfectly accurate due to race conditions.

Returns:



253
254
255
256
257
258
259
 
# File 'lib/counting_semaphore/redis_semaphore.rb', line 253

def drain_permits
  available = available_permits
  return nil if available <= 0

  # Try to acquire all available permits
  try_acquire(available, timeout: 0.1)
end

#release(lease) ⇒ nil

Releases a previously acquired lease, returning the permits to the semaphore.

Parameters:

Returns:

  • (nil)

Raises:

  • (ArgumentError) —

    if lease belongs to a different semaphore



203
204
205
206
207
208
209
210
 
# File 'lib/counting_semaphore/redis_semaphore.rb', line 203

def release(lease)
  unless lease.semaphore == self
    raise ArgumentError, "Lease belongs to a different semaphore"
  end

  release_lease(lease.id, lease.permits)
  nil
end

#try_acquire(permits = 1, timeout: nil) ⇒ CountingSemaphore::Lease?

Acquires the given number of permits from this semaphore, only if all are available at the time of invocation or within the timeout interval.

Parameters:

  • permits (Integer) (defaults to: 1) —

    Number of permits to acquire (default: 1)

  • timeout (Numeric, nil) (defaults to: nil) —

    Number of seconds to wait, or nil to return immediately (default: nil). The timeout value will be rounded up to the nearest whole second due to Redis BLPOP limitations.

Returns:

Raises:

  • (ArgumentError) —

    if permits is not an integer or is less than one



220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
 
# File 'lib/counting_semaphore/redis_semaphore.rb', line 220

def try_acquire(permits = 1, timeout: nil)
  raise ArgumentError, "Permits must be at least 1, got #{permits}" if permits < 1
  if permits > @capacity
    return nil
  end

  timeout_seconds = timeout.nil? ? 0.1 : timeout
  begin
    lease_key = acquire_lease_internal(permits, timeout_seconds: timeout_seconds)
    @logger.debug { "Acquired #{permits} permits (try) with lease #{lease_key}" }

    CountingSemaphore::Lease.new(
      semaphore: self,
      id: lease_key,
      permits: permits
    )
  rescue CountingSemaphore::LeaseTimeout
    nil
  end
end