Module: SimpleCov::ParallelResultMerger

Extended by:
ParallelResultMerger
Included in:
ParallelResultMerger
Defined in:
lib/simplecov/parallel_result_merger.rb,
lib/simplecov/parallel_result_merger/worker_payload.rb

Overview

ResultMerger.absorb_results fanned out across forked workers, driving SimpleCov.collate(..., processes: N). Each worker folds a contiguous slice of the file list and ships the pair back over a pipe.

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.

Defined Under Namespace

Modules: WorkerPayload

Instance Method Summary collapse

Instance Method Details

#abandon(workers) ⇒ Object 



86
87
88
89
90
91
 
# File 'lib/simplecov/parallel_result_merger.rb', line 86

def abandon(workers)
  workers.each do |worker|
    worker.fetch(:reader).close
    succeeded?(worker.fetch(:pid))
  end
end

#absorb_results(file_paths, processes:, ignore_timeout: false, tracked_files: Set.new, context_maps: ContextMap::Union.new) ⇒ Object

Answers nil when the work could not be fanned out and the caller should merge in this process instead. The tracked paths and context-map union a worker's slice carried come back with its payload rather than through a collector block, since that block would be mutating state in the wrong process.



42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/simplecov/parallel_result_merger.rb', line 42

def absorb_results(file_paths, processes:, ignore_timeout: false, tracked_files: Set.new,
  context_maps: ContextMap::Union.new)
  return nil if processes < 2 || file_paths.size < 2
  # JRuby cannot fork on the JVM and deliberately answers false here, but
  # still defines `Kernel#fork` and raises `NotImplementedError` from it,
  # so probing that instead would send it down the fan-out.
  return nil unless Process.respond_to?(:fork)

  fan_out(chunk(file_paths, processes), ignore_timeout: ignore_timeout, tracked_files: tracked_files,
    context_maps: context_maps)
end

#chunk(file_paths, processes) ⇒ Object 



54
55
56
57
58
59
60
 
# File 'lib/simplecov/parallel_result_merger.rb', line 54

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

#collect_payloads(workers) ⇒ Object

Answers nil if any worker failed, so the caller redoes the fold serially rather than report a subset of the resultsets as the whole.



128
129
130
131
132
133
134
135
136
 
# File 'lib/simplecov/parallel_result_merger.rb', line 128

def collect_payloads(workers)
  payloads = drain(workers)
  failed = workers.count { |worker| !succeeded?(worker.fetch(:pid)) }
  return warn_about_failed_workers(failed, workers.size) unless failed.zero? && payloads.all?

  payloads
ensure
  workers.each { |worker| worker.fetch(:reader).close }
end

#drain(workers) ⇒ Object

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.



142
143
144
 
# File 'lib/simplecov/parallel_result_merger.rb', line 142

def drain(workers)
  workers.map { |worker| Thread.new { read_payload(worker.fetch(:reader)) } }.map(&:value)
end

#fan_out(chunks, ignore_timeout:, tracked_files:, context_maps:) ⇒ Object

A fork that fails here raises, and is left to: the runtimes that never fork are already excluded, so what remains is the OS refusing a process we expected to get, which says something is wrong with the machine rather than with the merge.



66
67
68
69
70
71
72
73
 
# File 'lib/simplecov/parallel_result_merger.rb', line 66

def fan_out(chunks, ignore_timeout:, tracked_files:, context_maps:)
  workers = spawn_workers(chunks, ignore_timeout: ignore_timeout)
  payloads = collect_payloads(workers)
  return nil unless payloads

  payloads.each { |payload| WorkerPayload.absorb(payload, tracked_files, context_maps) }
  ResultMerger.merge_coverage(*payloads.map { |payload| WorkerPayload.pair(payload) })
end

#merge_and_store(*file_paths, processes:, ignore_timeout: false) ⇒ Object 



18
19
20
21
22
23
24
 
# File 'lib/simplecov/parallel_result_merger.rb', line 18

