Class: TurboTests::Runner

Inherits:
Object
  • Object
show all
Defined in:
lib/turbo_tests/runner.rb

Constant Summary collapse

DEFAULT_RUNTIME_LOG = 
"tmp/turbo_rspec_runtime.log"
DEFAULT_WORKER_OUTPUT_MODE = 
:warnings
WORKER_OUTPUT_MODES = 
%i[warnings stream buffered quiet].freeze
READER_DRAIN_TIMEOUT =

Shared wall-clock budget, in seconds, for draining a worker’s stdout and
stderr readers after the child has exited. Normally the pipe EOFs in
milliseconds, so this is not on the hot path; the bound exists only for a
wedged pipe whose write end is still held open (e.g. by a grandchild),
which can never EOF. Shared across both streams so teardown cannot take
twice the budget.

1
FORCE_STOP_JOIN =

Short join used only after the pipe is force-closed, to give the reader a
moment to notice EOF before it is killed.

0.1

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(**opts) ⇒ Runner

Returns a new instance of Runner.



237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
 
# File 'lib/turbo_tests/runner.rb', line 237

def initialize(**opts)
  @formatters = opts[:formatters]
  @reporter = opts[:reporter]
  @files = opts[:files]
  @test_selectors = opts[:test_selectors] || {}
  @tags = opts[:tags]
  @verbose = opts[:verbose]
  @fail_fast = opts[:fail_fast]
  @start_time = opts[:start_time]
  @count = opts[:count]
  @seed = opts[:seed]
  @seed_used = opts[:seed_used]
  @order = opts[:order]
  @nice = opts[:nice]
  @use_runtime_info = opts[:use_runtime_info]

  @load_time = 0
  @load_count = 0
  @failure_count = 0
  # 4.0 default: kill behavior on (900s of total worker silence). Set
  # TURBO_TESTS2_WATCHDOG_SECONDS=0 to disable explicitly (a long-hang
  # run then ends via whatever outer watcher exists). See README
  # "Stall watchdog".
  @watchdog_seconds = Float(ENV.fetch("TURBO_TESTS2_WATCHDOG_SECONDS", "900"))
  @watchdog_warned = false
  @last_message_monotonic = nil

  # Supports runtime_log as a top level option,
  #   but also nested inside parallel_options
  @runtime_log = opts[:runtime_log] || DEFAULT_RUNTIME_LOG
  @parallel_options = opts.fetch(:parallel_options, {})
  @parallel_options[:runtime_log] ||= @runtime_log
  @record_runtime = opts.key?(:record_runtime) ? opts[:record_runtime] : true

  @messages = Thread::Queue.new
  @threads = []
  @wait_threads = []
  @exited_process_ids = []
  @worker_output = Hash.new { |hash, process_id| hash[process_id] = {stdout: +"", stderr: +""} }
  @worker_output_mutex = Mutex.new
  @deferred_run_options_messages = Hash.new { |hash, message| hash[message] = [] }
  @error = false
  @closed_process_ids = {}
  @failed_worker_statuses = {}
  @print_failed_group = opts[:print_failed_group]
  @worker_output_mode = self.class.normalize_worker_output_mode(opts.fetch(:worker_output, DEFAULT_WORKER_OUTPUT_MODE))
end

Class Method Details

.create(count) ⇒ Object 



32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/turbo_tests/runner.rb', line 32

def create(count)
  # Rails bootstrap: prepare the database once per worker, in parallel.
  # Implemented directly with Process.spawn so the runner does not
  # depend on parallel_tests for anything.
  count = RuntimeGroups.process_count(count)
  pids = (1..count).map do |process_id|
    Process.spawn(
      {"TEST_ENV_NUMBER" => process_id.to_s, "PARALLEL_TEST_GROUPS" => count.to_s},
      "bundle", "exec", "rake", "db:create", "RAILS_ENV=#{ENV.fetch("RAILS_ENV", "test")}"
    )
  end
  statuses = pids.map { |pid| Process.wait2(pid).last }
  statuses.all?(&:success?) ? 0 : 1
