Class: Sidecloq::Locker

Inherits:
Object
  • Object
show all
Includes:
Utils
Defined in:
lib/sidecloq/locker.rb

Overview

Locker obtains or waits for an exclusive lock on a key in redis

Constant Summary collapse

DEFAULT_LOCK_KEY = 
'sidecloq_leader_lock'

Instance Method Summary collapse

Methods included from Utils

included, #logger, #redis, #will_never_run

Constructor Details

#initialize(options = {}) ⇒ Locker

Returns a new instance of Locker.



10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/sidecloq/locker.rb', line 10

def initialize(options = {})
  # we keep a connection from the pool by default
  @redis = options[:redis] || Sidekiq.redis_pool.checkout
  # for compatibility with sidekiq < 7, we need to convert the Redis
  # instance to a RedisClient instance, which redlock requires
  if defined?(Redis) && @redis.instance_of?(Redis)
    @redis = RedisClient.new(@redis.connection.except(:location))
  end
  @key = options[:lock_key] || DEFAULT_LOCK_KEY
  @ttl = options[:ttl] || 60
  @check_interval = options[:check_interval] || 15
  @lock_manager = Redlock::Client.new([@redis])
  @obtained_lock = Concurrent::Event.new
  @check_task = nil
  @lock = nil
  @stopping = false
  @in_with_lock = false
end

Instance Method Details

#locked? ⇒ Boolean

Returns:

  • (Boolean) 


62
63
64
 
# File 'lib/sidecloq/locker.rb', line 62

def locked?
  !@stopping && @lock && @obtained_lock.set?
end

#start ⇒ Object 



68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/sidecloq/locker.rb', line 68

def start
  logger.debug('Starting locker check task')
  @check_task = Concurrent::TimerTask.new(
    execution_interval: @check_interval,
    run_now: true
  ) do
    try_to_get_or_refresh_lock
  end
  @check_task.execute

  # return the check task, to help with tests
  @check_task
end

#stop(timeout = nil) ⇒ Object 



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/sidecloq/locker.rb', line 44

def stop(timeout = nil)
  return if @stopping

  @stopping = true

  if @check_task
    logger.debug('Stopping locker check task')
    @check_task.shutdown
    @check_task.wait_for_termination(timeout)
    logger.debug('Stopped locker check task')
  end

  # release the lock in case someone is blocked on with_lock
  @obtained_lock.set

  @stopping = false unless @in_with_lock
end

#try_to_get_or_refresh_lock ⇒ Object 



82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/sidecloq/locker.rb', line 82

def try_to_get_or_refresh_lock
  # redlock is in ms, not seconds
  if @lock
    @lock = @lock_manager.lock(@key, @ttl * 1000, extend: @lock)
  else
    @lock = @lock_manager.lock(@key, @ttl * 1000)
  end
  @obtained_lock.set if @lock
  logger.debug("Leader lock #{'not ' unless @lock}held")
  @lock
end

#with_lock ⇒ Object

blocks until lock is obtained, then yields



30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/sidecloq/locker.rb', line 30

def with_lock
  @in_with_lock = true

  start
  @obtained_lock.wait

  yield if locked?

  stop
  @stopping = false

  @in_with_lock = false
end