Pin benchmarks on Windows from the runner

harness/harness-common.rb

@@ -101,10 +101,6 @@ def is_macos
   RUBY_PLATFORM.match?(/darwin/)
 end
 
-def is_windows
-  RUBY_PLATFORM.match?(/mswin|mingw/)
-end
-
 def load_fiddle
   unless $LOAD_PATH.resolve_feature_path("fiddle")
     # In Ruby 3.5+, fiddle is no longer a default gem. Load the bundled-gem fiddle instead.
@@ -167,48 +163,6 @@ def macos_resident_size
   buffer[RI_RESIDENT_SIZE_OFFSET, 8].unpack1('Q')
 end
 
-# A dropped Fiddle::Handle is freed by a deferred finalizer, and its interrupt sends YJIT code back to the
-# interpreter for the rest of the loop the benchmark was in, so kernel32 stays open.
-module WindowsKernel32
-  def self.function(name, args, ret)
-    @handle ||= Fiddle.dlopen("kernel32")
-    Fiddle::Function.new(@handle[name], args, ret)
-  end
-end
-
-def pin_to_cpu(cpu)
-  load_fiddle
-  # Etc.nprocessors counts the CPUs of every processor group, but the mask holds those of one group only.
-  if cpu >= Fiddle::SIZEOF_VOIDP * 8
-    warn "Not pinning to CPU #{cpu}, which is out of the affinity mask"
-    return false
-  end
-  process = WindowsKernel32.function('GetCurrentProcess', [], Fiddle::TYPE_VOIDP).call
-  set_affinity = WindowsKernel32.function('SetProcessAffinityMask', [Fiddle::TYPE_VOIDP, Fiddle::TYPE_UINTPTR_T], Fiddle::TYPE_INT)
-  raise "SetProcessAffinityMask failed" if set_affinity.call(process, 1 << cpu).zero?
-  true
-end
-
-def raise_priority
-  load_fiddle
-  process = WindowsKernel32.function('GetCurrentProcess', [], Fiddle::TYPE_VOIDP).call
-  set_priority_class = WindowsKernel32.function('SetPriorityClass', [Fiddle::TYPE_VOIDP, Fiddle::TYPE_INT], Fiddle::TYPE_INT)
-  raise "SetPriorityClass failed" if set_priority_class.call(process, 0x80).zero? # HIGH_PRIORITY_CLASS
-
-  # PROCESS_POWER_THROTTLING_STATE that turns off the EXECUTION_SPEED throttling Windows 11 may apply to a windowless process.
-  set_information = WindowsKernel32.function('SetProcessInformation', [Fiddle::TYPE_VOIDP, Fiddle::TYPE_INT, Fiddle::TYPE_VOIDP, Fiddle::TYPE_INT], Fiddle::TYPE_INT)
-  state = [1, 1, 0].pack('L3')
-  raise "SetProcessInformation failed" if set_information.call(process, 4, state, state.bytesize).zero? # ProcessPowerThrottling
-end
-
-if is_windows && ENV["RUBY_BENCH_PIN_CPU"]
-  begin
-    raise_priority if pin_to_cpu(Integer(ENV["RUBY_BENCH_PIN_CPU"]))
-  rescue LoadError, StandardError => e
-    warn "Failed to pin the benchmark: #{e.message}"
-  end
-end
-
 def get_maxrss
   load_fiddle
 

lib/benchmark_runner.rb

@@ -5,6 +5,7 @@
 require 'rbconfig'
 require 'shellwords'
 require_relative 'table_formatter'
+require_relative 'windows_pinning'
 
 # Extracted helper methods from run_benchmarks.rb for testing
 module BenchmarkRunner
@@ -154,16 +155,18 @@ def render_graph(json_path)
 
     # Checked system - error or return info if the command fails.
     # An Array command with arguments runs without a shell, so Shellwords escapes never reach the program.