end

.generate_seed ⇒ Object 



172
173
174
 
# File 'lib/turbo_tests/runner.rb', line 172

def generate_seed
  (Random.new_seed % 65_535).to_s
end

.normalize_order(order) ⇒ Object

Raises:

  • (ArgumentError) 


164
165
166
167
168
169
170
 
# File 'lib/turbo_tests/runner.rb', line 164

def normalize_order(order)
  order = order.to_s.strip.downcase
  return "random" if order.empty?
  return order if %w[random defined].include?(order)

  raise ArgumentError, "Unsupported order #{order.inspect}; use random or defined"
end

.normalize_rspec_file_selection(files) ⇒ Object 



124
125
126
127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/turbo_tests/runner.rb', line 124

def normalize_rspec_file_selection(files)
  selectors = {}
  files.each do |entry|
    file, selector = split_rspec_location(entry)
    selectors[file] ||= []
    if selector == file
      selectors[file] = [file]
    elsif !selectors[file].include?(file)
      selectors[file] << selector unless selectors[file].include?(selector)
    end
  end

  {files: selectors.keys, selectors: selectors}
end

.normalize_worker_output_mode(mode) ⇒ Object

Raises:

  • (ArgumentError) 


113
114
115
116
117
118
119
120
121
122
 
# File 'lib/turbo_tests/runner.rb', line 113

def normalize_worker_output_mode(mode)
  value = mode.to_s.strip.downcase.tr("-", "_")
  return DEFAULT_WORKER_OUTPUT_MODE if value.empty?

  normalized = value.to_sym
  return normalized if WORKER_OUTPUT_MODES.include?(normalized)

  raise ArgumentError,
    "Unsupported worker output mode #{mode.inspect}; expected one of: #{WORKER_OUTPUT_MODES.join(", ")}"
end

.project_rspec_options(root = Dir.pwd) ⇒ Object 



227
228
229
230
231
232
233
234
 
# File 'lib/turbo_tests/runner.rb', line 227

def project_rspec_options(root = Dir.pwd)
  %w[.rspec .rspec-local].flat_map do |path|
    option_file = File.join(root, path)
    next [] unless File.file?(option_file)

    Shellwords.split(File.read(option_file))
  end
end

.rspec_configured_files_to_run ⇒ Object 



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
 
# File 'lib/turbo_tests/runner.rb', line 176

def rspec_configured_files_to_run
  configuration = RSpec::Core::Configuration.new
  RSpec::Core::ConfigurationOptions.new(["spec"]).configure(configuration)
  # Normalize against the cwd that owns the RSpec configuration being
  # read, NOT the nearest .rspec ancestor of each file. parallel_tests
  # stats these paths and spawns workers from this cwd, so the returned
  # paths must resolve from here. In a stacked monorepo the root .rspec
  # aggregates per-gem specs (each gem may also have its own .rspec);
  # walking to the nearest .rspec would strip those to bare
  # member-relative "spec/..." paths that do not exist at the root,
  # crashing File.stat during group sizing.
  #
  # Compare paths by filesystem identity (File.identical?) rather than by
  # string. On the Windows CI runner Dir.pwd reports the 8.3 short name
  # (C:/Users/RUNNER~1/...) while Dir.glob, which RSpec uses to expand
  # --pattern, reports the long name (C:/Users/runneradmin/...). Both
  # spellings name the same directory, but File.realpath does NOT expand
  # 8.3 names, so neither a string prefix nor Pathname#relative_path_from
  # can reconcile them. The same class of mismatch arises from a symlinked
  # ancestor. See TurboTests::Utils::Paths.
  root = File.expand_path(Dir.pwd)
  configuration.files_to_run.map do |path|
    TurboTests::Utils::Paths.relative_from(File.expand_path(path.to_s, root), root)
  end
end

