Class: Sidekiq::BasicFetch
- Inherits:
-
Object
- Object
- Sidekiq::BasicFetch
- Includes:
- Component
- Defined in:
- lib/sidekiq/fetch.rb
Defined Under Namespace
Classes: UnitOfWork
Constant Summary collapse
- TIMEOUT =
We want the fetch operation to timeout every few seconds so the thread can check if the process is shutting down.
2
Instance Attribute Summary
Attributes included from Component
Instance Method Summary collapse
- #bulk_requeue(inprogress) ⇒ Object
-
#initialize(cap) ⇒ BasicFetch
constructor
A new instance of BasicFetch.
-
#queues_cmd ⇒ Object
Creating the Redis#brpop command takes into account any configured queue weights.
- #retrieve_work ⇒ Object
Methods included from Component
#fire_event, #handle_exception, #hostname, #identity, #inspect, #logger, #process_nonce, #redis, #safe_thread, #tid, #watchdog
Constructor Details
#initialize(cap) ⇒ BasicFetch
Returns a new instance of BasicFetch.
30 31 32 33 34 35 36 |
# File 'lib/sidekiq/fetch.rb', line 30 def initialize(cap) raise ArgumentError, "missing queue list" unless cap.queues @config = cap @strictly_ordered_queues = cap.mode == :strict @queues = config.queues.map { |q| "queue:#{q}" } @queues.uniq! if @strictly_ordered_queues end |
Instance Method Details
#bulk_requeue(inprogress) ⇒ Object
51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 |
# File 'lib/sidekiq/fetch.rb', line 51 def bulk_requeue(inprogress) return if inprogress.empty? logger.debug { "Re-queueing terminated jobs" } jobs_to_requeue = {} inprogress.each do |unit_of_work| jobs_to_requeue[unit_of_work.queue] ||= [] jobs_to_requeue[unit_of_work.queue] << unit_of_work.job end redis do |conn| conn.pipelined do |pipeline| jobs_to_requeue.each do |queue, jobs| pipeline.rpush(queue, jobs) end end end logger.info("Pushed #{inprogress.size} jobs back to Redis") rescue => ex logger.warn("Failed to requeue #{inprogress.size} jobs: #{ex.}") end |
#queues_cmd ⇒ Object
Creating the Redis#brpop command takes into account any configured queue weights. By default Redis#brpop returns data from the first queue that has pending elements. We recreate the queue command each time we invoke Redis#brpop to honor weights and avoid queue starvation.
78 79 80 81 82 83 84 85 86 |
# File 'lib/sidekiq/fetch.rb', line 78 def queues_cmd if @strictly_ordered_queues @queues else permute = @queues.shuffle permute.uniq! permute end end |
#retrieve_work ⇒ Object
38 39 40 41 42 43 44 45 46 47 48 49 |
# File 'lib/sidekiq/fetch.rb', line 38 def retrieve_work qs = queues_cmd # 4825 Sidekiq Pro with all queues paused will return an # empty set of queues if qs.size <= 0 sleep(TIMEOUT) return nil end queue, job = redis { |conn| conn.blocking_call(TIMEOUT, "brpop", *qs, TIMEOUT) } UnitOfWork.new(queue, job, config) if queue end |