diff --git a/CHANGELOG.md b/CHANGELOG.md index 7bd937ff6..1349f4859 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. +* `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 84f1a7b3f..807da2b65 100644 --- a/README.md +++ b/README.md @@ -853,6 +853,57 @@ 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. + +Pass `processes:` to spread that fold across forked worker 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.collate Dir["simplecov-resultset-*/.resultset.json"], processes: 8 + end +end +``` + +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` 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. + +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 | +| ------------ | ----------- | +| 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. + ### 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..3c9480d59 100644 --- a/benchmarks/collate.rb +++ b/benchmarks/collate.rb @@ -20,10 +20,14 @@ # # 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.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/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..4072137dc 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.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 + # `ResultMerger.merge_results` 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..c625765b4 --- /dev/null +++ b/features/test_unit_parallel_collate.feature @@ -0,0 +1,43 @@ +@test_unit +Feature: + + 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" + + 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..cfc97b031 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" diff --git a/lib/simplecov/parallel_result_merger.rb b/lib/simplecov/parallel_result_merger.rb new file mode 100644 index 000000000..3a6c39773 --- /dev/null +++ b/lib/simplecov/parallel_result_merger.rb @@ -0,0 +1,179 @@ +# 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 + # 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 + + # 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 + + # 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 = chunks.map { |chunk| spawn_worker(chunk, ignore_timeout: ignore_timeout) } + payloads = collect(workers) + + payloads && ResultMerger.merge_coverage(*payloads) + end + + def spawn_worker(chunk, ignore_timeout:) + reader, writer = IO.pipe + pid = fork { run_in_child(reader, writer, chunk, ignore_timeout) } + writer.close + + {pid: pid, reader: reader} + end + + # 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. + 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); " \ + "merging the resultsets in this process instead." + 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..44786e424 100644 --- a/lib/simplecov/result_processing.rb +++ b/lib/simplecov/result_processing.rb @@ -13,13 +13,26 @@ class << self # 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, &) + # `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. Defaults to `SIMPLECOV_CONCURRENCY`, or 1 when that is + # unset. See `SimpleCov::ParallelResultMerger`. + # + 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, &) # Use the ResultMerger to produce a single, merged result, ready to use. - @result = ResultMerger.merge_and_store(*result_filenames, ignore_timeout: ignore_timeout) + @result = merge_collated(result_filenames, [1, processes].max, ignore_timeout) @collating_result = true run_exit_tasks! @@ -117,6 +130,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 new file mode 100644 index 000000000..1cd3f7a7d --- /dev/null +++ b/sig/internal/simplecov/parallel_result_merger.rbs @@ -0,0 +1,57 @@ +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 + + # 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 + + 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 + + # 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 + 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..a645445d3 100644 --- a/sig/simplecov.rbs +++ b/sig/simplecov.rbs @@ -107,8 +107,14 @@ 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 + # 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 + + # 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. 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/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 new file mode 100644 index 000000000..91f54ae89 --- /dev/null +++ b/spec/parallel_result_merger_spec.rb @@ -0,0 +1,278 @@ +# 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.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 + without_fork + + 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", 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", 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).to all(eq(serial)) + end + + 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) + + 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 + without_fork + + expect(described_class.merge_resultsets(paths, processes: 4)).to be_nil + end + 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", if: FORK_SUPPORTED 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 "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 + with_print_errors(false) { described_class.fan_out(chunks, ignore_timeout: true) } + end + + expect(output).to be_empty + end + end + + 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) + + described_class.run_in_child(reader, writer, paths, true) + + 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 + + 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 + + # 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 + 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..9df3453d0 100644 --- a/spec/simplecov_spec.rb +++ b/spec/simplecov_spec.rb @@ -1100,6 +1100,106 @@ def expect_merged end end + describe ".collate across processes" 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.collate([], processes: 2) } + .to raise_error("There are no reports to be merged") + 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 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.collate paths, processes: processes } + end + + expect(reports.uniq.size).to eq(1) + end + + 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 "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) + + expect(collate_with { |paths| described_class.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..929905e6f 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.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