.run(opts = {}) ⇒ Object 



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
 
# File 'lib/turbo_tests/runner.rb', line 47

def run(opts = {})
  default_file_discovery = !opts.key?(:files) || opts[:files].nil?
  files = default_file_discovery ? rspec_configured_files_to_run : opts[:files]
  file_selection = normalize_rspec_file_selection(files)
  formatters = opts[:formatters]
  tags = opts[:tags]
  parallel_options = opts[:parallel_options] || {}

  start_time = opts.fetch(:start_time) { RSpec::Core::Time.now }
  runtime_log = opts.fetch(:runtime_log, nil) || DEFAULT_RUNTIME_LOG
  example_status_log = opts.fetch(:example_status_log, nil)
  verbose = opts.fetch(:verbose, false)
  fail_fast = opts.fetch(:fail_fast, nil)
  count = opts.fetch(:count, nil)
  order = normalize_order(opts.fetch(:order, nil))
  seed = opts.fetch(:seed, nil)
  seed_used = order != "defined"
  seed = generate_seed if seed_used && seed.nil?
  print_failed_group = opts.fetch(:print_failed_group, false)
  nice = opts.fetch(:nice, false)
  worker_output = normalize_worker_output_mode(
    opts[:worker_output] || ENV["TURBO_TESTS2_WORKER_OUTPUT"]
  )

  use_runtime_info = default_file_discovery && !Gem.win_platform?
  record_runtime = !Gem.win_platform?
  parallel_options[:runtime_log] ||= runtime_log

  if example_status_log
    runtime_log = runtime_log_from_example_status(example_status_log)
    parallel_options[:runtime_log] = runtime_log
  elsif use_runtime_info
    parallel_options[:runtime_log] ||= runtime_log
  else
    parallel_options[:group_by] ||= :filesize
  end
  parallel_options[:group_by] ||= :filesize if parallel_options[:only_group]

  warn("VERBOSE") if verbose

  reporter = Reporter.from_config(formatters, start_time, seed, seed_used, files, parallel_options)

  new(
    reporter: reporter,
    formatters: formatters,
    start_time: start_time,
    files: file_selection.fetch(:files),
    test_selectors: file_selection.fetch(:selectors),
    tags: tags,
    runtime_log: runtime_log,
    example_status_log: example_status_log,
    verbose: verbose,
    fail_fast: fail_fast,
    count: count,
    seed: seed,
    seed_used: seed_used,
    order: order,
    print_failed_group: print_failed_group,
    use_runtime_info: use_runtime_info,
    parallel_options: parallel_options,
    nice: nice,
    worker_output: worker_output,
    record_runtime: record_runtime
  ).run
end

.runtime_log_from_example_status(example_status_log) ⇒ Object 



149
150
151
152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/turbo_tests/runner.rb', line 149

def runtime_log_from_example_status(example_status_log)
  statuses = RSpec::Core::ExampleStatusPersister.load_from(example_status_log)
  runtimes = statuses.each_with_object(Hash.new(0.0)) do |status, sums|
    next unless status.fetch(:status).match?(/pass/i)

    file_name = RSpec::Core::Example.parse_id(status.fetch(:example_id)).first
    sums[file_name] += status.fetch(:run_time).to_s[/\d+(\.\d+)?/].to_f
  end

  path = File.join("tmp", "turbo_tests2_example_status_runtime.log")
  FileUtils.mkdir_p(File.dirname(path))
  File.write(path, runtimes.sort.map { |file, runtime| "#{file}:#{runtime}" }.join("\n"))
  path
end

.split_rspec_location(entry) ⇒ Object 



139
140
141
142
143
144
145
146
147
 
# File 'lib/turbo_tests/runner.rb', line 139

def split_rspec_location(entry)
  value = entry.to_s
  return [value, value] if File.exist?(value)

  match = value.match(/\A(.+):(\d+)\z/)
  return [value, value] unless match && File.exist?(match[1])

  [match[1], value]
end

