1
0
Fork 0
mirror of https://github.com/mperham/sidekiq.git synced 2022-11-09 13:52:34 -05:00
mperham--sidekiq/lib/sidekiq/processor.rb

195 lines
5.5 KiB
Ruby
Raw Normal View History

require 'sidekiq/util'
require 'sidekiq/middleware/server/retry_jobs'
2012-02-25 16:43:53 -05:00
require 'sidekiq/middleware/server/logging'
module Sidekiq
2012-06-13 00:55:06 -04:00
##
# The Processor receives a message from the Manager and actually
# processes it. It instantiates the worker, runs the middleware
# chain and then calls Sidekiq::Worker#perform.
class Processor
2015-10-05 13:13:00 -04:00
# To prevent a memory leak, ensure that stats expire. However, they
# should take up a minimal amount of storage so keep them around
# for a long time.
STATS_TIMEOUT = 24 * 60 * 60 * 365 * 5
include Util
def self.default_middleware
Middleware::Chain.new do |m|
2012-02-25 16:43:53 -05:00
m.add Middleware::Server::Logging
2012-03-18 02:04:31 -04:00
m.add Middleware::Server::RetryJobs
if defined?(::ActiveRecord::Base)
require 'sidekiq/middleware/server/active_record'
m.add Sidekiq::Middleware::Server::ActiveRecord
end
end
end
2015-10-05 13:13:00 -04:00
attr_reader :thread
2015-10-05 13:13:00 -04:00
def initialize(mgr)
@mgr = mgr
@done = false
@work = ::Queue.new
end
def terminate(wait=false)
@done = true
@work << nil
2015-10-06 15:43:01 -04:00
@thread.value if wait
end
def kill(wait=false)
2015-10-05 13:13:00 -04:00
# 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.
2015-10-06 15:43:01 -04:00
@thread.raise ::Sidekiq::Shutdown
2015-10-05 13:13:00 -04:00
@thread.value if wait
end
2015-10-06 15:43:01 -04:00
def start
@thread ||= safe_thread("processor", &method(:run))
end
2015-10-06 15:43:01 -04:00
def request_process(work)
2015-10-05 13:13:00 -04:00
raise ArgumentError, "Processor is shut down!" if @done
2015-10-06 15:43:01 -04:00
raise ArgumentError, "Processor has not started!" unless @thread
2015-10-05 13:13:00 -04:00
@work << work
end
2015-10-06 15:43:01 -04:00
private unless $TESTING
2015-10-05 13:13:00 -04:00
def run
begin
while !@done
job = @work.pop
2015-10-06 15:43:01 -04:00
process(job) if job
2015-10-05 13:13:00 -04:00
end
rescue Exception => ex
Sidekiq.logger.warn(ex.message)
@mgr.processor_died(self, ex)
end
end
2015-10-06 15:43:01 -04:00
def process(work)
msgstr = work.message
queue = work.queue_name
ack = false
begin
msg = Sidekiq.load_json(msgstr)
klass = msg['class'.freeze].constantize
worker = klass.new
worker.jid = msg['jid'.freeze]
stats(worker, msg, queue) do
Sidekiq.server_middleware.invoke(worker, msg, queue) do
# Only ack if we either attempted to start this job or
# successfully completed it. This prevents us from
# losing jobs if a middleware raises an exception before yielding
ack = true
execute_job(worker, cloned(msg['args'.freeze]))
2012-08-16 21:12:25 -04:00
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.
ack = false
rescue Exception => ex
handle_exception(ex, msg || { :message => msgstr })
raise
ensure
work.acknowledge if ack
end
2015-10-05 13:13:00 -04:00
@mgr.processor_done(self)
end
def execute_job(worker, cloned_args)
worker.perform(*cloned_args)
end
def thread_identity
@str ||= Thread.current.object_id.to_s(36)
end
def stats(worker, msg, queue)
2014-02-24 23:06:48 -05:00
# Do not conflate errors from the job with errors caused by updating
# stats so calling code can react appropriately
2014-02-24 16:08:57 -05:00
retry_and_suppress_exceptions do
2014-02-24 23:06:48 -05:00
hash = Sidekiq.dump_json({:queue => queue, :payload => msg, :run_at => Time.now.to_i })
Sidekiq.redis do |conn|
2015-10-06 15:43:01 -04:00
conn.pipelined do
conn.hmset("#{identity}:workers", thread_identity, hash)
2015-10-05 13:13:00 -04:00
conn.expire("#{identity}:workers", 14400)
end
end
end
nowdate = Time.now.utc.strftime("%Y-%m-%d".freeze)
begin
yield
2012-05-12 16:23:23 -04:00
rescue Exception
2014-02-24 16:08:57 -05:00
retry_and_suppress_exceptions do
failed = "stat:failed:#{nowdate}"
Sidekiq.redis do |conn|
2015-10-06 15:43:01 -04:00
conn.pipelined do
conn.incrby("stat:failed".freeze, 1)
conn.incrby(failed, 1)
conn.expire(failed, STATS_TIMEOUT)
end
end
end
raise
ensure
2014-02-24 16:08:57 -05:00
retry_and_suppress_exceptions do
processed = "stat:processed:#{nowdate}"
Sidekiq.redis do |conn|
2015-10-06 15:43:01 -04:00
conn.pipelined do
conn.hdel("#{identity}:workers", thread_identity)
conn.incrby("stat:processed".freeze, 1)
conn.incrby(processed, 1)
conn.expire(processed, STATS_TIMEOUT)
end
end
end
end
end
# Deep clone the arguments passed to the worker so that if
# the message fails, what is pushed back onto Redis hasn't
# been mutated by the worker.
def cloned(ary)
Marshal.load(Marshal.dump(ary))
end
2014-02-24 16:10:07 -05:00
# If an exception occurs in the block passed to this method, that block will be retried up to max_retries times.
# All exceptions will be swallowed and logged.
def retry_and_suppress_exceptions(max_retries = 5)
retry_count = 0
begin
yield
2014-02-24 16:10:58 -05:00
rescue => e
retry_count += 1
if retry_count <= max_retries
2015-10-06 15:43:01 -04:00
Sidekiq.logger.info {"Suppressing and retrying error: #{e.inspect}"}
2015-05-20 13:31:16 -04:00
pause_for_recovery(retry_count)
retry
else
2014-02-24 23:47:44 -05:00
handle_exception(e, { :message => "Exhausted #{max_retries} retries"})
end
end
end
2015-05-20 13:31:16 -04:00
def pause_for_recovery(retry_count)
sleep(retry_count)
end
end
end