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
- NextTaskSQL =
'(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]+)/
- @@max_attempts =
25
- @@max_run_time =
4.hours
- @@worker_name =
Every worker has a unique name which by default is the pid of the process. There are some advantages to overriding this with something which survives worker retarts: Workers can safely resume working on tasks which are locked by themselves. The worker will assume that it crashed before.
nil
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.
-
.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).
-
.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.
- .worker_name ⇒ Object
- .worker_name=(val) ⇒ Object
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).
- #worker_name ⇒ Object
- #worker_name=(val) ⇒ Object
Class Method Details
.clear_locks! ⇒ Object
When a worker is exiting, make sure we don’t have any locked jobs.
56 57 58 |
# File 'lib/delayed/job.rb', line 56 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
127 128 129 130 131 132 133 134 135 136 137 138 |
# File 'lib/delayed/job.rb', line 127 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 |
.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).
141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 |
# File 'lib/delayed/job.rb', line 141 def self.find_available(limit = 5, max_run_time = max_run_time) time_now = db_time_now sql = NextTaskSQL.dup conditions = [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) ActiveRecord::Base.silence do find(:all, :conditions => conditions, :order => NextTaskOrder, :limit => limit) 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
168 169 170 171 172 173 174 175 176 177 178 |
# File 'lib/delayed/job.rb', line 168 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.
215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 |
# File 'lib/delayed/job.rb', line 215 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 |
.worker_name ⇒ Object
29 30 31 32 |
# File 'lib/delayed/job.rb', line 29 def self.worker_name return @@worker_name unless @@worker_name.nil? "host:#{Socket.gethostname} pid:#{Process.pid}" rescue "pid:#{Process.pid}" end |
.worker_name=(val) ⇒ Object
34 35 36 |
# File 'lib/delayed/job.rb', line 34 def self.worker_name=(val) @@worker_name = val end |
Instance Method Details
#failed? ⇒ Boolean Also known as: failed
60 61 62 |
# File 'lib/delayed/job.rb', line 60 def failed? failed_at end |
#invoke_job ⇒ Object
Moved into its own method so that new_relic can trace it.
234 235 236 |
# File 'lib/delayed/job.rb', line 234 def invoke_job payload_object.perform 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.
182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 |
# File 'lib/delayed/job.rb', line 182 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 < ?) and (run_at <= ?)", id, (now - max_run_time.to_i), now]) 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
208 209 210 211 |
# File 'lib/delayed/job.rb', line 208 def log_exception(error) logger.error "* [JOB] #{name} failed with #{error.class.name}: #{error.} - #{attempts} failed attempts" logger.error(error) end |
#name ⇒ Object
69 70 71 72 73 74 75 76 77 78 |
# File 'lib/delayed/job.rb', line 69 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
65 66 67 |
# File 'lib/delayed/job.rb', line 65 def payload_object @payload_object ||= deserialize(self['handler']) end |
#payload_object=(object) ⇒ Object
80 81 82 |
# File 'lib/delayed/job.rb', line 80 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.
86 87 88 89 90 91 92 93 94 95 96 97 98 99 |
# File 'lib/delayed/job.rb', line 86 def reschedule(, backtrace = [], time = nil) self.last_error = + "\n" + backtrace.join("\n") if (self.attempts += 1) < max_attempts time ||= Job.db_time_now + (attempts ** 4) + 5 self.run_at = time self.unlock save! else logger.info "* [JOB] PERMANENTLY removing #{self.name} because of #{attempts} consecutive failures." destroy_failed_jobs ? destroy : update_attribute(:failed_at, Delayed::Job.db_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.
103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 |
# File 'lib/delayed/job.rb', line 103 def run_with_lock(max_run_time, worker_name) logger.info "* [JOB] acquiring 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 acquire exclusive lock for #{name}" return nil # no work done end begin runtime = Benchmark.realtime do Timeout.timeout(max_run_time.to_i) { invoke_job } destroy 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)
202 203 204 205 |
# File 'lib/delayed/job.rb', line 202 def unlock self.locked_at = nil self.locked_by = nil end |
#worker_name ⇒ Object
38 39 40 |
# File 'lib/delayed/job.rb', line 38 def worker_name self.class.worker_name end |
#worker_name=(val) ⇒ Object
42 43 44 |
# File 'lib/delayed/job.rb', line 42 def worker_name=(val) @@worker_name = val end |