Class: Sidekiq::Processor
- Inherits:
-
Object
- Object
- Sidekiq::Processor
- Includes:
- Component
- Defined in:
- lib/sidekiq/processor.rb
Overview
The Processor is a standalone thread which:
-
fetches a job from Redis
-
executes the job
a. instantiate the job class
b. run the middleware chain
c. call #perform
A Processor can exit due to shutdown or due to an error during job execution.
If an error occurs in the job execution, the Processor calls the Manager to create a new one to replace itself and exits.
Defined Under Namespace
Classes: Counter, SharedWorkState
Constant Summary collapse
Instance Attribute Summary collapse
-
#capsule ⇒ Object
readonly
Returns the value of attribute capsule.
-
#job ⇒ Object
readonly
Returns the value of attribute job.
-
#thread ⇒ Object
readonly
Returns the value of attribute thread.
Attributes included from Component
Instance Method Summary collapse
- #dispatch(job_hash, queue, jobstr) ⇒ Object
- #execute_job(inst, cloned_args) ⇒ Object
- #fetch ⇒ Object
- #get_one ⇒ Object
- #handle_fetch_exception(ex) ⇒ Object
-
#initialize(capsule, &block) ⇒ Processor
constructor
A new instance of Processor.
- #kill(wait = false) ⇒ Object
- #process(uow) ⇒ Object
- #process_one(&block) ⇒ Object
- #run ⇒ Object
- #start ⇒ Object
- #stats(jobstr, queue) ⇒ Object
- #terminate(wait = false) ⇒ Object
Methods included from Component
#fire_event, #handle_exception, #hostname, #identity, #logger, #process_nonce, #redis, #safe_thread, #tid, #watchdog
Constructor Details
#initialize(capsule, &block) ⇒ Processor
Returns a new instance of Processor.
31 32 33 34 35 36 37 38 39 40 41 |
# File 'lib/sidekiq/processor.rb', line 31 def initialize(capsule, &block) @config = @capsule = capsule @callback = block @down = false @done = false @job = nil @thread = nil @reloader = Sidekiq.default_configuration[:reloader] @job_logger = (capsule.config[:job_logger] || Sidekiq::JobLogger).new(logger) @retrier = Sidekiq::JobRetry.new(capsule) end |
Instance Attribute Details
#capsule ⇒ Object (readonly)
Returns the value of attribute capsule.
29 30 31 |
# File 'lib/sidekiq/processor.rb', line 29 def capsule @capsule end |
#job ⇒ Object (readonly)
Returns the value of attribute job.
28 29 30 |
# File 'lib/sidekiq/processor.rb', line 28 def job @job end |
#thread ⇒ Object (readonly)
Returns the value of attribute thread.
27 28 29 |
# File 'lib/sidekiq/processor.rb', line 27 def thread @thread end |
Instance Method Details
#dispatch(job_hash, queue, jobstr) ⇒ Object
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 |
# File 'lib/sidekiq/processor.rb', line 118 def dispatch(job_hash, queue, jobstr) # since middleware can mutate the job hash # we need to clone it to report the original # job structure to the Web UI # or to push back to redis when retrying. # To avoid costly and, most of the time, useless cloning here, # we pass original String of JSON to respected methods # to re-parse it there if we need access to the original, untouched job @job_logger.prepare(job_hash) do @retrier.global(jobstr, queue) do @job_logger.call(job_hash, queue) do stats(jobstr, queue) do # Rails 5 requires a Reloader to wrap code execution. In order to # constantize the worker and instantiate an instance, we have to call # the Reloader. It handles code loading, db connection management, etc. # Effectively this block denotes a "unit of work" to Rails. @reloader.call do klass = Object.const_get(job_hash["class"]) inst = klass.new inst.jid = job_hash["jid"] @retrier.local(inst, jobstr, queue) do yield inst end end end end end end end |
#execute_job(inst, cloned_args) ⇒ Object
209 210 211 |
# File 'lib/sidekiq/processor.rb', line 209 def execute_job(inst, cloned_args) inst.perform(*cloned_args) end |
#fetch ⇒ Object
98 99 100 101 102 103 104 105 106 |
# File 'lib/sidekiq/processor.rb', line 98 def fetch j = get_one if j && @done j.requeue nil else j end end |
#get_one ⇒ Object
86 87 88 89 90 91 92 93 94 95 96 |
# File 'lib/sidekiq/processor.rb', line 86 def get_one uow = capsule.fetcher.retrieve_work if @down logger.info { "Redis is online, #{::Process.clock_gettime(::Process::CLOCK_MONOTONIC) - @down} sec downtime" } @down = nil end uow rescue Sidekiq::Shutdown rescue => ex handle_fetch_exception(ex) end |
#handle_fetch_exception(ex) ⇒ Object
108 109 110 111 112 113 114 115 116 |
# File 'lib/sidekiq/processor.rb', line 108 def handle_fetch_exception(ex) unless @down @down = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC) logger.error("Error fetching job: #{ex}") handle_exception(ex) end sleep(1) nil end |
#kill(wait = false) ⇒ Object
49 50 51 52 53 54 55 56 57 58 59 |
# File 'lib/sidekiq/processor.rb', line 49 def kill(wait = false) @done = true return unless @thread # unlike the other actors, terminate does not wait # for the thread to finish because we don't know how # long the job will take to finish. Instead we # provide a `kill` method to call after the shutdown # timeout passes. @thread.raise ::Sidekiq::Shutdown @thread.value if wait end |
#process(uow) ⇒ Object
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 179 180 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 |
# File 'lib/sidekiq/processor.rb', line 154 def process(uow) jobstr = uow.job queue = uow.queue_name # Treat malformed JSON as a special case: job goes straight to the morgue. job_hash = nil begin job_hash = Sidekiq.load_json(jobstr) rescue => ex handle_exception(ex, {context: "Invalid JSON for job", jobstr: jobstr}) now = Time.now.to_f redis do |conn| conn.multi do |xa| xa.zadd("dead", now.to_s, jobstr) xa.zremrangebyscore("dead", "-inf", now - @capsule.config[:dead_timeout_in_seconds]) xa.zremrangebyrank("dead", 0, - @capsule.config[:dead_max_jobs]) end end return uow.acknowledge end ack = false Thread.handle_interrupt(IGNORE_SHUTDOWN_INTERRUPTS) do Thread.handle_interrupt(ALLOW_SHUTDOWN_INTERRUPTS) do dispatch(job_hash, queue, jobstr) do |inst| config.server_middleware.invoke(inst, job_hash, queue) do execute_job(inst, job_hash["args"]) end end ack = true rescue Sidekiq::Shutdown # Had to force kill this job because it didn't finish # within the timeout. Don't acknowledge the work since # we didn't properly finish it. rescue Sidekiq::JobRetry::Handled => h # this is the common case: job raised error and Sidekiq::JobRetry::Handled # signals that we created a retry successfully. We can acknowledge the job. ack = true e = h.cause || h handle_exception(e, {context: "Job raised exception", job: job_hash}) raise e rescue Exception => ex # Unexpected error! This is very bad and indicates an exception that got past # the retry subsystem (e.g. network partition). We won't acknowledge the job # so it can be rescued when using Sidekiq Pro. handle_exception(ex, {context: "Internal exception!", job: job_hash, jobstr: jobstr}) raise ex end ensure if ack uow.acknowledge end end end |
#process_one(&block) ⇒ Object
80 81 82 83 84 |
# File 'lib/sidekiq/processor.rb', line 80 def process_one(&block) @job = fetch process(@job) if @job @job = nil end |
#run ⇒ Object
67 68 69 70 71 72 73 74 75 76 77 78 |
# File 'lib/sidekiq/processor.rb', line 67 def run # By setting this thread-local, Sidekiq.redis will access +Sidekiq::Capsule#redis_pool+ # instead of the global pool in +Sidekiq::Config#redis_pool+. Thread.current[:sidekiq_capsule] = @capsule process_one until @done @callback.call(self) rescue Sidekiq::Shutdown @callback.call(self) rescue Exception => ex @callback.call(self, ex) end |
#start ⇒ Object
61 62 63 |
# File 'lib/sidekiq/processor.rb', line 61 def start @thread ||= safe_thread("#{config.name}/processor", &method(:run)) end |
#stats(jobstr, queue) ⇒ Object
267 268 269 270 271 272 273 274 275 276 277 278 279 |
# File 'lib/sidekiq/processor.rb', line 267 def stats(jobstr, queue) WORK_STATE.set(tid, {queue: queue, payload: jobstr, run_at: Time.now.to_i}) begin yield rescue Exception FAILURE.incr raise ensure WORK_STATE.delete(tid) PROCESSED.incr end end |
#terminate(wait = false) ⇒ Object
43 44 45 46 47 |
# File 'lib/sidekiq/processor.rb', line 43 def terminate(wait = false) @done = true return unless @thread @thread.value if wait end |