Class: Nexo::Concurrent

Inherits:
Object
  • Object
show all
Defined in:
lib/nexo/concurrent.rb

Overview

Bounded fan-out driver for running many agent/workflow calls concurrently on Ruby's fiber scheduler (Spec 5). It is the value Nexo adds over a raw Async {} block: rate-bounded concurrency (so you stay under provider rate limits) with proper error propagation (the first task failure is re-raised, never swallowed) and results returned in submission order.

results = Nexo.concurrent(max_in_flight: 8) do |c|
docs.each { |d| c.add { SummarizeDocument.run(text: d.body).result } }
end
# => one result per doc, in doc order, never more than 8 calls in flight

async is a SOFT (optional) dependency: it is required lazily the moment a fan-out actually runs. With the gem absent, require "nexo" still loads cleanly; only calling Nexo.concurrent raises MissingDependencyError.

Defined Under Namespace

Classes: Task

Instance Method Summary collapse

Constructor Details

#initialize(max_in_flight:) ⇒ Concurrent

max_in_flight bounds how many submitted blocks run concurrently once the collector is run inside an async reactor.



25
26
27
28
 
# File 'lib/nexo/concurrent.rb', line 25

def initialize(max_in_flight:)
  @max_in_flight = max_in_flight
  @tasks = []
end

Instance Method Details

#add(&block) ⇒ Object

Registers a block to run. Called via the yielded collector inside Nexo.concurrent. Order of add calls is the order of the results array.



32
33
34
 
# File 'lib/nexo/concurrent.rb', line 32

def add(&block)
  @tasks << Task.new(block)
end

#run ⇒ Object

Runs every added block inside ONE reactor, bounding in-flight work with an Async::Semaphore and coordinating with an Async::Barrier. Results are index-assigned, so the returned Array is in submission order regardless of completion order. On the first task that raises, the error propagates out of barrier.wait and the ensure stops the remaining in-flight tasks — errors are never swallowed.



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/nexo/concurrent.rb', line 42

def run
  require_async!
  results = Array.new(@tasks.size)

  Async do |task|
    semaphore = Async::Semaphore.new(@max_in_flight, parent: task)
    barrier = Async::Barrier.new(parent: semaphore)

    @tasks.each_with_index do |t, i|
      barrier.async { results[i] = t.block.call }
    end

    # barrier.wait re-raises the first task failure (no silent swallowing).
    barrier.wait
  ensure
    barrier&.stop
  end.wait

  results
end