-    def check_call(command, env: {}, raise_error: true, quiet: ENV['BENCHMARK_QUIET'] == '1')
+    # pin_cpu pins the command to a CPU on Windows, where there is no taskset to put in the command.
+    def check_call(command, env: {}, raise_error: true, quiet: ENV['BENCHMARK_QUIET'] == '1', pin_cpu: nil)
       command_str = command.is_a?(Array) ? command.shelljoin : command
       puts("+ #{command_str}") unless quiet
 
       result = {}
 
-      if quiet
-        result[:success] = system(env, *command, out: File::NULL, err: File::NULL)
+      options = quiet ? { out: File::NULL, err: File::NULL } : {}
+      result[:success] = if pin_cpu
+        WindowsPinning.system(pin_cpu, env, *command, **options)
       else
-        result[:success] = system(env, *command)
+        system(env, *command, **options)
       end
       result[:status] = $?
 

lib/benchmark_suite.rb

@@ -12,14 +12,14 @@
 require_relative 'benchmark_discovery'
 require_relative 'ractor_breakdown'
 require_relative 'ractor_counts'
+require_relative 'windows_pinning'
 
 # BenchmarkSuite runs a collection of benchmarks and collects their results
 class BenchmarkSuite
   BENCHMARKS_DIR = "benchmarks"
   RACTOR_CATEGORY = ["ractor"].freeze
   RACTOR_ONLY_CATEGORY = ["ractor-only"].freeze
   RACTOR_HARNESS = "harness-ractor"
-  PIN_CPU_ENV = "RUBY_BENCH_PIN_CPU"
 
   attr_reader :categories, :name_filters, :excludes, :out_path, :harness, :harness_explicit, :pre_init, :no_pinning, :force_pinning, :bench_dir
 
@@ -48,18 +48,19 @@ def harness_for(benchmark_name)
   # Run a single benchmark entry on a single executable.
   # Returns { name:, data: } on success, { name:, failure: } on error.
   def run_benchmark(entry, ruby:, ruby_description:)
-    env = benchmark_env(ruby).merge(pinning_env(ruby_description, entry.name))
+    env = benchmark_env(ruby)
     caller_json_path = ENV["RESULT_JSON_PATH"]
     quiet = ENV['BENCHMARK_QUIET'] == '1'
     cmd_prefix = base_cmd(ruby_description, entry.name)
+    pin_cpu = windows_pin_cpu(ruby_description, entry.name)
     benchmark_harness = benchmark_harness_for(entry.name)
 
     if benchmark_harness == RACTOR_HARNESS
-      return run_ractor_benchmark(entry, ruby, cmd_prefix, env, benchmark_harness, caller_json_path, quiet: quiet)
+      return run_ractor_benchmark(entry, ruby, cmd_prefix, env, benchmark_harness, caller_json_path, quiet: quiet, pin_cpu: pin_cpu)
     end
 
     result_json_path = caller_json_path || File.join(out_path, "temp#{Process.pid}.json")
-    result = run_benchmark_process(entry.script_path, result_json_path, ruby, cmd_prefix, env, benchmark_harness, quiet: quiet)
+    result = run_benchmark_process(entry.script_path, result_json_path, ruby, cmd_prefix, env, benchmark_harness, quiet: quiet, pin_cpu: pin_cpu)
 
     if result[:success]
       { name: entry.name, data: process_benchmark_result(result_json_path, result[:command], delete_file: !caller_json_path), harness: benchmark_harness }
@@ -103,13 +104,13 @@ def setup_benchmark_directories
     end
   end
 
-  def run_ractor_benchmark(entry, ruby, cmd_prefix, env, benchmark_harness, caller_json_path, quiet: false)
+  def run_ractor_benchmark(entry, ruby, cmd_prefix, env, benchmark_harness, caller_json_path, quiet: false, pin_cpu: nil)
     blobs_by_count = {}
 
     ractor_counts_for(entry.name).each do |count|
       result_json_path = File.join(out_path, "temp#{Process.pid}_r#{count}.json")
       count_env = env.merge(RactorCounts::ENV_VAR => count.to_s)
