Class: Delayed::Job
- Inherits:
-
ActiveRecord::Base
- Object
- ActiveRecord::Base
- Delayed::Job
- Extended by:
- JobDeprecations
- Defined in:
- lib/delayed_on_steroids/job.rb
Overview
A job object that is persisted to the database. Contains the work object as a YAML field handler
.
Constant Summary collapse
- MAX_ATTEMPTS =
25
- MAX_RUN_TIME =
4.hours
- ParseObjectFromYaml =
/\!ruby\/\w+\:([^\s]+)/
Class Method Summary collapse
-
.db_time_now ⇒ Object
Get the current time (GMT or local depending on DB) Note: This does not ping the DB to get the time, so all your clients must have syncronized clocks.
-
.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.
Instance Method Summary collapse
-
#failed? ⇒ Boolean
(also: #failed)
Returns
true
if current job failed. -
#invoke_job ⇒ Object
Moved into its own method so that new_relic can trace it.
-
#lock_exclusively!(max_run_time = MAX_RUN_TIME, worker_name = Worker.name) ⇒ Object
Lock this job for this worker.
-
#locked? ⇒ Boolean
(also: #locked)
Returns
true
if current job locked. -
#log_exception(e) ⇒ Object
This is a good hook if you need to report job processing errors in additional or different ways.
-
#name ⇒ Object
Returns job name.
- #payload_object ⇒ Object
- #payload_object=(object) ⇒ Object
-
#reschedule(message, backtrace = [], time = nil) ⇒ Object
Reschedule the job to run at
time
(when a job fails). -
#run_with_lock(max_run_time = MAX_RUN_TIME, worker_name = Worker.name) ⇒ Object
Try to run one job.
Methods included from JobDeprecations
Class Method Details
.db_time_now ⇒ Object
Get the current time (GMT or local depending on DB) Note: This does not ping the DB to get the time, so all your clients must have syncronized clocks.
222 223 224 225 226 227 228 229 230 |
# File 'lib/delayed_on_steroids/job.rb', line 222 def self.db_time_now if Time.zone Time.zone.now elsif ActiveRecord::Base.default_timezone == :utc Time.now.utc else Time.now end end |
.enqueue(*args, &block) ⇒ Object
Add a job to the queue. Arguments: priority, run_at.
95 96 97 98 99 100 101 102 103 104 105 106 |
# File 'lib/delayed_on_steroids/job.rb', line 95 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[0] || 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).
109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 |
# File 'lib/delayed_on_steroids/job.rb', line 109 def self.find_available(limit = 5, max_run_time = MAX_RUN_TIME) time_now = db_time_now sql = '' conditions = [] # 1) not scheduled in the future sql << '(run_at <= ?)' conditions << time_now # 2) and job is not failed yet sql << ' AND (failed_at IS NULL)' # 3a) and already locked by same worker sql << ' AND (' sql << '(locked_by = ?)' conditions << Worker.name # 3b) or not locked yet sql << ' OR (locked_at IS NULL)' # 3c) or lock expired sql << ' OR (locked_at < ?)' sql << ')' conditions << time_now - max_run_time if Worker.min_priority sql << ' AND (priority >= ?)' conditions << Worker.min_priority end if Worker.max_priority sql << ' AND (priority <= ?)' conditions << Worker.max_priority end if Worker.job_types sql << ' AND (job_type IN (?))' conditions << Worker.job_types end conditions.unshift(sql) find(:all, :conditions => conditions, :order => 'priority ASC, run_at ASC', :limit => limit) 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
155 156 157 158 159 160 161 162 163 164 165 |
# File 'lib/delayed_on_steroids/job.rb', line 155 def self.reserve_and_run_one_job(max_run_time = MAX_RUN_TIME) # We get up to 20 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(20, 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 20 were not lockable end |
.work_off(num = 100) ⇒ Object
Do num jobs and return stats on success/failure. Exit early if interrupted.
196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 |
# File 'lib/delayed_on_steroids/job.rb', line 196 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
Returns true
if current job failed.
22 23 24 |
# File 'lib/delayed_on_steroids/job.rb', line 22 def failed? not failed_at.nil? end |
#invoke_job ⇒ Object
Moved into its own method so that new_relic can trace it.
215 216 217 |
# File 'lib/delayed_on_steroids/job.rb', line 215 def invoke_job payload_object.perform end |
#lock_exclusively!(max_run_time = MAX_RUN_TIME, worker_name = Worker.name) ⇒ Object
Lock this job for this worker. Returns true if we have the lock, false otherwise.
169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 |
# File 'lib/delayed_on_steroids/job.rb', line 169 def lock_exclusively!(max_run_time = MAX_RUN_TIME, worker_name = Worker.name) now = self.class.db_time_now affected_rows = if locked_by != worker_name # 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_name], ["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_name]) end if affected_rows == 1 self.locked_at = now self.locked_by = worker_name return true else return false end end |
#locked? ⇒ Boolean Also known as: locked
Returns true
if current job locked.
28 29 30 |
# File 'lib/delayed_on_steroids/job.rb', line 28 def locked? not locked_at.nil? end |
#log_exception(e) ⇒ Object
This is a good hook if you need to report job processing errors in additional or different ways
189 190 191 192 |
# File 'lib/delayed_on_steroids/job.rb', line 189 def log_exception(e) Worker.logger.error("! [#{Worker.name}] #{name} failed with #{e.class.name}: #{e.} - #{attempts} failed attempts") Worker.logger.error(e) end |
#name ⇒ Object
Returns job name.
43 44 45 46 47 48 49 50 51 52 |
# File 'lib/delayed_on_steroids/job.rb', line 43 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
33 34 35 |
# File 'lib/delayed_on_steroids/job.rb', line 33 def payload_object @payload_object ||= deserialize(self['handler']) end |
#payload_object=(object) ⇒ Object
37 38 39 40 |
# File 'lib/delayed_on_steroids/job.rb', line 37 def payload_object=(object) self['job_type'] = object.class.to_s self['handler'] = object.to_yaml end |
#reschedule(message, backtrace = [], time = nil) ⇒ Object
Reschedule the job to run at time
(when a job fails). If time
is nil it uses an exponential scale depending on the number of failed attempts.
56 57 58 59 60 61 62 63 64 65 66 67 68 69 |
# File 'lib/delayed_on_steroids/job.rb', line 56 def reschedule(, backtrace = [], time = nil) if (self.attempts += 1) < MAX_ATTEMPTS time ||= Job.db_time_now + (attempts ** 4) + 5 self.run_at = time self.last_error = + "\n" + backtrace.join("\n") self.locked_at = nil self.locked_by = nil save! else Worker.logger.warn("* [#{Worker.name}] PERMANENTLY removing #{self.name} because of #{attempts} consequetive failures.") Worker.destroy_failed_jobs ? destroy : update_attribute(:failed_at, self.class.db_time_now) end end |
#run_with_lock(max_run_time = MAX_RUN_TIME, worker_name = Worker.name) ⇒ Object
Try to run one job. Returns true/false (work done/work failed) or nil if job can’t be locked.
72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
# File 'lib/delayed_on_steroids/job.rb', line 72 def run_with_lock(max_run_time = MAX_RUN_TIME, worker_name = Worker.name) Worker.logger.info("* [#{Worker.name}] acquiring lock on #{name}") unless lock_exclusively!(max_run_time, worker_name) # We did not get the lock, some other worker process must have Worker.logger.warn("* [#{Worker.name}] 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 Worker.logger.info("* [#{Worker.name}] #{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 |