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 caller in completion order. After all workers join, the first cleanup error, or otherwise the first input-order error, is raised.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(max_concurrency: 8, runner: TaskRunner.new, wait_through_interruptions: false) ⇒ Executor

Sets the maximum number of batch workers and their execution policy.

Raises:

  • (ArgumentError) 


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

def initialize(max_concurrency: 8, runner: TaskRunner.new, wait_through_interruptions: false)
  raise ArgumentError, "max_concurrency must be at least 1" if max_concurrency < 1

  @max_concurrency = max_concurrency
  @runner = runner
  @wait_through_interruptions = wait_through_interruptions
end

Instance Attribute Details

#runner ⇒ Object (readonly)

:nodoc:



22
23
24
 
# File 'lib/little_ghost/support/executor.rb', line 22

def runner
  @runner
end

Class Method Details

.blocking ⇒ Object

:nodoc:



152
 
# File 'lib/little_ghost/support/executor.rb', line 152

def blocking = BLOCKING # :nodoc:

Instance Method Details

#call(&work) ⇒ Object

Runs one unit of work and returns its result.



30
31
32
 
# File 'lib/little_ghost/support/executor.rb', line 30

def call(&work)
  resolve(submit(&work))
end

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

Maps values with at most the configured number of workers.



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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/little_ghost/support/executor.rb', line 44

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
  workers = []
  begin
    worker_count.times do
      workers << submit 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
      ensure
        completions << :worker_finished
      end
    end
  rescue
    cancellation_token.cancel
    workers.each(&:wait)
    raise
  end
  callback_error = nil
  begin
    completed_items = 0
    finished_workers = 0
    until completed_items == items.length || finished_workers == workers.length
      completion = completions.pop
      if completion == :worker_finished
        finished_workers += 1
        next
      end

      completed_items += 1
      begin
        on_result.call(completion, results[completion]) if on_result && !errors[completion]
      rescue => error
        callback_error = error
        cancellation_token.cancel
        break
      end
    end
  ensure
    workers.each(&:wait)
  end

  raise callback_error if callback_error
  task_error = workers.filter_map(&:error).first
  raise task_error if task_error

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

  results
end

#submit(&work) ⇒ Object

Submits one unit of work and returns its Task.



25
26
27
 
# File 'lib/little_ghost/support/executor.rb', line 25

def submit(&work)
  @runner.spawn(&work)
end

#try_call(&work) ⇒ Object

Runs one unit of work only when runner capacity is immediately available. Returns an accepted flag and the result.



36
37
38
39
40
41
 
# File 'lib/little_ghost/support/executor.rb', line 36

def try_call(&work)
  task = @runner.try_spawn(&work)
  return [false, nil] unless task

  [true, resolve(task)]
end