Class: Philiprehberger::TaskQueue::Queue

Inherits:
Object
  • Object
show all
Defined in:
lib/philiprehberger/task_queue/queue.rb

Overview

In-process async job queue with concurrency control.

Tasks are enqueued as blocks or callable objects and executed by a pool of worker threads. The queue is fully thread-safe.

Constant Summary collapse

BACKOFF_POLICIES =

Supported retry backoff policies.

%i[none fixed exponential].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(concurrency: 4, max_retries: 0, retry_backoff: :none, retry_base_delay: 0.1) ⇒ Queue

Returns a new instance of Queue.

Parameters:

  • concurrency (Integer) (defaults to: 4) —

    maximum number of concurrent worker threads

  • max_retries (Integer) (defaults to: 0) —

    how many times a failing task is requeued before it is counted as failed (default 0, i.e. no retries)

  • retry_backoff (Symbol) (defaults to: :none) —

    delay policy between retries — one of :none, :fixed, or :exponential

  • retry_base_delay (Numeric) (defaults to: 0.1) —

    base delay in seconds used by the :fixed and :exponential backoff policies

Raises:

  • (ArgumentError) —

    if max_retries is negative or retry_backoff is not a recognized policy



31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 31

def initialize(concurrency: 4, max_retries: 0, retry_backoff: :none, retry_base_delay: 0.1)
  @concurrency = concurrency
  @max_retries = max_retries
  @retry_backoff = retry_backoff
  @retry_base_delay = retry_base_delay
  validate_retry_options!
  @tasks = []
  @mutex = Mutex.new
  @condition = ConditionVariable.new
  @drain_condition = ConditionVariable.new
  @workers = []
  @running = true
  @started = false
  @paused = false
  @pause_condition = ConditionVariable.new
  @error_handler = nil
  @complete_handler = nil
  @stats = { completed: 0, failed: 0, in_flight: 0, retried: 0 }
end

Instance Attribute Details

#concurrency ⇒ Integer (readonly)

Returns the maximum number of concurrent worker threads.

Returns:

  • (Integer) —

    the maximum number of concurrent worker threads



20
21
22
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 20

def concurrency
  @concurrency
end

Instance Method Details

#busy? ⇒ Boolean

Whether the queue has any pending tasks or any in-flight tasks.

Convenient inverse of "idle" for callers polling without going through stats. Equivalent to: !empty? || in_flight > 0.

Returns:

  • (Boolean) 


212
213
214
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 212

def busy?
  @mutex.synchronize { !@tasks.empty? || @stats[:in_flight].positive? }
end

#clear ⇒ Integer

Remove all pending tasks from the queue.

Returns:

  • (Integer) —

    number of tasks cleared



123
124
125
126
127
128
129
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 123

def clear
  @mutex.synchronize do
    count = @tasks.size
    @tasks.clear
    count
  end
end

#drain(timeout: 30) ⇒ void

This method returns an undefined value.

Block until all pending tasks are complete without shutting down.

Parameters:

  • timeout (Numeric) (defaults to: 30) —

    seconds to wait before returning



150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 150

def drain(timeout: 30)
  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
  @mutex.synchronize do
    while !@tasks.empty? || @stats[:in_flight].positive?
      remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
      break if remaining <= 0

      @drain_condition.wait(@mutex, remaining)
    end
  end
  nil
end

#empty? ⇒ Boolean

Whether there are no pending tasks waiting to be started.

In-flight tasks are not considered; use drain to wait for them.

Returns:

  • (Boolean) 


202
203
204
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 202

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

#on_complete {|result| ... } ⇒ self

Register a callback invoked after each successful task completion.

The callback receives the return value of the completed task.

Yields:

  • (result) —

    called on task success

Returns:

  • (self) 


73
74
75
76
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 73

def on_complete(&block)
  @mutex.synchronize { @complete_handler = block }
  self
end

#on_error {|exception, task, attempt| ... } ⇒ self

