Collect Ractor GC samples in harness-ractor-mem

With --ractor-gc (RUBY_BENCH_RACTOR_GC=1), harness-ractor-mem checks
target support before the baseline and writes gc_by_ractors in the
same shape as harness-ractor. Scenarios spawn their own workers, so
each worker wraps its body in measure_worker_gc and the main Ractor
passes the sample to record_worker_gc. A trial fails when its recorded
worker indexes are not 0...count.

ractor-idle-garbage takes its sample while the set is still live and
sends a shareable message, so sampling cannot sweep the garbage
before the worker parks.

Author: eightbitraptor

README.md

@@ -198,6 +198,19 @@ On a target with Ractor-local GC, a plain `GC.start` collects only the main
 Ractor's object space. The JSON field `ractor_mem_settle` records `global` or
 `default`.
 
+With `--ractor-gc` (`RUBY_BENCH_RACTOR_GC=1`), every worker samples its own
+Ractor-local GC activity. The results then carry the same `gc_by_ractors` data
+as `harness-ractor`. A scenario wraps each worker body in
+`measure_worker_gc { ... }`, which returns `[result, sample]`. The main Ractor
+passes each sample to `record_worker_gc(worker_index, sample)`. A trial fails
+when its recorded worker indexes are not `0...count`.
+
+Worker samples cover only the workers' own object spaces during the scenario.
+They do not include allocation by the main Ractor, such as the payloads that
+`ractor-msg-backlog` sends. They also do not include the GCs that the harness
+runs to measure retention. The JSON field `gc_controller_samples` covers the
+main Ractor during the scenario.
+
 ## Ruby options
 
 By default, ruby-bench benchmarks the Ruby used for `run_benchmarks.rb`.
@@ -337,7 +350,8 @@ process's lifetime peak from `getrusage`.
 ## Measuring Ractor GC activity
 
 The `--ractor-gc` option of `run_benchmarks.rb` collects Ractor-local GC
