Class: OllamaChat::Utils::OrderedQueue

Inherits:
Object
  • Object
show all
Defined in:
lib/ollama_chat/utils/ordered_queue.rb

Overview

A thread-safe priority queue backed by a Min-Heap.

This class ensures that items are retrieved in the order of their IDs, which can be tuples like [thread_id, chunk_id].

Instance Method Summary collapse

Constructor Details

#initialize ⇒ OrderedQueue

Initializes a new OrderedQueue.



7
8
9
10
11
 
# File 'lib/ollama_chat/utils/ordered_queue.rb', line 7

def initialize
  @heap = []
  @mutex = Mutex.new
  @cv = ConditionVariable.new
end

Instance Method Details

#empty? ⇒ Boolean

Checks if the queue is empty.

Returns:

  • (Boolean) —

    true if empty



43
44
45
 
# File 'lib/ollama_chat/utils/ordered_queue.rb', line 43

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

#extract_min ⇒ Array? (private)

Removes the minimum element from the heap.

Returns:

  • (Array, nil) —

    the [id, payload] pair or nil if empty



57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/ollama_chat/utils/ordered_queue.rb', line 57

def extract_min
  min = @heap[0]
  if @heap.size > 1
    last = @heap.pop
    @heap[0] = last
    heapify_down(0)
  else
    @heap.clear
  end
  min
end

#heapify_down(index) ⇒ Object (private)

Restores the min-heap property by moving an element down.



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/ollama_chat/utils/ordered_queue.rb', line 81

def heapify_down(index)
  size = @heap.size
  while true
    smallest = index
    left = 2 * index + 1
    right = 2 * index + 2

    smallest = left if left < size && @heap[left][0] < @heap[smallest][0]
    smallest = right if right < size && @heap[right][0] < @heap[smallest][0]

    break if smallest == index

    @heap[index], @heap[smallest] = @heap[smallest], @heap[index]
    index = smallest
  end
end

#heapify_up(index) ⇒ Object (private)

Restores the min-heap property by moving an element up.



70
71
72
73
74
75
76
77
78
 
# File 'lib/ollama_chat/utils/ordered_queue.rb', line 70

def heapify_up(index)
  while index > 0
    parent = (index - 1) / 2
    break if @heap[index][0] >= @heap[parent][0]

    @heap[index], @heap[parent] = @heap[parent], @heap[index]
    index = parent
  end
end

#peek ⇒ Array?

Peeks at the item with the lowest ID without removing it.

Returns:

  • (Array, nil) —

    the [id, payload] pair or nil if empty



29
30
31
 
# File 'lib/ollama_chat/utils/ordered_queue.rb', line 29

def peek
  @mutex.synchronize { @heap[0] }
end

#pop ⇒ Array?

Removes and returns the item with the lowest ID.

Returns:

  • (Array, nil) —

    the [id, payload] pair or nil if empty



36
37
38
 
# File 'lib/ollama_chat/utils/ordered_queue.rb', line 36

def pop
  @mutex.synchronize { extract_min }
end

#push(id, payload) ⇒ Object

Pushes an item into the queue.

Parameters:

  • id (Array<Integer>, Integer) —

    the ID for ordering (e.g., [thread_id, chunk_id])

  • payload (Object) —

    the data associated with the ID



17
18
19
20
21
22
23
24
 
# File 'lib/ollama_chat/utils/ordered_queue.rb', line 17

def push(id, payload)
  @mutex.synchronize do
    id.extend(Comparable) rescue nil
    @heap << [id, payload]
    heapify_up(@heap.size - 1)
    @cv.signal
  end
end

#wait ⇒ Object

Waits for a signal from a producer thread.



48
49
50
 
# File 'lib/ollama_chat/utils/ordered_queue.rb', line 48

def wait
  @mutex.synchronize { @cv.wait(@mutex) }
end