Class: Binpacker::Orchestrator

Inherits:
Object
  • Object
show all
Defined in:
lib/binpacker/orchestrator.rb,
sig/binpacker/orchestrator.rbs

Constant Summary collapse

MIN_BATCH_WEIGHT =

Dynamic batches halve the queue's remaining predicted weight, floored at a per-run minimum: early batches are large to amortize the per-batch test-runner boot (each batch is a fresh rspec process), tail batches stay small so workers finish together.

The floor comes in two regimes (computed by #min_batch_weight in #run and threaded into #drain_batch):

- Calibrated: weights are measured seconds, so the floor is a
fixed MIN_BATCH_WEIGHT (~30s), an order of magnitude above the
~3s runner boot — boot overhead stays in the noise.
- Cold start: weights are filesize KB, not seconds, so a fixed
30 is meaningless. We instead derive a floor that targets
~COLD_START_BATCHES_PER_WORKER batches per worker (below).

Returns:

  • (Float) 
30.0
COLD_START_BATCHES_PER_WORKER =

On a pure cold start predicted weights are filesize KB, not seconds, so the 30s MIN_BATCH_WEIGHT floor has no meaning. Instead we size the floor so the total predicted weight divides into about this many batches per worker — matching the ~5 runner boots per worker that gave the best makespan in the PR #12 simulation.

Returns:

  • (Integer) 
5

Instance Method Summary collapse

Constructor Details

#initialize(config, passthrough: [], quiet: false, report_path: nil) ⇒ Orchestrator

Returns a new instance of Orchestrator.

Parameters:

  • (Object)
  • passthrough: (Object) (defaults to: [])
  • quiet: (Object) (defaults to: false)
  • report_path: (Object) (defaults to: nil) 


27
28
29
30
31
32
 
# File 'lib/binpacker/orchestrator.rb', line 27

def initialize(config, passthrough: [], quiet: false, report_path: nil)
  @config = config
  @passthrough = passthrough
  @quiet = quiet
  @report_path = report_path
end

Instance Method Details

#discover ⇒ Object

Returns:

  • (Object) 


64
65
66
67
68
69
70
71
72
73
 
# File 'lib/binpacker/orchestrator.rb', line 64

def discover
  case @config.test_runner
  when 'rspec'
    RSpecDiscovery.new(@config).enumerate
  when 'minitest'
    MinitestDiscovery.new(@config).enumerate
  else
    raise ConfigError, "unsupported runner: #{@config.test_runner}"
  end
end

#drain_batch(queue, floor) ⇒ Array[untyped]

Drains the next batch off queue: enough tests to halve its remaining predicted weight, but never less than floor (see #min_batch_weight) and never fewer than one test.

Parameters:

Returns:

  • (Array[untyped]) 


270
271
272
273
274
275
276
277
278
279
280
281
 
# File 'lib/binpacker/orchestrator.rb', line 270

def drain_batch(queue, floor)
  return [] if queue.nil? || queue.empty?

  target = [queue.total_weight(@timings) / 2.0, floor].max
  batch = []
  weight = 0.0
  while (batch.empty? || weight < target) && (test = queue.pop)
    batch << test
    weight += @timings.fetch(test.key, Timing::DEFAULT_WEIGHT)
  end
  batch
end

#finalize(timing, all_timings, all_passed, total_examples, passed_examples, tests) ⇒ { passed: untyped, total: untyped, passed_count: untyped, timings: Array[untyped], empty_filter: untyped }

Parameters:

Returns:

  • ({ passed: untyped, total: untyped, passed_count: untyped, timings: Array[untyped], empty_filter: untyped }) 


283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
 
# File 'lib/binpacker/orchestrator.rb', line 283

def finalize(timing, all_timings, all_passed, total_examples, passed_examples, tests)
  unless all_timings.empty?
    timing.append_all(all_timings)
    timing.compact!
  end
  empty_filter = minitest_empty_filter?(tests, total_examples)
  all_passed = false if empty_filter

  {
    passed: all_passed,
    total: total_examples,
    passed_count: passed_examples,
    timings: all_timings,
    empty_filter: empty_filter
  }
end

#min_batch_weight(timing, timings) ⇒ Float

Per-run batch-weight floor. Calibrated runs use the fixed seconds-scale MIN_BATCH_WEIGHT. On a cold start weights are KB, so that floor is meaningless; derive one that splits the total predicted weight into ~COLD_START_BATCHES_PER_WORKER batches per worker, falling back to MIN_BATCH_WEIGHT when there is nothing to divide (zero total weight or zero workers).

Parameters:

Returns:

  • (Float) 


257
258
259
260
261
262
263
264
265
 
# File 'lib/binpacker/orchestrator.rb', line 257

def min_batch_weight(timing, timings)
  return MIN_BATCH_WEIGHT if timing.calibrated?

  total = timings.values.sum(0.0)
  workers = @config.worker_count
  return MIN_BATCH_WEIGHT if total.zero? || workers.zero?

  total / (workers * COLD_START_BATCHES_PER_WORKER)
end

#minitest_empty_filter?(tests, total_examples) ⇒ Boolean

Parameters:

  • (Object)
  • (Object)

Returns:

  • (Boolean) 


300
301
302
303
304
305
306
 
# File 'lib/binpacker/orchestrator.rb', line 300

def minitest_empty_filter?(tests, total_examples)
  return false unless @config.test_runner == 'minitest'
  return false unless tests.any?
  return false unless total_examples.zero?

  minitest_include_filter?
end

#minitest_include_filter? ⇒ Boolean

Returns:

  • (Boolean) 


308
309
310
311
312
313
314
 
# File 'lib/binpacker/orchestrator.rb', line 308

def minitest_include_filter?
  @passthrough.any? do |arg|
    %w[--name --include -n -i].include?(arg) ||
      arg.start_with?('--name=', '--include=') ||
      (arg.start_with?('-n', '-i') && arg.length > 2)
  end
end

#run ⇒ { passed: untyped, total: untyped, passed_count: untyped, timings: Array[untyped], empty_filter: untyped }

Returns:

  • ({ passed: untyped, total: untyped, passed_count: untyped, timings: Array[untyped], empty_filter: untyped }) 


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
 
# File 'lib/binpacker/orchestrator.rb', line 34

def run
  tests = discover
  timing = Timing.new(@config.timing_file)
  timings = timing.load_with_fallback(tests)
  @timings = timings

  scheduler = Scheduler.for(@config.scheduler['algorithm'])
  queues = scheduler.partition(
    tests: tests,
    worker_count: @config.worker_count,
    timings: timings
  )

  # Capture predicted per-Worker loads before execution drains the queues.
  @predicted_loads = queues.map { |q| q.total_weight(timings) }

  runner_class = TestRunner.for(@config.test_runner)
  workers = queues.map.with_index do |_queue, idx|
    Worker.new(idx, runner_class, passthrough: @passthrough, quiet: @quiet).tap(&:start)
  end

  if @config.scheduler['steal_enabled']
    run_dynamic(workers, queues, timing, tests, min_batch_weight(timing, timings))
  else
    run_static(workers, queues, timing, tests)
  end
end

#run_dynamic(workers, queues, timing, tests, floor) ⇒ Object

Parameters:

Returns:

  • (Object) 


129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
 
# File 'lib/binpacker/orchestrator.rb', line 129

def run_dynamic(workers, queues, timing, tests, floor)
  all_timings = []
  all_passed = true
  active = []

  queue_totals = queues.map(&:size)
  worker_done = Array.new(workers.size, 0)
  batch_sizes = Array.new(workers.size, 0)
  worker_time = Array.new(workers.size, 0.0)
  worker_examples = Array.new(workers.size, 0)
  worker_passed = Array.new(workers.size, 0)

  progress = ProgressDisplay.new(workers.size)

  workers.zip(queues).each do |worker, queue|
    batch = drain_batch(queue, floor)
    if batch.empty?
      worker.signal_done
      worker.collect_results
      all_timings.concat(worker.timings)
      all_passed &&= worker.success?
      worker_examples[worker.id] = worker.example_count
      worker_passed[worker.id] = worker.passed_count
      worker.cleanup
      worker_done[worker.id] = queue_totals[worker.id]
      progress.update(worker.id, done: worker_done[worker.id], total: queue_totals[worker.id], file: 'done')
    else
      worker.send_tests(batch)
      worker.batch_done
      active << worker
      batch_sizes[worker.id] = batch.size
      current_file = batch.first&.file || ''
      progress.update(worker.id, done: 0, total: queue_totals[worker.id], file: current_file)
    end
  end

  until active.empty?
    ready = active.find { |w| w.wait_for_batch }
    unless ready
      active.reject! { |w| %i[crashed error].include?(w.status) }
      sleep 0.1
      next
    end

    begin
      all_passed &&= ready.success?

      worker_done[ready.id] += batch_sizes[ready.id]
      # Worker#example_count accumulates across batches, so these
      # are running totals per worker; the grand totals are summed
      # once after the loop, never added per batch.
      worker_examples[ready.id] = ready.example_count
      worker_passed[ready.id] = ready.passed_count

      own_queue = queues[ready.id]
      next_batch = drain_batch(own_queue, floor)

      if next_batch.empty?
        donor = queues.reject(&:empty?).max_by { |q| q.total_weight(@timings) }
        if donor
          next_batch = drain_batch(donor, floor)
          queue_totals[ready.id] += next_batch.size
          queue_totals[donor.worker_id] -= next_batch.size
        end
      end

      if next_batch.any?
        ready.send_tests(next_batch)
        ready.batch_done
        batch_sizes[ready.id] = next_batch.size
        current_file = next_batch.first&.file || ''
        progress.update(ready.id, done: worker_done[ready.id], total: queue_totals[ready.id], file: current_file)
        progress.refresh
      else
        ready.signal_done
        all_timings.concat(ready.timings)
        active.delete(ready)
        worker_done[ready.id] = queue_totals[ready.id]
        progress.update(ready.id, done: queue_totals[ready.id], total: queue_totals[ready.id], file: 'done')
        progress.refresh
      end
    rescue WorkerError => e
      warn "worker #{ready.id} error: #{e.message}"
      all_passed = false
      active.delete(ready)
    end
  end

  progress.finish

  total_examples = worker_examples.sum
  passed_examples = worker_passed.sum

  worker_stats = workers.map.with_index do |w, i|
    tw = w.timings.sum { |t| t[:time] }
    {
      files: queue_totals[i],
      total_time: tw > 0 ? tw : worker_time[i],
      examples: worker_examples[i],
      passed: worker_passed[i]
    }
  end
  progress.summary(worker_stats)
  write_report(worker_stats, all_timings)

  workers.each(&:cleanup)
  finalize(timing, all_timings, all_passed, total_examples, passed_examples, tests)
end

#run_static(workers, queues, timing, tests) ⇒ Object

Parameters:

Returns:

  • (Object) 


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
124
125
126
127
 
# File 'lib/binpacker/orchestrator.rb', line 75

def run_static(workers, queues, timing, tests)
  queue_totals = queues.map(&:size)
  progress = ProgressDisplay.new(workers.size)

  workers.zip(queues).each do |worker, queue|
    worker.send_tests(queue.remaining)
    progress.update(worker.id, done: 0, total: queue_totals[worker.id], file: queue.remaining.first&.file || '')
  end
  progress.refresh

  workers.each(&:signal_done)

  all_timings = []
  all_passed = true
  total_examples = 0
  passed_examples = 0
  worker_time = Array.new(workers.size, 0.0)
  worker_examples = Array.new(workers.size, 0)
  worker_passed = Array.new(workers.size, 0)

  workers.each do |worker|
    worker.collect_results
    all_timings.concat(worker.timings)
    all_passed &&= worker.success?
    total_examples += worker.example_count
    passed_examples += worker.passed_count
    worker_examples[worker.id] = worker.example_count
    worker_passed[worker.id] = worker.passed_count
    worker_time[worker.id] = worker.timings.sum { |t| t[:time] }
    progress.update(worker.id, done: queue_totals[worker.id], total: queue_totals[worker.id], file: 'done')
    progress.refresh
  rescue WorkerError => e
    warn "worker #{worker.id} error: #{e.message}"
    all_passed = false
  ensure
    worker.cleanup
  end

  progress.finish

  worker_stats = workers.map.with_index do |_w, i|
    {
      files: queue_totals[i],
      total_time: worker_time[i],
      examples: worker_examples[i],
      passed: worker_passed[i]
    }
  end
  progress.summary(worker_stats)
  write_report(worker_stats, all_timings)

  finalize(timing, all_timings, all_passed, total_examples, passed_examples, tests)
end

#write_report(worker_stats, all_timings) ⇒ void

This method returns an undefined value.

Parameters:

  • (Object)
  • (Array[untyped]) 


238
239
240
241
242
243
244
245
246
247
248
249
 
# File 'lib/binpacker/orchestrator.rb', line 238

def write_report(worker_stats, all_timings)
  return unless @report_path

  Report.new(
    profile: @config.profile,
    algorithm: @config.scheduler['algorithm'],
    predicted_loads: @predicted_loads,
    worker_stats: worker_stats,
    all_timings: all_timings,
    timings: @timings
  ).write(@report_path)
end