Module: Resque::Plugins::LockTimeout

Defined in:
lib/resque/plugins/lock_timeout.rb

Overview

If you want only one instance of your job running at a time, extend it with this module:

require 'resque-lock-timeout'

class UpdateNetworkGraph
extend Resque::Plugins::LockTimeout
@queue = :network_graph

def self.perform(repo_id)
  heavy_lifting
end
end

If you wish to limit the duration a lock may be held for, you can set/override lock_timeout. e.g.

class UpdateNetworkGraph
extend Resque::Plugins::LockTimeout
@queue = :network_graph

# lock may be held for upto an hour.
@lock_timeout = 3600

def self.perform(repo_id)
  heavy_lifting
end
end

If you wish that only one instance of the job defined by #identifier may be enqueued or running, you can set/override loner. e.g.

class PdfExport
extend Resque::Plugins::LockTimeout
@queue = :exports

# only one job can be running/enqueued at a time. For instance a button
# to run a PDF export. If the user clicks several times on it, enqueue
# the job if and only if
#   - the same export is not currently running
#   - the same export is not currently queued.
# ('same' being defined by `identifier`)
@loner = true

def self.perform(repo_id)
  heavy_lifting
end
end

Instance Method Summary collapse

Instance Method Details

#acquire_lock!(*args) ⇒ Boolean, Fixnum

Try to acquire a lock for running the job.

Returns:

  • (Boolean, Fixnum) 


185
186
187
 
# File 'lib/resque/plugins/lock_timeout.rb', line 185

def acquire_lock!(*args)
  acquire_lock_impl!(:redis_lock_key, :lock_failed, *args)
end

#acquire_lock_algorithm!(lock_key, *args) ⇒ Object

Attempts to acquire the lock using a timeout / deadlock algorithm.

Locking algorithm: http://code.google.com/p/redis/wiki/SetnxCommand

Parameters:

  • lock_key (String) —

    redis lock key

  • args (Array) —

    job arguments



228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
 
# File 'lib/resque/plugins/lock_timeout.rb', line 228

def acquire_lock_algorithm!(lock_key, *args)
  now = Time.now.to_i
  lock_until = now + lock_timeout(*args)
  acquired = false

  return [true, lock_until] if lock_redis.setnx(lock_key, lock_until)
  # Can't acquire the lock, see if it has expired.
  lock_expiration = lock_redis.get(lock_key)
  if lock_expiration && lock_expiration.to_i < now
    # expired, try to acquire.
    lock_expiration = lock_redis.getset(lock_key, lock_until)
    if lock_expiration.nil? || lock_expiration.to_i < now
      acquired = true
    end
  else
    # Try once more...
    acquired = true if lock_redis.setnx(lock_key, lock_until)
  end

  [acquired, lock_until]
end

#acquire_lock_impl!(lock_key_method, failed_hook, *args) ⇒ Boolean, Fixnum

Generic implementation of the locking logic

Returns false; when unable to acquire the lock.

  • Returns true; when lock acquired, without a timeout.
  • Returns timestamp; when lock acquired with a timeout, timestamp is when the lock timeout expires.

Parameters:

  • lock_key_method (Symbol) —

    the method returning redis key to lock

  • failed_hook (Symbol) —

    the method called if lock failed

  • args (Array) —

    job arguments

Returns:

  • (Boolean, Fixnum) 


206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
 
# File 'lib/resque/plugins/lock_timeout.rb', line 206

def acquire_lock_impl!(lock_key_method, failed_hook, *args)
  acquired = false
  lock_key = self.send lock_key_method, *args

  unless lock_timeout(*args) > 0
    # Acquire without using a timeout.
    acquired = true if lock_redis.setnx(lock_key, true)
  else
    # Acquire using the timeout algorithm.
    acquired, lock_until = acquire_lock_algorithm!(lock_key, *args)
  end

  self.send(failed_hook, *args) if !acquired
  lock_until && acquired ? lock_until : acquired
