Class: LittleGhost::Support::Executor

Inherits:
Object
  • Object
show all
Defined in:
lib/little_ghost/support/executor.rb

Overview

Executor runs independent work concurrently while preserving input order in the final results. It gives framework extensions bounded parallelism without losing cancellation or request-scoped state.

ExecutionState is copied to workers. on_result runs on the calling thread in completion order. After all workers join, the first cleanup error, or otherwise the first input-order error, is raised.

Instance Method Summary collapse

Constructor Details

#initialize(max_concurrency: 8) ⇒ Executor

Sets the maximum number of worker threads.

Raises:

  • (ArgumentError) 


14
15
16
17
18
 
# File 'lib/little_ghost/support/executor.rb', line 14

def initialize(max_concurrency: 8)
  raise ArgumentError, "max_concurrency must be at least 1" if max_concurrency < 1

  @max_concurrency = max_concurrency
end

Instance Method Details

#map(values, cancellation_token: CancellationToken.new, on_result: nil, &block) ⇒ Object

Maps values with at most the configured number of workers.



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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/little_ghost/support/executor.rb', line 21

def map(values, cancellation_token: CancellationToken.new, on_result: nil, &block)
  unless on_result.nil? || on_result.respond_to?(:call)
    raise ArgumentError, "on_result must be callable"
  end

  items = values.to_a
  return [] if items.empty?

  queue = Queue.new
  completions = Queue.new
  items.each_index { |index| queue << index }
  results = Array.new(items.length)
  errors = Array.new(items.length)
  worker_count = [@max_concurrency, items.length].min
  execution_state = ExecutionState.capture

  workers = worker_count.times.map do
    Thread.new do
      ExecutionState.with(execution_state) do
        loop do
          index = begin
            queue.pop(true)
          rescue ThreadError
            break
          end

          begin
            cancellation_token.raise_if_cancelled!
            results[index] = block.call(items[index])
          rescue => error
            errors[index] = error
          ensure
            completions << index
          end
        end
      end
    end
  end
  begin
    items.length.times do
      index = completions.pop
      on_result.call(index, results[index]) if on_result && !errors[index]
    end
  ensure
    workers.each(&:join)
  end

  first_error = errors.compact.find { |error| error.is_a?(CleanupError) } || errors.compact.first
  raise first_error if first_error

  results
end