0
@@ -4,142 +4,192 @@ module Delayed
0
class Job < ActiveRecord::Base
0
- ParseObjectFromYaml = /\!ruby\/\w+\:([^\s]+)/
0
set_table_name :delayed_jobs
0
- attr_accessor :logger, :jobs
0
- attr_accessor :runs, :success, :failure
0
+ cattr_accessor :worker_name
0
+ self.worker_name = "pid:#{Process.pid}"
0
- def initialize(jobs, logger = nil)
0
- self.runs = self.success = self.failure = 0
0
- ActiveRecord::Base.cache do
0
- ActiveRecord::Base.transaction do
0
- time = Benchmark.measure do
0
- ActiveRecord::Base.uncached { job.destroy }
0
- logger.debug "Executed job in #{time.real}"
0
- rescue DeserializationError, StandardError, RuntimeError => e
0
- logger.error "Job #{job.id}: #{e.class} #{e.message}"
0
- logger.error e.backtrace.join("\n")
0
- ActiveRecord::Base.uncached { job.reshedule e.message }
0
- def self.enqueue(object, priority = 0)
0
- raise ArgumentError, 'Cannot enqueue items which do not respond to perform' unless object.respond_to?(:perform)
0
- Job.create(:handler => object, :priority => priority)
0
- self['handler'] = object.to_yaml
0
+ NextTaskSQL = '`run_at` <= ? AND (`locked_until` IS NULL OR `locked_until` < ?) OR (`locked_by`=?)'
0
+ NextTaskOrder = 'priority DESC, run_at ASC'
0
+ ParseObjectFromYaml = /\!ruby\/\w+\:([^\s]+)/
0
+ class LockError < StandardError
0
+ connection.execute "UPDATE #{table_name} SET `locked_by`=NULL, `locked_until`=NULL WHERE `locked_by`=#{quote_value(worker_name)}"
0
- @handler ||= deserialize(self['handler'])
0
+ @payload_object ||= deserialize(self['handler'])
0
+ def payload_object=(object)
0
+ self['handler'] = object.to_yaml
0
- def reshedule(message)
0
- self.run_at = self.class.time_now + (attempts ** 4).seconds
0
- self.last_error = message
0
+ def reshedule(message, time = nil)
0
+ time ||= Job.db_time_now + (attempts ** 4).seconds + 1
0
+ self.last_error = message
0
- def self.peek(limit = 1)
0
- find(:first, :order => "priority DESC, run_at ASC", :conditions => ['run_at <= ?', time_now])
0
- find(:all, :order => "priority DESC, run_at ASC", :limit => limit, :conditions => ['run_at <= ?', time_now])
0
+ def self.enqueue(object, priority = 0)
0
+ unless object.respond_to?(:perform)
0
+ raise ArgumentError, 'Cannot enqueue items which do not respond to perform'
0
- def self.work_off(limit = 100)
0
- jobs = Job.find(:all, :conditions => ['run_at <= ?', time_now], :order => "priority DESC, run_at ASC", :limit => limit)
0
+ Job.create(:payload_object => object, :priority => priority)
0
- Job::Runner.new(jobs, logger).run
0
+ def self.find_available(limit = 5)
0
+ time_now = db_time_now
0
+ find(:all, :conditions => [NextTaskSQL, time_now, time_now, worker_name], :order => NextTaskOrder, :limit => 5)
0
+ # Get the payload of the next job we can get an exclusive lock on.
0
+ # If no jobs are left we return nil
0
+ def self.reserve(timeout = 5 * 60)
0
+ # We get up to 5 jobs from the db. In face we cannot get exclusive access to a job we try the next.
0
+ # this leads to a more even distribution of jobs across the worker processes
0
+ find_available(5).each do |job|
0
+ job.lock_exclusively!(self.db_time_now + timeout, worker_name)
0
+ yield job.payload_object
0
+ # We did not get the lock, some other worker process must have
0
+ puts "failed to aquire exclusive lock for #{job.id}"
0
+ rescue StandardError => e
0
+ job.reshedule e.message
0
+ # This method is used internally by reserve method to ensure exclusive access
0
+ # to the given job. It will rise a LockError if it cannot get this lock.
0
+ def lock_exclusively!(lock_until, worker = worker_name)
0
+ affected_rows = if locked_by != worker
0
+ # We don't own this job so we will update the locked_by name and the locked_until
0
+ connection.update(<<-end_sql, "#{self.class.name} Update to aquire exclusive lock")
0
+ UPDATE #{self.class.table_name}
0
+ SET `locked_until`=#{quote_value(lock_until)}, `locked_by`=#{quote_value(worker)}
0
+ WHERE #{self.class.primary_key} = #{quote_value(id)} AND (`locked_until`<#{quote_value(self.class.db_time_now)} OR `locked_until` IS NULL)
0
+ # We alrady own this job, this may happen if the job queue crashes.
0
+ # Simply update the lock timeout
0
+ connection.update(<<-end_sql, "#{self.class.name} Update exclusive lock")
0
+ UPDATE #{self.class.table_name}
0
+ SET `locked_until`=#{quote_value(lock_until)}
0
+ WHERE #{self.class.primary_key} = #{quote_value(id)} AND (`locked_by`=#{quote_value(worker)})
0
+ unless affected_rows == 1
0
+ raise LockError, "Attempted to aquire exclusive lock failed"
0
+ self.locked_until = lock_until
0
+ self.locked_by = worker
0
- (ActiveRecord::Base.default_timezone == :utc) ? Time.now.utc : Time.now
0
+ self.locked_until = nil
0
- self.run_at ||= self.class.time_now
0
+ def self.work_off(num = 100)
0
+ success, failure = 0, 0
0
+ job = self.reserve do |j|
0
+ return [success, failure]
0
def deserialize(source)
0
attempt_to_load_file = true
0
handler = YAML.load(source) rescue nil
0
return handler if handler.respond_to?(:perform)
0
if source =~ ParseObjectFromYaml
0
# Constantize the object so that ActiveSupport can attempt
0
# its auto loading magic. Will raise LoadError if not successful.
0
# If successful, retry the yaml.load
0
handler = YAML.load(source)
0
return handler if handler.respond_to?(:perform)
0
if handler.is_a?(YAML::Object)
0
# Constantize the object so that ActiveSupport can attempt
0
# its auto loading magic. Will raise LoadError if not successful.
0
attempt_to_load(handler.class)
0
# If successful, retry the yaml.load
0
handler = YAML.load(source)
0
return handler if handler.respond_to?(:perform)
0
raise DeserializationError, 'Job failed to load: Unknown handler. Try to manually require the appropiate file.'
0
rescue TypeError, LoadError, NameError => e
0
raise DeserializationError, "Job failed to load: #{e.message}. Try to manually require the required file."
0
def attempt_to_load(klass)
0
+ (ActiveRecord::Base.default_timezone == :utc) ? Time.now.utc : Time.now
0
+ self.run_at ||= self.class.db_time_now
0
\ No newline at end of file
this repo is viewable by everyone
Comments
No one has commented yet.