.worker_spec_opts(spec_opts) ⇒ Object 



202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
 
# File 'lib/turbo_tests/runner.rb', line 202

def worker_spec_opts(spec_opts)
  args = project_rspec_options + Array(spec_opts)
  filtered = []
  skip_next = false

  args.each do |arg|
    if skip_next
      skip_next = false
      next
    end

    case arg
    when "--pattern", "-P", "--default-path", "--format", "-f", "--out", "-o"
      skip_next = true
    when /\A--pattern=/, /\A-P.+/, /\A--default-path=/,
      /\A--format=/, /\A-f.+/, /\A--out=/, /\A-o.+/
      next
    else
      filtered << arg
    end
  end

  filtered
end

Instance Method Details

#run ⇒ Object 



285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
 
# File 'lib/turbo_tests/runner.rb', line 285

def run
  parallel_tests_options = @parallel_options.reject { |key, _value| key == :only_group }
  @num_processes = [
    RuntimeGroups.process_count(@count),
    @files.size
  ].min

  if @num_processes.zero?
    @tests_in_groups = []
    @reporter.report([]) { |_reporter| }
    return 0
  end

  tests_in_groups =
    RuntimeGroups::Grouping.in_groups(
      @files,
      @num_processes,
      group_by: parallel_tests_options[:group_by] || :runtime,
      runtime_log: parallel_tests_options[:runtime_log],
      allowed_missing: (parallel_tests_options[:allowed_missing_percent] || 100).to_f / 100,
      unknown_runtime: parallel_tests_options[:unknown_runtime]
    )
  tests_in_groups = selected_groups(tests_in_groups) if @parallel_options[:only_group]
  @tests_in_groups = tests_in_groups

  subprocess_opts = {
    record_runtime: @record_runtime
  }

  pid_file_was_set = ENV.key?("PARALLEL_PID_FILE")
  pid_file_before = ENV["PARALLEL_PID_FILE"]
  begin
    RuntimeGroups::PidFile.with do |pid_path|
      ENV["PARALLEL_PID_FILE"] = pid_path

      exit_status = nil
      report_coverage = false

      @reporter.report(tests_in_groups) do |_reporter|
        old_signal = Signal.trap(:INT) { handle_interrupt }

        @wait_threads = tests_in_groups.map.with_index do |tests, process_id|
          start_regular_subprocess(tests, process_id + 1, **subprocess_opts)
        end.compact
        @interrupt_handled = false

        handle_messages

        if @watchdog_fired
          # Workers were SIGKILLed by the watchdog; their reader threads end
          # at EOF and waiters return as the OS reaps them — but if a kill
          # failed (permissions, zombie, pid reuse) an unbounded join here
          # would hang the run anyway, defeating the watchdog. Bound every
          # join and treat any worker that still has not exited as failed.
          @threads.each do |thread|
            thread.join(FORCE_STOP_JOIN)
            thread.kill if thread.alive?
          end
          @wait_threads.each do |wait_thr|
            wait_thr.join(FORCE_STOP_JOIN)
          end
          flush_worker_output unless stream_worker_output?
          exit_status = 1
        else
          @threads.each(&:join)

          report_workers_without_results

          report_failed_group(tests_in_groups) if @print_failed_group

          Signal.trap(:INT, old_signal)

          statuses = @wait_threads.map(&:value)

          if @reporter.failed_examples.empty? && statuses.all?(&:success?)
            flush_successful_worker_output
            report_coverage = true
            exit_status = 0
          else
            flush_worker_output unless stream_worker_output?
            # From https://github.com/galtzo-floss/turbo_tests2/pull/20/
            exit_status = statuses.map(&:exitstatus).max
          end
        end
      end

      flush_coverage_summary if report_coverage
      exit_status
    end
  ensure
    if pid_file_was_set
      ENV["PARALLEL_PID_FILE"] = pid_file_before
    else
      ENV.delete("PARALLEL_PID_FILE")
    end
  end
end