Class: Sidekiq::BasicFetch

Inherits:
Object
  • Object
show all
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

#config

Instance Method Summary collapse

Methods included from Component

#fire_event, #handle_exception, #hostname, #identity, #logger, #process_nonce, #redis, #safe_thread, #tid, #watchdog

Constructor Details

#initialize(config) ⇒ BasicFetch

Returns a new instance of BasicFetch.

Raises:

  • (ArgumentError) 


29
30
31
32
33
34
35
36
37
38
 
# File 'lib/sidekiq/fetch.rb', line 29

def initialize(config)
  raise ArgumentError, "missing queue list" unless config[:queues]
  @config = config
  @strictly_ordered_queues = !!@config[:strict]
  @queues = @config[:queues].map { |q| "queue:#{q}" }
  if @strictly_ordered_queues
    @queues.uniq!
    @queues << TIMEOUT
  end
end

Instance Method Details

#bulk_requeue(inprogress, options) ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/sidekiq/fetch.rb', line 53

def bulk_requeue(inprogress, options)
  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.message}")
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.



80
81
82
83
84
85
86
87
88
89
 
# File 'lib/sidekiq/fetch.rb', line 80

def queues_cmd
  if @strictly_ordered_queues
    @queues
  else
    permute = @queues.shuffle
    permute.uniq!
    permute << TIMEOUT
    permute
  end
end

#retrieve_work ⇒ Object 



40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/sidekiq/fetch.rb', line 40

def retrieve_work
  qs = queues_cmd
  # 4825 Sidekiq Pro with all queues paused will return an
  # empty set of queues with a trailing TIMEOUT value.
  if qs.size <= 1
    sleep(TIMEOUT)
    return nil
  end

  queue, job = redis { |conn| conn.brpop(*qs) }
  UnitOfWork.new(queue, job, config) if queue
end