Register a callback invoked when a task raises an exception.

The callback receives the exception, the task that raised it, and the attempt number (1 for the first try, incrementing on each retry). It fires on every failed attempt, including the ones that are retried.

The callback may be registered at any time — even after tasks have been pushed — and is read live when the task runs.

Yields:

  • (exception, task, attempt) —

    called on each task failure

Returns:

  • (self) 


62
63
64
65
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 62

def on_error(&block)
  @mutex.synchronize { @error_handler = block }
  self
end

#pause ⇒ self

Pause the queue so workers stop dequeuing new tasks.

In-flight tasks will finish, but no new tasks will be picked up until resume is called.

Returns:

  • (self) 


95
96
97
98
99
100
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 95

def pause
  @mutex.synchronize do
    @paused = true
  end
  self
end

#paused? ⇒ Boolean

Whether the queue is currently paused.

Returns:

  • (Boolean) 


116
117
118
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 116

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

#push(callable = nil, priority: 0) { ... } ⇒ self Also known as: <<

Enqueue a task to be processed asynchronously.

Higher-priority tasks are dequeued before lower-priority ones; tasks that share a priority run in FIFO (insertion) order. The << alias always enqueues at the default priority of 0.

Parameters:

  • callable (#call, nil) (defaults to: nil) —

    a callable object (used by <<)

  • priority (Integer) (defaults to: 0) —

    dequeue priority; higher runs first (default 0)

Yields:

  • the block to execute (takes precedence over callable)

Returns:

  • (self)

Raises:

  • (ArgumentError) 


173
174
175
176
177
178
179
180
181
182
183
184
185
186
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 173

def push(callable = nil, priority: 0, &block)
  task = block || callable
  raise ArgumentError, 'a block is required' unless task

  @mutex.synchronize do
    raise 'queue is shut down' unless @running

    start_workers unless @started
    @tasks << Entry.new(task, priority, 0)
    @condition.signal
  end

  self
end

#resume ⇒ self

Resume a paused queue, waking workers to continue processing.

Returns:

  • (self) 


105
106
107
108
109
110
111
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 105

def resume
  @mutex.synchronize do
    @paused = false
    @pause_condition.broadcast
  end
  self
end

#running? ⇒ Boolean

Whether the queue is accepting new tasks.

Returns:

  • (Boolean) 


219
220
221
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 219

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

#shutdown(timeout: 30) ⇒ void

This method returns an undefined value.

Gracefully shut down the queue.

Signals all workers to finish their current task and drain remaining tasks, then waits up to timeout seconds for threads to exit.

Parameters:

  • timeout (Numeric) (defaults to: 30) —

    seconds to wait for workers to finish



230
231
232
233
234
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 230

def shutdown(timeout: 30)
  signal_shutdown
  wait_for_workers(timeout)
  nil
end

#size ⇒ Integer

Number of pending (not yet started) tasks.

Returns:

  • (Integer) 


193
194
195
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 193

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

#stats ⇒ Hash{Symbol => Integer}

Return statistics about processed tasks.

Returns:

  • (Hash{Symbol => Integer}) —

    counts for :completed, :failed, :pending, :in_flight, and :retried (total retry attempts made)



82
83
84
85
86
87
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 82

def stats
  @mutex.synchronize do
    { completed: @stats[:completed], failed: @stats[:failed], pending: @tasks.size,
      in_flight: @stats[:in_flight], retried: @stats[:retried] }
  end
end

#stats_reset! ⇒ self

Atomically zero the completed and failed counters.

Leaves pending, in_flight, worker threads, and registered callbacks untouched. Useful for resetting metrics between reporting intervals without shutting down the pool.

Returns:

  • (self) 


138
139
140
141
142
143
144
 
# File 'lib/philiprehberger/task_queue/queue.rb', line 138

def stats_reset!
  @mutex.synchronize do
    @stats[:completed] = 0
    @stats[:failed] = 0
  end
  self
end