Class: Delayed::Job
- Inherits:
-
ActiveRecord::Base
- Object
- ActiveRecord::Base
- Delayed::Job
- Defined in:
- lib/delayed/job.rb
Overview
A job object that is persisted to the database. Contains the work object as a YAML field.
Constant Summary collapse
- MAX_ATTEMPTS =
25
- MAX_RUN_TIME =
4.hours
- NextTaskSQL =
'completed = ? AND (run_at <= ? AND (locked_at IS NULL OR locked_at < ?) OR (locked_by = ?)) AND failed_at IS NULL'
- NextTaskOrder =
'priority DESC, run_at ASC'
- ParseObjectFromYaml =
/\!ruby\/\w+\:([^\s]+)/
Class Method Summary collapse
-
.clear_locks! ⇒ Object
When a worker is exiting, make sure we don’t have any locked jobs.
-
.enqueue(*args, &block) ⇒ Object
Add a job to the queue.
- .enqueue_for(owned_by, *args, &block) ⇒ Object
-
.find_available(limit = 5, max_run_time = MAX_RUN_TIME) ⇒ Object
Find a few candidate jobs to run (in case some immediately get locked by others).
- .owner_key_for(owned_by) ⇒ Object
-
.reserve_and_run_one_job(max_run_time = MAX_RUN_TIME) ⇒ Object
Run the next job we can get an exclusive lock on.
-
.work_off(num = 100) ⇒ Object
Do num jobs and return stats on success/failure.
Instance Method Summary collapse
- #failed? ⇒ Boolean (also: #failed)
-
#invoke_job ⇒ Object
Moved into its own method so that new_relic can trace it.
-
#lock_exclusively!(max_run_time, worker = worker_name) ⇒ Object
Lock this job for this worker.
-
#log_exception(error) ⇒ Object
This is a good hook if you need to report job processing errors in additional or different ways.
- #name ⇒ Object
- #payload_object ⇒ Object
- #payload_object=(object) ⇒ Object
-
#reschedule(message, backtrace = [], time = nil) ⇒ Object
Reschedule the job in the future (when a job fails).
-
#run_with_lock(max_run_time, worker_name) ⇒ Object
Try to run one job.
-
#unlock ⇒ Object
Unlock this job (note: not saved to DB).
Class Method Details
.clear_locks! ⇒ Object
When a worker is exiting, make sure we don’t have any locked jobs.
38 39 40 |
# File 'lib/delayed/job.rb', line 38 def self.clear_locks! update_all("locked_by = null, locked_at = null", ["locked_by = ?", worker_name]) end |
.enqueue(*args, &block) ⇒ Object
Add a job to the queue
114 115 116 117 118 119 120 121 122 123 124 125 |
# File 'lib/delayed/job.rb', line 114 def self.enqueue(*args, &block) object = block_given? ? EvaledJob.new(&block) : args.shift unless object.respond_to?(:perform) || block_given? raise ArgumentError, 'Cannot enqueue items which do not respond to perform' end priority = args.first || 0 run_at = args[1] Job.create!(:payload_object => object, :priority => priority.to_i, :run_at => run_at) end |
.enqueue_for(owned_by, *args, &block) ⇒ Object
127 128 129 130 131 132 133 134 135 136 137 |
# File 'lib/delayed/job.rb', line 127 def self.enqueue_for(owned_by, *args, &block) object = block_given? ? EvaledJob.new(&block) : args.shift owner_key = self.owner_key_for(owned_by) priority = args.first || 0 run_at = args[1] Job.create!(:payload_object => object, :priority => priority.to_i, :run_at => run_at, :owner => owner_key) end |
.find_available(limit = 5, max_run_time = MAX_RUN_TIME) ⇒ Object
Find a few candidate jobs to run (in case some immediately get locked by others). Return in random order prevent everyone trying to do same head job at once.
153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 |
# File 'lib/delayed/job.rb', line 153 def self.find_available(limit = 5, max_run_time = MAX_RUN_TIME) time_now = db_time_now sql = NextTaskSQL.dup conditions = [false, time_now, time_now - max_run_time, worker_name] if self.min_priority sql << ' AND (priority >= ?)' conditions << min_priority end if self.max_priority sql << ' AND (priority <= ?)' conditions << max_priority end conditions.unshift(sql) records = ActiveRecord::Base.silence do find(:all, :conditions => conditions, :order => NextTaskOrder, :limit => limit) end records.sort_by { rand() } end |
.owner_key_for(owned_by) ⇒ Object
139 140 141 142 143 144 145 146 147 148 149 |
# File 'lib/delayed/job.rb', line 139 def self.owner_key_for(owned_by) case owned_by when String owned_by when ActiveRecord::Base raise(ArgumentError, 'Owner cannot be a new_record') if owned_by.new_record? "#{owned_by.class}_#{owned_by.id}" else raise ArgumentError, 'Only strings and descendants of ActiveRecord::Base are allowed' end end |
.reserve_and_run_one_job(max_run_time = MAX_RUN_TIME) ⇒ Object
Run the next job we can get an exclusive lock on. If no jobs are left we return nil
182 183 184 185 186 187 188 189 190 191 192 |
# File 'lib/delayed/job.rb', line 182 def self.reserve_and_run_one_job(max_run_time = MAX_RUN_TIME) # We get up to 5 jobs from the db. In case 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, max_run_time).each do |job| t = job.run_with_lock(max_run_time, worker_name) return t unless t == nil # return if we did work (good or bad) end nil # we didn't do any work, all 5 were not lockable end |
.work_off(num = 100) ⇒ Object
Do num jobs and return stats on success/failure. Exit early if interrupted.
229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 |
# File 'lib/delayed/job.rb', line 229 def self.work_off(num = 100) success, failure = 0, 0 num.times do case self.reserve_and_run_one_job when true success += 1 when false failure += 1 else break # leave if no work could be done end break if $exit # leave if we're exiting end return [success, failure] end |
Instance Method Details
#failed? ⇒ Boolean Also known as: failed
42 43 44 |
# File 'lib/delayed/job.rb', line 42 def failed? failed_at end |
#invoke_job ⇒ Object
Moved into its own method so that new_relic can trace it.
248 249 250 251 252 253 254 255 256 |
# File 'lib/delayed/job.rb', line 248 def invoke_job res = payload_object.perform if res.respond_to? :to_yaml self.result = res.to_yaml else logger.info "* [JOB] res does not respond to #to_yaml" end end |
#lock_exclusively!(max_run_time, worker = worker_name) ⇒ Object
Lock this job for this worker. Returns true if we have the lock, false otherwise.
196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 |
# File 'lib/delayed/job.rb', line 196 def lock_exclusively!(max_run_time, worker = worker_name) now = self.class.db_time_now affected_rows = if locked_by != worker # We don't own this job so we will update the locked_by name and the locked_at self.class.update_all(["locked_at = ?, locked_by = ?", now, worker], ["id = ? and (locked_at is null or locked_at < ?)", id, (now - max_run_time.to_i)]) else # We already own this job, this may happen if the job queue crashes. # Simply resume and update the locked_at self.class.update_all(["locked_at = ?", now], ["id = ? and locked_by = ?", id, worker]) end if affected_rows == 1 self.locked_at = now self.locked_by = worker return true else return false end end |
#log_exception(error) ⇒ Object
This is a good hook if you need to report job processing errors in additional or different ways
222 223 224 225 |
# File 'lib/delayed/job.rb', line 222 def log_exception(error) logger.error "* [JOB] #{name} failed with #{error.class.name}: #{error.} - #{attempts} failed attempts" logger.error(error) end |
#name ⇒ Object
51 52 53 54 55 56 57 58 59 60 |
# File 'lib/delayed/job.rb', line 51 def name @name ||= begin payload = payload_object if payload.respond_to?(:display_name) payload.display_name else payload.class.name end end end |
#payload_object ⇒ Object
47 48 49 |
# File 'lib/delayed/job.rb', line 47 def payload_object @payload_object ||= deserialize(self['handler']) end |
#payload_object=(object) ⇒ Object
62 63 64 |
# File 'lib/delayed/job.rb', line 62 def payload_object=(object) self['handler'] = object.to_yaml end |
#reschedule(message, backtrace = [], time = nil) ⇒ Object
Reschedule the job in the future (when a job fails). Uses an exponential scale depending on the number of failed attempts.
68 69 70 71 72 73 74 75 76 77 78 79 80 81 |
# File 'lib/delayed/job.rb', line 68 def reschedule(, backtrace = [], time = nil) if self.attempts < MAX_ATTEMPTS time ||= Job.db_time_now + (attempts ** 4) + 5 self.attempts += 1 self.run_at = time self.last_error = + "\n" + backtrace.join("\n") self.unlock save! else logger.info "* [JOB] PERMANENTLY removing #{self.name} because of #{attempts} consequetive failures." destroy_failed_jobs ? destroy : update_attribute(:failed_at, Time.now) end end |
#run_with_lock(max_run_time, worker_name) ⇒ Object
Try to run one job. Returns true/false (work done/work failed) or nil if job can’t be locked.
85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 |
# File 'lib/delayed/job.rb', line 85 def run_with_lock(max_run_time, worker_name) logger.info "* [JOB] aquiring lock on #{name}" unless lock_exclusively!(max_run_time, worker_name) # We did not get the lock, some other worker process must have logger.warn "* [JOB] failed to aquire exclusive lock for #{name}" return nil # no work done end begin runtime = Benchmark.realtime do invoke_job # TODO: raise error if takes longer than max_run_time if destroy_completed_jobs destroy else self.completed = true save end end # TODO: warn if runtime > max_run_time ? logger.info "* [JOB] #{name} completed after %.4f" % runtime return true # did work rescue Exception => e reschedule e., e.backtrace log_exception(e) return false # work failed end end |
#unlock ⇒ Object
Unlock this job (note: not saved to DB)
216 217 218 219 |
# File 'lib/delayed/job.rb', line 216 def unlock self.locked_at = nil self.locked_by = nil end |