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 
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.

5

Instance Method Summary collapse

Constructor Details

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



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

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

Instance Method Details

#discover ⇒ Object 



71
72
73
74
75
76
77
78
79
80
 
# File 'lib/binpacker/orchestrator.rb', line 71

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.



280
281
282
283
284
285
286
287
288
289
290
291
 
# File 'lib/binpacker/orchestrator.rb', line 280

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 } 



293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
 
# File 'lib/binpacker/orchestrator.rb', line 293

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).



267
268
269
270
271
272
273
274
275
 
# File 'lib/binpacker/orchestrator.rb', line 267

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 



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

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 



318
319
320
321
322
323
324
 
# File 'lib/binpacker/orchestrator.rb', line 318

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 } 



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

def run
  tests = discover
  timing = Timing.new(@config.timing_file)
  # Weights are computed over the WHOLE suite before slicing, so every shard cuts against the same
  # numbers and the partition they agree on covers every test exactly once.
  timings = timing.load_with_fallback(tests)
  @timings = timings

  scheduler = Scheduler.for(@config.scheduler['algorithm'])
  @discovered_count = tests.size
  tests = @shard.select(tests: tests, timings: timings, scheduler: scheduler) if @shard
  @selected_count = tests.size

  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 



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
237
238
239
240
241
242
243
 
# File 'lib/binpacker/orchestrator.rb', line 136

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 



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
128
129
130
131
132
133
134
 
# File 'lib/binpacker/orchestrator.rb', line 82

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 



245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
 
# File 'lib/binpacker/orchestrator.rb', line 245

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,
    shard: @shard,
    discovered: @discovered_count,
    selected: @selected_count
  ).write(@report_path)
end