From 2d67a100221c83797d58b3e894eb87cf0ee6da48 Mon Sep 17 00:00:00 2001 From: Daniel Westendorf Date: Thu, 30 Jul 2026 10:31:07 -0600 Subject: [PATCH 1/5] Add SimpleCov.parallel_collate for multi-process merge Add SimpleCov.parallel_collate to fan the merge out across processes Collating a large CI matrix's resultsets reads, parses and folds every one of them in sequence, and nearly all the wall clock goes into that fold. `SimpleCov.parallel_collate` is `SimpleCov.collate` with the fold spread across forked workers. Measured with `PROCESSES=N ruby benchmarks/collate.rb` (160 resultsets, 1836 files, 147,875 lines, 8,205 branch conditions, branch coverage enabled, 14 cores; store / format / thresholds skipped, since the fan-out only touches the merge phase): processes merge serial 8.53s 4 2.65s -68.9% 8 2.04s -76.1% It takes `collate`'s arguments plus a required `processes:`. The count is deliberately not clamped to the core count nor gated on a minimum number of resultsets - only the caller knows what a collate job is allowed to use - and asking for more processes than there are result files just gives one file per process. Below 1 it raises rather than quietly merging serially. The report is identical to `collate`'s for the same inputs, not merely equivalent. Each worker folds a *contiguous* slice and the parent folds the slices back in index order, so the resultsets are visited in the order the serial fold visits them. That matters because visiting order is observable: `MethodsCombiner` retains the first key it sees for a given source identity, so a round-robin split would have produced a report differing from `collate`'s in its method keys. Verified byte-identical against the serial fold over all 160 fixture resultsets at processes = 2, 3, 7, 8, 13, 160 and 400, and `features/test_unit_parallel_collate.feature` pins the same percentages the existing collate feature asserts. Notes: - Every failure path returns nil rather than a partial merge, and the caller redoes the fold serially: reporting coverage for a subset of the resultsets would silently understate it. That covers a runtime that cannot fork (JRuby, TruffleRuby, Windows - detected by the NotImplementedError the call raises, not by `respond_to?`), a worker that died, and a payload that came back truncated. - Workers ship their folded pair back over a pipe, deserialized on a thread per worker so every pipe is drained while the workers are still writing. A payload larger than the pipe buffer would otherwise block its worker mid-write, and the parent would block reaping a worker that can never finish. - A worker folds its slice one file at a time, as `merge_results` does, so memory scales with the worker count rather than the resultset count. - Children end with `exit!` so they never fall through to the collating process's at_exit handlers. `run_worker` returns the status rather than exiting itself, which keeps it exercisable in-process. - `collate` and `parallel_collate` move to `lib/simplecov/collation.rb`, sharing the validate / configure / finalize scaffolding. `collate` is unchanged, including that it still merges via `ResultMerger.merge_and_store`. - `benchmarks/collate.rb` gains a `PROCESSES` knob so a parallel run can be compared against a serial baseline. --- CHANGELOG.md | 1 + README.md | 45 +++ benchmarks/collate.rb | 3 + benchmarks/collate/cli.rb | 22 +- benchmarks/collate/report.rb | 10 +- benchmarks/collate/runner.rb | 32 ++- features/test_unit_parallel_collate.feature | 42 +++ lib/simplecov.rb | 2 + lib/simplecov/collation.rb | 69 +++++ lib/simplecov/parallel_result_merger.rb | 201 +++++++++++++ lib/simplecov/result_merger.rb | 35 ++- lib/simplecov/result_processing.rb | 28 +- .../simplecov/parallel_result_merger.rbs | 61 ++++ sig/internal/simplecov/result_merger.rbs | 4 + sig/simplecov.rbs | 28 +- spec/combine_differential_spec.rb | 109 ++++++++ spec/parallel_result_merger_spec.rb | 263 ++++++++++++++++++ spec/result_merger_spec.rb | 27 ++ spec/simplecov_spec.rb | 82 ++++++ spec/support/merge_fuzzer.rb | 119 ++++++++ spec/support/merge_reference.rb | 109 ++++++++ test_projects/faked_project/Rakefile | 5 + 22 files changed, 1241 insertions(+), 56 deletions(-) create mode 100644 features/test_unit_parallel_collate.feature create mode 100644 lib/simplecov/collation.rb create mode 100644 lib/simplecov/parallel_result_merger.rb create mode 100644 sig/internal/simplecov/parallel_result_merger.rbs create mode 100644 spec/combine_differential_spec.rb create mode 100644 spec/parallel_result_merger_spec.rb create mode 100644 spec/support/merge_fuzzer.rb create mode 100644 spec/support/merge_reference.rb diff --git a/CHANGELOG.md b/CHANGELOG.md index 7bd937ff6..ceeba3fe8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ Unreleased ## Enhancements * The favicon (a solid square in the overall coverage band's colour) is now drawn by the viewer from the report's own palette instead of shipping as fixed PNGs, so it matches the report's green/yellow/red exactly and follows the light/dark theme, including the in-page toggle. +* New `SimpleCov.parallel_collate`, which is `SimpleCov.collate` with the resultset merge fanned out across forked worker processes. It takes the same arguments plus a required `processes:`, and produces a report identical to `collate`'s — each worker folds a contiguous slice of the file list and the collating process folds the slices back in order, so the resultsets are visited in the order the serial merge visits them. This addresses the wall clock of a large CI matrix's collate step, where the collating process reads, parses and folds hundreds of resultsets in sequence and nearly all the time goes into that fold: merging 160 resultsets covering 1,836 files on a 14-core machine took 8.53s serially and 2.04s across 8 workers via a new `SimpleCov::ParallelResultMerger`, which mirrors `ResultMerger`'s `merge_and_store` / `merge_results` / `merge_resultsets` entry points. `processes` is deliberately not clamped to the core count nor gated on a minimum number of resultsets — only the caller knows what a collate job is allowed to use — and asking for more processes than there are result files just gives one file per process. Merging falls back to the collating process, with the same report and no error, when the runtime cannot fork (JRuby, TruffleRuby, Windows), when there is only one resultset, or when a worker dies. A `benchmarks/collate.rb` harness (`PROCESSES=N`) measures the phases against a saved baseline. 1.0.3 (2026-07-26) ================== diff --git a/README.md b/README.md index 84f1a7b3f..b2634cb6f 100644 --- a/README.md +++ b/README.md @@ -853,6 +853,51 @@ namespace :coverage do end ``` +#### Fanning the merge out across processes + +Collating a handful of resultsets is quick. Collating a few hundred is not: the collating process reads, parses and +folds every one of them in sequence, and on a large CI matrix that fold is where nearly all the wall clock goes. + +`SimpleCov.parallel_collate` is `SimpleCov.collate` with that fold spread across forked worker processes. It takes the +same arguments — result filenames, an optional profile, an optional configuration block, `ignore_timeout:` — plus a +required `processes:`: + +```ruby +# lib/tasks/coverage_report.rake +namespace :coverage do + desc "Collates all result sets generated by the different test runners" + task :report do + require 'simplecov' + + SimpleCov.parallel_collate Dir["simplecov-resultset-*/.resultset.json"], processes: 8 + end +end +``` + +The report is identical to the one `collate` produces for the same inputs, not merely equivalent: each worker folds a +contiguous slice of the file list and the collating process folds the slices back in order, so the resultsets are +visited in the same order the serial merge visits them. + +`processes` is required and is deliberately not clamped to your core count, nor gated on some minimum number of +resultsets — how many processes a collate job can afford is something only you know. Asking for more processes than +there are result files simply gives one file per process. A `processes` below 1 raises `ArgumentError`, so use +`[n, 1].max` if the count comes from arithmetic that can reach zero. + +It falls back to merging in the collating process — same report, no error — when the runtime cannot fork (JRuby, +TruffleRuby, Windows), when there is only one resultset to fold, or when a worker dies. + +Merging 160 resultsets covering 1,836 files on a 14-core machine (`benchmarks/collate.rb`, so reproduce it on your own +hardware before budgeting for it): + +| `processes:` | merge phase | +| --------------------- | ----------- | +| serial (`collate`) | 8.53s | +| 4 | 2.65s | +| 8 | 2.04s | + +Memory scales with the worker count rather than the resultset count: each worker folds its slice one file at a time, so +it holds one resultset plus its own running total, and the collating process holds one folded total per worker. + ### Forked subprocesses `SimpleCov.merge_subprocesses true` lets SimpleCov observe subprocesses started with `Process.fork`. It wraps Ruby's diff --git a/benchmarks/collate.rb b/benchmarks/collate.rb index 050db7dff..b827e6887 100644 --- a/benchmarks/collate.rb +++ b/benchmarks/collate.rb @@ -20,10 +20,13 @@ # # ruby benchmarks/collate.rb baseline # COUNT=8 ruby benchmarks/collate.rb faster --baseline baseline +# PROCESSES=8 ruby benchmarks/collate.rb parallel --baseline baseline # # Environment: # COUNT merge only the first N resultsets — the knob for a fast # iteration loop; merge cost grows with N (default: 160) +# PROCESSES fan the merge phase out across N forked workers, as +# `SimpleCov.parallel_collate` does; 1 merges serially (default: 1) # SCALE divide `Shape::FILES` by this (default: 4, giving ~1,836 files; # SCALE=1 generates the full 7,345) # SKIP comma-separated trailing phases to skip, e.g. SKIP=format,store diff --git a/benchmarks/collate/cli.rb b/benchmarks/collate/cli.rb index 4a61f086c..64dbf4e1a 100644 --- a/benchmarks/collate/cli.rb +++ b/benchmarks/collate/cli.rb @@ -9,8 +9,10 @@ module CollateBenchmark module CLI DEFAULT_SCALE = 4 + DEFAULT_PROCESSES = 1 + # `resultsets` rather than `count`, which would shadow `Struct#count`. - Options = Struct.new(:label, :resultsets, :scale, :skip, :rebuild, :baseline, :breakdown, + Options = Struct.new(:label, :resultsets, :scale, :skip, :rebuild, :baseline, :breakdown, :processes, keyword_init: true) class << self @@ -21,16 +23,24 @@ def run(argv) def options(argv) argv = argv.dup Options.new( - label: label(argv), - baseline: flag_value(argv, "--baseline"), + label: label(argv), baseline: flag_value(argv, "--baseline"), resultsets: ENV.fetch("COUNT", Shape::RESULTSETS).to_i, scale: ENV.fetch("SCALE", DEFAULT_SCALE).to_i, - skip: skip, - rebuild: ENV["REBUILD"] == "1", - breakdown: ENV["BREAKDOWN"] == "1" + skip: skip, rebuild: ENV["REBUILD"] == "1", + breakdown: ENV["BREAKDOWN"] == "1", processes: processes ) end + # Above 1, the merge phase runs the fan-out `SimpleCov.parallel_collate` + # runs instead of the serial fold. Every later phase is unchanged, so a + # PROCESSES run is directly comparable to a serial baseline. + def processes + count = ENV.fetch("PROCESSES", DEFAULT_PROCESSES).to_i + return count if count >= 1 + + raise ArgumentError, "PROCESSES must be at least 1 (got #{count})" + end + def label(argv) argv.first && !argv.first.start_with?("-") ? argv.shift : "run" end diff --git a/benchmarks/collate/report.rb b/benchmarks/collate/report.rb index 0fca769e7..d1e7e8731 100644 --- a/benchmarks/collate/report.rb +++ b/benchmarks/collate/report.rb @@ -22,10 +22,16 @@ def initialize(run:, timings:, baseline_label:) def header(fixture) puts puts "SimpleCov collate benchmark — #{@run.label}" + print_summary(fixture) + puts + end + + def print_summary(fixture) + merge = @run.processes > 1 ? "across #{@run.processes} forked workers" : "serial, in this process" puts " resultsets: #{fixture.resultset_paths.size}" + puts " merge: #{merge}" puts " fixture: #{fixture_summary(fixture)}" puts " skipping: #{@run.skip.to_a.join(', ')}" if @run.skip.any? - puts end def fixture_summary(fixture) @@ -112,7 +118,7 @@ def write(peak_rss, files_reported) FileUtils.mkdir_p(Fixture::TIMINGS_DIR) timings = { "label" => @run.label, "scale" => @run.scale, "resultsets" => @run.resultsets_used, - "phases" => @timings, "total" => total, "peak_rss" => peak_rss, + "processes" => @run.processes, "phases" => @timings, "total" => total, "peak_rss" => peak_rss, "files_reported" => files_reported, "ruby" => RUBY_DESCRIPTION, # Instrumented runs carry a few percent of wrapper overhead; flagged so # a future comparison knows not to trust this as a baseline. diff --git a/benchmarks/collate/runner.rb b/benchmarks/collate/runner.rb index 92cd2ddb4..823d82bba 100644 --- a/benchmarks/collate/runner.rb +++ b/benchmarks/collate/runner.rb @@ -20,7 +20,7 @@ class Runner # only the three trailing phases can be dropped from a run. SKIPPABLE_PHASES = %w[store format thresholds].freeze - attr_reader :label, :scale, :skip, :breakdown, :resultsets_used + attr_reader :label, :scale, :skip, :breakdown, :processes, :resultsets_used def initialize(options) @label = options.label @@ -30,6 +30,7 @@ def initialize(options) @rebuild = options.rebuild @baseline_label = options.baseline @breakdown = options.breakdown + @processes = options.processes @timings = {} end @@ -37,7 +38,7 @@ def call fixture = Fixture.prepare(scale: @scale, resultsets: @requested_resultsets, force: @rebuild) @resultsets_used = fixture.resultset_paths.size configure(fixture) - Breakdown.install! if @breakdown + install_breakdown if @breakdown report = Report.new(run: self, timings: @timings, baseline_label: @baseline_label) report.header(fixture) @@ -47,6 +48,17 @@ def call private + # The counters live in whichever process ran the wrapped method, so a + # forked worker's attribution dies with it and the merge row would come + # back near-empty. Say so rather than print a misleading table. + def install_breakdown + if processes > 1 + warn "[#{@label}] BREAKDOWN only attributes work done in this process; " \ + "the workers' share of the merge will be missing" + end + Breakdown.install! + end + def measure(fixture) sampler = RssSampler.new run_phases(fixture) @@ -83,9 +95,21 @@ def run_phases(fixture) @files_reported = result&.files&.size end - # The read-and-fold loop out of `ResultMerger.merge_results`, stopping short - # of `create_result` so source-file building is timed separately. + # With PROCESSES > 1, the fan-out `SimpleCov.parallel_collate` performs — + # same merge, same visiting order, spread over forked workers. Falls + # through to the in-process loop if the fan-out bails, which is what the + # real `parallel_collate` does too. def merge_coverage(paths) + return serial_merge_coverage(paths) if processes < 2 + + SimpleCov::ParallelResultMerger.merge_resultsets(paths, processes: processes, ignore_timeout: true) || + serial_merge_coverage(paths) + end + + # `ResultMerger.merge_resultsets`, reproduced here so the per-resultset + # progress line can be printed, and stopping short of `create_result` so + # source-file building is timed separately. + def serial_merge_coverage(paths) remaining = paths.dup initial = valid_results(remaining.shift) diff --git a/features/test_unit_parallel_collate.feature b/features/test_unit_parallel_collate.feature new file mode 100644 index 000000000..3ef81aa2c --- /dev/null +++ b/features/test_unit_parallel_collate.feature @@ -0,0 +1,42 @@ +@test_unit +Feature: + + Using SimpleCov.parallel_collate should get the user the same coverage report + SimpleCov.collate does, with the merge fanned out across forked workers. + + Background: + Given I'm working on the project "faked_project" + + Scenario: + Given SimpleCov for Test/Unit is configured with: + """ + require 'simplecov' + SimpleCov.start + """ + + When I successfully run `bundle exec rake part1` + Then a coverage report should have been generated + When I successfully run `mv coverage/.resultset.json coverage/resultset1.json` + And I successfully run `rm coverage/index.html` + + When I successfully run `bundle exec rake part2` + Then a coverage report should have been generated + When I successfully run `mv coverage/.resultset.json coverage/resultset2.json` + And I successfully run `rm coverage/index.html` + + # Identical to the figures in test_unit_collate.feature: fanning the merge + # out changes how the resultsets are folded, not what they fold to. + When I open the coverage report generated with `bundle exec rake parallel_collate` + Then I should see the groups: + | name | coverage | files | + | All Files | 88.09% | 4 | + + And I should see the source files: + | name | coverage | + | lib/faked_project.rb | 100.00% | + | lib/faked_project/some_class.rb | 80.00% | + | lib/faked_project/framework_specific.rb | 75.00% | + | lib/faked_project/meta_magic.rb | 100.00% | + + And the report should be based upon: + | Unit Tests | diff --git a/lib/simplecov.rb b/lib/simplecov.rb index 4772604df..14e3e2d69 100644 --- a/lib/simplecov.rb +++ b/lib/simplecov.rb @@ -247,6 +247,7 @@ def warn_if_jruby_full_trace_disabled require_relative "simplecov/last_run" require_relative "simplecov/lines_classifier" require_relative "simplecov/result_merger" +require_relative "simplecov/parallel_result_merger" require_relative "simplecov/parallel_adapters" require_relative "simplecov/command_guesser" require_relative "simplecov/version" @@ -261,6 +262,7 @@ def warn_if_jruby_full_trace_disabled require_relative "simplecov/useless_results_remover" require_relative "simplecov/simulate_coverage" require_relative "simplecov/result_processing" +require_relative "simplecov/collation" require_relative "simplecov/exit_handling" require_relative "simplecov/parallel_coordination" diff --git a/lib/simplecov/collation.rb b/lib/simplecov/collation.rb new file mode 100644 index 000000000..a721baf44 --- /dev/null +++ b/lib/simplecov/collation.rb @@ -0,0 +1,69 @@ +# frozen_string_literal: true + +# The `collate` entry points: stitch the resultsets written by separate +# test runs — parallel CI jobs, several build machines, a matrix of Ruby +# versions — into one report, either in this process or fanned out across +# forked workers. +module SimpleCov + class << self + # + # Collate a series of SimpleCov result files into a single SimpleCov output. + # + # See README for usage. By default `collate` ignores the merge_timeout + # so all results in all files specified will be merged. Pass + # `ignore_timeout: false` to honor it. + # + def collate(result_filenames, profile = nil, ignore_timeout: true, &config) + collating(result_filenames, profile, config) do + # Use the ResultMerger to produce a single, merged result, ready to use. + ResultMerger.merge_and_store(*result_filenames, ignore_timeout: ignore_timeout) + end + end + + # + # `collate`, with the merge fanned out across `processes` forked worker + # processes. Takes the same arguments and produces the same report: the + # workers fold contiguous slices of `result_filenames` in the order the + # serial merge visits them, so the merged result is identical, not merely + # equivalent. Only the wall clock differs, and only for a collate big + # enough that reading and parsing the resultsets dominates it. + # + # `processes` is required, and is deliberately not clamped to the machine's + # core count nor gated on some minimum number of resultsets: how many + # processes a collate job can afford is the caller's call, not SimpleCov's. + # Asking for more processes than there are result files simply gives one + # file per process. A `processes` below 1 raises rather than quietly + # merging serially, so compute it with `[n, 1].max` if it comes from + # arithmetic that can reach zero. + # + # Falls back to merging in this process — same report, no error — when the + # runtime cannot fork (JRuby, TruffleRuby, Windows), when there is nothing + # worth splitting, or when a worker dies. + # + def parallel_collate(result_filenames, profile = nil, processes:, ignore_timeout: true, &config) + raise ArgumentError, "processes must be at least 1, got #{processes}" if processes < 1 + + collating(result_filenames, profile, config) do + ParallelResultMerger.merge_and_store(*result_filenames, processes: processes, ignore_timeout: ignore_timeout) + end + end + + private + + # The scaffolding both entry points share: validate, apply the caller's + # profile and configuration block, then run the finalizer over whatever + # merged result the given strategy produced. `config` is the caller's + # configuration block, passed as an object because the merge strategy + # occupies the block slot. + def collating(result_filenames, profile, config) + raise ArgumentError, "There are no reports to be merged" if result_filenames.empty? + + initial_setup(profile, &config) + @result = yield + @collating_result = true + run_exit_tasks! + ensure + @collating_result = false + end + end +end diff --git a/lib/simplecov/parallel_result_merger.rb b/lib/simplecov/parallel_result_merger.rb new file mode 100644 index 000000000..e28aa6485 --- /dev/null +++ b/lib/simplecov/parallel_result_merger.rb @@ -0,0 +1,201 @@ +# frozen_string_literal: true + +module SimpleCov + # + # Folds a list of resultset files into one merged coverage table across + # forked worker processes. Drives `SimpleCov.parallel_collate`. + # + # `ResultMerger.merge_resultsets` is a fold over N independent + # read-parse-combine steps, so it splits cleanly: each worker runs that + # same fold over a contiguous slice of the file list and ships the pair + # back over a pipe, and the parent combines the handful of per-worker + # pairs it gets back. Reading and parsing the shards — where a collate + # over a few hundred CI jobs spends most of its time — is what actually + # parallelises. + # + # The slices are contiguous and merged back in order, so the fold visits + # the resultsets in the order the serial fold visits them and the merged + # result is identical to `SimpleCov.collate`'s, not merely equivalent. + # + # Every failure path returns nil rather than a partial merge, so the caller + # can redo the fold serially: reporting coverage for a subset of the + # resultsets would silently understate it. + # + module ParallelResultMerger + module_function + + # + # `ResultMerger.merge_and_store` across `processes` forked workers. + # + def merge_and_store(*file_paths, processes:, ignore_timeout: false) + result = merge_results(*file_paths, processes: processes, ignore_timeout: ignore_timeout) + ResultMerger.store_result(result) if result + result + end + + # + # `ResultMerger.merge_results` across `processes` forked workers, merging + # in this process instead whenever the fan-out did not produce a complete + # merge — a runtime that cannot fork, nothing worth splitting, or a worker + # that died. The result is the same either way; only the time it took to + # get there differs. + # + def merge_results(*file_paths, processes:, ignore_timeout: false) + command_names, coverage = + merge_resultsets(file_paths, processes: processes, ignore_timeout: ignore_timeout) || + ResultMerger.merge_resultsets(file_paths, ignore_timeout: ignore_timeout) + + ResultMerger.create_result(command_names, coverage) + end + + # + # `ResultMerger.merge_resultsets` across at most `processes` forked + # workers: same arguments, same `[command_names, coverage]` return. + # + # @return [Array(Array, Hash), nil] the pair + # `ResultMerger.create_result` consumes, or nil when the work could not + # be fanned out and the caller should merge in this process instead. + # + def merge_resultsets(file_paths, processes:, ignore_timeout: false) + # One worker folds the whole list anyway, and one file is a fold of + # one — in both cases the fork and the round trip are pure overhead. + return nil if processes < 2 || file_paths.size < 2 + return nil unless fork_supported? + + fan_out(chunk(file_paths, processes), ignore_timeout: ignore_timeout) + end + + def fork_supported? + Process.respond_to?(:fork) + end + + # Contiguous slices whose sizes differ by at most one, so no worker is + # left folding twice its share while the others idle. There are never + # more slices than files: asking for more processes than there are + # resultsets just gives one resultset per process. + def chunk(file_paths, processes) + groups = [processes, file_paths.size].min + base, remainder = file_paths.size.divmod(groups) + remaining = file_paths.dup + + Array.new(groups) { |index| remaining.shift(base + (index < remainder ? 1 : 0)) } + end + + def fan_out(chunks, ignore_timeout:) + workers = spawn_workers(chunks, ignore_timeout: ignore_timeout) + return nil unless workers + + payloads = collect(workers) + payloads && ResultMerger.merge_coverage(*payloads) + end + + # nil when this runtime turns out not to support forking after all. + # JRuby, TruffleRuby and Windows define `Process.fork` and raise only + # when it is called, so this — rather than the `respond_to?` probe — is + # what actually detects them. Anything already spawned is torn down. + def spawn_workers(chunks, ignore_timeout:) + workers = [] #: Array[Hash[Symbol, untyped]] + # Accumulated rather than mapped so the rescue below can still see the + # workers spawned before the failing fork. + chunks.each { |chunk| workers << spawn_worker(chunk, ignore_timeout: ignore_timeout) } # rubocop:disable Style/MapIntoArray + workers + rescue NotImplementedError + # @type var workers: Array[Hash[Symbol, untyped]] + shut_down(workers) + nil + end + + def spawn_worker(chunk, ignore_timeout:) + reader, writer = IO.pipe + + pid = fork do + # simplecov:disable — the child's lines are measured in the child, + # which exits via `exit!` without reporting; `run_worker` itself is + # covered by calling it directly. + reader.close + exit!(run_worker(chunk, writer, ignore_timeout: ignore_timeout)) + # simplecov:enable + end + + writer.close + {pid: pid, reader: reader} + end + + # The body of a worker: fold the slice, ship it back, and report the exit + # status the child should terminate with. Kept free of the exit itself so + # it can be exercised in-process. + # + # A child must never fall through to the collating process's `at_exit` + # handlers — SimpleCov's own report generation included — which is why + # `spawn_worker` ends it with `exit!` rather than `exit`. + def run_worker(chunk, writer, ignore_timeout:) + Marshal.dump(ResultMerger.merge_resultsets(chunk, ignore_timeout: ignore_timeout), writer) + writer.close + 0 + rescue StandardError => e + warn "[SimpleCov]: parallel merge worker failed: #{e.class}: #{e.message}" if SimpleCov.print_errors + 1 + end + + # Deserializes on a thread per worker so every pipe is drained while the + # workers are still writing. A payload larger than the pipe buffer would + # otherwise block its worker mid-write, and the parent would block + # reaping a worker that can never finish. + # + # Returns nil if any worker failed, so the caller can fall back to the + # serial fold rather than report a subset of the resultsets as the whole. + def collect(workers) + payloads = drain(workers) + failed = workers.count { |worker| !succeeded?(worker[:pid]) } + return payloads if failed.zero? && payloads.all? + + warn_about_failed_workers(failed, workers.size) + nil + ensure + workers.each { |worker| worker[:reader].close } + end + + def drain(workers) + workers.map { |worker| Thread.new { read_payload(worker[:reader]) } }.map(&:value) + end + + def read_payload(reader) + # The writer is a fork of this very process and the pipe never leaves + # it, so this is our own data coming back through our own kernel + # buffer, not input. A worker that died mid-write leaves the stream + # truncated, which Marshal reports by raising rather than returning. + # RBS types `Marshal.load`'s source as `_Source`, which IO satisfies + # structurally but not nominally. + # steep:ignore:start + Marshal.load(reader) # rubocop:disable Security/MarshalLoad + # steep:ignore:end + rescue StandardError + nil + end + + def succeeded?(pid) + _pid, status = Process.wait2(pid) + status.success? + rescue SystemCallError + # Errno::ECHILD — nothing left to reap, so there is no status to judge + # this worker's slice by and we have to assume it did not finish. + false + end + + def warn_about_failed_workers(failed, total) + return unless SimpleCov.print_errors + + warn "[SimpleCov]: parallel merge did not complete (#{failed} of #{total} workers failed); " \ + "falling back to merging the resultsets in this process." + end + + # Best-effort teardown for workers spawned before a fork failed. Closing + # the read end makes a worker still writing die of EPIPE. + def shut_down(workers) + workers.each do |worker| + worker[:reader].close + succeeded?(worker[:pid]) + end + end + end +end diff --git a/lib/simplecov/result_merger.rb b/lib/simplecov/result_merger.rb index 9f0f6e826..3985ee81b 100644 --- a/lib/simplecov/result_merger.rb +++ b/lib/simplecov/result_merger.rb @@ -23,23 +23,32 @@ def merge_and_store(*file_paths, ignore_timeout: false) end def merge_results(*file_paths, ignore_timeout: false) - # It is intentional here that files are only read in and parsed one at a time. - # - # In big CI setups you might deal with 100s of CI jobs and each one producing Megabytes - # of data. Reading them all in easily produces Gigabytes of memory consumption which - # we want to avoid. - # - # For similar reasons a SimpleCov::Result is only created in the end as that'd create - # even more data especially when it also reads in all source files. - initial_memo = valid_results(file_paths.shift, ignore_timeout: ignore_timeout) - - command_names, coverage = file_paths.reduce(initial_memo) do |memo, file_path| - merge_coverage(memo, valid_results(file_path, ignore_timeout: ignore_timeout)) - end + command_names, coverage = merge_resultsets(file_paths, ignore_timeout: ignore_timeout) create_result(command_names, coverage) end + # + # Read the given resultset files and combine them into a single + # `[command_names, coverage]` pair. + # + # It is intentional here that files are only read in and parsed one at a time. + # + # In big CI setups you might deal with 100s of CI jobs and each one producing Megabytes + # of data. Reading them all in easily produces Gigabytes of memory consumption which + # we want to avoid. + # + # For similar reasons a SimpleCov::Result is only created in the end as that'd create + # even more data especially when it also reads in all source files. + # + def merge_resultsets(file_paths, ignore_timeout: false) + initial_memo = valid_results(file_paths.first, ignore_timeout: ignore_timeout) + + file_paths.drop(1).reduce(initial_memo) do |memo, file_path| + merge_coverage(memo, valid_results(file_path, ignore_timeout: ignore_timeout)) + end + end + def valid_results(file_path, ignore_timeout: false) merge_valid_results(ResultsetFile.parse(file_path), ignore_timeout: ignore_timeout) end diff --git a/lib/simplecov/result_processing.rb b/lib/simplecov/result_processing.rb index ad4562a8b..aa5893017 100644 --- a/lib/simplecov/result_processing.rb +++ b/lib/simplecov/result_processing.rb @@ -1,32 +1,12 @@ # frozen_string_literal: true # Result-building façade: turns the raw `Coverage.result` hash into a -# `SimpleCov::Result`, applies filters and groups, drives merging -# across test suites via `SimpleCov::ResultMerger`, and exposes the -# `collate` entry point for stitching disparate resultsets together. +# `SimpleCov::Result`, applies filters and groups, and drives merging +# across test suites via `SimpleCov::ResultMerger`. The `collate` entry +# points for stitching disparate resultsets together live alongside it +# in `simplecov/collation`. module SimpleCov class << self - # - # Collate a series of SimpleCov result files into a single SimpleCov output. - # - # See README for usage. By default `collate` ignores the merge_timeout - # so all results in all files specified will be merged. Pass - # `ignore_timeout: false` to honor it. - # - def collate(result_filenames, profile = nil, ignore_timeout: true, &) - raise ArgumentError, "There are no reports to be merged" if result_filenames.empty? - - initial_setup(profile, &) - - # Use the ResultMerger to produce a single, merged result, ready to use. - @result = ResultMerger.merge_and_store(*result_filenames, ignore_timeout: ignore_timeout) - - @collating_result = true - run_exit_tasks! - ensure - @collating_result = false - end - # # Returns the result for the current coverage run, merging it across test suites # from cache using SimpleCov::ResultMerger if use_merging is activated (default) diff --git a/sig/internal/simplecov/parallel_result_merger.rbs b/sig/internal/simplecov/parallel_result_merger.rbs new file mode 100644 index 000000000..87f0fe421 --- /dev/null +++ b/sig/internal/simplecov/parallel_result_merger.rbs @@ -0,0 +1,61 @@ +module SimpleCov + # + # Merges a list of resultset files into one coverage table across forked + # worker processes. Drives `SimpleCov.parallel_collate`. + # + # `ResultMerger.merge_resultsets` is a fold over N independent + # read-parse-combine steps, so it splits cleanly: each worker runs that + # same fold over a contiguous slice of the file list and ships the pair + # back over a pipe, and the parent combines the per-worker pairs. + # + # Every failure path returns nil rather than a partial merge, so the caller + # can redo the merge in this process. + # + module ParallelResultMerger + # `ResultMerger.merge_and_store` across `processes` forked workers. + def self?.merge_and_store: (*untyped file_paths, processes: Integer, ?ignore_timeout: bool) -> untyped + + # `ResultMerger.merge_results` across `processes` forked workers, merging + # in this process when the fan-out did not produce a complete merge. + def self?.merge_results: (*untyped file_paths, processes: Integer, ?ignore_timeout: bool) -> untyped + + # + # `ResultMerger.merge_resultsets` across at most `processes` forked + # workers: same arguments, same `[command_names, coverage]` return, or + # nil when the work could not be fanned out. + # + def self?.merge_resultsets: (Array[String] file_paths, processes: Integer, ?ignore_timeout: bool) -> untyped + + def self?.fork_supported?: () -> bool + + # Contiguous slices whose sizes differ by at most one. There are never + # more slices than files. + def self?.chunk: (Array[String] file_paths, Integer processes) -> Array[Array[String]] + + def self?.fan_out: (Array[Array[String]] chunks, ignore_timeout: bool) -> untyped + + # nil when this runtime turns out not to support forking after all. + def self?.spawn_workers: (Array[Array[String]] chunks, ignore_timeout: bool) -> Array[Hash[Symbol, untyped]]? + + def self?.spawn_worker: (Array[String] chunk, ignore_timeout: bool) -> Hash[Symbol, untyped] + + # The body of a worker: merge the slice, ship it back, and report the exit + # status the child should terminate with. + def self?.run_worker: (Array[String] chunk, IO writer, ignore_timeout: bool) -> Integer + + # Returns nil if any worker failed, so the caller can fall back to merging + # in this process. + def self?.collect: (Array[Hash[Symbol, untyped]] workers) -> Array[untyped]? + + def self?.drain: (Array[Hash[Symbol, untyped]] workers) -> Array[untyped] + + def self?.read_payload: (IO reader) -> untyped + + def self?.succeeded?: (Integer pid) -> bool + + def self?.warn_about_failed_workers: (Integer failed, Integer total) -> void + + # Best-effort teardown for workers spawned before a fork failed. + def self?.shut_down: (Array[Hash[Symbol, untyped]] workers) -> void + end +end diff --git a/sig/internal/simplecov/result_merger.rbs b/sig/internal/simplecov/result_merger.rbs index 8f238bc2f..82cad973a 100644 --- a/sig/internal/simplecov/result_merger.rbs +++ b/sig/internal/simplecov/result_merger.rbs @@ -11,6 +11,10 @@ module SimpleCov def self.merge_results: (*untyped file_paths, ?ignore_timeout: bool) -> untyped + # Read the given resultset files and combine them into a single + # `[command_names, coverage]` pair, one file at a time. + def self.merge_resultsets: (Array[String] file_paths, ?ignore_timeout: bool) -> untyped + def self.valid_results: (untyped file_path, ?ignore_timeout: bool) -> untyped def self.merge_valid_results: (untyped results, ?ignore_timeout: bool) -> untyped diff --git a/sig/simplecov.rbs b/sig/simplecov.rbs index eaf2b72e8..1f8bf9b2d 100644 --- a/sig/simplecov.rbs +++ b/sig/simplecov.rbs @@ -96,20 +96,34 @@ module SimpleCov VERSION: String end -# ----- result_processing ----- +# ----- collation ----- -# Result-building façade: turns the raw `Coverage.result` hash into a -# `SimpleCov::Result`, applies filters and groups, drives merging, and -# exposes the `collate` entry point. +# The `collate` entry points: stitch the resultsets written by separate +# test runs into one report, in this process or across forked workers. module SimpleCov - self.@result: Result? - self.@collating_result: bool? - # Collate a series of resultset files into a single merged report. By # default ignores `merge_timeout` so every listed resultset merges; # pass `ignore_timeout: false` to honor it. def self.collate: (Array[String] result_filenames, ?(String | Symbol)? profile, ?ignore_timeout: bool) ?{ () [self: singleton(SimpleCov)] -> void } -> void + # `collate`, with the merge fanned out across `processes` forked worker + # processes. Same arguments, same report. `processes` is required and is + # not clamped; below 1 it raises. + def self.parallel_collate: (Array[String] result_filenames, ?(String | Symbol)? profile, processes: Integer, ?ignore_timeout: bool) ?{ () [self: singleton(SimpleCov)] -> void } -> void + + # Implementation detail of the two entry points above (private in Ruby; + # RBS cannot mark singleton methods private). + def self.collating: (Array[String] result_filenames, (String | Symbol)? profile, (^() [self: singleton(SimpleCov)] -> void)? config) { () -> Result? } -> void +end + +# ----- result_processing ----- + +# Result-building façade: turns the raw `Coverage.result` hash into a +# `SimpleCov::Result`, applies filters and groups, and drives merging. +module SimpleCov + self.@result: Result? + self.@collating_result: bool? + # The result for the current coverage run, merged across test suites # when merging is active. nil when no coverage was tracked. def self.result: () -> Result? diff --git a/spec/combine_differential_spec.rb b/spec/combine_differential_spec.rb new file mode 100644 index 000000000..a22bff1da --- /dev/null +++ b/spec/combine_differential_spec.rb @@ -0,0 +1,109 @@ +# frozen_string_literal: true + +require "helper" +require "support/merge_fuzzer" +require "support/merge_reference" + +# Differential test for the whole merge, end to end: `Combine.combine`'s +# short-circuit, `ResultsCombiner`'s fold over files, `FilesCombiner`'s +# reconciliation, and the three leaf combiners. The unit specs for each of +# those live in spec/combine/; this one is about what they add up to. +# +# `ResultsCombiner` folds resultsets pairwise, so what it does to N of them is +# emergent rather than written down anywhere. `MergeReference` states the N-way +# rules directly; this compares the two over generated shard sets built to hit +# the cases the rules turn on (see `MergeFuzzer`). +# +# Its job is to pin the pairwise fold's behaviour precisely enough that it can +# be replaced by a single-pass accumulator without silently changing an answer - +# in particular the `reconcile_synthesized` rule, which is defined on a pair and +# has to be restated N-way to survive that change. +# +# Seeds are fixed so CI is deterministic. SIMPLECOV_MERGE_SEEDS raises the +# count for a local soak; a failure prints the seed to replay. +RSpec.describe SimpleCov::Combine do + around do |example| + SimpleCov.enable_coverage(:branch) + SimpleCov.enable_coverage(:method) + example.run + SimpleCov.clear_coverage_criteria + end + + describe "merging N resultsets", + if: SimpleCov.branch_coverage_supported? && SimpleCov.method_coverage_supported? do + it "agrees with the reference merge on every generated shard set" do + mismatches = seeds.filter_map { |seed| mismatch_for(seed, :fold, saturate: true) } + + expect(mismatches).to be_empty, -> { describe_mismatches(mismatches) } + end + + # Identity collapse only happens inside a combiner, and `Combine.combine` + # short-circuits before reaching one when either side is nil - handing the + # other side back verbatim. So a file whose coverage never meets a non-nil + # counterpart keeps duplicate identities, and the #1233 / #1234 dedup does + # not apply to it. A resultset with a single command_name is the common + # case: `merge_coverage` returns it unchanged, so nothing collapses. + # + # Pinned here rather than folded into the differential above (which + # saturates its shard sets to avoid it) so that a change to it has to be + # deliberate. The consequence is a phantom uncovered method for the #1234 + # shape - one `define_method` landing on two receivers - in a single-suite + # run. + it "leaves duplicate identities uncollapsed when the counterpart is nil" do + duplicate = {["Foo", :call, 2, 2, 4, 10] => 3, ["Bar", :call, 2, 2, 4, 10] => 0} + + expect(described_class.combine(SimpleCov::Combine::MethodsCombiner, duplicate, nil)) + .to eq(duplicate) + expect(described_class.combine(SimpleCov::Combine::MethodsCombiner, duplicate, {})) + .to eq(["Foo", :call, 2, 2, 4, 10] => 3) + end + end + + def seeds + 1..Integer(ENV.fetch("SIMPLECOV_MERGE_SEEDS", "300")) + end + + # Returns nil when the fold and the reference agree, otherwise the detail + # needed to reproduce and read the disagreement. + def mismatch_for(seed, strategy, saturate:) + shards = MergeFuzzer.shards(seed, saturate: saturate) + actual = normalize(send(strategy, shards)) + expected = normalize(MergeReference.call(shards, branches: true, methods: true)) + return nil if actual == expected + + files = expected.keys.union(actual.keys).reject { |file| actual[file] == expected[file] } + {seed: seed, strategy: strategy, shards: shards, files: files, actual: actual, expected: expected} + end + + def fold(shards) + shards.reduce { |a, b| described_class.combine(SimpleCov::Combine::ResultsCombiner, a, b) } + end + + # Only the criteria matter, not which keys a coverage happens to carry. + def normalize(files) + files.to_h do |file, entry| + [file, {"lines" => entry["lines"], + "branches" => entry["branches"] || {}, + "methods" => entry["methods"] || {}}] + end + end + + def describe_mismatches(mismatches) + detail = mismatches.first(3).map { |mismatch| describe_mismatch(mismatch) } + "#{mismatches.length} seed(s) disagreed with the reference merge " \ + "(#{mismatches.map { |m| m[:seed] }.first(20).join(', ')}):\n\n#{detail.join("\n")}" + end + + def describe_mismatch(mismatch) + seed, strategy, files = mismatch.values_at(:seed, :strategy, :files) + detail = files.flat_map { |file| describe_file(file, mismatch) } + ["seed #{seed} (#{strategy}) — files: #{files.join(', ')}", *detail].join("\n") + end + + def describe_file(file, mismatch) + [" #{file}", + " shards: #{mismatch[:shards].map { |shard| shard[file] }.inspect}", + " actual: #{mismatch[:actual][file].inspect}", + " reference: #{mismatch[:expected][file].inspect}"] + end +end diff --git a/spec/parallel_result_merger_spec.rb b/spec/parallel_result_merger_spec.rb new file mode 100644 index 000000000..c1892fed7 --- /dev/null +++ b/spec/parallel_result_merger_spec.rb @@ -0,0 +1,263 @@ +# frozen_string_literal: true + +require "helper" +require "tmpdir" + +RSpec.describe SimpleCov::ParallelResultMerger do + let(:resultset_dir) { Dir.mktmpdir("simplecov-parallel-merge") } + + # `sample.rb` is executed by every shard and the second file by exactly one, + # so the merged table exercises both summing and union. Every file has to + # exist on disk: these resultsets get built into real `SimpleCov::Result`s, + # which warn about coverage for files that have since been removed. + let(:shards) do + [ + ["resultset1.rb", {"lines" => [nil, 1, 1, nil]}], + ["resultset2.rb", {"lines" => [nil, 1, nil, 2]}], + ["three.rb", {"lines" => [nil, 0, 1, nil]}], + ["never.rb", {"lines" => [1, nil, nil, nil]}], + ["inline.rb", {"lines" => [nil, 2, nil, nil]}] + ] + end + + let(:paths) do + shards.each_with_index.map do |(fixture, lines), index| + write_resultset( + "shard#{index}", + { + source_fixture("sample.rb") => lines, + source_fixture(fixture) => {"lines" => [index, nil]} + } + ) + end + end + + let(:serial) { SimpleCov::ResultMerger.merge_resultsets(paths, ignore_timeout: true) } + + after { FileUtils.remove_entry(resultset_dir) } + + describe ".merge_and_store" do + before { FileUtils.mkdir_p(File.dirname(SimpleCov::ResultMerger.resultset_path)) } + + after { FileUtils.rm_f(SimpleCov::ResultMerger.resultset_path) } + + it "produces the result ResultMerger.merge_and_store produces" do + result = described_class.merge_and_store(*paths, processes: 3, ignore_timeout: true) + + expect(result.to_hash).to eq(SimpleCov::ResultMerger.create_result(*serial).to_hash) + end + + it "has the result stored" do + described_class.merge_and_store(*paths, processes: 3, ignore_timeout: true) + + expect(SimpleCov::ResultMerger.read_resultset.keys).to eq([serial.first.sort.join(", ")]) + end + + # The workers do the dropping, so the "older than merge_timeout" warning + # comes from them and reaches the inherited stderr rather than anything + # this process can capture. + it "stores nothing when every resultset is outdated" do + allow(SimpleCov::ResultMerger).to receive(:store_result) + stale = Array.new(3) do |i| + write_resultset("old#{i}", {source_fixture("sample.rb") => {"lines" => [1]}}, outdated: true) + end + + expect(described_class.merge_and_store(*stale, processes: 2)).to be_nil + expect(SimpleCov::ResultMerger).not_to have_received(:store_result) + end + end + + describe ".merge_results" do + it "produces the result ResultMerger.merge_results produces" do + result = described_class.merge_results(*paths, processes: 3, ignore_timeout: true) + + expect(result.to_hash).to eq(SimpleCov::ResultMerger.merge_results(*paths, ignore_timeout: true).to_hash) + end + + it "merges in this process when the fan-out cannot run" do + allow(described_class).to receive(:fork_supported?).and_return(false) + + result = described_class.merge_results(*paths, processes: 3, ignore_timeout: true) + + expect(result.to_hash).to eq(SimpleCov::ResultMerger.merge_results(*paths, ignore_timeout: true).to_hash) + end + end + + describe ".merge_resultsets" do + it "produces the pair ResultMerger.merge_resultsets produces" do + expect(described_class.merge_resultsets(paths, processes: 3, ignore_timeout: true)).to eq(serial) + end + + it "produces the same pair however many processes it is given" do + merges = [2, 3, 4, 5, 12].map do |processes| + described_class.merge_resultsets(paths, processes: processes, ignore_timeout: true) + end + + expect(merges.uniq.size).to eq(1) + end + + it "honours ignore_timeout: false by dropping expired resultsets" do + expired = write_resultset("stale", {source_fixture("sample.rb") => {"lines" => [9, 9, 9, 9]}}, outdated: true) + fresh = paths.first(2) + + command_names, = described_class.merge_resultsets([*fresh, expired], processes: 3, ignore_timeout: false) + + expect(command_names).to contain_exactly("shard0", "shard1", "") + end + + it "returns nil when a single process was requested" do + expect(described_class.merge_resultsets(paths, processes: 1)).to be_nil + end + + it "returns nil when there is only one resultset to merge" do + expect(described_class.merge_resultsets(paths.first(1), processes: 4)).to be_nil + end + + it "returns nil when the runtime cannot fork" do + allow(described_class).to receive(:fork_supported?).and_return(false) + + expect(described_class.merge_resultsets(paths, processes: 4)).to be_nil + end + end + + describe ".fork_supported?" do + # False on JRuby, TruffleRuby and Windows, where `fork` is defined but + # raises; there is no CRuby-side way to reach the false case. + it { expect(described_class.fork_supported?).to eq(Process.respond_to?(:fork)) } + end + + describe ".chunk" do + it "splits into contiguous slices, in order" do + expect(described_class.chunk(%w[a b c d e f], 3)).to eq([%w[a b], %w[c d], %w[e f]]) + end + + it "gives the remainder to the leading slices so no worker folds twice its share" do + expect(described_class.chunk(%w[a b c d e f g], 3)).to eq([%w[a b c], %w[d e], %w[f g]]) + end + + it "makes no more slices than there are files" do + expect(described_class.chunk(%w[a b c], 10)).to eq([%w[a], %w[b], %w[c]]) + end + + it "leaves the caller's list alone" do + files = %w[a b c d] + described_class.chunk(files, 2) + + expect(files).to eq(%w[a b c d]) + end + end + + describe ".run_worker" do + let(:pipe) { IO.pipe } + + after { pipe.each { |io| io.close unless io.closed? } } + + it "writes the merged pair and reports success" do + reader, writer = pipe + + expect(described_class.run_worker(paths, writer, ignore_timeout: true)).to eq(0) + expect(Marshal.load(reader)).to eq(serial) # rubocop:disable Security/MarshalLoad + end + + it "reports failure and warns when the merged pair cannot be shipped back" do + _reader, writer = pipe + writer.close + + output = capture_stderr do + expect(described_class.run_worker(paths, writer, ignore_timeout: true)).to eq(1) + end + + expect(output).to include("parallel merge worker failed", "IOError") + end + + it "stays quiet about a failed worker when print_errors is off" do + _reader, writer = pipe + writer.close + + output = capture_stderr do + with_print_errors(false) { described_class.run_worker(paths, writer, ignore_timeout: true) } + end + + expect(output).to be_empty + end + end + + describe ".fan_out" do + let(:chunks) { described_class.chunk(paths, 3) } + + it "returns nil and warns when a worker dies without shipping its slice" do + allow(described_class).to receive(:run_worker).and_return(1) + + output = capture_stderr do + expect(described_class.fan_out(chunks, ignore_timeout: true)).to be_nil + end + + expect(output).to include("parallel merge did not complete (3 of 3 workers failed)") + end + + it "returns nil when the workers could not be spawned at all" do + allow(described_class).to receive(:spawn_workers).and_return(nil) + + expect(described_class.fan_out(chunks, ignore_timeout: true)).to be_nil + end + + it "stays quiet about a failed fan-out when print_errors is off" do + allow(described_class).to receive(:run_worker).and_return(1) + + output = capture_stderr do + with_print_errors(false) { described_class.fan_out(chunks, ignore_timeout: true) } + end + + expect(output).to be_empty + end + end + + describe ".spawn_workers" do + it "tears down the workers it did spawn and returns nil when a fork is refused" do + spawned = [] + allow(described_class).to receive(:spawn_worker).and_wrap_original do |original, *args, **options| + raise NotImplementedError, "fork is not available on this platform" if spawned.any? + + original.call(*args, **options).tap { |worker| spawned << worker } + end + + expect(described_class.spawn_workers(described_class.chunk(paths, 3), ignore_timeout: true)).to be_nil + expect(spawned.first[:reader]).to be_closed + end + end + + describe ".read_payload" do + it "returns nil for a stream a worker never finished writing" do + reader, writer = IO.pipe + writer.write(Marshal.dump([["shard0"], {}]).byteslice(0, 4)) + writer.close + + expect(described_class.read_payload(reader)).to be_nil + ensure + reader.close + end + end + + describe ".succeeded?" do + it "is false for a pid it cannot reap" do + expect(described_class.succeeded?(-1)).to be false + end + end + +private + + def write_resultset(command_name, coverage, outdated: false) + timestamp = Time.now.to_i - (outdated ? SimpleCov.merge_timeout * 2 : 0) + path = File.join(resultset_dir, ".resultset-#{command_name}.json") + File.write(path, JSON.generate(command_name => {"coverage" => coverage, "timestamp" => timestamp})) + path + end + + def with_print_errors(value) + previous = SimpleCov.print_errors + SimpleCov.print_errors value + yield + ensure + SimpleCov.print_errors previous + end +end diff --git a/spec/result_merger_spec.rb b/spec/result_merger_spec.rb index dd7986255..270f2988b 100644 --- a/spec/result_merger_spec.rb +++ b/spec/result_merger_spec.rb @@ -276,6 +276,33 @@ end end + describe ".merge_resultsets" do + let(:resultset_prefix) { "fold_test_resultset" } + let(:paths) { [1, 2].map { |index| "#{resultset_prefix}#{index}.json" } } + + before do + store_result(first_result, path: paths.first) + store_result(second_result, path: paths.last) + end + + after do + FileUtils.rm Dir.glob("#{resultset_prefix}*.json") + end + + it "combines the resultsets into a command_names / coverage pair" do + command_names, coverage = described_class.merge_resultsets(paths, ignore_timeout: true) + + expect(command_names).to eq(%w[result1 result2]) + expect(coverage).to eq(merged_resultsets) + end + + it "leaves the caller's list of paths alone" do + described_class.merge_resultsets(paths, ignore_timeout: true) + + expect(paths).to eq(["#{resultset_prefix}1.json", "#{resultset_prefix}2.json"]) + end + end + describe ".store_result" do it "refreshes the resultset" do set = described_class.read_resultset diff --git a/spec/simplecov_spec.rb b/spec/simplecov_spec.rb index 6ee378030..a6603ed4c 100644 --- a/spec/simplecov_spec.rb +++ b/spec/simplecov_spec.rb @@ -1100,6 +1100,88 @@ def expect_merged end end + describe ".parallel_collate" do + let(:resultset_path) { SimpleCov::ResultMerger.resultset_path } + + let(:resultset_folder) { File.dirname(resultset_path) } + + let(:collated) do + JSON.parse(File.read(resultset_path)).transform_values { |data| data.reject { |key| key == "timestamp" } } + end + + context "when no files to be merged" do + it "shows an error message" do + expect { described_class.parallel_collate([], processes: 2) } + .to raise_error("There are no reports to be merged") + end + end + + context "when fewer than one process is requested" do + it "shows an error message rather than quietly merging in this process" do + expect { described_class.parallel_collate(["#{resultset_folder}/.resultset.json"], processes: 0) } + .to raise_error(ArgumentError, "processes must be at least 1, got 0") + end + end + + context "when files to be merged" do + before do + allow(described_class).to receive(:run_exit_tasks!) + 5.times { |index| create_mergeable_report("result#{index}", index) } + end + + after do + described_class.clear_result + FileUtils.rm Dir.glob("#{resultset_path}*") + end + + it "produces the report .collate produces for the same inputs" do + expect(collate_with { |paths| described_class.parallel_collate paths, processes: 3 }) + .to eq(collate_with { |paths| described_class.collate paths }) + expect(described_class).to have_received(:run_exit_tasks!).twice + end + + it "produces that report however many processes it is given" do + reports = [1, 2, 4, 9].map do |processes| + collate_with { |paths| described_class.parallel_collate paths, processes: processes } + end + + expect(reports.uniq.size).to eq(1) + end + + it "falls back to merging in this process when the fan-out fails" do + serial = collate_with { |paths| described_class.collate paths } + allow(SimpleCov::ParallelResultMerger).to receive(:merge_resultsets).and_return(nil) + + expect(collate_with { |paths| described_class.parallel_collate paths, processes: 3 }).to eq(serial) + end + + private + + # Each shard covers `sample.rb` plus one file only it loaded, so the + # merged table exercises both summing and union. + def create_mergeable_report(name, index) + coverage = { + source_fixture("sample.rb") => {"lines" => [nil, 1, index, nil, nil, nil, 1, 1, nil, nil]}, + source_fixture("resultset#{index}.rb") => {"lines" => [1, index, nil, 1]} + } + result = SimpleCov::Result.new(coverage) + result.command_name = name + SimpleCov::ResultMerger.store_result(result) + FileUtils.mv resultset_path, "#{resultset_path}#{name}.final" + end + + # Collate the stored shards from a clean slate and return the resultset + # that produced, so two strategies can be compared on equal terms — + # `store_result` merges into whatever is already at `resultset_path`. + def collate_with + FileUtils.rm_f(resultset_path) + described_class.clear_result + yield Dir.glob("#{resultset_folder}/*.final", File::FNM_DOTMATCH) + collated + end + end + end + # Normally wouldn't test private methods but just start has side effects that # cause errors so for time this is pragmatic (tm) describe ".start_coverage_measurement" do diff --git a/spec/support/merge_fuzzer.rb b/spec/support/merge_fuzzer.rb new file mode 100644 index 000000000..dfc1eeda2 --- /dev/null +++ b/spec/support/merge_fuzzer.rb @@ -0,0 +1,119 @@ +# frozen_string_literal: true + +# Deterministic generator of small, adversarial resultset shard sets for +# spec/merge_differential_spec.rb. Each seed reproduces exactly, so a mismatch +# can be replayed and shrunk by hand. +# +# The cases it deliberately produces, since these are the ones the merge rules +# actually turn on and the ones a real-world fixture tends not to contain: +# +# - files missing from some shards entirely (the nil short-circuit in +# `Combine.combine`, which passes a coverage through verbatim) +# - entries with no positive line, i.e. simulated / never-loaded files, mixed +# with executed ones for the same file (`reconcile_synthesized`, #1233) +# - the same source span carrying a different branch id in each shard (the +# drift #1233 is about) +# - two keys in *one* entry sharing an identity, so grouping has to collapse +# within an entry and not only across them +# - class and name varying over one method location (#1234) +# - keys in both array and `inspect`-string form, mixed within a shard +# - `branches` / `methods` keys absent, and line arrays of differing lengths +module MergeFuzzer +module_function + + # A small pool, so collisions and drift arise naturally rather than by + # special-casing. Two `:if` spans differ only in end column, as in #1233. + CONDITION_SPANS = [ + [:if, 2, 2, 4, 10], [:if, 2, 2, 4, 12], [:unless, 6, 4, 8, 20], + [:"&.", 9, 2, 9, 18], [:case, 11, 2, 15, 8] + ].freeze + ARM_SPANS = [ + [:then, 3, 4, 3, 10], [:else, 4, 4, 4, 10], [:when, 12, 4, 12, 9], [:body, 7, 6, 7, 14] + ].freeze + CLASSES = ["Foo", "Bar", "#"].freeze + NAMES = %i[call run inspect].freeze + FILES = %w[a.rb b.rb c.rb d.rb].freeze + + # Returns an array of shards, each a `{file => entry}` coverage hash. + # + # `saturate` puts every file in every shard with both criterion keys always + # present, so `Combine.combine` never takes its nil short-circuit. That + # short-circuit hands a coverage back verbatim, skipping identity collapse + # entirely, which is a divergence in its own right rather than anything about + # how N results combine - see the characterization example in the spec. + def shards(seed, saturate: false) + rng = Random.new(seed) + count = rng.rand(2..5) + files = FILES.first(rng.rand(1..FILES.size)) + membership = files.to_h { |file| [file, present_in(rng, count)] } + built = Array.new(count) do |index| + files.filter_map { |file| [file, entry(rng)] if membership[file].include?(index) }.to_h + end + saturate ? saturated(built) : built + end + + # A stand-in entry for an absent file is inert: no branches or methods to + # contribute, and an all-nil line array merges to nil. + def saturated(shards) + files = shards.flat_map(&:keys).uniq + shards.map do |shard| + files.to_h do |file| + entry = shard[file] || {"lines" => [nil]} + [file, {"lines" => entry["lines"], "branches" => entry["branches"] || {}, + "methods" => entry["methods"] || {}}] + end + end + end + + # Every file lands in at least one shard, but often not all of them. + def present_in(rng, count) + chosen = (0...count).select { rng.rand < 0.75 } + chosen.empty? ? [rng.rand(count)] : chosen + end + + def entry(rng) + entry = {"lines" => lines(rng, executed: rng.rand < 0.6)} + entry["branches"] = branches(rng) unless rng.rand < 0.2 + entry["methods"] = methods(rng) unless rng.rand < 0.2 + entry + end + + # A non-executed entry has no positive count, which is what makes its branch + # and method tuples non-authoritative next to an executed entry. + def lines(rng, executed:) + counts = Array.new(rng.rand(1..6)) { rng.rand < 0.4 ? nil : 0 } + counts[rng.rand(counts.size)] = rng.rand(1..5) if executed + counts + end + + def branches(rng) + Array.new(rng.rand(0..3)) do + [key(rng, CONDITION_SPANS.sample(random: rng)), arms(rng)] + end.to_h + end + + def arms(rng) + Array.new(rng.rand(1..3)) { [key(rng, ARM_SPANS.sample(random: rng)), rng.rand(0..4)] }.to_h + end + + def methods(rng) + Array.new(rng.rand(0..3)) { [method_key(rng), rng.rand(0..4)] }.to_h + end + + # The random id is the per-process counter that differs between shards for + # the same source span. + def key(rng, span) + type, *rest = span + serialize(rng, [type, rng.rand(0..9), *rest]) + end + + def method_key(rng) + _type, *location = CONDITION_SPANS.sample(random: rng) + serialize(rng, [CLASSES.sample(random: rng), NAMES.sample(random: rng), *location]) + end + + # Half of the keys go out in the `inspect` form a JSON round-trip produces. + def serialize(rng, tuple) + rng.rand < 0.5 ? tuple : tuple.inspect + end +end diff --git a/spec/support/merge_reference.rb b/spec/support/merge_reference.rb new file mode 100644 index 000000000..305702a19 --- /dev/null +++ b/spec/support/merge_reference.rb @@ -0,0 +1,109 @@ +# frozen_string_literal: true + +# An independent statement of what merging N resultsets should produce, used as +# the oracle for spec/merge_differential_spec.rb. +# +# `ResultsCombiner` folds resultsets pairwise, which makes the rules it applies +# hard to read off the code. `reconcile_synthesized` especially: it is defined +# on a *pair*, and its N-way meaning only emerges from the fold. This module +# states the N-way rules directly, written for obviousness rather than speed. +# +# - Lines merge across every entry for a file: `nil` only where every entry is +# `nil` (or ended earlier), otherwise the sum with `nil` read as 0. Lines are +# never dropped, whatever `executed?` says (issue #1059). +# - Branches and methods come only from *executed* entries when any entry for +# the file was executed, and from all entries otherwise. A simulated entry +# synthesizes its tuples statically, so merging them into a real run's would +# keep any drifted location as a phantom, permanently-missed branch (#1233). +# - Keys group by identity - location, ignoring the id, and ignoring class and +# name for methods (#1233, #1234). Counts sum, and the first key seen among +# the contributing entries is retained. +module MergeReference +module_function + + def call(shards, branches:, methods:) + shards.flat_map(&:keys).uniq.to_h do |file| + entries = shards.filter_map { |shard| shard[file] } + [file, merge_entries(entries, branches: branches, methods: methods)] + end + end + + def merge_entries(entries, branches:, methods:) + sources = contributing(entries) + merged = {"lines" => merge_lines(entries)} + merged["branches"] = merge_branches(sources) if branches + merged["methods"] = merge_methods(sources) if methods + merged + end + + # The entries whose branch and method tuples are authoritative. + def contributing(entries) + executed = entries.select { |entry| executed?(entry) } + executed.empty? ? entries : executed + end + + # A file some process actually loaded has at least one executed line. + def executed?(entry) + Array(entry["lines"]).any? { |count| count&.positive? } + end + + # Nil when no entry carried lines at all, matching the short-circuit in + # `Combine.combine` for two absent coverages. + def merge_lines(entries) + arrays = entries.filter_map { |entry| entry["lines"] } + return nil if arrays.empty? + + Array.new(arrays.map(&:size).max) { |index| merge_line(arrays, index) } + end + + # A line stays `nil` only where every entry had `nil` there; a `0` on any + # entry makes it relevant-but-uncovered rather than absent (issue #1059). + def merge_line(arrays, index) + counts = arrays.map { |array| array[index] } + counts.all?(&:nil?) ? nil : counts.sum(&:to_i) + end + + def merge_branches(entries) + conditions = entries.flat_map { |entry| (entry["branches"] || {}).to_a } + group(conditions) { |key| span(key) }.to_h do |first_key, pairs| + arms = pairs.flat_map { |(_condition, table)| table.to_a } + [first_key, sum_counts(arms) { |key| span(key) }] + end + end + + def merge_methods(entries) + sum_counts(entries.flat_map { |entry| (entry["methods"] || {}).to_a }) { |key| location(key) } + end + + def sum_counts(pairs, &identity) + group(pairs, &identity).to_h { |first_key, group| [first_key, group.sum { |(_key, count)| count }] } + end + + # Groups `[key, value]` pairs by their key's identity, as + # `[[first_key_seen, [pair, ...]], ...]` in first-seen order. + def group(pairs) + grouped = {} + pairs.each do |pair| + slot = grouped[yield(pair.first)] ||= [pair.first, []] + slot[1] << pair + end + grouped.values + end + + # `[type, id, sl, sc, el, ec]` -> `[type, sl, sc, el, ec]` + def span(key) + tuple = tuple(key) + [tuple[0], *tuple[2..5]] + end + + # `[class, name, sl, sc, el, ec]` -> `[sl, sc, el, ec]` + def location(key) + tuple(key)[2..5] + end + + # Keys are arrays in-process and their `inspect` form once a resultset has + # been through JSON; both must resolve to the same identity. + def tuple(key) + key.is_a?(Array) ? key : SimpleCov::SourceFile::RubyDataParser.call(key) + end +end diff --git a/test_projects/faked_project/Rakefile b/test_projects/faked_project/Rakefile index cf3d9c197..8158b74ff 100644 --- a/test_projects/faked_project/Rakefile +++ b/test_projects/faked_project/Rakefile @@ -26,6 +26,11 @@ task :collate do SimpleCov.collate Dir["coverage/resultset*.json"] end +task :parallel_collate do + require "simplecov" + SimpleCov.parallel_collate Dir["coverage/resultset*.json"], processes: 2 +end + Rake::TestTask.new(:minitest) do |test| test.libs << "minitest" test.test_files = FileList["minitest/**/*_test.rb"].sort From 4870770588d89aa6aa61f57046c7c57dbfa0eede Mon Sep 17 00:00:00 2001 From: Daniel Westendorf Date: Thu, 30 Jul 2026 17:39:41 -0600 Subject: [PATCH 2/5] `SimpleCov.collate` support for processes arg --- CHANGELOG.md | 2 +- README.md | 30 ++++---- benchmarks/collate.rb | 3 +- benchmarks/collate/runner.rb | 6 +- features/test_unit_parallel_collate.feature | 5 +- lib/simplecov.rb | 1 - lib/simplecov/collation.rb | 69 ------------------- lib/simplecov/parallel_result_merger.rb | 24 ------- lib/simplecov/result_merger.rb | 15 ++-- lib/simplecov/result_processing.rb | 40 +++++++++-- .../simplecov/parallel_result_merger.rbs | 7 -- sig/internal/simplecov/result_merger.rbs | 6 +- sig/simplecov.rbs | 30 +++----- spec/parallel_result_merger_spec.rb | 47 ------------- spec/result_merger_spec.rb | 7 ++ spec/simplecov_spec.rb | 28 ++++---- test_projects/faked_project/Rakefile | 2 +- 17 files changed, 104 insertions(+), 218 deletions(-) delete mode 100644 lib/simplecov/collation.rb diff --git a/CHANGELOG.md b/CHANGELOG.md index ceeba3fe8..76a20b005 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,7 @@ Unreleased ## Enhancements * The favicon (a solid square in the overall coverage band's colour) is now drawn by the viewer from the report's own palette instead of shipping as fixed PNGs, so it matches the report's green/yellow/red exactly and follows the light/dark theme, including the in-page toggle. -* New `SimpleCov.parallel_collate`, which is `SimpleCov.collate` with the resultset merge fanned out across forked worker processes. It takes the same arguments plus a required `processes:`, and produces a report identical to `collate`'s — each worker folds a contiguous slice of the file list and the collating process folds the slices back in order, so the resultsets are visited in the order the serial merge visits them. This addresses the wall clock of a large CI matrix's collate step, where the collating process reads, parses and folds hundreds of resultsets in sequence and nearly all the time goes into that fold: merging 160 resultsets covering 1,836 files on a 14-core machine took 8.53s serially and 2.04s across 8 workers via a new `SimpleCov::ParallelResultMerger`, which mirrors `ResultMerger`'s `merge_and_store` / `merge_results` / `merge_resultsets` entry points. `processes` is deliberately not clamped to the core count nor gated on a minimum number of resultsets — only the caller knows what a collate job is allowed to use — and asking for more processes than there are result files just gives one file per process. Merging falls back to the collating process, with the same report and no error, when the runtime cannot fork (JRuby, TruffleRuby, Windows), when there is only one resultset, or when a worker dies. A `benchmarks/collate.rb` harness (`PROCESSES=N`) measures the phases against a saved baseline. +* `SimpleCov.collate` takes a new `processes:` argument that fans the resultset merge out across that many forked worker processes. This addresses the wall clock of a large CI matrix's collate step, where the collating process reads, parses and folds hundreds of resultsets in sequence and nearly all the time goes into that fold: merging 160 resultsets covering 1,836 files on a 14-core machine took 8.53s at the default `processes: 1` and 2.04s across 8 workers. The report is identical either way, not merely equivalent — each worker folds a contiguous slice of the file list and the collating process folds the slices back in order, so the resultsets are visited in the same order a single-process merge visits them. The fan-out lives in a new `SimpleCov::ParallelResultMerger`, whose `merge_resultsets` mirrors the `ResultMerger.merge_resultsets` extracted alongside it. `processes` defaults to 1 and never forks at that value, so existing `collate` calls are unaffected; it is deliberately not clamped to the core count nor gated on a minimum number of resultsets — only the caller knows what a collate job is allowed to use — asking for more processes than there are result files just gives one file per process, and anything below 1 is taken as 1. Merging falls back to the collating process, with the same report and no error, when the runtime cannot fork (JRuby, TruffleRuby, Windows), when there is only one resultset, or when a worker dies. A `benchmarks/collate.rb` harness (`PROCESSES=N`) measures the phases against a saved baseline. 1.0.3 (2026-07-26) ================== diff --git a/README.md b/README.md index b2634cb6f..b3e827c95 100644 --- a/README.md +++ b/README.md @@ -858,9 +858,7 @@ end Collating a handful of resultsets is quick. Collating a few hundred is not: the collating process reads, parses and folds every one of them in sequence, and on a large CI matrix that fold is where nearly all the wall clock goes. -`SimpleCov.parallel_collate` is `SimpleCov.collate` with that fold spread across forked worker processes. It takes the -same arguments — result filenames, an optional profile, an optional configuration block, `ignore_timeout:` — plus a -required `processes:`: +Pass `processes:` to spread that fold across forked worker processes: ```ruby # lib/tasks/coverage_report.rake @@ -869,19 +867,19 @@ namespace :coverage do task :report do require 'simplecov' - SimpleCov.parallel_collate Dir["simplecov-resultset-*/.resultset.json"], processes: 8 + SimpleCov.collate Dir["simplecov-resultset-*/.resultset.json"], processes: 8 end end ``` -The report is identical to the one `collate` produces for the same inputs, not merely equivalent: each worker folds a -contiguous slice of the file list and the collating process folds the slices back in order, so the resultsets are -visited in the same order the serial merge visits them. +The report is identical to the one a single-process collate produces for the same inputs, not merely equivalent: each +worker folds a contiguous slice of the file list and the collating process folds the slices back in order, so the +resultsets are visited in the same order they would be otherwise. -`processes` is required and is deliberately not clamped to your core count, nor gated on some minimum number of -resultsets — how many processes a collate job can afford is something only you know. Asking for more processes than -there are result files simply gives one file per process. A `processes` below 1 raises `ArgumentError`, so use -`[n, 1].max` if the count comes from arithmetic that can reach zero. +`processes` defaults to 1, which never forks — existing `collate` calls behave exactly as before. It is deliberately +not clamped to your core count, nor gated on some minimum number of resultsets: how many processes a collate job can +afford is something only you know. Asking for more processes than there are result files simply gives one file per +process, and anything below 1 is taken as 1, so a count computed from arithmetic that can reach zero needs no guarding. It falls back to merging in the collating process — same report, no error — when the runtime cannot fork (JRuby, TruffleRuby, Windows), when there is only one resultset to fold, or when a worker dies. @@ -889,11 +887,11 @@ TruffleRuby, Windows), when there is only one resultset to fold, or when a worke Merging 160 resultsets covering 1,836 files on a 14-core machine (`benchmarks/collate.rb`, so reproduce it on your own hardware before budgeting for it): -| `processes:` | merge phase | -| --------------------- | ----------- | -| serial (`collate`) | 8.53s | -| 4 | 2.65s | -| 8 | 2.04s | +| `processes:` | merge phase | +| ------------ | ----------- | +| 1 (default) | 8.53s | +| 4 | 2.65s | +| 8 | 2.04s | Memory scales with the worker count rather than the resultset count: each worker folds its slice one file at a time, so it holds one resultset plus its own running total, and the collating process holds one folded total per worker. diff --git a/benchmarks/collate.rb b/benchmarks/collate.rb index b827e6887..3c9480d59 100644 --- a/benchmarks/collate.rb +++ b/benchmarks/collate.rb @@ -26,7 +26,8 @@ # COUNT merge only the first N resultsets — the knob for a fast # iteration loop; merge cost grows with N (default: 160) # PROCESSES fan the merge phase out across N forked workers, as -# `SimpleCov.parallel_collate` does; 1 merges serially (default: 1) +# `SimpleCov.collate processes: N` does; 1 merges in this +# process (default: 1) # SCALE divide `Shape::FILES` by this (default: 4, giving ~1,836 files; # SCALE=1 generates the full 7,345) # SKIP comma-separated trailing phases to skip, e.g. SKIP=format,store diff --git a/benchmarks/collate/runner.rb b/benchmarks/collate/runner.rb index 823d82bba..4072137dc 100644 --- a/benchmarks/collate/runner.rb +++ b/benchmarks/collate/runner.rb @@ -95,10 +95,10 @@ def run_phases(fixture) @files_reported = result&.files&.size end - # With PROCESSES > 1, the fan-out `SimpleCov.parallel_collate` performs — + # With PROCESSES > 1, the fan-out `SimpleCov.collate processes: N` performs — # same merge, same visiting order, spread over forked workers. Falls - # through to the in-process loop if the fan-out bails, which is what the - # real `parallel_collate` does too. + # through to the in-process loop if the fan-out bails, which is what + # `ResultMerger.merge_results` does too. def merge_coverage(paths) return serial_merge_coverage(paths) if processes < 2 diff --git a/features/test_unit_parallel_collate.feature b/features/test_unit_parallel_collate.feature index 3ef81aa2c..c625765b4 100644 --- a/features/test_unit_parallel_collate.feature +++ b/features/test_unit_parallel_collate.feature @@ -1,8 +1,9 @@ @test_unit Feature: - Using SimpleCov.parallel_collate should get the user the same coverage report - SimpleCov.collate does, with the merge fanned out across forked workers. + Using SimpleCov.collate with processes: > 1 should get the user the same + coverage report a single-process collate does, with the merge fanned out + across forked workers. Background: Given I'm working on the project "faked_project" diff --git a/lib/simplecov.rb b/lib/simplecov.rb index 14e3e2d69..cfc97b031 100644 --- a/lib/simplecov.rb +++ b/lib/simplecov.rb @@ -262,7 +262,6 @@ def warn_if_jruby_full_trace_disabled require_relative "simplecov/useless_results_remover" require_relative "simplecov/simulate_coverage" require_relative "simplecov/result_processing" -require_relative "simplecov/collation" require_relative "simplecov/exit_handling" require_relative "simplecov/parallel_coordination" diff --git a/lib/simplecov/collation.rb b/lib/simplecov/collation.rb deleted file mode 100644 index a721baf44..000000000 --- a/lib/simplecov/collation.rb +++ /dev/null @@ -1,69 +0,0 @@ -# frozen_string_literal: true - -# The `collate` entry points: stitch the resultsets written by separate -# test runs — parallel CI jobs, several build machines, a matrix of Ruby -# versions — into one report, either in this process or fanned out across -# forked workers. -module SimpleCov - class << self - # - # Collate a series of SimpleCov result files into a single SimpleCov output. - # - # See README for usage. By default `collate` ignores the merge_timeout - # so all results in all files specified will be merged. Pass - # `ignore_timeout: false` to honor it. - # - def collate(result_filenames, profile = nil, ignore_timeout: true, &config) - collating(result_filenames, profile, config) do - # Use the ResultMerger to produce a single, merged result, ready to use. - ResultMerger.merge_and_store(*result_filenames, ignore_timeout: ignore_timeout) - end - end - - # - # `collate`, with the merge fanned out across `processes` forked worker - # processes. Takes the same arguments and produces the same report: the - # workers fold contiguous slices of `result_filenames` in the order the - # serial merge visits them, so the merged result is identical, not merely - # equivalent. Only the wall clock differs, and only for a collate big - # enough that reading and parsing the resultsets dominates it. - # - # `processes` is required, and is deliberately not clamped to the machine's - # core count nor gated on some minimum number of resultsets: how many - # processes a collate job can afford is the caller's call, not SimpleCov's. - # Asking for more processes than there are result files simply gives one - # file per process. A `processes` below 1 raises rather than quietly - # merging serially, so compute it with `[n, 1].max` if it comes from - # arithmetic that can reach zero. - # - # Falls back to merging in this process — same report, no error — when the - # runtime cannot fork (JRuby, TruffleRuby, Windows), when there is nothing - # worth splitting, or when a worker dies. - # - def parallel_collate(result_filenames, profile = nil, processes:, ignore_timeout: true, &config) - raise ArgumentError, "processes must be at least 1, got #{processes}" if processes < 1 - - collating(result_filenames, profile, config) do - ParallelResultMerger.merge_and_store(*result_filenames, processes: processes, ignore_timeout: ignore_timeout) - end - end - - private - - # The scaffolding both entry points share: validate, apply the caller's - # profile and configuration block, then run the finalizer over whatever - # merged result the given strategy produced. `config` is the caller's - # configuration block, passed as an object because the merge strategy - # occupies the block slot. - def collating(result_filenames, profile, config) - raise ArgumentError, "There are no reports to be merged" if result_filenames.empty? - - initial_setup(profile, &config) - @result = yield - @collating_result = true - run_exit_tasks! - ensure - @collating_result = false - end - end -end diff --git a/lib/simplecov/parallel_result_merger.rb b/lib/simplecov/parallel_result_merger.rb index e28aa6485..9fcca5aec 100644 --- a/lib/simplecov/parallel_result_merger.rb +++ b/lib/simplecov/parallel_result_merger.rb @@ -24,30 +24,6 @@ module SimpleCov module ParallelResultMerger module_function - # - # `ResultMerger.merge_and_store` across `processes` forked workers. - # - def merge_and_store(*file_paths, processes:, ignore_timeout: false) - result = merge_results(*file_paths, processes: processes, ignore_timeout: ignore_timeout) - ResultMerger.store_result(result) if result - result - end - - # - # `ResultMerger.merge_results` across `processes` forked workers, merging - # in this process instead whenever the fan-out did not produce a complete - # merge — a runtime that cannot fork, nothing worth splitting, or a worker - # that died. The result is the same either way; only the time it took to - # get there differs. - # - def merge_results(*file_paths, processes:, ignore_timeout: false) - command_names, coverage = - merge_resultsets(file_paths, processes: processes, ignore_timeout: ignore_timeout) || - ResultMerger.merge_resultsets(file_paths, ignore_timeout: ignore_timeout) - - ResultMerger.create_result(command_names, coverage) - end - # # `ResultMerger.merge_resultsets` across at most `processes` forked # workers: same arguments, same `[command_names, coverage]` return. diff --git a/lib/simplecov/result_merger.rb b/lib/simplecov/result_merger.rb index 3985ee81b..70e0e3fb2 100644 --- a/lib/simplecov/result_merger.rb +++ b/lib/simplecov/result_merger.rb @@ -16,14 +16,21 @@ def resultset_path ResultsetStore.resultset_path end - def merge_and_store(*file_paths, ignore_timeout: false) - result = merge_results(*file_paths, ignore_timeout: ignore_timeout) + def merge_and_store(*file_paths, processes: 1, ignore_timeout: false) + result = merge_results(*file_paths, processes: processes, ignore_timeout: ignore_timeout) store_result(result) if result result end - def merge_results(*file_paths, ignore_timeout: false) - command_names, coverage = merge_resultsets(file_paths, ignore_timeout: ignore_timeout) + # `processes:` above 1 asks `ParallelResultMerger` to fan the read-and- + # combine out across that many forked workers. It answers nil — and we + # merge here instead — when it could not: a runtime that cannot fork, + # nothing worth splitting, or a worker that died. The result is the same + # either way; only the time it took to get there differs. + def merge_results(*file_paths, processes: 1, ignore_timeout: false) + command_names, coverage = + ParallelResultMerger.merge_resultsets(file_paths, processes: processes, ignore_timeout: ignore_timeout) || + merge_resultsets(file_paths, ignore_timeout: ignore_timeout) create_result(command_names, coverage) end diff --git a/lib/simplecov/result_processing.rb b/lib/simplecov/result_processing.rb index aa5893017..c3c267a15 100644 --- a/lib/simplecov/result_processing.rb +++ b/lib/simplecov/result_processing.rb @@ -1,12 +1,44 @@ # frozen_string_literal: true # Result-building façade: turns the raw `Coverage.result` hash into a -# `SimpleCov::Result`, applies filters and groups, and drives merging -# across test suites via `SimpleCov::ResultMerger`. The `collate` entry -# points for stitching disparate resultsets together live alongside it -# in `simplecov/collation`. +# `SimpleCov::Result`, applies filters and groups, drives merging +# across test suites via `SimpleCov::ResultMerger`, and exposes the +# `collate` entry point for stitching disparate resultsets together. module SimpleCov class << self + # + # Collate a series of SimpleCov result files into a single SimpleCov output. + # + # See README for usage. By default `collate` ignores the merge_timeout + # so all results in all files specified will be merged. Pass + # `ignore_timeout: false` to honor it. + # + # `processes:` above 1 fans the merge out across that many forked worker + # processes, for a collate big enough that reading and parsing the + # resultsets dominates it. The report is identical either way, not merely + # equivalent — the workers visit the resultsets in the order the + # single-process merge visits them — and one process never forks at all. + # The count is deliberately not clamped to the machine's core count nor + # gated on some minimum number of resultsets: how many processes a collate + # job can afford is the caller's call, not SimpleCov's. Anything below 1 is + # taken as 1, so a count computed from arithmetic that can reach zero needs + # no guarding. See `SimpleCov::ParallelResultMerger`. + # + def collate(result_filenames, profile = nil, processes: 1, ignore_timeout: true, &) + raise ArgumentError, "There are no reports to be merged" if result_filenames.empty? + + initial_setup(profile, &) + + # Use the ResultMerger to produce a single, merged result, ready to use. + @result = ResultMerger.merge_and_store(*result_filenames, processes: [1, processes].max, + ignore_timeout: ignore_timeout) + + @collating_result = true + run_exit_tasks! + ensure + @collating_result = false + end + # # Returns the result for the current coverage run, merging it across test suites # from cache using SimpleCov::ResultMerger if use_merging is activated (default) diff --git a/sig/internal/simplecov/parallel_result_merger.rbs b/sig/internal/simplecov/parallel_result_merger.rbs index 87f0fe421..4ceb954d8 100644 --- a/sig/internal/simplecov/parallel_result_merger.rbs +++ b/sig/internal/simplecov/parallel_result_merger.rbs @@ -12,13 +12,6 @@ module SimpleCov # can redo the merge in this process. # module ParallelResultMerger - # `ResultMerger.merge_and_store` across `processes` forked workers. - def self?.merge_and_store: (*untyped file_paths, processes: Integer, ?ignore_timeout: bool) -> untyped - - # `ResultMerger.merge_results` across `processes` forked workers, merging - # in this process when the fan-out did not produce a complete merge. - def self?.merge_results: (*untyped file_paths, processes: Integer, ?ignore_timeout: bool) -> untyped - # # `ResultMerger.merge_resultsets` across at most `processes` forked # workers: same arguments, same `[command_names, coverage]` return, or diff --git a/sig/internal/simplecov/result_merger.rbs b/sig/internal/simplecov/result_merger.rbs index 82cad973a..a149ee839 100644 --- a/sig/internal/simplecov/result_merger.rbs +++ b/sig/internal/simplecov/result_merger.rbs @@ -7,9 +7,11 @@ module SimpleCov module ResultMerger def self.resultset_path: () -> untyped - def self.merge_and_store: (*untyped file_paths, ?ignore_timeout: bool) -> untyped + def self.merge_and_store: (*untyped file_paths, ?processes: Integer, ?ignore_timeout: bool) -> untyped - def self.merge_results: (*untyped file_paths, ?ignore_timeout: bool) -> untyped + # `processes:` above 1 fans the read-and-combine out across that many + # forked workers, merging here when the fan-out could not run. + def self.merge_results: (*untyped file_paths, ?processes: Integer, ?ignore_timeout: bool) -> untyped # Read the given resultset files and combine them into a single # `[command_names, coverage]` pair, one file at a time. diff --git a/sig/simplecov.rbs b/sig/simplecov.rbs index 1f8bf9b2d..b30f9984d 100644 --- a/sig/simplecov.rbs +++ b/sig/simplecov.rbs @@ -96,34 +96,22 @@ module SimpleCov VERSION: String end -# ----- collation ----- - -# The `collate` entry points: stitch the resultsets written by separate -# test runs into one report, in this process or across forked workers. -module SimpleCov - # Collate a series of resultset files into a single merged report. By - # default ignores `merge_timeout` so every listed resultset merges; - # pass `ignore_timeout: false` to honor it. - def self.collate: (Array[String] result_filenames, ?(String | Symbol)? profile, ?ignore_timeout: bool) ?{ () [self: singleton(SimpleCov)] -> void } -> void - - # `collate`, with the merge fanned out across `processes` forked worker - # processes. Same arguments, same report. `processes` is required and is - # not clamped; below 1 it raises. - def self.parallel_collate: (Array[String] result_filenames, ?(String | Symbol)? profile, processes: Integer, ?ignore_timeout: bool) ?{ () [self: singleton(SimpleCov)] -> void } -> void - - # Implementation detail of the two entry points above (private in Ruby; - # RBS cannot mark singleton methods private). - def self.collating: (Array[String] result_filenames, (String | Symbol)? profile, (^() [self: singleton(SimpleCov)] -> void)? config) { () -> Result? } -> void -end - # ----- result_processing ----- # Result-building façade: turns the raw `Coverage.result` hash into a -# `SimpleCov::Result`, applies filters and groups, and drives merging. +# `SimpleCov::Result`, applies filters and groups, drives merging, and +# exposes the `collate` entry point. module SimpleCov self.@result: Result? self.@collating_result: bool? + # Collate a series of resultset files into a single merged report. By + # default ignores `merge_timeout` so every listed resultset merges; + # pass `ignore_timeout: false` to honor it. `processes:` above 1 fans the + # merge out across that many forked workers, for an identical report; + # anything below 1 is taken as 1. + def self.collate: (Array[String] result_filenames, ?(String | Symbol)? profile, ?processes: Integer, ?ignore_timeout: bool) ?{ () [self: singleton(SimpleCov)] -> void } -> void + # The result for the current coverage run, merged across test suites # when merging is active. nil when no coverage was tracked. def self.result: () -> Result? diff --git a/spec/parallel_result_merger_spec.rb b/spec/parallel_result_merger_spec.rb index c1892fed7..c5f4275ed 100644 --- a/spec/parallel_result_merger_spec.rb +++ b/spec/parallel_result_merger_spec.rb @@ -36,53 +36,6 @@ after { FileUtils.remove_entry(resultset_dir) } - describe ".merge_and_store" do - before { FileUtils.mkdir_p(File.dirname(SimpleCov::ResultMerger.resultset_path)) } - - after { FileUtils.rm_f(SimpleCov::ResultMerger.resultset_path) } - - it "produces the result ResultMerger.merge_and_store produces" do - result = described_class.merge_and_store(*paths, processes: 3, ignore_timeout: true) - - expect(result.to_hash).to eq(SimpleCov::ResultMerger.create_result(*serial).to_hash) - end - - it "has the result stored" do - described_class.merge_and_store(*paths, processes: 3, ignore_timeout: true) - - expect(SimpleCov::ResultMerger.read_resultset.keys).to eq([serial.first.sort.join(", ")]) - end - - # The workers do the dropping, so the "older than merge_timeout" warning - # comes from them and reaches the inherited stderr rather than anything - # this process can capture. - it "stores nothing when every resultset is outdated" do - allow(SimpleCov::ResultMerger).to receive(:store_result) - stale = Array.new(3) do |i| - write_resultset("old#{i}", {source_fixture("sample.rb") => {"lines" => [1]}}, outdated: true) - end - - expect(described_class.merge_and_store(*stale, processes: 2)).to be_nil - expect(SimpleCov::ResultMerger).not_to have_received(:store_result) - end - end - - describe ".merge_results" do - it "produces the result ResultMerger.merge_results produces" do - result = described_class.merge_results(*paths, processes: 3, ignore_timeout: true) - - expect(result.to_hash).to eq(SimpleCov::ResultMerger.merge_results(*paths, ignore_timeout: true).to_hash) - end - - it "merges in this process when the fan-out cannot run" do - allow(described_class).to receive(:fork_supported?).and_return(false) - - result = described_class.merge_results(*paths, processes: 3, ignore_timeout: true) - - expect(result.to_hash).to eq(SimpleCov::ResultMerger.merge_results(*paths, ignore_timeout: true).to_hash) - end - end - describe ".merge_resultsets" do it "produces the pair ResultMerger.merge_resultsets produces" do expect(described_class.merge_resultsets(paths, processes: 3, ignore_timeout: true)).to eq(serial) diff --git a/spec/result_merger_spec.rb b/spec/result_merger_spec.rb index 270f2988b..12623f0b1 100644 --- a/spec/result_merger_spec.rb +++ b/spec/result_merger_spec.rb @@ -138,6 +138,13 @@ expect_resultset_1_and_2_merged(described_class.read_resultset) end + + it "merges to the same result across forked workers" do + result = described_class.merge_and_store(resultset1_path, resultset2_path, processes: 2) + + expect_resultset_1_and_2_merged(result.to_hash) + expect_resultset_1_and_2_merged(described_class.read_resultset) + end end context "when 1 resultset is outdated" do diff --git a/spec/simplecov_spec.rb b/spec/simplecov_spec.rb index a6603ed4c..9697eba47 100644 --- a/spec/simplecov_spec.rb +++ b/spec/simplecov_spec.rb @@ -612,7 +612,7 @@ def self.after_run(&block) described_class.collate(["coverage/worker/.resultset.json"]) expect(SimpleCov::ResultMerger).to have_received(:merge_and_store) - .with("coverage/worker/.resultset.json", ignore_timeout: true) + .with("coverage/worker/.resultset.json", processes: 1, ignore_timeout: true) expect(described_class).to have_received(:write_last_run).with(result) end end @@ -1100,7 +1100,7 @@ def expect_merged end end - describe ".parallel_collate" do + describe ".collate across processes" do let(:resultset_path) { SimpleCov::ResultMerger.resultset_path } let(:resultset_folder) { File.dirname(resultset_path) } @@ -1111,18 +1111,11 @@ def expect_merged context "when no files to be merged" do it "shows an error message" do - expect { described_class.parallel_collate([], processes: 2) } + expect { described_class.collate([], processes: 2) } .to raise_error("There are no reports to be merged") end end - context "when fewer than one process is requested" do - it "shows an error message rather than quietly merging in this process" do - expect { described_class.parallel_collate(["#{resultset_folder}/.resultset.json"], processes: 0) } - .to raise_error(ArgumentError, "processes must be at least 1, got 0") - end - end - context "when files to be merged" do before do allow(described_class).to receive(:run_exit_tasks!) @@ -1134,25 +1127,30 @@ def expect_merged FileUtils.rm Dir.glob("#{resultset_path}*") end - it "produces the report .collate produces for the same inputs" do - expect(collate_with { |paths| described_class.parallel_collate paths, processes: 3 }) + it "produces the report a single-process collate produces" do + expect(collate_with { |paths| described_class.collate paths, processes: 3 }) .to eq(collate_with { |paths| described_class.collate paths }) expect(described_class).to have_received(:run_exit_tasks!).twice end it "produces that report however many processes it is given" do reports = [1, 2, 4, 9].map do |processes| - collate_with { |paths| described_class.parallel_collate paths, processes: processes } + collate_with { |paths| described_class.collate paths, processes: processes } end expect(reports.uniq.size).to eq(1) end - it "falls back to merging in this process when the fan-out fails" do + it "takes a process count below 1 as 1 rather than raising" do + expect(collate_with { |paths| described_class.collate paths, processes: 0 }) + .to eq(collate_with { |paths| described_class.collate paths }) + end + + it "merges in this process when the fan-out cannot run" do serial = collate_with { |paths| described_class.collate paths } allow(SimpleCov::ParallelResultMerger).to receive(:merge_resultsets).and_return(nil) - expect(collate_with { |paths| described_class.parallel_collate paths, processes: 3 }).to eq(serial) + expect(collate_with { |paths| described_class.collate paths, processes: 3 }).to eq(serial) end private diff --git a/test_projects/faked_project/Rakefile b/test_projects/faked_project/Rakefile index 8158b74ff..929905e6f 100644 --- a/test_projects/faked_project/Rakefile +++ b/test_projects/faked_project/Rakefile @@ -28,7 +28,7 @@ end task :parallel_collate do require "simplecov" - SimpleCov.parallel_collate Dir["coverage/resultset*.json"], processes: 2 + SimpleCov.collate Dir["coverage/resultset*.json"], processes: 2 end Rake::TestTask.new(:minitest) do |test| From f9754571f611ed8651d615c7c14c5fe107782afa Mon Sep 17 00:00:00 2001 From: Daniel Westendorf Date: Fri, 31 Jul 2026 17:00:14 -0600 Subject: [PATCH 3/5] Clean up API so that we aren't leaking processes variable all over --- CHANGELOG.md | 2 +- README.md | 6 +-- lib/simplecov/parallel_result_merger.rb | 24 +++++++++ lib/simplecov/result_merger.rb | 15 ++---- lib/simplecov/result_processing.rb | 11 +++- .../simplecov/parallel_result_merger.rbs | 7 +++ sig/internal/simplecov/result_merger.rbs | 6 +-- sig/simplecov.rbs | 4 ++ spec/parallel_result_merger_spec.rb | 50 +++++++++++++++++++ spec/result_merger_spec.rb | 7 --- spec/simplecov_spec.rb | 2 +- 11 files changed, 105 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 76a20b005..d8b3e906f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,7 @@ Unreleased ## Enhancements * The favicon (a solid square in the overall coverage band's colour) is now drawn by the viewer from the report's own palette instead of shipping as fixed PNGs, so it matches the report's green/yellow/red exactly and follows the light/dark theme, including the in-page toggle. -* `SimpleCov.collate` takes a new `processes:` argument that fans the resultset merge out across that many forked worker processes. This addresses the wall clock of a large CI matrix's collate step, where the collating process reads, parses and folds hundreds of resultsets in sequence and nearly all the time goes into that fold: merging 160 resultsets covering 1,836 files on a 14-core machine took 8.53s at the default `processes: 1` and 2.04s across 8 workers. The report is identical either way, not merely equivalent — each worker folds a contiguous slice of the file list and the collating process folds the slices back in order, so the resultsets are visited in the same order a single-process merge visits them. The fan-out lives in a new `SimpleCov::ParallelResultMerger`, whose `merge_resultsets` mirrors the `ResultMerger.merge_resultsets` extracted alongside it. `processes` defaults to 1 and never forks at that value, so existing `collate` calls are unaffected; it is deliberately not clamped to the core count nor gated on a minimum number of resultsets — only the caller knows what a collate job is allowed to use — asking for more processes than there are result files just gives one file per process, and anything below 1 is taken as 1. Merging falls back to the collating process, with the same report and no error, when the runtime cannot fork (JRuby, TruffleRuby, Windows), when there is only one resultset, or when a worker dies. A `benchmarks/collate.rb` harness (`PROCESSES=N`) measures the phases against a saved baseline. +* `SimpleCov.collate` takes a new `processes:` argument that fans the resultset merge out across that many forked worker processes. This addresses the wall clock of a large CI matrix's collate step, where the collating process reads, parses and folds hundreds of resultsets in sequence and nearly all the time goes into that fold: merging 160 resultsets covering 1,836 files on a 14-core machine took 4.53s at the default `processes: 1` and 1.35s across 8 workers. The report is identical either way, not merely equivalent — each worker folds a contiguous slice of the file list and the collating process folds the slices back in order, so the resultsets are visited in the same order a single-process merge visits them. The fan-out lives in a new `SimpleCov::ParallelResultMerger`, whose `merge_resultsets` mirrors the `ResultMerger.merge_resultsets` extracted alongside it. `processes` defaults to 1 and never forks at that value, so existing `collate` calls are unaffected; it is deliberately not clamped to the core count nor gated on a minimum number of resultsets — only the caller knows what a collate job is allowed to use — asking for more processes than there are result files just gives one file per process, and anything below 1 is taken as 1. Merging falls back to the collating process, with the same report and no error, when the runtime cannot fork (JRuby, TruffleRuby, Windows), when there is only one resultset, or when a worker dies. A `benchmarks/collate.rb` harness (`PROCESSES=N`) measures the phases against a saved baseline. 1.0.3 (2026-07-26) ================== diff --git a/README.md b/README.md index b3e827c95..775f57f96 100644 --- a/README.md +++ b/README.md @@ -889,9 +889,9 @@ hardware before budgeting for it): | `processes:` | merge phase | | ------------ | ----------- | -| 1 (default) | 8.53s | -| 4 | 2.65s | -| 8 | 2.04s | +| 1 (default) | 4.53s | +| 4 | 1.70s | +| 8 | 1.35s | Memory scales with the worker count rather than the resultset count: each worker folds its slice one file at a time, so it holds one resultset plus its own running total, and the collating process holds one folded total per worker. diff --git a/lib/simplecov/parallel_result_merger.rb b/lib/simplecov/parallel_result_merger.rb index 9fcca5aec..e28aa6485 100644 --- a/lib/simplecov/parallel_result_merger.rb +++ b/lib/simplecov/parallel_result_merger.rb @@ -24,6 +24,30 @@ module SimpleCov module ParallelResultMerger module_function + # + # `ResultMerger.merge_and_store` across `processes` forked workers. + # + def merge_and_store(*file_paths, processes:, ignore_timeout: false) + result = merge_results(*file_paths, processes: processes, ignore_timeout: ignore_timeout) + ResultMerger.store_result(result) if result + result + end + + # + # `ResultMerger.merge_results` across `processes` forked workers, merging + # in this process instead whenever the fan-out did not produce a complete + # merge — a runtime that cannot fork, nothing worth splitting, or a worker + # that died. The result is the same either way; only the time it took to + # get there differs. + # + def merge_results(*file_paths, processes:, ignore_timeout: false) + command_names, coverage = + merge_resultsets(file_paths, processes: processes, ignore_timeout: ignore_timeout) || + ResultMerger.merge_resultsets(file_paths, ignore_timeout: ignore_timeout) + + ResultMerger.create_result(command_names, coverage) + end + # # `ResultMerger.merge_resultsets` across at most `processes` forked # workers: same arguments, same `[command_names, coverage]` return. diff --git a/lib/simplecov/result_merger.rb b/lib/simplecov/result_merger.rb index 70e0e3fb2..3985ee81b 100644 --- a/lib/simplecov/result_merger.rb +++ b/lib/simplecov/result_merger.rb @@ -16,21 +16,14 @@ def resultset_path ResultsetStore.resultset_path end - def merge_and_store(*file_paths, processes: 1, ignore_timeout: false) - result = merge_results(*file_paths, processes: processes, ignore_timeout: ignore_timeout) + def merge_and_store(*file_paths, ignore_timeout: false) + result = merge_results(*file_paths, ignore_timeout: ignore_timeout) store_result(result) if result result end - # `processes:` above 1 asks `ParallelResultMerger` to fan the read-and- - # combine out across that many forked workers. It answers nil — and we - # merge here instead — when it could not: a runtime that cannot fork, - # nothing worth splitting, or a worker that died. The result is the same - # either way; only the time it took to get there differs. - def merge_results(*file_paths, processes: 1, ignore_timeout: false) - command_names, coverage = - ParallelResultMerger.merge_resultsets(file_paths, processes: processes, ignore_timeout: ignore_timeout) || - merge_resultsets(file_paths, ignore_timeout: ignore_timeout) + def merge_results(*file_paths, ignore_timeout: false) + command_names, coverage = merge_resultsets(file_paths, ignore_timeout: ignore_timeout) create_result(command_names, coverage) end diff --git a/lib/simplecov/result_processing.rb b/lib/simplecov/result_processing.rb index c3c267a15..5fc28b874 100644 --- a/lib/simplecov/result_processing.rb +++ b/lib/simplecov/result_processing.rb @@ -30,8 +30,7 @@ def collate(result_filenames, profile = nil, processes: 1, ignore_timeout: true, initial_setup(profile, &) # Use the ResultMerger to produce a single, merged result, ready to use. - @result = ResultMerger.merge_and_store(*result_filenames, processes: [1, processes].max, - ignore_timeout: ignore_timeout) + @result = merge_collated(result_filenames, [1, processes].max, ignore_timeout) @collating_result = true run_exit_tasks! @@ -129,6 +128,14 @@ def round_coverage(coverage) private + # A single worker would merge the whole list in this process anyway, so the + # default takes exactly the path `collate` has always taken and never forks. + def merge_collated(result_filenames, processes, ignore_timeout) + return ResultMerger.merge_and_store(*result_filenames, ignore_timeout: ignore_timeout) if processes < 2 + + ParallelResultMerger.merge_and_store(*result_filenames, processes: processes, ignore_timeout: ignore_timeout) + end + def initial_setup(profile, &block) load_profile(profile) if profile configure(&block) if block diff --git a/sig/internal/simplecov/parallel_result_merger.rbs b/sig/internal/simplecov/parallel_result_merger.rbs index 4ceb954d8..87f0fe421 100644 --- a/sig/internal/simplecov/parallel_result_merger.rbs +++ b/sig/internal/simplecov/parallel_result_merger.rbs @@ -12,6 +12,13 @@ module SimpleCov # can redo the merge in this process. # module ParallelResultMerger + # `ResultMerger.merge_and_store` across `processes` forked workers. + def self?.merge_and_store: (*untyped file_paths, processes: Integer, ?ignore_timeout: bool) -> untyped + + # `ResultMerger.merge_results` across `processes` forked workers, merging + # in this process when the fan-out did not produce a complete merge. + def self?.merge_results: (*untyped file_paths, processes: Integer, ?ignore_timeout: bool) -> untyped + # # `ResultMerger.merge_resultsets` across at most `processes` forked # workers: same arguments, same `[command_names, coverage]` return, or diff --git a/sig/internal/simplecov/result_merger.rbs b/sig/internal/simplecov/result_merger.rbs index a149ee839..82cad973a 100644 --- a/sig/internal/simplecov/result_merger.rbs +++ b/sig/internal/simplecov/result_merger.rbs @@ -7,11 +7,9 @@ module SimpleCov module ResultMerger def self.resultset_path: () -> untyped - def self.merge_and_store: (*untyped file_paths, ?processes: Integer, ?ignore_timeout: bool) -> untyped + def self.merge_and_store: (*untyped file_paths, ?ignore_timeout: bool) -> untyped - # `processes:` above 1 fans the read-and-combine out across that many - # forked workers, merging here when the fan-out could not run. - def self.merge_results: (*untyped file_paths, ?processes: Integer, ?ignore_timeout: bool) -> untyped + def self.merge_results: (*untyped file_paths, ?ignore_timeout: bool) -> untyped # Read the given resultset files and combine them into a single # `[command_names, coverage]` pair, one file at a time. diff --git a/sig/simplecov.rbs b/sig/simplecov.rbs index b30f9984d..a645445d3 100644 --- a/sig/simplecov.rbs +++ b/sig/simplecov.rbs @@ -112,6 +112,10 @@ module SimpleCov # anything below 1 is taken as 1. def self.collate: (Array[String] result_filenames, ?(String | Symbol)? profile, ?processes: Integer, ?ignore_timeout: bool) ?{ () [self: singleton(SimpleCov)] -> void } -> void + # Implementation detail of `collate` (private in Ruby; RBS cannot mark + # singleton methods private). + def self.merge_collated: (Array[String] result_filenames, Integer processes, bool ignore_timeout) -> Result? + # The result for the current coverage run, merged across test suites # when merging is active. nil when no coverage was tracked. def self.result: () -> Result? diff --git a/spec/parallel_result_merger_spec.rb b/spec/parallel_result_merger_spec.rb index c5f4275ed..73bf637d5 100644 --- a/spec/parallel_result_merger_spec.rb +++ b/spec/parallel_result_merger_spec.rb @@ -36,6 +36,56 @@ after { FileUtils.remove_entry(resultset_dir) } + describe ".merge_and_store" do + before { FileUtils.mkdir_p(File.dirname(SimpleCov::ResultMerger.resultset_path)) } + + after { FileUtils.rm_f(SimpleCov::ResultMerger.resultset_path) } + + it "produces the result ResultMerger.merge_and_store produces" do + result = described_class.merge_and_store(*paths, processes: 3, ignore_timeout: true) + + expect(result.original_result) + .to eq(SimpleCov::ResultMerger.merge_results(*paths, ignore_timeout: true).original_result) + end + + it "has the result stored" do + described_class.merge_and_store(*paths, processes: 3, ignore_timeout: true) + + expect(SimpleCov::ResultMerger.read_resultset.keys).to eq([serial.first.sort.join(", ")]) + end + + # The workers do the dropping, so the "older than merge_timeout" warning + # comes from them and reaches the inherited stderr rather than anything + # this process can capture. + it "stores nothing when every resultset is outdated" do + allow(SimpleCov::ResultMerger).to receive(:store_result) + stale = Array.new(3) do |index| + write_resultset("old#{index}", {source_fixture("sample.rb") => {"lines" => [1]}}, outdated: true) + end + + expect(described_class.merge_and_store(*stale, processes: 2)).to be_nil + expect(SimpleCov::ResultMerger).not_to have_received(:store_result) + end + end + + describe ".merge_results" do + it "produces the result ResultMerger.merge_results produces" do + result = described_class.merge_results(*paths, processes: 3, ignore_timeout: true) + + expect(result.original_result) + .to eq(SimpleCov::ResultMerger.merge_results(*paths, ignore_timeout: true).original_result) + end + + it "merges in this process when the fan-out cannot run" do + allow(described_class).to receive(:fork_supported?).and_return(false) + + result = described_class.merge_results(*paths, processes: 3, ignore_timeout: true) + + expect(result.original_result) + .to eq(SimpleCov::ResultMerger.merge_results(*paths, ignore_timeout: true).original_result) + end + end + describe ".merge_resultsets" do it "produces the pair ResultMerger.merge_resultsets produces" do expect(described_class.merge_resultsets(paths, processes: 3, ignore_timeout: true)).to eq(serial) diff --git a/spec/result_merger_spec.rb b/spec/result_merger_spec.rb index 12623f0b1..270f2988b 100644 --- a/spec/result_merger_spec.rb +++ b/spec/result_merger_spec.rb @@ -138,13 +138,6 @@ expect_resultset_1_and_2_merged(described_class.read_resultset) end - - it "merges to the same result across forked workers" do - result = described_class.merge_and_store(resultset1_path, resultset2_path, processes: 2) - - expect_resultset_1_and_2_merged(result.to_hash) - expect_resultset_1_and_2_merged(described_class.read_resultset) - end end context "when 1 resultset is outdated" do diff --git a/spec/simplecov_spec.rb b/spec/simplecov_spec.rb index 9697eba47..5650e526a 100644 --- a/spec/simplecov_spec.rb +++ b/spec/simplecov_spec.rb @@ -612,7 +612,7 @@ def self.after_run(&block) described_class.collate(["coverage/worker/.resultset.json"]) expect(SimpleCov::ResultMerger).to have_received(:merge_and_store) - .with("coverage/worker/.resultset.json", processes: 1, ignore_timeout: true) + .with("coverage/worker/.resultset.json", ignore_timeout: true) expect(described_class).to have_received(:write_last_run).with(result) end end From 0bd2d3f3c81b8f8699861864a2f6b994b3eb99fe Mon Sep 17 00:00:00 2001 From: Daniel Westendorf Date: Sun, 2 Aug 2026 19:37:41 -0600 Subject: [PATCH 4/5] Clean up fork detection --- lib/simplecov/parallel_result_merger.rb | 76 +++++++------------ .../simplecov/parallel_result_merger.rbs | 12 +-- spec/helper.rb | 13 ++++ spec/parallel_result_merger_spec.rb | 70 ++++++++++------- 4 files changed, 85 insertions(+), 86 deletions(-) diff --git a/lib/simplecov/parallel_result_merger.rb b/lib/simplecov/parallel_result_merger.rb index e28aa6485..3a6c39773 100644 --- a/lib/simplecov/parallel_result_merger.rb +++ b/lib/simplecov/parallel_result_merger.rb @@ -60,15 +60,18 @@ def merge_resultsets(file_paths, processes:, ignore_timeout: false) # One worker folds the whole list anyway, and one file is a fold of # one — in both cases the fork and the round trip are pure overhead. return nil if processes < 2 || file_paths.size < 2 - return nil unless fork_supported? + # The portable feature test, and the one the rest of the ecosystem uses. + # CRuby leaves `fork` undefined on Windows; JRuby and TruffleRuby cannot + # fork on the JVM and deliberately answer false here so libraries can + # detect that without rescuing an exception — TruffleRuby's compatibility + # guide names this as the correct check. They do still define + # `Kernel#fork` and raise `NotImplementedError` from it, so probing that + # instead would answer true and send them down the fan-out. + return nil unless Process.respond_to?(:fork) fan_out(chunk(file_paths, processes), ignore_timeout: ignore_timeout) end - def fork_supported? - Process.respond_to?(:fork) - end - # Contiguous slices whose sizes differ by at most one, so no worker is # left folding twice its share while the others idle. There are never # more slices than files: asking for more processes than there are @@ -81,53 +84,37 @@ def chunk(file_paths, processes) Array.new(groups) { |index| remaining.shift(base + (index < remainder ? 1 : 0)) } end + # A `fork` that fails here raises, and is left to. `merge_resultsets` has + # already excluded the runtimes that never fork, so what remains is the OS + # refusing a process we expected to get — EAGAIN at RLIMIT_NPROC, ENOMEM + # under memory pressure. That says something is wrong with the machine + # rather than with the merge, and quietly absorbing it would hide it. def fan_out(chunks, ignore_timeout:) - workers = spawn_workers(chunks, ignore_timeout: ignore_timeout) - return nil unless workers - + workers = chunks.map { |chunk| spawn_worker(chunk, ignore_timeout: ignore_timeout) } payloads = collect(workers) - payloads && ResultMerger.merge_coverage(*payloads) - end - # nil when this runtime turns out not to support forking after all. - # JRuby, TruffleRuby and Windows define `Process.fork` and raise only - # when it is called, so this — rather than the `respond_to?` probe — is - # what actually detects them. Anything already spawned is torn down. - def spawn_workers(chunks, ignore_timeout:) - workers = [] #: Array[Hash[Symbol, untyped]] - # Accumulated rather than mapped so the rescue below can still see the - # workers spawned before the failing fork. - chunks.each { |chunk| workers << spawn_worker(chunk, ignore_timeout: ignore_timeout) } # rubocop:disable Style/MapIntoArray - workers - rescue NotImplementedError - # @type var workers: Array[Hash[Symbol, untyped]] - shut_down(workers) - nil + payloads && ResultMerger.merge_coverage(*payloads) end def spawn_worker(chunk, ignore_timeout:) reader, writer = IO.pipe - - pid = fork do - # simplecov:disable — the child's lines are measured in the child, - # which exits via `exit!` without reporting; `run_worker` itself is - # covered by calling it directly. - reader.close - exit!(run_worker(chunk, writer, ignore_timeout: ignore_timeout)) - # simplecov:enable - end - + pid = fork { run_in_child(reader, writer, chunk, ignore_timeout) } writer.close + {pid: pid, reader: reader} end - # The body of a worker: fold the slice, ship it back, and report the exit + # Everything the child does. `exit!` rather than `exit` because it must + # never fall through to the collating process's inherited `at_exit` + # handlers — SimpleCov's own report generation included. + def run_in_child(reader, writer, chunk, ignore_timeout) + reader.close + exit!(run_worker(chunk, writer, ignore_timeout: ignore_timeout)) + end + + # The body of a worker: merge the slice, ship it back, and report the exit # status the child should terminate with. Kept free of the exit itself so # it can be exercised in-process. - # - # A child must never fall through to the collating process's `at_exit` - # handlers — SimpleCov's own report generation included — which is why - # `spawn_worker` ends it with `exit!` rather than `exit`. def run_worker(chunk, writer, ignore_timeout:) Marshal.dump(ResultMerger.merge_resultsets(chunk, ignore_timeout: ignore_timeout), writer) writer.close @@ -186,16 +173,7 @@ def warn_about_failed_workers(failed, total) return unless SimpleCov.print_errors warn "[SimpleCov]: parallel merge did not complete (#{failed} of #{total} workers failed); " \ - "falling back to merging the resultsets in this process." - end - - # Best-effort teardown for workers spawned before a fork failed. Closing - # the read end makes a worker still writing die of EPIPE. - def shut_down(workers) - workers.each do |worker| - worker[:reader].close - succeeded?(worker[:pid]) - end + "merging the resultsets in this process instead." end end end diff --git a/sig/internal/simplecov/parallel_result_merger.rbs b/sig/internal/simplecov/parallel_result_merger.rbs index 87f0fe421..1cd3f7a7d 100644 --- a/sig/internal/simplecov/parallel_result_merger.rbs +++ b/sig/internal/simplecov/parallel_result_merger.rbs @@ -26,19 +26,18 @@ module SimpleCov # def self?.merge_resultsets: (Array[String] file_paths, processes: Integer, ?ignore_timeout: bool) -> untyped - def self?.fork_supported?: () -> bool - # Contiguous slices whose sizes differ by at most one. There are never # more slices than files. def self?.chunk: (Array[String] file_paths, Integer processes) -> Array[Array[String]] + # Raises if the OS refuses a fork; see the method comment. def self?.fan_out: (Array[Array[String]] chunks, ignore_timeout: bool) -> untyped - # nil when this runtime turns out not to support forking after all. - def self?.spawn_workers: (Array[Array[String]] chunks, ignore_timeout: bool) -> Array[Hash[Symbol, untyped]]? - def self?.spawn_worker: (Array[String] chunk, ignore_timeout: bool) -> Hash[Symbol, untyped] + # Everything the child does; ends the process with `exit!`. + def self?.run_in_child: (IO reader, IO writer, Array[String] chunk, bool ignore_timeout) -> bot + # The body of a worker: merge the slice, ship it back, and report the exit # status the child should terminate with. def self?.run_worker: (Array[String] chunk, IO writer, ignore_timeout: bool) -> Integer @@ -54,8 +53,5 @@ module SimpleCov def self?.succeeded?: (Integer pid) -> bool def self?.warn_about_failed_workers: (Integer failed, Integer total) -> void - - # Best-effort teardown for workers spawned before a fork failed. - def self?.shut_down: (Array[Hash[Symbol, untyped]] workers) -> void end end diff --git a/spec/helper.rb b/spec/helper.rb index dbae7310f..796a15783 100644 --- a/spec/helper.rb +++ b/spec/helper.rb @@ -71,6 +71,8 @@ # doesn't support, so the lib/ lines those specs would have hit stay # uncovered there — set the line threshold a hair below today's # actual to act as a regression guard rather than a strict ceiling. + # (Files that are wholly unreachable on an engine are filtered out + # below instead, so they don't drag this number down.) # Engines absent from this hash get an informational report only, # no threshold enforcement. DOGFOOD_THRESHOLDS = { @@ -82,6 +84,11 @@ RSpec.configure do |config| config.after(:suite) do extra_filters = %w[/spec/ /features/ /test_projects/ /tmp/].map { |path| SimpleCov::StringFilter.new(path) } + # `ParallelResultMerger`'s fan-out forks, so where the runtime cannot + # (JRuby, TruffleRuby, CRuby on Windows) its worker lines are unreachable + # rather than untested. Drop the file on those engines instead of + # lowering the bar for every other file; CRuby still holds it to 100%. + extra_filters << SimpleCov::StringFilter.new("parallel_result_merger.rb") unless FORK_SUPPORTED raw = SimpleCov::UselessResultsRemover.call(Coverage.result) adapted = SimpleCov::ResultAdapter.call(raw) @@ -141,6 +148,12 @@ end end +# Specs that need real worker processes (see +# spec/parallel_result_merger_spec.rb) are skipped where this is false: CRuby +# on Windows, and JRuby / TruffleRuby, which cannot fork on the JVM. The +# in-process merge those runtimes fall back to is covered on every engine. +FORK_SUPPORTED = Process.respond_to?(:fork) + def source_fixture(filename) File.join(source_fixture_base_directory, "fixtures", filename) end diff --git a/spec/parallel_result_merger_spec.rb b/spec/parallel_result_merger_spec.rb index 73bf637d5..91f54ae89 100644 --- a/spec/parallel_result_merger_spec.rb +++ b/spec/parallel_result_merger_spec.rb @@ -77,7 +77,7 @@ end it "merges in this process when the fan-out cannot run" do - allow(described_class).to receive(:fork_supported?).and_return(false) + without_fork result = described_class.merge_results(*paths, processes: 3, ignore_timeout: true) @@ -87,19 +87,19 @@ end describe ".merge_resultsets" do - it "produces the pair ResultMerger.merge_resultsets produces" do + it "produces the pair ResultMerger.merge_resultsets produces", if: FORK_SUPPORTED do expect(described_class.merge_resultsets(paths, processes: 3, ignore_timeout: true)).to eq(serial) end - it "produces the same pair however many processes it is given" do + it "produces the same pair however many processes it is given", if: FORK_SUPPORTED do merges = [2, 3, 4, 5, 12].map do |processes| described_class.merge_resultsets(paths, processes: processes, ignore_timeout: true) end - expect(merges.uniq.size).to eq(1) + expect(merges).to all(eq(serial)) end - it "honours ignore_timeout: false by dropping expired resultsets" do + it "honours ignore_timeout: false by dropping expired resultsets", if: FORK_SUPPORTED do expired = write_resultset("stale", {source_fixture("sample.rb") => {"lines" => [9, 9, 9, 9]}}, outdated: true) fresh = paths.first(2) @@ -117,18 +117,12 @@ end it "returns nil when the runtime cannot fork" do - allow(described_class).to receive(:fork_supported?).and_return(false) + without_fork expect(described_class.merge_resultsets(paths, processes: 4)).to be_nil end end - describe ".fork_supported?" do - # False on JRuby, TruffleRuby and Windows, where `fork` is defined but - # raises; there is no CRuby-side way to reach the false case. - it { expect(described_class.fork_supported?).to eq(Process.respond_to?(:fork)) } - end - describe ".chunk" do it "splits into contiguous slices, in order" do expect(described_class.chunk(%w[a b c d e f], 3)).to eq([%w[a b], %w[c d], %w[e f]]) @@ -188,7 +182,7 @@ describe ".fan_out" do let(:chunks) { described_class.chunk(paths, 3) } - it "returns nil and warns when a worker dies without shipping its slice" do + it "returns nil and warns when a worker dies without shipping its slice", if: FORK_SUPPORTED do allow(described_class).to receive(:run_worker).and_return(1) output = capture_stderr do @@ -198,13 +192,7 @@ expect(output).to include("parallel merge did not complete (3 of 3 workers failed)") end - it "returns nil when the workers could not be spawned at all" do - allow(described_class).to receive(:spawn_workers).and_return(nil) - - expect(described_class.fan_out(chunks, ignore_timeout: true)).to be_nil - end - - it "stays quiet about a failed fan-out when print_errors is off" do + it "stays quiet about a failed fan-out when print_errors is off", if: FORK_SUPPORTED do allow(described_class).to receive(:run_worker).and_return(1) output = capture_stderr do @@ -215,17 +203,34 @@ end end - describe ".spawn_workers" do - it "tears down the workers it did spawn and returns nil when a fork is refused" do - spawned = [] - allow(described_class).to receive(:spawn_worker).and_wrap_original do |original, *args, **options| - raise NotImplementedError, "fork is not available on this platform" if spawned.any? + describe ".run_in_child" do + # Driven directly, with `exit!` stubbed, because a child's own coverage + # dies with it: it exits without reporting, so nothing it executes ever + # merges back into this process's result. + it "closes the read end and exits with the status the worker reports" do + reader, writer = IO.pipe + allow(described_class).to receive(:exit!) + allow(described_class).to receive(:run_worker).and_return(1) - original.call(*args, **options).tap { |worker| spawned << worker } - end + described_class.run_in_child(reader, writer, paths, true) - expect(described_class.spawn_workers(described_class.chunk(paths, 3), ignore_timeout: true)).to be_nil - expect(spawned.first[:reader]).to be_closed + expect(reader).to be_closed + expect(described_class).to have_received(:exit!).with(1) + expect(described_class).to have_received(:run_worker).with(paths, writer, ignore_timeout: true) + ensure + writer.close + end + end + + describe ".fan_out when the OS refuses a fork" do + # A machine out of process slots is not something to paper over: the merge + # is fine, the box is not, and swallowing it would turn that into a + # mysteriously slow collate. + it "lets the error through rather than merging in this process", if: FORK_SUPPORTED do + allow(described_class).to receive(:fork).and_raise(Errno::EAGAIN) + + expect { described_class.merge_results(*paths, processes: 3, ignore_timeout: true) } + .to raise_error(Errno::EAGAIN) end end @@ -256,6 +261,13 @@ def write_resultset(command_name, coverage, outdated: false) path end + # What CRuby on Windows, JRuby and TruffleRuby report. Every other + # `respond_to?` has to keep working, hence the `and_call_original` first. + def without_fork + allow(Process).to receive(:respond_to?).and_call_original + allow(Process).to receive(:respond_to?).with(:fork).and_return(false) + end + def with_print_errors(value) previous = SimpleCov.print_errors SimpleCov.print_errors value From d8df170e8e9fb71fa4c25371888c6c37b4d32e82 Mon Sep 17 00:00:00 2001 From: Daniel Westendorf Date: Mon, 3 Aug 2026 18:25:52 -0600 Subject: [PATCH 5/5] Add support for ENV var configuration of processes --- CHANGELOG.md | 2 +- README.md | 16 ++++++++++++---- lib/simplecov/result_processing.rb | 6 ++++-- spec/simplecov_spec.rb | 20 ++++++++++++++++++++ 4 files changed, 37 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d8b3e906f..1349f4859 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,7 @@ Unreleased ## Enhancements * The favicon (a solid square in the overall coverage band's colour) is now drawn by the viewer from the report's own palette instead of shipping as fixed PNGs, so it matches the report's green/yellow/red exactly and follows the light/dark theme, including the in-page toggle. -* `SimpleCov.collate` takes a new `processes:` argument that fans the resultset merge out across that many forked worker processes. This addresses the wall clock of a large CI matrix's collate step, where the collating process reads, parses and folds hundreds of resultsets in sequence and nearly all the time goes into that fold: merging 160 resultsets covering 1,836 files on a 14-core machine took 4.53s at the default `processes: 1` and 1.35s across 8 workers. The report is identical either way, not merely equivalent — each worker folds a contiguous slice of the file list and the collating process folds the slices back in order, so the resultsets are visited in the same order a single-process merge visits them. The fan-out lives in a new `SimpleCov::ParallelResultMerger`, whose `merge_resultsets` mirrors the `ResultMerger.merge_resultsets` extracted alongside it. `processes` defaults to 1 and never forks at that value, so existing `collate` calls are unaffected; it is deliberately not clamped to the core count nor gated on a minimum number of resultsets — only the caller knows what a collate job is allowed to use — asking for more processes than there are result files just gives one file per process, and anything below 1 is taken as 1. Merging falls back to the collating process, with the same report and no error, when the runtime cannot fork (JRuby, TruffleRuby, Windows), when there is only one resultset, or when a worker dies. A `benchmarks/collate.rb` harness (`PROCESSES=N`) measures the phases against a saved baseline. +* `SimpleCov.collate` takes a new `processes:` argument that fans the resultset merge out across that many forked worker processes. This addresses the wall clock of a large CI matrix's collate step, where the collating process reads, parses and folds hundreds of resultsets in sequence and nearly all the time goes into that fold: merging 160 resultsets covering 1,836 files on a 14-core machine took 4.53s at the default `processes: 1` and 1.35s across 8 workers. The report is identical either way, not merely equivalent — each worker folds a contiguous slice of the file list and the collating process folds the slices back in order, so the resultsets are visited in the same order a single-process merge visits them. The fan-out lives in a new `SimpleCov::ParallelResultMerger`, whose `merge_resultsets` mirrors the `ResultMerger.merge_resultsets` extracted alongside it. `processes` defaults to the `SIMPLECOV_CONCURRENCY` environment variable (1 when unset), so one rake task can serve CI runners of different sizes without being edited, and an explicit argument wins over the variable. It never forks at 1, so existing `collate` calls are unaffected; it is deliberately not clamped to the core count nor gated on a minimum number of resultsets — only the caller knows what a collate job is allowed to use — asking for more processes than there are result files just gives one file per process, and anything below 1 is taken as 1. Merging falls back to the collating process, with the same report and no error, when the runtime cannot fork (JRuby, TruffleRuby, Windows), when there is only one resultset, or when a worker dies. A `benchmarks/collate.rb` harness (`PROCESSES=N`) measures the phases against a saved baseline. 1.0.3 (2026-07-26) ================== diff --git a/README.md b/README.md index 775f57f96..807da2b65 100644 --- a/README.md +++ b/README.md @@ -876,10 +876,18 @@ The report is identical to the one a single-process collate produces for the sam worker folds a contiguous slice of the file list and the collating process folds the slices back in order, so the resultsets are visited in the same order they would be otherwise. -`processes` defaults to 1, which never forks — existing `collate` calls behave exactly as before. It is deliberately -not clamped to your core count, nor gated on some minimum number of resultsets: how many processes a collate job can -afford is something only you know. Asking for more processes than there are result files simply gives one file per -process, and anything below 1 is taken as 1, so a count computed from arithmetic that can reach zero needs no guarding. +`processes` defaults to the `SIMPLECOV_CONCURRENCY` environment variable, or 1 when that is unset — and 1 never forks, +so existing `collate` calls behave exactly as before. Setting it in the environment lets one rake task serve runners of +different sizes without editing the task: + +```sh +SIMPLECOV_CONCURRENCY=8 bundle exec rake coverage:report +``` + +An explicit `processes:` argument wins over the environment variable. The count is deliberately not clamped to your core +count, nor gated on some minimum number of resultsets: how many processes a collate job can afford is something only you +know. Asking for more processes than there are result files simply gives one file per process, and anything below 1 is +taken as 1, so a count computed from arithmetic that can reach zero needs no guarding. It falls back to merging in the collating process — same report, no error — when the runtime cannot fork (JRuby, TruffleRuby, Windows), when there is only one resultset to fold, or when a worker dies. diff --git a/lib/simplecov/result_processing.rb b/lib/simplecov/result_processing.rb index 5fc28b874..44786e424 100644 --- a/lib/simplecov/result_processing.rb +++ b/lib/simplecov/result_processing.rb @@ -22,9 +22,11 @@ class << self # gated on some minimum number of resultsets: how many processes a collate # job can afford is the caller's call, not SimpleCov's. Anything below 1 is # taken as 1, so a count computed from arithmetic that can reach zero needs - # no guarding. See `SimpleCov::ParallelResultMerger`. + # no guarding. Defaults to `SIMPLECOV_CONCURRENCY`, or 1 when that is + # unset. See `SimpleCov::ParallelResultMerger`. # - def collate(result_filenames, profile = nil, processes: 1, ignore_timeout: true, &) + def collate(result_filenames, profile = nil, processes: ENV.fetch("SIMPLECOV_CONCURRENCY", 1).to_i, + ignore_timeout: true, &) raise ArgumentError, "There are no reports to be merged" if result_filenames.empty? initial_setup(profile, &) diff --git a/spec/simplecov_spec.rb b/spec/simplecov_spec.rb index 5650e526a..9df3453d0 100644 --- a/spec/simplecov_spec.rb +++ b/spec/simplecov_spec.rb @@ -1146,6 +1146,26 @@ def expect_merged .to eq(collate_with { |paths| described_class.collate paths }) end + it "defaults the process count to SIMPLECOV_CONCURRENCY" do + allow(SimpleCov::ParallelResultMerger).to receive(:merge_resultsets).and_call_original + + with_env("SIMPLECOV_CONCURRENCY" => "3") { collate_with { |paths| described_class.collate paths } } + + expect(SimpleCov::ParallelResultMerger).to have_received(:merge_resultsets) + .with(anything, hash_including(processes: 3)) + end + + it "prefers an explicit process count over SIMPLECOV_CONCURRENCY" do + allow(SimpleCov::ParallelResultMerger).to receive(:merge_resultsets).and_call_original + + with_env("SIMPLECOV_CONCURRENCY" => "3") do + collate_with { |paths| described_class.collate paths, processes: 2 } + end + + expect(SimpleCov::ParallelResultMerger).to have_received(:merge_resultsets) + .with(anything, hash_including(processes: 2)) + end + it "merges in this process when the fan-out cannot run" do serial = collate_with { |paths| described_class.collate paths } allow(SimpleCov::ParallelResultMerger).to receive(:merge_resultsets).and_return(nil)