end

#acquire_loner_lock!(*args) ⇒ Boolean, Fixnum

Try to acquire a lock to enqueue a loner job.

Returns:

  • (Boolean, Fixnum) 


191
192
193
 
# File 'lib/resque/plugins/lock_timeout.rb', line 191

def acquire_loner_lock!(*args)
  acquire_lock_impl!(:redis_loner_lock_key, :loner_enqueue_failed, *args)
end

#around_perform_lock(*args) ⇒ Object

Where the magic happens.

Parameters:

  • args (Array) —

    job arguments



276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
 
# File 'lib/resque/plugins/lock_timeout.rb', line 276

def around_perform_lock(*args)
  lock_until = acquire_lock!(*args)

  # Release loner lock as job has been dequeued
  release_loner_lock!(*args) if loner

  # Abort if another job holds the lock.
  return unless lock_until

  begin
    yield
  ensure
    # Release the lock on success and error. Unless a lock_timeout is
    # used, then we need to be more careful before releasing the lock.
    now = Time.now.to_i
    if lock_until != true and lock_until < now
      # Eeek! Lock expired before perform finished. Trigger callback.
      lock_expired_before_release(*args)
    else
      release_lock!(*args)
    end
  end
end

#before_enqueue_lock(*args) ⇒ Object

This method is abstract.

if the job is a loner, enqueue only if no other same job is already running/enqueued

Parameters:

  • args (Array) —

    job arguments



171
172
173
174
175
176
177
178
179
180
181
 
# File 'lib/resque/plugins/lock_timeout.rb', line 171

def before_enqueue_lock(*args)
  if loner
    if locked?(*args)
      # Same job is currently running
      loner_enqueue_failed(*args)
      false
    else
      acquire_loner_lock!(*args)
    end
  end
end

#enqueued?(*args) ⇒ Boolean

Convenience method to check if a loner job is queued and lock did not expire.

Returns:

  • (Boolean) —

    true if the job is already queued



130
131
132
 
# File 'lib/resque/plugins/lock_timeout.rb', line 130

def enqueued?(*args)
  inspect_lock(:redis_loner_lock_key, *args)
end

#identifier(*args) ⇒ String?

This method is abstract.

You may override to implement a custom identifier, you should consider doing this if your job arguments are many/long or may not cleanly cleanly to strings.

Builds an identifier using the job arguments. This identifier is used as part of the redis lock key.

Parameters:

  • args (Array) —

    job arguments

Returns:

  • (String, nil) —

    job identifier



61
62
63
 
# File 'lib/resque/plugins/lock_timeout.rb', line 61

def identifier(*args)
  args.join('-')
end

#inspect_lock(lock_key_method, *args) ⇒ Boolean

Check for existence of given key.

Parameters:

  • args (Array) —

    job arguments

  • lock_key_method (Symbol) —

    the method returning redis key to lock

Returns:

  • (Boolean) —

    true if the lock exists



139
140
141
142
143
 
# File 'lib/resque/plugins/lock_timeout.rb', line 139

def inspect_lock(lock_key_method, *args)
  lock_until = lock_redis.get(self.send(lock_key_method, *args))
  return (lock_until.to_i > Time.now.to_i) if lock_timeout(*args) > 0
  !lock_until.nil?
end

#lock_expired_before_release(*args) ⇒ Object

This method is abstract.

Hook method; called when the lock expired before we released it.

Parameters:

  • args (Array) —

    job arguments



163
164
 
# File 'lib/resque/plugins/lock_timeout.rb', line 163

def lock_expired_before_release(*args)
end

#lock_failed(*args) ⇒ Object

This method is abstract.

Hook method; called when a were unable to acquire the lock.

Parameters:

  • args (Array) —

    job arguments



149
150
 
# File 'lib/resque/plugins/lock_timeout.rb', line 149

def lock_failed(*args)
end

#lock_redis ⇒ Redis

Override to fully control the redis object used for storing the locks.

