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
73
74
75
76
77
78
79
80
81
82
83
84
 
# 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)
  error_mutex = Mutex.new
  first_worker_error = nil
  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
            error_mutex.synchronize { first_worker_error ||= error }
            cancellation_token.cancel
          ensure
            completions << index
          end
        end
      end
    end
  end
  callback_error = nil
  begin
    items.length.times do
      index = completions.pop
      begin
        on_result.call(index, results[index]) if on_result && !errors[index]
      rescue => error
        callback_error = error
        cancellation_token.cancel
        break
      end
    end
  ensure
    workers.each(&:join)
  end

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

  results
end