Class: Farce::Abstract::PriorityQueue Abstract

Inherits:
Queue
  • Object
show all
Includes:
Internal::BlockingPriorityQueue
Defined in:
lib/farce/abstract/priority_queue.rb

Overview

This class is abstract.

Shared priority ordering and blocking behavior.

Instance Attribute Summary collapse

Attributes inherited from Queue

#capacity, #mode, #num_waiting, #size

Instance Method Summary collapse

Methods inherited from Queue

#age_tracking?, #clear, #close, #closed?, #deq, #duplicable?, #empty?, #enq, #full?, #generation, #length, #max, #oldest_age, #oldest_enqueued_at, #seal, #sealed?, #wait_push

Constructor Details

#initialize(capacity: nil, default_priority: 0, order: :ascending, track_age: false) ⇒ PriorityQueue

Returns a new instance of PriorityQueue.

Parameters:

  • capacity (Integer, nil) (defaults to: nil) —

    the maximum number of values, or nil for an unbounded queue

  • default_priority (BasicObject) (defaults to: 0) —

    the priority used when push or try_push is called without an explicit priority

  • order (:ascending, :descending) (defaults to: :ascending) —

    the priority order



22
23
24
25
26
 
# File 'lib/farce/abstract/priority_queue.rb', line 22

def initialize(capacity: nil, default_priority: 0, order: :ascending, track_age: false)
  @order            = normalize_order(order)
  @default_priority = default_priority
  super(capacity:, reverse_order: @order == :descending, track_age:)
end

Instance Attribute Details

#default_priority ⇒ BasicObject (readonly)

Returns the priority used when none is passed to push.

Returns:

  • (BasicObject) —

    the priority used when none is passed to push



12
13
14
 
# File 'lib/farce/abstract/priority_queue.rb', line 12

def default_priority
  @default_priority
end

#order ⇒ :ascending, :descending (readonly)

Returns the priority order.

Returns:

  • (:ascending, :descending) —

    the priority order



15
16
17
 
# File 'lib/farce/abstract/priority_queue.rb', line 15

def order
  @order
end

Instance Method Details

#delete(value, priority:, compare_by_identity: false) ⇒ Boolean

Delete the oldest matching value at the exact priority.

Parameters:

  • value (BasicObject) —

    the value to delete

  • priority (BasicObject) —

    the exact priority to search

  • compare_by_identity (Boolean) (defaults to: false) —

    compare values by identity instead of equality

Returns:

  • (Boolean) —

    whether a value was deleted



98
99
100
 
# File 'lib/farce/abstract/priority_queue.rb', line 98

def delete(value, priority:, compare_by_identity: false)
  delete_from_storage(priority, value, compare_by_identity:)
end

#first_priority { ... } ⇒ BasicObject?

Return the next priority without removing it.

Yields:

  • called when the queue is empty

Returns:

  • (BasicObject, nil) —

    the priority or the fallback result



82
83
84
 
# File 'lib/farce/abstract/priority_queue.rb', line 82

def first_priority(&)
  internal_reverse_order? ? internal_queue.peek_last_priority(&) : internal_queue.peek_priority(&)
end

#last_priority { ... } ⇒ BasicObject?

Return the last priority without removing it.

Yields:

  • called when the queue is empty

Returns:

  • (BasicObject, nil) —

    the priority or the fallback result



89
90
91
 
# File 'lib/farce/abstract/priority_queue.rb', line 89

def last_priority(&)
  internal_reverse_order? ? internal_queue.peek_priority(&) : internal_queue.peek_last_priority(&)
end

#peek { ... } ⇒ BasicObject?

Return the next value without removing it.

Yields:

  • called when the queue is empty

Returns:

  • (BasicObject, nil) —

    the next value or the fallback result



31
 
# File 'lib/farce/abstract/priority_queue.rb', line 31

def peek(&) = internal_reverse_order? ? internal_queue.peek_last(&) : internal_queue.peek(&)

#pop(non_block = false, timeout: nil) ⇒ BasicObject

Remove the oldest value at the lowest or highest priority, depending on #order, waiting when empty.



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
# File 'lib/farce/abstract/priority_queue.rb', line 34

def pop(non_block = false, timeout: nil) # rubocop:disable Style/OptionalBooleanParameter
  return try_pop { raise ThreadError, "queue empty" } if non_block
  deadline = timeout_at(timeout) unless timeout.nil?

  while true
    empty  = false
    result = if internal_reverse_order?
               internal_queue.pop_last { empty = true }
             else
               internal_queue.pop { empty = true }
             end

    return result unless empty
    remaining = remaining_timeout(deadline)
    return block_given? ? yield : nil unless wait_pop(timeout: remaining)
  end
end

#push(value, non_block = false, priority: default_priority, timeout: nil) ⇒ Boolean

Add a value, waiting for capacity when the queue is bounded and full.

Parameters:

  • value (BasicObject) —

    the value to add

  • non_block (Boolean) (defaults to: false) —

    whether to raise an exception when the queue is at capacity

  • priority (BasicObject) (defaults to: default_priority) —

    the value used to order the entry

  • timeout (Numeric, nil) (defaults to: nil) —

    maximum number of seconds to wait

Returns:

  • (Boolean) —

    whether the value was added

Raises:

  • (ThreadError) —

    when the queue is at capacity and non_block is true



59
60
61
 
# File 'lib/farce/abstract/priority_queue.rb', line 59

def push(value, non_block = false, priority: default_priority, timeout: nil) # rubocop:disable Style/OptionalBooleanParameter
  push_to_storage(priority, non_block, value, timeout:)
end

#try_pop { ... } ⇒ BasicObject?

Try to remove the oldest value at the lowest priority without waiting.

Yields:

  • called when the queue is empty

Returns:

  • (BasicObject, nil) —

    the next value or the fallback result



66
 
# File 'lib/farce/abstract/priority_queue.rb', line 66

def try_pop(&) = internal_reverse_order? ? internal_queue.pop_last(&) : internal_queue.pop(&)

#try_push(value, priority: default_priority) { ... } ⇒ Boolean, BasicObject

Try to add a value without waiting.

Parameters:

  • value (BasicObject) —

    the value to add

  • priority (BasicObject) (defaults to: default_priority) —

    the value used to order the entry

Yields:

  • called when the queue is at capacity

Returns:

  • (Boolean, BasicObject) —

    true, or the fallback result when full



73
74
75
76
77
 
# File 'lib/farce/abstract/priority_queue.rb', line 73

def try_push(value, priority: default_priority)
  return true if internal_queue.push(priority, value)

  block_given? ? yield : false
end

#wait_pop(timeout: nil) ⇒ BasicObject

Wait until a value is available without removing it.



103
 
# File 'lib/farce/abstract/priority_queue.rb', line 103

def wait_pop(timeout: nil) = wait(timeout) { size.positive? }