Class: Riffer::Runner::Threaded

Inherits:
Riffer::Runner show all
Defined in:
lib/riffer/runner/threaded.rb,
sig/generated/riffer/runner/threaded.rbs

Overview

Processes items concurrently using a thread pool of up to max_concurrency workers pulling from a shared queue, so a slow item doesn't block others. If multiple workers raise, only the first exception is re-raised after all finish.

Constant Summary collapse

DEFAULT_MAX_CONCURRENCY =

Returns:

  • (Integer) 
5

Instance Method Summary collapse

Constructor Details

#initialize(max_concurrency: DEFAULT_MAX_CONCURRENCY) ⇒ Threaded

-- : (?max_concurrency: Integer) -> void

Parameters:

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


14
15
16
 
# File 'lib/riffer/runner/threaded.rb', line 14

def initialize(max_concurrency: DEFAULT_MAX_CONCURRENCY)
  @max_concurrency = max_concurrency
end

Instance Method Details

#map(items, context:) {|arg0| ... } ⇒ Array[untyped]

-- : (Array, context: Riffer::Agent::Context?) { (untyped) -> untyped } -> Array

Parameters:

Yields:

Yield Parameters:

  • arg0 (Object)

Yield Returns:

  • (Object)

Returns:

  • (Array[untyped]) 


20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/riffer/runner/threaded.rb', line 20

def map(items, context:, &block)
  return [] if items.empty?

  results = Array.new(items.size)
  errors = Array.new(items.size)
  queue = Queue.new
  items.each_with_index { |item, i| queue << [item, i] }

  workers = [items.size, @max_concurrency].min.times.map do
    Thread.new do
      loop do
        pair = begin
          queue.pop(true)
        rescue ThreadError
          break
        end
        item, index = pair
        begin
          results[index] = block.call(item)
        rescue => e
          errors[index] = e
        end
      end
    end
  end

  workers.each(&:join)

  first_error = errors.compact.first
  raise first_error if first_error

  results
end