-metrics for benchmarks that use the Ractor harness (`--category ractor`).
+metrics for benchmarks that use `harness-ractor` or `harness-ractor-mem`
+(`--category ractor`).
 The target must use Ruby 4.1 or newer with per-Ractor global GC attribution
 ([ruby/ruby#19147](https://github.com/ruby/ruby/pull/19147)); older targets
 fail before warmup.

benchmarks/ractor-dead-set/benchmark.rb

@@ -11,17 +11,24 @@
 run_benchmark(3) do |num_ractors|
   workers = num_ractors.times.map do |worker|
     Ractor.new(worker, DEAD_SET_ITEMS) do |worker_id, items|
-      keep = []
-      i = 0
-      while i < items
-        keep << "worker #{worker_id} item #{i} " + ("y" * 100)
-        i += 1
+      measure_worker_gc do
+        keep = []
+        i = 0
+        while i < items
+          keep << "worker #{worker_id} item #{i} " + ("y" * 100)
+          i += 1
+        end
+        keep.size
       end
-      keep.size
     end
   end
 
-  total = workers.sum(&:value)
+  total = 0
+  workers.each_with_index do |worker, worker_id|
+    size, sample = worker.value
+    record_worker_gc(worker_id, sample)
+    total += size
+  end
   raise "unexpected dead-set size" unless total == num_ractors * DEAD_SET_ITEMS
   nil
 end

benchmarks/ractor-idle-garbage/benchmark.rb

@@ -12,26 +12,30 @@
 
 run_benchmark(3) do |num_ractors|
   drained = Ractor.new(num_ractors) do |count|
-    count.times { Ractor.receive }
-    :all_drained
+    Array.new(count) { Ractor.receive }
   end
 
   workers = num_ractors.times.map do |worker|
     Ractor.new(drained, worker, IDLE_GARBAGE_ITEMS) do |ack, worker_id, items|
       keep = []
-      i = 0
-      while i < items
-        keep << "worker #{worker_id} garbage #{i} " + ("y" * 100)
-        i += 1
+      _, sample = measure_worker_gc do
+        i = 0
+        while i < items
+          keep << "worker #{worker_id} garbage #{i} " + ("y" * 100)
+          i += 1
+        end
       end
+      message = Ractor.make_shareable([worker_id, sample])
       keep = nil
-      ack.send :drained
+      ack.send message
       Ractor.receive
       :worker_done
     end
   end
 
-  raise "unexpected drain barrier result" unless drained.value == :all_drained
+  drained_workers = drained.value
+  raise "unexpected drain barrier result" unless drained_workers.size == num_ractors
+  drained_workers.each { |worker_id, sample| record_worker_gc(worker_id, sample) }
 
   proc do
     workers.each do |worker|

benchmarks/ractor-msg-backlog/benchmark.rb

@@ -16,14 +16,16 @@
 run_benchmark(3) do |num_ractors|
   consumers = num_ractors.times.map do |consumer|
     Ractor.new(consumer, BACKLOG_GATE_SLEEP) do |consumer_id, gate|
-      sleep gate
-      taken = 0
-      loop do
-        message = Ractor.receive
-        break if message == :done
-        taken += 1
+      measure_worker_gc do
+        sleep gate
+        taken = 0
+        loop do
+          message = Ractor.receive
+          break if message == :done
+          taken += 1
+        end
+        taken
       end
-      [consumer_id, taken]
     end
   end
 
@@ -36,7 +38,12 @@
     consumer.send :done
   end
 
-  results = consumers.map(&:value)
-  raise "unexpected backlog drain" unless results.sum { |_, taken| taken } == num_ractors * BACKLOG_MESSAGES
+  total = 0
+  consumers.each_with_index do |consumer, consumer_id|
+    taken, sample = consumer.value
+    record_worker_gc(consumer_id, sample)
+    total += taken
+  end
+  raise "unexpected backlog drain" unless total == num_ractors * BACKLOG_MESSAGES
   nil
 end

harness-ractor-mem/harness.rb

@@ -3,6 +3,9 @@
 
 Warning[:experimental] = false
 
+RACTOR_GC_ENABLED = ENV["RUBY_BENCH_RACTOR_GC"] == "1"
+require_relative "../lib/gc_stats" if RACTOR_GC_ENABLED
+
 default_ractors = [1, 2, 4, 6, 8]
 if rs = ENV["RUBY_BENCH_RACTORS"]
   rs = rs.split(",").map(&:to_i)
@@ -17,6 +20,7 @@
 PEAK_SAMPLE_INTERVAL = Float(ENV.fetch("RACTOR_MEM_PEAK_SAMPLE_INTERVAL", 0.005))
 
 GLOBAL_GC_START = GC.method(:start).parameters.include?([:key, :global])
+WORKER_GC_SAMPLES = []
 
 puts RUBY_DESCRIPTION
 
@@ -58,33 +62,65 @@ def measure_peak_rss
   [result, peak]
 end
 
+def measure_worker_gc
+  return [yield, nil] unless RACTOR_GC_ENABLED
+
+  result = nil
+  sample = GCStats.measure { result = yield }
+  [result, sample]
+end
+
+def record_worker_gc(worker_index, sample)
+  return unless sample
+
+  WORKER_GC_SAMPLES << sample.merge("worker_index" => worker_index)
+end
+
 def run_benchmark(num_itrs_hint, &scenario)
   bench_itrs = Integer(ENV.fetch("MIN_BENCH_ITRS", num_itrs_hint))
   bench_itrs = MAX_ITERS if bench_itrs > MAX_ITERS
   bench_itrs = 1 if bench_itrs < 1
 
+  if RACTOR_GC_ENABLED
+    GCStats.check_ractor_gc_support!
+    gc_config = GC.config.transform_keys(&:to_s) if GC.respond_to?(:config)
+    GCStats.with_measure_total_time { run_memory_trials(bench_itrs, gc_config, &scenario) }
+  else
+    run_memory_trials(bench_itrs, nil, &scenario)
+  end
+end
+
+def run_memory_trials(bench_itrs, gc_config, &scenario)
   gc_settle
   base_rss = get_rss
   puts format("base RSS: %.1f MiB", base_rss / 2**20.0)
 
   metrics = Hash.new { |h, count| h[count] = { retained: [], peak: [] } }
   times = Hash.new { |h, count| h[count] = [] }
+  gc_groups = Hash.new do |h, count|
+    h[count] = { "gc_worker_samples" => [], "gc_controller_samples" => [] }
+  end
 
   RACTORS.each do |count|
     puts "ractors: #{count}"
     itr = 0
     while itr < bench_itrs
+      WORKER_GC_SAMPLES.clear
+      controller_before = GCStats.snapshot if RACTOR_GC_ENABLED
       t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
       finish, peak = measure_peak_rss { scenario.call(count) }
       times[count] << Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0
+      controller_sample = GCStats.delta(controller_before, GCStats.snapshot) if RACTOR_GC_ENABLED
       gc_settle
       retained = get_rss - base_rss
       finish&.call
       gc_settle
       metrics[count][:retained] << retained
       metrics[count][:peak] << peak
-      puts format("  itr #%d: retained %.1f MiB, peak %.1f MiB",
+      line = format("  itr #%d: retained %.1f MiB, peak %.1f MiB",
         itr + 1, retained / 2**20.0, peak / 2**20.0)
+      line << record_trial_gc(gc_groups[count], count, controller_sample) if RACTOR_GC_ENABLED
+      puts line
       itr += 1
     end
   end
@@ -109,11 +145,38 @@ def run_benchmark(num_itrs_hint, &scenario)
   puts format("BENCH_METRIC peak_mib=%.1f", worst_peak / 2**20.0)
   puts format("BENCH_METRIC worst_ractor_count=%d", worst_retained_count)
 
-  return_results([], times.values.flatten,
+  extra = {
     bench_by_ractors: times,
     ractor_mem_settle: GLOBAL_GC_START ? "global" : "default",
     ractor_mem_base_rss: base_rss,
     ractor_mem_medians: medians,
     ractor_mem_samples: metrics,
-  )
+  }
+  if RACTOR_GC_ENABLED
+    extra[:gc_scope] = "ractor-local-workload"
+    extra[:gc_stat_scope] = "ractor-local"
+    extra[:gc_measure_total_time_scope] = "ractor-local"
+    extra[:gc_by_ractors] = gc_groups
+    extra[:gc_config] = gc_config if gc_config
+  end
+  return_results([], times.values.flatten, **extra)
+end
+
+def record_trial_gc(group, count, controller_sample)
+  workers = WORKER_GC_SAMPLES.sort_by { |s| s["worker_index"] }
+  unless workers.map { |s| s["worker_index"] } == (0...count).to_a
+    raise "scenario recorded worker GC samples #{workers.map { |s| s["worker_index"] }.inspect} for #{count} ractors"
+  end
+
+  group["gc_worker_samples"] << workers
+  group["gc_controller_samples"] << controller_sample
+  agg = GCStats.aggregate(workers)
+  GCStats::RACTOR_SERIES.each do |series, field|
+    value = field == GCStats::TOTAL_TIME_FIELD ? agg[field]&.fdiv(1_000_000) : agg[field]
+    (group[series] ||= []) << value
+  end
+
+  format(", worker gc %s (minor %s major %s global %s) %s",
+    agg["gc_count"], agg["gc_minor_count"], agg["gc_major_count"], agg["gc_global_count"],
+    agg["gc_total_time_ns"] ? format("%.1fms", agg["gc_total_time_ns"] / 1_000_000.0) : "N/A")
 end

test/ractor_gc_harness_test.rb

@@ -103,6 +103,32 @@ def GCStats.global_gc_attributed?(*) = false
     puts "workload_ran=#{workload_ran}"
   RUBY
 
+  MEM_WORKLOAD_BODY = <<~'RUBY'
+    run_benchmark(2) do |count|
+      workers = count.times.map do |worker|
+        Ractor.new(worker) do |worker_id|
+          measure_worker_gc do
+            20_000.times { Object.new }
+            GC.start(full_mark: true, immediate_sweep: true)
+            worker_id
+          end
+        end
+      end
+      workers.each_with_index do |worker, worker_id|
+        _, sample = worker.value
+        record_worker_gc(worker_id, sample)
+      end
+      nil
+    end
+  RUBY
+
+  MEM_UNRECORDED_BODY = <<~'RUBY'
+    run_benchmark(2) do |count|
+      count.times.map { Ractor.new { 10_000.times { Object.new } } }.each(&:value)
+      nil
+    end
+  RUBY
+
   before do
     @explicit_target = !ENV['RACTOR_GC_TEST_RUBY'].nil?
     @ruby = ENV['RACTOR_GC_TEST_RUBY'] || RbConfig.ruby
@@ -132,7 +158,7 @@ def GCStats.global_gc_attributed?(*) = false
     end
   end
 
-  def run_workload(body)
+  def run_workload(body, harness: 'harness-ractor')
     Dir.mktmpdir do |dir|
       result_path = File.join(dir, 'results.json')
       script = File.join(dir, 'workload.rb')
@@ -146,7 +172,7 @@ def run_workload(body)
         'MIN_BENCH_TIME' => '0',
         'RESULT_JSON_PATH' => result_path
       )
-      stdout, stderr, status = Open3.capture3(env, @ruby, "-I#{File.join(ROOT, 'harness-ractor')}", script, chdir: ROOT)
+      stdout, stderr, status = Open3.capture3(env, @ruby, "-I#{File.join(ROOT, harness)}", script, chdir: ROOT)
       yield stdout, stderr, status, result_path
     end
   end
@@ -294,6 +320,70 @@ def run_workload(body)
     end
   end
 
+  it 'collects per-worker GC samples from a ractor memory scenario' do
+    run_workload(MEM_WORKLOAD_BODY, harness: 'harness-ractor-mem') do |stdout, stderr, status, result_path|
+      assert status.success?, "workload failed:\n#{stdout}\n#{stderr}"
+
+      data = JSON.parse(File.read(result_path))
+      assert_equal 'ractor-local-workload', data['gc_scope']
+      assert_equal 'global', data['ractor_mem_settle']
+      assert_equal %w[1 2], data['bench_by_ractors'].keys.sort
+      assert_equal %w[1 2], data['gc_by_ractors'].keys.sort
+
+      data['gc_by_ractors'].each do |count, group|
+        assert_equal 2, data['bench_by_ractors'][count].length, "count #{count} timing samples"
+        assert_equal 2, group['gc_worker_samples'].length, "count #{count} measured iterations"
+        assert_equal 2, group['gc_controller_samples'].length, "count #{count} controller samples"
+        group['gc_worker_samples'].each_with_index do |workers, i|
+          assert_equal (0...count.to_i).to_a, workers.map { |w| w['worker_index'] }, 'spawn-index order'
+          workers.each { |w| assert_operator w['gc_count'], :>, 0, 'every worker records GC activity' }
+          assert_equal workers.sum { |w| w['gc_count'] }, group['gc_count_bench'][i]
+          assert_in_delta workers.sum { |w| w['gc_total_time_ns'] } / 1_000_000.0, group['gc_total_time_bench'][i], 1e-9
+        end
+      end
+    end
+  end
+
+  it 'fails a ractor memory scenario that does not record its worker GC samples' do
+    run_workload(MEM_UNRECORDED_BODY, harness: 'harness-ractor-mem') do |stdout, stderr, status, result_path|
+      refute status.success?, "expected an unrecorded scenario to fail:\n#{stdout}\n#{stderr}"
+      assert_includes stderr, 'scenario recorded worker GC samples [] for 1 ractors'
+      refute File.exist?(result_path), 'no results file may be written for a failed benchmark'
+    end
+  end
+
+  it 'reports every worker of each ractor memory benchmark' do
+    %w[ractor-dead-set ractor-idle-garbage ractor-msg-backlog].each do |bench|
+      Dir.mktmpdir do |dir|
+        result_path = File.join(dir, 'results.json')
+        env = CLEAN_ENV.merge(
+          'RUBY_BENCH_RACTOR_GC' => '1',
+          'RUBY_BENCH_RACTORS' => '1,2',
+          'MIN_BENCH_ITRS' => '1',
+          'MAX_BENCH_ITRS' => '1',
+          'RACTOR_MEM_SETTLE_SLEEP' => '0',
+          'RACTOR_DEAD_SET_ITEMS' => '5000',
+          'RACTOR_IDLE_GARBAGE_ITEMS' => '5000',
+          'RACTOR_BACKLOG_MESSAGES' => '500',
+          'RACTOR_BACKLOG_GATE_SLEEP' => '0.05',
+          'RESULT_JSON_PATH' => result_path
+        )
+        script = File.join(ROOT, 'benchmarks', bench, 'benchmark.rb')
+        stdout, stderr, status = Open3.capture3(env, @ruby, "-I#{File.join(ROOT, 'harness-ractor-mem')}", script, chdir: ROOT)
+        assert status.success?, "#{bench} failed:\n#{stdout}\n#{stderr}"
+
+        data = JSON.parse(File.read(result_path))
+        assert_equal %w[1 2], data['gc_by_ractors'].keys.sort, bench
+        data['gc_by_ractors'].each do |count, group|
+          assert_equal 1, group['gc_controller_samples'].length, "#{bench} count #{count} controller samples"
+          workers = group['gc_worker_samples'].fetch(0)
+          assert_equal (0...count.to_i).to_a, workers.map { |w| w['worker_index'] }, "#{bench} count #{count} workers"
+          workers.each { |w| assert_kind_of Integer, w['gc_count'], "#{bench} count #{count} worker gc_count" }
+        end
+      end
+    end
+  end
+
   it 'propagates a worker failure without writing partial or dummy results' do
     run_workload(FAILING_WORKLOAD_BODY) do |stdout, stderr, status, result_path|
       refute status.success?, "expected worker failure to fail the run:\n#{stdout}\n#{stderr}"