Class: Delayed::Worker
- Inherits:
-
Object
show all
- Includes:
- Logging
- Defined in:
- lib/delayed/worker.rb,
lib/delayed/worker/health_check.rb,
lib/delayed/worker/process_helper.rb,
lib/delayed/worker/null_health_check.rb,
lib/delayed/worker/consul_health_check.rb
Defined Under Namespace
Modules: ProcessHelper
Classes: ConsulHealthCheck, HealthCheck, NullHealthCheck
Constant Summary
collapse
- SIGNALS =
%i{INT TERM QUIT}
- LINUX_PAGE_SIZE =
(size = `getconf PAGESIZE`.to_i; size > 0 ? size : 4096)
Instance Attribute Summary collapse
Class Method Summary
collapse
Instance Method Summary
collapse
Methods included from Logging
logger, #logger, #say
Constructor Details
#initialize(options = {}) ⇒ Worker
Returns a new instance of Worker.
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
|
# File 'lib/delayed/worker.rb', line 58
def initialize(options = {})
@exit = false
@parent_pid = options[:parent_pid]
@queue_name = options[:queue] ||= Settings.queue
@min_priority = options[:min_priority]
@max_priority = options[:max_priority]
@max_job_count = options[:worker_max_job_count].to_i
@max_memory_usage = options[:worker_max_memory_usage].to_i
@work_queue = options.delete(:work_queue) || WorkQueue::InProcess.new
@health_check_type = Settings.worker_health_check_type
@health_check_config = Settings.worker_health_check_config
@config = options
@job_count = 0
@self_pipe = IO.pipe
@signal_queue = []
app = Rails.application
if app && !app.config.cache_classes
Delayed::Worker.lifecycle.around(:perform) do |worker, job, &block|
reload = app.config.reload_classes_only_on_change != true || app.reloaders.map(&:updated?).any?
if reload
if defined?(ActiveSupport::Reloader)
Rails.application.reloader.reload!
else
ActionDispatch::Reloader.prepare!
end
end
begin
block.call(worker, job)
ensure
ActionDispatch::Reloader.cleanup! if reload && !defined?(ActiveSupport::Reloader)
end
end
end
plugins.each { |plugin| plugin.inject! }
end
|
Instance Attribute Details
#config ⇒ Object
Returns the value of attribute config.
25
26
27
|
# File 'lib/delayed/worker.rb', line 25
def config
@config
end
|
#max_priority ⇒ Object
Returns the value of attribute max_priority.
25
26
27
|
# File 'lib/delayed/worker.rb', line 25
def max_priority
@max_priority
end
|
#min_priority ⇒ Object
Returns the value of attribute min_priority.
25
26
27
|
# File 'lib/delayed/worker.rb', line 25
def min_priority
@min_priority
end
|
#queue_name ⇒ Object
Returns the value of attribute queue_name.
25
26
27
|
# File 'lib/delayed/worker.rb', line 25
def queue_name
@queue_name
end
|
#work_queue ⇒ Object
Returns the value of attribute work_queue.
25
26
27
|
# File 'lib/delayed/worker.rb', line 25
def work_queue
@work_queue
end
|
Class Method Details
.current_job ⇒ Object
47
48
49
|
# File 'lib/delayed/worker.rb', line 47
def self.current_job
Thread.current[:running_delayed_job]
end
|
.lifecycle ⇒ Object
43
44
45
|
# File 'lib/delayed/worker.rb', line 43
def self.lifecycle
@lifecycle ||= Delayed::Lifecycle.new
end
|
.on_max_failures=(block) ⇒ Object
Callback to fire when a delayed job fails max_attempts times. If this callback is defined, then the value of destroy_failed_jobs is ignored, and the job is destroyed if this block returns true.
This allows for destroying “uninteresting” failures, while keeping around interesting failures to be investigated later.
The block is called with args(job, last_exception)
35
36
37
|
# File 'lib/delayed/worker.rb', line 35
def self.on_max_failures=(block)
@@on_max_failures = block
end
|
.running_job(job) ⇒ Object
51
52
53
54
55
56
|
# File 'lib/delayed/worker.rb', line 51
def self.running_job(job)
Thread.current[:running_delayed_job] = job
yield
ensure
Thread.current[:running_delayed_job] = nil
end
|
Instance Method Details
#cleaned? ⇒ Boolean
177
178
179
|
# File 'lib/delayed/worker.rb', line 177
def cleaned?
@cleaned
end
|
#cleanup! ⇒ Object
167
168
169
170
171
172
173
174
175
|
# File 'lib/delayed/worker.rb', line 167
def cleanup!
return if cleaned?
health_check.stop
work_queue.close
Delayed::Job.clear_locks!(name)
@cleaned = true
end
|
set up the session context information, so that it gets logged with the job log lines also set up a unique tmpdir, which will get removed at the end of the job.
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
|
# File 'lib/delayed/worker.rb', line 287
def configure_for_job(job)
previous_tmpdir = ENV['TMPDIR']
self.class.running_job(job) do
dir = Dir.mktmpdir("job-#{job.id}-#{self.name.gsub(/[^\w\.]/, '.')}-")
begin
ENV['TMPDIR'] = dir
yield
ensure
FileUtils.remove_entry(dir, true)
end
end
ensure
ENV['TMPDIR'] = previous_tmpdir
end
|
#exit? ⇒ Boolean
107
108
109
|
# File 'lib/delayed/worker.rb', line 107
def exit?
!!@exit || parent_exited?
end
|
#handle_failed_job(job, error) ⇒ Object
266
267
268
269
270
|
# File 'lib/delayed/worker.rb', line 266
def handle_failed_job(job, error)
job.last_error = "#{error.message}\n#{error.backtrace.join("\n")}"
logger.error("Failed with #{error.class} [#{error.message}] (#{job.attempts} attempts)")
job.reschedule(error)
end
|
#health_check ⇒ Object
303
304
305
306
307
308
309
|
# File 'lib/delayed/worker.rb', line 303
def health_check
@health_check ||= HealthCheck.build(
type: @health_check_type,
worker_name: name,
config: @health_check_config
)
end
|
#id ⇒ Object
272
273
274
|
# File 'lib/delayed/worker.rb', line 272
def id
Process.pid
end
|
#log_job(job, format = :short) ⇒ Object
276
277
278
279
280
281
282
283
|
# File 'lib/delayed/worker.rb', line 276
def log_job(job, format = :short)
case format
when :long
"#{job.full_name} #{ Settings.job_detailed_log_format.call(job) }"
else
job.full_name
end
end
|
#name ⇒ Object
99
100
101
|
# File 'lib/delayed/worker.rb', line 99
def name
@name ||= "#{Socket.gethostname rescue "X"}:#{self.id}"
end
|
#parent_exited? ⇒ Boolean
111
112
113
|
# File 'lib/delayed/worker.rb', line 111
def parent_exited?
@parent_pid && @parent_pid != Process.ppid
end
|
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
|
# File 'lib/delayed/worker.rb', line 215
def perform(job)
begin
count = 1
raise Delayed::Backend::JobExpired, "job expired at #{job.expires_at}" if job.expired?
self.class.lifecycle.run_callbacks(:perform, self, job) do
set_process_name("run:#{Settings.worker_procname_prefix}#{job.id}:#{job.name}")
logger.info("Processing #{log_job(job, :long)}")
runtime = Benchmark.realtime do
if job.batch?
count = perform_batch(job)
else
job.invoke_job
end
job.destroy
end
logger.info("Completed #{log_job(job)} #{"%.0fms" % (runtime * 1000)}")
end
rescue ::Delayed::RetriableError => re
can_retry = job.attempts + 1 < job.inferred_max_attempts
callback_type = can_retry ? :retry : :error
self.class.lifecycle.run_callbacks(callback_type, self, job, re) do
handle_failed_job(job, re)
end
rescue SystemExit => se
job.reschedule(se)
rescue Exception => e
self.class.lifecycle.run_callbacks(:error, self, job, e) do
handle_failed_job(job, e)
end
end
count
end
|
252
253
254
255
256
257
258
259
260
261
262
263
264
|
# File 'lib/delayed/worker.rb', line 252
def perform_batch(parent_job)
batch = parent_job.payload_object
if batch.mode == :serial
batch.jobs.each do |job|
job.source = parent_job.source
job.create_and_lock!(name)
configure_for_job(job) do
perform(job)
end
end
batch.items.size
end
end
|
#run ⇒ Object
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
|
# File 'lib/delayed/worker.rb', line 181
def run
return if exit?
self.class.lifecycle.run_callbacks(:loop, self) do
set_process_name("pop:#{Settings.worker_procname_prefix}#{@queue_name}:#{min_priority || 0}:#{max_priority || 'max'}")
job = self.class.lifecycle.run_callbacks(:pop, self) do
work_queue.get_and_lock_next_available(name, config)
end
if job
configure_for_job(job) do
@job_count += perform(job)
if @max_job_count > 0 && @job_count >= @max_job_count
logger.debug "Max job count of #{@max_job_count} exceeded, dying"
@exit = true
end
if @max_memory_usage > 0
memory = sample_memory
if memory > @max_memory_usage
logger.debug "Memory usage of #{memory} exceeds max of #{@max_memory_usage}, dying"
@exit = true
else
logger.debug "Memory usage: #{memory}"
end
end
end
else
set_process_name("wait:#{Settings.worker_procname_prefix}#{@queue_name}:#{min_priority || 0}:#{max_priority || 'max'}")
sleep(Settings.sleep_delay + (rand * Settings.sleep_delay_stagger)) unless exit?
end
end
end
|
#sample_memory ⇒ Object
321
322
323
324
|
# File 'lib/delayed/worker.rb', line 321
def sample_memory
s = File.read("/proc/#{Process.pid}/statm").to_i rescue 0
s * LINUX_PAGE_SIZE / 1024
end
|
#set_process_name(new_name) ⇒ Object
103
104
105
|
# File 'lib/delayed/worker.rb', line 103
def set_process_name(new_name)
$0 = "delayed:#{new_name}"
end
|
#start ⇒ Object
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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
# File 'lib/delayed/worker.rb', line 120
def start
logger.info "Starting worker"
set_process_name("start:#{Settings.worker_procname_prefix}#{@queue_name}:#{min_priority || 0}:#{max_priority || 'max'}")
work_thread = Thread.current
SIGNALS.each do |sig|
trap(sig) { @signal_queue << sig; wake_up }
end
raise 'Could not register health_check' unless health_check.start
signal_processor = Thread.new do
loop do
@self_pipe[0].read(1)
case @signal_queue.pop
when :INT, :TERM
@exit = true work_thread.raise(SystemExit) cleanup! break
when :QUIT
@exit = true
else
logger.error "Unknown signal '#{sig}' received"
end
end
end
self.class.lifecycle.run_callbacks(:execute, self) do
until exit? do
run
end
end
logger.info "Stopping worker"
rescue => e
Rails.logger.fatal("Child process died: #{e.inspect}") rescue nil
self.class.lifecycle.run_callbacks(:exceptional_exit, self, e) { }
ensure
cleanup!
if signal_processor
signal_processor.kill
signal_processor.join
end
end
|
#wake_up ⇒ Object
115
116
117
118
|
# File 'lib/delayed/worker.rb', line 115
def wake_up
@self_pipe[1].write_nonblock('.', exception: false)
work_queue.wake_up
end
|