Class: Fractor::PriorityWorkQueue

Inherits:
Object
  • Object
show all
Defined in:
lib/fractor/priority_work_queue.rb

Overview

PriorityWorkQueue manages work items with priority-based scheduling

Features:

  • Priority levels from :critical to :background
  • FIFO within same priority level
  • Optional priority aging to prevent starvation
  • Thread-safe operations

Examples:

Basic usage

queue = Fractor::PriorityWorkQueue.new
queue.push(PriorityWork.new("urgent", priority: :critical))
queue.push(PriorityWork.new("normal", priority: :normal))
work = queue.pop # Returns critical priority work first

With priority aging

queue = Fractor::PriorityWorkQueue.new(aging_enabled: true, aging_threshold: 60)

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(aging_enabled: false, aging_threshold: 60) ⇒ PriorityWorkQueue

Initialize a new PriorityWorkQueue

Parameters:

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

    Enable priority aging to prevent starvation

  • aging_threshold (Integer) (defaults to: 60) —

    Seconds before a work item gets priority boost



27
28
29
30
31
32
33
34
 
# File 'lib/fractor/priority_work_queue.rb', line 27

def initialize(aging_enabled: false, aging_threshold: 60)
  @queue = []
  @mutex = Mutex.new
  @condition = ConditionVariable.new
  @aging_enabled = aging_enabled
  @aging_threshold = aging_threshold
  @closed = false
end

Instance Attribute Details

#aging_enabled ⇒ Object (readonly)

Returns the value of attribute aging_enabled.



21
22
23
 
# File 'lib/fractor/priority_work_queue.rb', line 21

def aging_enabled
  @aging_enabled
end

#aging_threshold ⇒ Object (readonly)

Returns the value of attribute aging_threshold.



21
22
23
 
# File 'lib/fractor/priority_work_queue.rb', line 21

def aging_threshold
  @aging_threshold
end

Instance Method Details

#clear ⇒ Array<PriorityWork>

Clear all items from the queue

Returns:



124
125
126
127
128
129
130
 
# File 'lib/fractor/priority_work_queue.rb', line 124

def clear
  @mutex.synchronize do
    items = @queue.dup
    @queue.clear
    items
  end
end

#close ⇒ Object

Close the queue No new items can be added, but existing items can be popped



107
108
109
110
111
112
 
# File 'lib/fractor/priority_work_queue.rb', line 107

def close
  @mutex.synchronize do
    @closed = true
    @condition.broadcast
  end
end

#closed? ⇒ Boolean

Check if queue is closed

Returns:

  • (Boolean) —

    true if queue is closed



117
118
119
 
# File 'lib/fractor/priority_work_queue.rb', line 117

def closed?
  @mutex.synchronize { @closed }
end

#empty? ⇒ Boolean

Check if queue is empty

Returns:

  • (Boolean) —

    true if queue is empty



101
102
103
 
# File 'lib/fractor/priority_work_queue.rb', line 101

def empty?
  @mutex.synchronize { @queue.empty? }
end

#pop ⇒ PriorityWork? Also known as: dequeue, shift

Remove and return highest priority work Blocks if queue is empty

Returns:

  • (PriorityWork, nil) —

    Highest priority work or nil if queue closed



61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/fractor/priority_work_queue.rb', line 61

def pop
  @mutex.synchronize do
    loop do
      return nil if @closed && @queue.empty?

      unless @queue.empty?
        apply_aging! if @aging_enabled
        return @queue.shift
      end

      @condition.wait(@mutex)
    end
  end
end

#pop_non_blocking ⇒ PriorityWork?

Try to remove and return highest priority work without blocking

Returns:

  • (PriorityWork, nil) —

    Highest priority work or nil if empty



81
82
83
84
85
86
87
88
 
# File 'lib/fractor/priority_work_queue.rb', line 81

def pop_non_blocking
  @mutex.synchronize do
    return nil if @queue.empty?

    apply_aging! if @aging_enabled
    @queue.shift
  end
end

#push(work) ⇒ Object Also known as: <<, enqueue

Add work to the queue

Parameters:

Raises:

  • (ArgumentError) —

    if work is not a PriorityWork instance

  • (ClosedQueueError) —

    if queue is closed



41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/fractor/priority_work_queue.rb', line 41

def push(work)
  unless work.is_a?(PriorityWork)
    raise ArgumentError,
          "Work must be a PriorityWork"
  end
  raise ClosedQueueError, "Queue is closed" if @closed

  @mutex.synchronize do
    @queue << work
    sort_queue!
    @condition.signal
  end
end

#size ⇒ Integer Also known as: length

Get current queue size

Returns:

  • (Integer) —

    Number of items in queue



93
94
95
 
# File 'lib/fractor/priority_work_queue.rb', line 93

def size
  @mutex.synchronize { @queue.size }
end

#stats ⇒ Hash

Get queue statistics

Returns:

  • (Hash) —

    Statistics including count by priority



135
136
137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/fractor/priority_work_queue.rb', line 135

def stats
  @mutex.synchronize do
    priority_counts = Hash.new(0)
    @queue.each { |work| priority_counts[work.priority] += 1 }

    {
      total: @queue.size,
      by_priority: priority_counts,
      oldest_age: @queue.empty? ? 0 : @queue.max_by(&:age).age,
      closed: @closed,
    }
  end
end