-      result = run_benchmark_process(entry.script_path, result_json_path, ruby, cmd_prefix, count_env, benchmark_harness, quiet: quiet)
+      result = run_benchmark_process(entry.script_path, result_json_path, ruby, cmd_prefix, count_env, benchmark_harness, quiet: quiet, pin_cpu: pin_cpu)
 
       unless result[:success]
         FileUtils.rm_f(result_json_path)
@@ -140,13 +141,13 @@ def ractor_counts_for(benchmark_name)
 
   # Clear project-level Bundler environment so benchmarks run in a clean context.
   # Benchmarks that need Bundler (e.g., railsbench) set up their own via use_gemfile.
-  def run_benchmark_process(script_path, result_json_path, ruby, cmd_prefix, env, benchmark_harness, quiet: false)
+  def run_benchmark_process(script_path, result_json_path, ruby, cmd_prefix, env, benchmark_harness, quiet: false, pin_cpu: nil)
     if defined?(Bundler)
       Bundler.with_unbundled_env do
-        run_single_benchmark(script_path, result_json_path, ruby, cmd_prefix, env, benchmark_harness, quiet: quiet)
+        run_single_benchmark(script_path, result_json_path, ruby, cmd_prefix, env, benchmark_harness, quiet: quiet, pin_cpu: pin_cpu)
       end
     else
-      run_single_benchmark(script_path, result_json_path, ruby, cmd_prefix, env, benchmark_harness, quiet: quiet)
+      run_single_benchmark(script_path, result_json_path, ruby, cmd_prefix, env, benchmark_harness, quiet: quiet, pin_cpu: pin_cpu)
     end
   end
 
@@ -195,7 +196,7 @@ def filter_entries(entries, categories:, name_filters:, excludes:, directory_map
     entries.select { |entry| filter.match?(entry.name) }
   end
 
-  def run_single_benchmark(script_path, result_json_path, ruby, cmd_prefix, env, benchmark_harness, quiet: false)
+  def run_single_benchmark(script_path, result_json_path, ruby, cmd_prefix, env, benchmark_harness, quiet: false, pin_cpu: nil)
     # Fix for jruby/jruby#7394 in JRuby 9.4.2.0
     script_path = File.expand_path(script_path)
 
@@ -213,7 +214,7 @@ def run_single_benchmark(script_path, result_json_path, ruby, cmd_prefix, env, b
     ].compact).map(&:to_s)
 
     # Do the benchmarking
-    result = BenchmarkRunner.check_call(cmd, env: env, raise_error: false, quiet: quiet)
+    result = BenchmarkRunner.check_call(cmd, env: env, raise_error: false, quiet: quiet, pin_cpu: pin_cpu)
     result[:command] = cmd.shelljoin
     result
   ensure
@@ -288,9 +289,9 @@ def base_cmd(ruby_description, benchmark_name)
     end
   end
 
-  # Windows has no taskset, so the harness pins itself to the CPU given in this variable.
-  def pinning_env(ruby_description, benchmark_name)
-    { PIN_CPU_ENV => (pinned_cpu.to_s if windows? && pin?(ruby_description, benchmark_name)) }
+  # Windows has no taskset, so the runner pins the benchmark process to this CPU when it spawns it.
+  def windows_pin_cpu(ruby_description, benchmark_name)
+    pinned_cpu if windows? && pin?(ruby_description, benchmark_name)
   end
 
   # Pin the process to one given core to improve caching and reduce variance on CRuby

lib/windows_pinning.rb

