Class: RedisSingleFile::Semaphore

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

Overview

This class acts as the main synchronization engine for distributed logic execution by utilizing the redis blpop command to facilitate a distributed synchronous queue.

Examples:

Default lock name and infinite blocking

semaphore = RedisSingleFile::Semaphore.new
semaphore.synchronize do
   # synchronized logic defined here...
end

Named locks can provide exclusive synchronization

semaphore = RedisSingleFile::Semaphore.new(name: :user_cache_update)
semaphore.synchronize do
   # synchronized logic defined here...
end

Prevent deadlocks by providing a timeout

semaphore = RedisSingleFile::Semaphore.new(name: s3_file_upload)
semaphore.synchronize(timeout: 15) do
   # synchronized logic defined here...
end

Support concurrent worker processing

semaphore = RedisSingleFile::Semaphore.new(name: :concurrent_queue)
semaphore.synchronize(concurrency: 3) do
  # synchronized logic defined here...
end

Use your own redis client instance

redis = Redis.new(...)
semaphore = RedisSingleFile::Semaphore.new(redis:)
semaphore.synchronize do
  # synchronized logic defined here...
end

Returns:

  • (self) —

    the semaphore instance

Author:

  • lifeBCE

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(redis: nil, name: Configuration.name, host: Configuration.host, port: Configuration.port, concurrency: Configuration.concurrency) ⇒ self

Note:

redis: Any more advanced configuration than host and port should be applied to an instance outside of redis single file and passed in via this attribute.

Note:

name: Distributed semaphores are coordinated by name. Each client that wishes to synchronize a particular block should do so under the same name.

Note:

port: Each synchronized execution can be done on a different redis port than globally configured. Passing a value for this attribute will redirect to that port.

Note:

concurrency: When a distributed lock needs to allow multiple executions at once, concurrency can be used. When this settings is set to more than the default value of 1, the queue will allow that many simultaneous slots in the distributed queue.

#note host:

Each synchronized execution can be done on a different redis server
than globally configured. Passing a value for this attribute will
redirect to that host.


77
78
79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/redis_single_file/semaphore.rb', line 77

def initialize(
  redis: nil,               # provide your own redis instance
  name: Configuration.name, # designate queue name per session
  host: Configuration.host, # designate redis host per session
  port: Configuration.port, # designate redis port per session
  concurrency: Configuration.concurrency # concurrent workers
)
  @redis = redis || Redis.new(host:, port:)

  @mutex_val = name
  @mutex_key = format(Configuration.mutex_key, @mutex_val)
  @queue_key = format(Configuration.queue_key, @mutex_val)
  @concurrency = concurrency.to_i
end

Instance Attribute Details

#concurrency ⇒ Integer

simultaneous slots allowed

Returns:

  • (Integer) —

    the current value of concurrency



49
50
51
 
# File 'lib/redis_single_file/semaphore.rb', line 49

def concurrency
  @concurrency
end

#host ⇒ String

host for redis server

Returns:

  • (String) —

    the current value of host



49
50
51
 
# File 'lib/redis_single_file/semaphore.rb', line 49

def host
  @host
end

#name ⇒ String

custom sync queue name

Returns:

  • (String) —

    the current value of name



49
50
51
 
# File 'lib/redis_single_file/semaphore.rb', line 49

def name
  @name
end

#port ⇒ String

port for redis server

Returns:

  • (String) —

    the current value of port



49
50
51
 
# File 'lib/redis_single_file/semaphore.rb', line 49

def port
  @port
end

#redis=(value) ⇒ ...

redis client instance

Parameters:

  • value (...) —

    the value to set the attribute redis to.

Returns:

  • (...) —

    the newly set value



49
50
51
 
# File 'lib/redis_single_file/semaphore.rb', line 49

def redis=(value)
  @redis = value
end

Instance Method Details

#synchronize(timeout: 0, concurrency: @concurrency) ⇒ nil

Queues up client and waits for turn to execute. Returns nil when queue wait time expires.

Parameters:

  • timeout (Integer) (defaults to: 0) —

    seconds for client to wait in queue

  • concurrency (Integer) (defaults to: @concurrency) —

    override concurrent workers

Yield Returns:

  • (...) —

    response from synchronized block execution

Returns:

  • (nil) —

    redis blpop timeout



99
100
101
102
103
 
# File 'lib/redis_single_file/semaphore.rb', line 99

def synchronize(timeout: 0, concurrency: @concurrency, &)
  synchronize!(timeout:, concurrency:, &)
rescue QueueTimeoutError => _e
  nil
end

#synchronize!(timeout: 0, concurrency: @concurrency) ⇒ Object

Queues up client and waits for turn to execute. Raise exception when queue wait time expires.

Parameters:

  • timeout (Integer) (defaults to: 0) —

    seconds for blpop to wait in queue

  • concurrency (Integer) (defaults to: @concurrency) —

    override concurrent workers

Yield Returns:

  • (...) —

    response from synchronized block execution

Raises:



112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
 
# File 'lib/redis_single_file/semaphore.rb', line 112

def synchronize!(timeout: 0, concurrency: @concurrency)
  return unless block_given?

  token_acquired = false

  with_retry_protection do
    prime_queue(concurrency) unless redis.getset(mutex_key, mutex_val)
    raise QueueTimeoutError  unless redis.blpop(queue_key, timeout:)

    token_acquired = true

    redis.multi do
      redis.persist(mutex_key) # unexpire during execution
      redis.persist(queue_key) # unexpire during execution
    end
  end

  yield
ensure
  # only return a token this client successfully acquired
  unlock_queue(concurrency) if token_acquired
end