The default is Resque.redis

Returns:

  • (Redis) —

    redis object



71
72
73
 
# File 'lib/resque/plugins/lock_timeout.rb', line 71

def lock_redis
  Resque.redis
end

#lock_timeout(*args) ⇒ Fixnum

Number of seconds the lock may be held for. A value of 0 or below will lock without a timeout.

Parameters:

  • args (Array) —

    job arguments

Returns:

  • (Fixnum) 


101
102
103
 
# File 'lib/resque/plugins/lock_timeout.rb', line 101

def lock_timeout(*args)
  @lock_timeout ||= 0
end

#locked?(*args) ⇒ Boolean

Convenience method to check if job is locked and lock did not expire.

Returns:

  • (Boolean) —

    true if the job is locked by someone



123
124
125
 
# File 'lib/resque/plugins/lock_timeout.rb', line 123

def locked?(*args)
  inspect_lock(:redis_lock_key, *args)
end

#loner(*args) ⇒ TrueClass || FalseClass

Whether one instance of the job should be running or enqueued.

Parameters:

  • args (Array) —

    job arguments

Returns:

  • (TrueClass || FalseClass) 


109
110
111
 
# File 'lib/resque/plugins/lock_timeout.rb', line 109

def loner(*args)
  @loner ||= false
end

#loner_enqueue_failed(*args) ⇒ Object

This method is abstract.

Hook method; called when a were unable to enqueue loner job.

Parameters:

  • args (Array) —

    job arguments



156
157
 
# File 'lib/resque/plugins/lock_timeout.rb', line 156

def loner_enqueue_failed(*args)
end

#loner_locked?(*args) ⇒ Boolean

Checks if job is locked or loner locked (if applicable).

Returns:

  • (Boolean) —

    true if the job is locked by someone



116
117
118
 
# File 'lib/resque/plugins/lock_timeout.rb', line 116

def loner_locked?(*args)
  locked?(*args) || (loner && enqueued?(*args))
end

#redis_lock_key(*args) ⇒ String

Override to fully control the lock key used. It is passed the job arguments.

The default looks like this: lock:<class name>:<identifier>

Parameters:

  • args (Array) —

    job arguments

Returns:

  • (String) —

    redis key



82
83
84
 
# File 'lib/resque/plugins/lock_timeout.rb', line 82

def redis_lock_key(*args)
  ['lock', name, identifier(*args)].compact.join(':')
end

#redis_loner_lock_key(*args) ⇒ String

Builds lock key used by @loner option. Passed job arguments.

The default looks like this: loner:lock:<class name>:<identifier>

Parameters:

  • args (Array) —

    job arguments

Returns:

  • (String) —

    redis key



92
93
94
 
# File 'lib/resque/plugins/lock_timeout.rb', line 92

def redis_loner_lock_key(*args)
  ['loner', redis_lock_key(*args)].compact.join(':')
end

#refresh_lock!(*args) ⇒ Object

Refresh the lock.

Parameters:

  • args (Array) —

    job arguments



267
268
269
270
271
 
# File 'lib/resque/plugins/lock_timeout.rb', line 267

def refresh_lock!(*args)
  now = Time.now.to_i
  lock_until = now + lock_timeout(*args)
  lock_redis.set(redis_lock_key(*args), lock_until)
end

#release_lock!(*args) ⇒ Object

Release the lock.

Parameters:

  • args (Array) —

    job arguments



253
254
255
 
# File 'lib/resque/plugins/lock_timeout.rb', line 253

def release_lock!(*args)
  lock_redis.del(redis_lock_key(*args))
end

#release_loner_lock!(*args) ⇒ Object

Release the enqueue lock for loner jobs

Parameters:

  • args (Array) —

    job arguments



260
261
262
 
# File 'lib/resque/plugins/lock_timeout.rb', line 260

def release_loner_lock!(*args)
  lock_redis.del(redis_loner_lock_key(*args))
end