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

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) ⇒ 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 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.



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

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
)
  @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)
end

Instance Attribute Details

#host ⇒ String

host for redis server

Returns:

  • (String) —

    the current value of host



42
43
44
 
# File 'lib/redis_single_file/semaphore.rb', line 42

def host
  @host
end

#name ⇒ String

custom sync queue name

Returns:

  • (String) —

    the current value of name



42
43
44
 
# File 'lib/redis_single_file/semaphore.rb', line 42

def name
  @name
end

#port ⇒ String

port for redis server

Returns:

  • (String) —

    the current value of port



42
43
44
 
# File 'lib/redis_single_file/semaphore.rb', line 42

def port
  @port
end

#redis=(value) ⇒ ...

redis client instance

Parameters:

  • value (...) —

    the value to set the attribute redis to.

Returns:

  • (...) —

    the newly set value



42
43
44
 
# File 'lib/redis_single_file/semaphore.rb', line 42

def redis=(value)
  @redis = value
end

Instance Method Details

#synchronize(timeout: 0) ⇒ 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

Yield Returns:

  • (...) —

    response from synchronized block execution

Returns:

  • (nil) —

    redis blpop timeout



83
84
85
86
87
 
# File 'lib/redis_single_file/semaphore.rb', line 83

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

#synchronize!(timeout: 0) ⇒ 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

Yield Returns:

  • (...) —

    response from synchronized block execution

Raises:



95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/redis_single_file/semaphore.rb', line 95

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

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

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

  yield
ensure
  # always cycle the queue when exiting
  unlock_queue if block_given?
end