Merge pull request #4 from miyagawa/master

Support unlocking deadlocks with setnx/getset

lib/resque-lonely_job.rb

@@ -6,7 +6,7 @@ module LonelyJob
       LOCK_TIMEOUT = 60 * 60 * 24 * 5 # 5 days
 
       def lock_timeout
-        Time.now.utc + LOCK_TIMEOUT + 1
+        Time.now.to_i + LOCK_TIMEOUT + 1
       end
 
       # Overwrite this method to uniquely identify which mutex should be used
@@ -16,7 +16,15 @@ def redis_key(*args)
       end
 
       def can_lock_queue?(*args)
-        Resque.redis.setnx(redis_key(*args), lock_timeout)
+        now = Time.now.to_i
+        key = redis_key(*args)
+        timeout = lock_timeout
+
+        # Per http://redis.io/commands/setnx
+        return true  if Resque.redis.setnx(key, timeout)
+        return false if Resque.redis.get(key).to_i > now
+        return true  if Resque.redis.getset(key, timeout).to_i <= now
+        return false
       end
 
       def unlock_queue(*args)

resque-lonely_job.gemspec

@@ -14,11 +14,12 @@ Gem::Specification.new do |gem|
   gem.require_paths = ["lib"]
   gem.version       = Resque::Plugins::LonelyJob::VERSION
 
-  gem.add_dependency 'resque', '~> 1.20.0'
+  gem.add_dependency 'resque', '>= 1.20.0'
   gem.add_development_dependency 'mock_redis', '~> 0.4.1'
   gem.add_development_dependency 'rake'
   gem.add_development_dependency 'rspec'
   gem.add_development_dependency 'debugger'
+  gem.add_development_dependency 'timecop'
 
   gem.description   = <<desc
 Ensures that for a given queue, only one worker is working on a job at any given time.

spec/lib/lonely_job_spec.rb

@@ -32,6 +32,37 @@ def self.perform(account_id, *args); end
       SerialJob.can_lock_queue?(:serial_work).should be_true
       SerialJob.can_lock_queue?(:serial_work).should be_false
     end
+
+    it 'cannot lock a queue with active lock' do
+      SerialJob.can_lock_queue?(:serial_work).should be_true
+      Timecop.travel(Date.today + 1) do
+        SerialJob.can_lock_queue?(:serial_work).should be_false
+      end
+    end
+
+    it 'can relock a queue with expired lock' do
+      SerialJob.can_lock_queue?(:serial_work).should be_true
+
+      Timecop.travel(Date.today + 10) do
+        SerialJob.can_lock_queue?(:serial_work).should be_true
+      end
+    end
+
+    it 'solves race condition with getset' do
+      SerialJob.can_lock_queue?(:serial_work).should be_true
+
+      Timecop.travel(Date.today + 10) do
+        threads = (1..10).to_a.map {
+          Thread.new {
+            Thread.current[:locked] = SerialJob.can_lock_queue?(:serial_work)
+          }
+        }
+
+        # Only one worker should acquire lock
+        locks = threads.map {|t| t.join; t[:locked] }
+        locks.count(true).should == 1
+      end
+    end
   end
 
   describe ".perform" do

spec/spec_helper.rb

@@ -5,6 +5,7 @@
 require 'mock_redis'
 require 'resque'
 require 'resque-lonely_job'
+require 'timecop'
 
 RSpec.configure do |config|
   config.before(:suite) do