Class: Raif::Evals::WorkerPool

Inherits:
Object
  • Object
show all
Defined in:
lib/raif/evals/worker_pool.rb

Overview

Runs a list of work items across a bounded pool of threads.

An eval run is almost entirely waiting on provider HTTP responses, so overlapping the waiting is where the wall clock goes. Threads rather than fibers: Net::HTTP only yields to a fiber scheduler, and Raif's inference path is stateless per call (Raif.llm builds a fresh client per call, and nothing in Raif keeps per-thread state).

Defined Under Namespace

Classes: State

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(concurrency: 1) ⇒ WorkerPool

Returns a new instance of WorkerPool.



14
15
16
 
# File 'lib/raif/evals/worker_pool.rb', line 14

def initialize(concurrency: 1)
  @concurrency = [concurrency.to_i, 1].max
end

Instance Attribute Details

#concurrency ⇒ Object (readonly)

Returns the value of attribute concurrency.



12
13
14
 
# File 'lib/raif/evals/worker_pool.rb', line 12

def concurrency
  @concurrency
end

Instance Method Details

#run(items, &block) ⇒ Object

Runs the block once per item and returns its values in item order, whatever order the items actually completed in.

On Ctrl-C, workers stop before taking their next item and the in-flight ones are joined rather than killed, so their results still reach whatever the caller records them in. The Interrupt is then re-raised for the caller to report on.



24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/raif/evals/worker_pool.rb', line 24

def run(items, &block)
  # Not just an optimization: a serial run stays on the main thread, with no executor
  # wrapping and no second connection.
  return items.map(&block) if concurrency == 1 || items.size <= 1

  state = State.new(items: items)
  workers = [concurrency, items.size].min.times.map { start_worker(state, &block) }

  begin
    workers.each(&:join)
  rescue Interrupt
    state.stop!
    workers.each(&:join)
    raise
  end

  raise state.failure if state.failure

  state.results
end