@@ -0,0 +1,113 @@
+# frozen_string_literal: true
+
+# Windows has no taskset, so the runner pins a benchmark from outside the benchmark process, as
+# taskset does on Linux. A child process inherits the affinity mask of its parent when it is
+# created, so the runner narrows its own mask for the duration of the spawn. The priority class
+# is not inherited, so it and the power throttling opt-out are set on the child by its PID.
+module WindowsPinning
+  HIGH_PRIORITY_CLASS = 0x80
+  PROCESS_SET_INFORMATION = 0x0200
+  PROCESS_POWER_THROTTLING = 4 # PROCESS_INFORMATION_CLASS::ProcessPowerThrottling
+  # PROCESS_POWER_THROTTLING_STATE that turns off the EXECUTION_SPEED throttling Windows 11 may
+  # apply to a windowless process: Version, ControlMask and StateMask.
+  NO_EXECUTION_SPEED_THROTTLING = [1, 1, 0].pack('L3').freeze
+
+  class << self
+    # Like Kernel#system, but the command runs pinned to the given CPU. If pinning fails, the
+    # command runs unpinned with a warning.
+    def system(cpu, env, *command, **options)
+      pid = spawn(cpu, env, *command, **options)
+    rescue SystemCallError
+      # Let Kernel#system fail the same way, so the result and $? match an unpinned run.
+      Kernel.system(env, *command, **options)
+    else
+      Process.wait(pid)
+      $?.success?
+    end
+
+    def spawn(cpu, env, *command, **options)
+      previous_mask = pin_self(cpu)
+      begin
+        pid = Process.spawn(env, *command, **options)
+      ensure
+        restore_affinity_mask(previous_mask) if previous_mask
+      end
+      raise_priority(pid) if previous_mask
+      pid
+    end
+
+    private
+
+    # Narrows the affinity of this process to the CPU, and returns its previous mask.
+    def pin_self(cpu)
+      load_fiddle
+      # Etc.nprocessors counts the CPUs of every processor group, but the mask holds those of one group only.
+      if cpu >= Fiddle::SIZEOF_VOIDP * 8
+        warn "Not pinning to CPU #{cpu}, which is out of the affinity mask"
+        return nil
+      end
+      previous_mask = affinity_mask
+      set_affinity_mask(1 << cpu)
+      previous_mask
+    rescue LoadError, StandardError => e
+      warn "Failed to pin the benchmark: #{e.message}"
+      nil
+    end
+
+    # The child keeps the mask it inherited, so only the runner is affected if this fails.
+    def restore_affinity_mask(mask)
+      set_affinity_mask(mask)
+    rescue StandardError => e
+      warn "Failed to restore the affinity of the runner: #{e.message}"
+    end
+
+    def raise_priority(pid)
+      process = function('OpenProcess', [Fiddle::TYPE_INT, Fiddle::TYPE_INT, Fiddle::TYPE_INT], Fiddle::TYPE_VOIDP)
+        .call(PROCESS_SET_INFORMATION, 0, pid)
+      raise "OpenProcess failed" if process.null?
+      begin
+        set_priority_class = function('SetPriorityClass', [Fiddle::TYPE_VOIDP, Fiddle::TYPE_INT], Fiddle::TYPE_INT)
+        raise "SetPriorityClass failed" if set_priority_class.call(process, HIGH_PRIORITY_CLASS).zero?
+
+        set_information = function('SetProcessInformation', [Fiddle::TYPE_VOIDP, Fiddle::TYPE_INT, Fiddle::TYPE_VOIDP, Fiddle::TYPE_INT], Fiddle::TYPE_INT)
+        state = NO_EXECUTION_SPEED_THROTTLING.dup
+        raise "SetProcessInformation failed" if set_information.call(process, PROCESS_POWER_THROTTLING, state, state.bytesize).zero?
+      ensure
+        function('CloseHandle', [Fiddle::TYPE_VOIDP], Fiddle::TYPE_INT).call(process)
+      end
+    rescue StandardError => e
+      warn "Failed to raise the priority of the benchmark: #{e.message}"
+    end
+
+    def affinity_mask
+      process_mask = [0].pack('J')
+      system_mask = [0].pack('J')
+      get_affinity = function('GetProcessAffinityMask', [Fiddle::TYPE_VOIDP, Fiddle::TYPE_VOIDP, Fiddle::TYPE_VOIDP], Fiddle::TYPE_INT)
+      raise "GetProcessAffinityMask failed" if get_affinity.call(current_process, process_mask, system_mask).zero?
+      process_mask.unpack1('J')
+    end
+
+    def set_affinity_mask(mask)
+      set_affinity = function('SetProcessAffinityMask', [Fiddle::TYPE_VOIDP, Fiddle::TYPE_UINTPTR_T], Fiddle::TYPE_INT)
+      raise "SetProcessAffinityMask failed" if set_affinity.call(current_process, mask).zero?
+    end
+
+    def current_process
+      function('GetCurrentProcess', [], Fiddle::TYPE_VOIDP).call
+    end
+
+    def function(name, args, ret)
+      @kernel32 ||= Fiddle.dlopen("kernel32")
+      Fiddle::Function.new(@kernel32[name], args, ret)
+    end
+
+    def load_fiddle
+      require 'fiddle'
+    rescue LoadError
+      # fiddle is a bundled gem since Ruby 3.5, which Bundler hides unless the Gemfile lists it.
+      raise unless defined?(Bundler) && defined?(Gem::BUNDLED_GEMS.force_activate)
+      Gem::BUNDLED_GEMS.force_activate("fiddle")
+      require 'fiddle'
+    end
+  end
+end