def merge_and_store(*file_paths, processes:, ignore_timeout: false)
  return ResultMerger.merge_and_store(*file_paths, ignore_timeout: ignore_timeout) if processes < 2

  result = merge_results(*file_paths, processes: processes, ignore_timeout: ignore_timeout)
  ResultMerger.store_result(result) if result
  result
end

#merge_results(*file_paths, processes:, ignore_timeout: false) ⇒ Object 



26
27
28
29
30
31
32
33
34
35
 
# File 'lib/simplecov/parallel_result_merger.rb', line 26

def merge_results(*file_paths, processes:, ignore_timeout: false)
  tracked_files = Set.new
  context_maps = ContextMap::Union.new
  pair = absorb_results(file_paths, processes: processes, ignore_timeout: ignore_timeout,
    tracked_files: tracked_files, context_maps: context_maps)
  return ResultMerger.merge_results(*file_paths, ignore_timeout: ignore_timeout) unless pair

  command_names, coverage = pair
  ResultMerger.create_result(command_names, coverage, tracked_files: tracked_files, contexts: context_maps.map)
end

#read_payload(reader) ⇒ Object 



146
147
148
149
150
151
152
153
 
# File 'lib/simplecov/parallel_result_merger.rb', line 146

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.
  Marshal.load(reader) # rubocop:disable Security/MarshalLoad
rescue
  nil
end

#run_in_child(reader, writer, chunk, ignore_timeout) ⇒ Object

exit! rather than exit because the child must never fall through to the collating process's inherited at_exit handlers, SimpleCov's own report generation included.



111
112
113
114
 
# File 'lib/simplecov/parallel_result_merger.rb', line 111

def run_in_child(reader, writer, chunk, ignore_timeout)
  reader.close
  exit!(run_worker(chunk, writer, ignore_timeout: ignore_timeout))
end

#run_worker(chunk, writer, ignore_timeout:) ⇒ Object

Kept free of the exit itself so it can be exercised in-process.



117
118
119
120
121
122
123
124
 
# File 'lib/simplecov/parallel_result_merger.rb', line 117

def run_worker(chunk, writer, ignore_timeout:)
  Marshal.dump(WorkerPayload.build(chunk, ignore_timeout: ignore_timeout), writer)
  writer.close
  0
rescue => e
  warn "[SimpleCov]: parallel merge worker failed: #{e.class}: #{e}" if SimpleCov.print_errors
  1
end

#spawn_worker(chunk, ignore_timeout:) ⇒ Object

The caller cleans up the workers it knows about, but this pipe is ours. The safe navigation keeps the cleanup well-defined when IO.pipe itself raised, and Steep cannot type body locals inside an ensure.



96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/simplecov/parallel_result_merger.rb', line 96

def spawn_worker(chunk, ignore_timeout:)
  reader, writer = IO.pipe
  pid = fork { run_in_child(reader, writer, chunk, ignore_timeout) }
  {pid: pid, reader: reader}
ensure
  # @type var writer: IO?
  # @type var reader: IO?
  # @type var pid: Integer?
  writer&.close
  reader&.close unless pid
end

#spawn_workers(chunks, ignore_timeout:) ⇒ Object 



75
76
77
78
79
80
81
82
83
84
 
# File 'lib/simplecov/parallel_result_merger.rb', line 75

def spawn_workers(chunks, ignore_timeout:)
  workers = [] #: Array[Hash[Symbol, untyped]]
  chunks.each do |chunk|
    workers << spawn_worker(chunk, ignore_timeout: ignore_timeout)
  rescue
    abandon(workers)
    raise
  end
  workers
end

#succeeded?(pid) ⇒ Boolean

Returns:

  • (Boolean) 


155
156
157
158
159
160
161
162
 
# File 'lib/simplecov/parallel_result_merger.rb', line 155

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

#warn_about_failed_workers(failed, total) ⇒ Object 



164
165
166
167
168
169
 
# File 'lib/simplecov/parallel_result_merger.rb', line 164

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