Class: Clacky::Fanout

Inherits:
Object
  • Object
show all
Defined in:
lib/clacky/fanout.rb

Overview

Runs a batch of blocking jobs on a bounded thread pool and collects their results in the order the jobs were given, regardless of completion order.

Built for subagent fan-out: each job blocks inside Agent#run, so the pool is sized for concurrency rather than CPU count. A job that raises is captured as a failed slot instead of tearing down its siblings.

Defined Under Namespace

Classes: Result

Constant Summary collapse

DEFAULT_MAX_CONCURRENCY = 
4
INTERRUPT_GRACE_SECONDS =

Forward an interrupt to workers still running a job so they unwind through their own ensure blocks, then give them a bounded window to finish. A worker that ignores the raise is killed so the caller isn't held hostage.

5.0

Instance Method Summary collapse

Constructor Details

#initialize(max_concurrency: DEFAULT_MAX_CONCURRENCY, timeout: nil) ⇒ Fanout

Returns a new instance of Fanout.

Parameters:

  • max_concurrency (Integer) (defaults to: DEFAULT_MAX_CONCURRENCY) —

    jobs allowed to run at once

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

    wall-clock budget for the whole batch

Raises:

  • (ArgumentError) 


21
22
23
24
25
26
 
# File 'lib/clacky/fanout.rb', line 21

def initialize(max_concurrency: DEFAULT_MAX_CONCURRENCY, timeout: nil)
  raise ArgumentError, "max_concurrency must be positive" unless max_concurrency.to_i.positive?

  @max_concurrency = max_concurrency.to_i
  @timeout = timeout
end

Instance Method Details

#run(jobs, on_cancel: nil) ⇒ Array<Result>

If the calling thread is interrupted (e.g. Thread#raise AgentInterrupted when a newer task supersedes this one) while waiting on the workers, the interrupt is forwarded to every live worker so their jobs — subagents that would otherwise keep calling the LLM on detached threads — unwind through their own ensure blocks (phase_end, transcript capture). The optional on_cancel callback fires first so callers can flip a shared cancel flag.

Parameters:

  • jobs (Array<#call>) —

    each job is invoked with no arguments

Returns:

  • (Array<Result>) —

    one entry per job, aligned to the input order



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
# File 'lib/clacky/fanout.rb', line 37

def run(jobs, on_cancel: nil)
  return [] if jobs.empty?

  pending = build_queue(jobs)
  results = Array.new(jobs.size)
  deadline = @timeout && (monotonic_now + @timeout)

  workers = Array.new([@max_concurrency, jobs.size].min) do
    Clacky::ThreadRegistry.spawn(name: "fanout-worker") { drain(pending, results, deadline) }
  end

  begin
    join_all(workers, deadline)
  rescue Exception => e # rubocop:disable Lint/RescueException
    # Includes AgentInterrupted. The cleanup below must not itself be cut
    # short by a second asynchronous Thread#raise, so defer any further
    # interrupts on this thread until the workers have been unwound.
    Thread.handle_interrupt(Object => :never) do
      on_cancel&.call
      drain_queue(pending)
      interrupt_workers(workers)
    end
    raise e
  end

  fill_unfinished(results)
end