test/benchmark_suite_test.rb

@@ -742,41 +742,39 @@ def base_cmd_for(ruby_description)
       before do
         File.write('benchmarks/pin_cpu.rb', <<~RUBY)
           require 'json'
-          File.write(ENV['RESULT_JSON_PATH'], JSON.generate('warmup' => [], 'bench' => [0.001], 'pin_cpu' => ENV['RUBY_BENCH_PIN_CPU']))
+          File.write(ENV['RESULT_JSON_PATH'], JSON.generate('warmup' => [], 'bench' => [0.001]))
         RUBY
       end
 
-      def pin_cpu_seen_by_benchmark(ruby_description: 'ruby 3.2.0', **options)
+      # Returns the CPUs WindowsPinning was asked to pin the benchmark to, without pinning anything.
+      def cpus_pinned_by_runner(ruby_description: 'ruby 3.2.0', **options)
         suite = BenchmarkSuite.new(categories: [], name_filters: ['pin_cpu'], out_path: @out_path, harness: 'harness', **options)
+        cpus = []
         result = nil
-        suite.stub(:windows?, true) do
-          suite.stub(:linux?, false) do
-            capture_io do
-              result = suite.run_benchmark(suite.benchmarks.first, ruby: [RbConfig.ruby], ruby_description: ruby_description)
+        pinned_system = ->(cpu, env, *command, **opts) { cpus << cpu; Kernel.system(env, *command, **opts) }
+        WindowsPinning.stub(:system, pinned_system) do
+          suite.stub(:windows?, true) do
+            suite.stub(:linux?, false) do
+              capture_io do
+                result = suite.run_benchmark(suite.benchmarks.first, ruby: [RbConfig.ruby], ruby_description: ruby_description)
+              end
             end
           end
         end
-        result[:data]['pin_cpu']
+        assert result[:data], 'the benchmark should succeed'
+        cpus
       end
 
-      it 'passes the CPU to pin to through the environment' do
-        assert_equal [(Etc.nprocessors / 3) - 1, 0].max.to_s, pin_cpu_seen_by_benchmark
+      it 'pins the benchmark from the runner' do
+        assert_equal [[(Etc.nprocessors / 3) - 1, 0].max], cpus_pinned_by_runner
       end
 
       it 'does not pin with no_pinning' do
-        assert_nil pin_cpu_seen_by_benchmark(no_pinning: true)
-      end
-
-      it 'does not pass on a CPU inherited from the environment' do
-        original = ENV['RUBY_BENCH_PIN_CPU']
-        ENV['RUBY_BENCH_PIN_CPU'] = '3'
-        assert_nil pin_cpu_seen_by_benchmark(no_pinning: true)
-      ensure
-        ENV['RUBY_BENCH_PIN_CPU'] = original
+        assert_empty cpus_pinned_by_runner(no_pinning: true)
       end
 
       it 'does not pin other Ruby implementations' do
-        assert_nil pin_cpu_seen_by_benchmark(ruby_description: 'truffleruby 24.0.0')
+        assert_empty cpus_pinned_by_runner(ruby_description: 'truffleruby 24.0.0')
       end
     end
   end