Class: Shoryuken::Polling::WeightedRoundRobin

Inherits:
BaseStrategy
  • Object
show all
Defined in:
lib/shoryuken/polling/weighted_round_robin.rb

Overview

A polling strategy that processes queues in weighted round-robin order. Queue weights determine how often each queue is polled relative to others. Queues are temporarily paused when no messages are found.

Instance Method Summary collapse

Methods inherited from BaseStrategy

#==, #delay

Methods included from Util

#elapsed, #fire_event, #logger, #unparse_queues, #worker_name

Constructor Details

#initialize(queues, delay = nil) ⇒ WeightedRoundRobin

Initializes a new WeightedRoundRobin polling strategy

Parameters:

  • queues (Array<String>) —

    array of queue names, with weights indicated by repetition

  • delay (Float, nil) (defaults to: nil) —

    delay in seconds before unpausing empty queues



13
14
15
16
17
18
19
20
21
22
 
# File 'lib/shoryuken/polling/weighted_round_robin.rb', line 13

def initialize(queues, delay = nil)
  @initial_queues = queues
  @queues = queues.dup.uniq
  @paused_queues = []
  @delay = delay
  # next_queue/messages_found run on the dispatch thread while
  # message_processed runs on processor-completion threads (for FIFO
  # queues), so all access to the shared state is serialized.
  @mutex = Mutex.new
end

Instance Method Details

#active_queues ⇒ Array<Array>

Returns the list of active queues with their current weights

Returns:

  • (Array<Array>) —

    array of [queue_name, weight] pairs



63
64
65
 
# File 'lib/shoryuken/polling/weighted_round_robin.rb', line 63

def active_queues
  @mutex.synchronize { unparse_queues(@queues) }
end

#message_processed(queue) ⇒ void

This method returns an undefined value.

Called when a message from a queue has been processed

Parameters:

  • queue (String) —

    the queue name



71
72
73
74
75
76
 
# File 'lib/shoryuken/polling/weighted_round_robin.rb', line 71

def message_processed(queue)
  @mutex.synchronize do
    paused_queue = @paused_queues.find { |_time, name| name == queue }
    paused_queue[0] = Time.at(0) if paused_queue
  end
end

#messages_found(queue, messages_found) ⇒ void

This method returns an undefined value.

Handles the result of polling a queue, adjusting weight if messages were found

Parameters:

  • queue (String) —

    the queue name

  • messages_found (Integer) —

    number of messages found



45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/shoryuken/polling/weighted_round_robin.rb', line 45

def messages_found(queue, messages_found)
  @mutex.synchronize do
    if messages_found == 0
      pause(queue)
    else
      maximum_weight = maximum_queue_weight(queue)
      current_weight = current_queue_weight(queue)
      if maximum_weight > current_weight
        logger.info { "Increasing #{queue} weight to #{current_weight + 1}, max: #{maximum_weight}" }
        @queues << queue
      end
    end
  end
end

#next_queue ⇒ QueueConfiguration?

Returns the next queue to poll in round-robin order

Returns:



27
28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/shoryuken/polling/weighted_round_robin.rb', line 27

def next_queue
  @mutex.synchronize do
    unpause_queues
    queue = @queues.shift
    if queue.nil?
      nil
    else
      @queues << queue
      QueueConfiguration.new(queue, {})
    end
  end
end