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

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) ⇒ Array<Result>

Returns one entry per job, aligned to the input order.

Parameters:

  • jobs (Array<#call>) —

    each job is invoked with no arguments

Returns:

  • (Array<Result>) —

    one entry per job, aligned to the input order



30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/clacky/fanout.rb', line 30

def run(jobs)
  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
    Thread.new { drain(pending, results, deadline) }
  end
  join_all(workers, deadline)

  fill_unfinished(results)
end