Skip to content
Browse files

Renamed locked_until to locked_at. We now store when we start a given…

… task instead of how long it will be locked by the worker. This allows us to get a reading on how long a task took to execute.
  • Loading branch information...
1 parent bf90faf commit 59989313afa5f0b431263e37a994d895a3cc669d Tobias Lütke committed
Showing with 36 additions and 25 deletions.
  1. +2 −1 README.textile
  2. +15 −13 lib/delayed/job.rb
  3. +1 −1 spec/database.rb
  4. +18 −10 spec/job_spec.rb
View
3 README.textile
@@ -13,7 +13,8 @@ It is a direct extraction from Shopify where the job table is responsible for a
* spam checks
h2. Changes
-
+
+* 1.6 Renamed locked_until to locked_at. We now store when we start a given task instead of how long it will be locked by the worker. This allows us to get a reading on how long a task took to execute.
* 1.5 Job runners can now be run in parallel. Two new database columns are needed: locked_until and locked_by. This allows us to use pessimistic locking, which enables us to run as many worker processes as we need to speed up queue processing.
* 1.0 Initial release
View
28 lib/delayed/job.rb
@@ -10,7 +10,7 @@ class Job < ActiveRecord::Base
self.worker_name = "pid:#{Process.pid}"
- NextTaskSQL = '`run_at` <= ? AND (`locked_until` IS NULL OR `locked_until` < ?) OR (`locked_by`=?)'
+ NextTaskSQL = '`run_at` <= ? AND (`locked_at` IS NULL OR `locked_at` < ?) OR (`locked_by` = ?)'
NextTaskOrder = 'priority DESC, run_at ASC'
ParseObjectFromYaml = /\!ruby\/\w+\:([^\s]+)/
@@ -18,7 +18,7 @@ class LockError < StandardError
end
def self.clear_locks!
- connection.execute "UPDATE #{table_name} SET `locked_by`=NULL, `locked_until`=NULL WHERE `locked_by`=#{quote_value(worker_name)}"
+ connection.execute "UPDATE #{table_name} SET `locked_by`=NULL, `locked_at`=NULL WHERE `locked_by`=#{quote_value(worker_name)}"
end
def payload_object
@@ -57,13 +57,13 @@ def self.find_available(limit = 5)
# Get the payload of the next job we can get an exclusive lock on.
# If no jobs are left we return nil
- def self.reserve(timeout = 4 * 60 * 60)
+ def self.reserve(max_run_time = 4.hours)
# We get up to 5 jobs from the db. In face we cannot get exclusive access to a job we try the next.
# this leads to a more even distribution of jobs across the worker processes
find_available(5).each do |job|
begin
- job.lock_exclusively!(self.db_time_now + timeout, worker_name)
+ job.lock_exclusively!(max_run_time, worker_name)
yield job.payload_object
job.destroy
return job
@@ -81,24 +81,26 @@ def self.reserve(timeout = 4 * 60 * 60)
# This method is used internally by reserve method to ensure exclusive access
# to the given job. It will rise a LockError if it cannot get this lock.
- def lock_exclusively!(lock_until, worker = worker_name)
+ def lock_exclusively!(max_run_time, worker = worker_name)
+ now = self.class.db_time_now
- affected_rows = if locked_by != worker
+ affected_rows = if locked_by != worker
- # We don't own this job so we will update the locked_by name and the locked_until
+
+ # We don't own this job so we will update the locked_by name and the locked_at
connection.update(<<-end_sql, "#{self.class.name} Update to aquire exclusive lock")
UPDATE #{self.class.table_name}
- SET `locked_until`=#{quote_value(lock_until)}, `locked_by`=#{quote_value(worker)}
- WHERE #{self.class.primary_key} = #{quote_value(id)} AND (`locked_until`<#{quote_value(self.class.db_time_now)} OR `locked_until` IS NULL)
+ SET `locked_at`=#{quote_value(now)}, `locked_by`=#{quote_value(worker)}
+ WHERE #{self.class.primary_key} = #{quote_value(id)} AND (`locked_at` IS NULL OR `locked_at` < #{quote_value(now + max_run_time)})
end_sql
else
# We alrady own this job, this may happen if the job queue crashes.
- # Simply update the lock timeout
+ # Simply resume and update the locked_at
connection.update(<<-end_sql, "#{self.class.name} Update exclusive lock")
UPDATE #{self.class.table_name}
- SET `locked_until`=#{quote_value(lock_until)}
+ SET `locked_at`=#{quote_value(now)}
WHERE #{self.class.primary_key} = #{quote_value(id)} AND (`locked_by`=#{quote_value(worker)})
end_sql
@@ -108,12 +110,12 @@ def lock_exclusively!(lock_until, worker = worker_name)
raise LockError, "Attempted to aquire exclusive lock failed"
end
- self.locked_until = lock_until
+ self.locked_at = now
self.locked_by = worker
end
def unlock
- self.locked_until = nil
+ self.locked_at = nil
self.locked_by = nil
end
View
2 spec/database.rb
@@ -17,7 +17,7 @@ def reset_db
table.text :handler
table.string :last_error
table.datetime :run_at
- table.datetime :locked_until
+ table.datetime :locked_at
table.string :locked_by
table.timestamps
end
View
28 spec/job_spec.rb
@@ -91,23 +91,31 @@ def perform; raise 'did not work'; end
end
- describe "when two workers are running" do
+ describe "when another worker is already performing an task, it" do
before :each do
Delayed::Job.worker_name = 'worker1'
- Delayed::Job.create :payload_object => SimpleJob.new, :locked_by => 'worker1', :locked_until => Time.now + 360
+ @job = Delayed::Job.create :payload_object => SimpleJob.new, :locked_by => 'worker1', :locked_at => Time.now.utc
end
- it "should give exclusive access only to a single worker" do
- job = Delayed::Job.find_available.first
- lambda { job.lock_exclusively! Time.now + 20, 'worker2' }.should raise_error(Delayed::Job::LockError)
- end
+ it "should not allow a second worker to get exclusive access" do
+ lambda { @job.lock_exclusively! 4.hours, 'worker2' }.should raise_error(Delayed::Job::LockError)
+ end
+
+ it "should be able to get access to the task if it was started more then max_age ago" do
+ @job.locked_at = 5.hours.ago
+ @job.save
+
+ @job.lock_exclusively! 4.hours, 'worker2'
+ @job.reload
+ @job.locked_by.should == 'worker2'
+ @job.locked_at.should > 1.minute.ago
+ end
it "should be able to get exclusive access again when the worker name is the same" do
- job = Delayed::Job.find_available.first
- job.lock_exclusively! Time.now + 20, 'worker1'
- job.lock_exclusively! Time.now + 21, 'worker1'
- job.lock_exclusively! Time.now + 22, 'worker1'
+ @job.lock_exclusively! Time.now + 20, 'worker1'
+ @job.lock_exclusively! Time.now + 21, 'worker1'
+ @job.lock_exclusively! Time.now + 22, 'worker1'
end
end

0 comments on commit 5998931

Please sign in to comment.
Something went wrong with that request. Please try again.