python: judge the CPU throttle by duty cycle instead of wall-clock ratio

python/tests/test_sandbox.py

@@ -5,9 +5,9 @@
 
 import json
 import os
+import resource
 import signal
 import socket
-import statistics
 import sys
 import threading
 import time
@@ -736,46 +736,26 @@ class TestCpuThrottle:
         "print(total)"
     )
 
-    def test_throttle_slows_execution(self):
-        # A 50% CPU cap should roughly double the wall-clock time of a CPU-bound
-        # loop. Comparing raw wall-clock times directly is flaky for two reasons:
-        #   * Fixed interpreter-startup cost is added to both runs but is not
-        #     throttled like the sustained loop, biasing the ratio toward 1.0
-        #     (this is what makes a tight lower bound fail intermittently).
-        #   * Scheduler jitter and background load add per-run noise.
-        # Remove the startup bias by subtracting the cost of a no-op run, and
-        # damp the jitter with a median over a few samples. With both corrections
-        # the ratio reliably centers on ~2.0.
-        SAMPLES = 3
-
-        def timed(argv, *, max_cpu=None):
-            kwargs = {} if max_cpu is None else {"max_cpu": max_cpu}
+    def test_throttle_halves_duty_cycle(self):
+        # The throttle promises a duty cycle, not a wall-clock slowdown. A core
+        # that idles every 50ms runs the awake half slower under frequency
+        # scaling, so wall-clock ratios drift well past 2x on such hosts.
+        def duty_cycle(**kwargs):
+            r0 = resource.getrusage(resource.RUSAGE_CHILDREN)
             t0 = time.monotonic()
-            result = _policy(**kwargs).run(argv)
+            result = _policy(**kwargs).run(["python3", "-c", self._BURN_CODE])
+            wall = time.monotonic() - t0
+            r1 = resource.getrusage(resource.RUSAGE_CHILDREN)
             assert result.success
-            return time.monotonic() - t0
+            cpu = (r1.ru_utime - r0.ru_utime) + (r1.ru_stime - r0.ru_stime)
+            return cpu / wall
 
-        def median_time(argv, *, max_cpu=None):
-            return statistics.median(
-                timed(argv, max_cpu=max_cpu) for _ in range(SAMPLES)
-            )
-
-        burn = ["python3", "-c", self._BURN_CODE]
-        overhead = median_time(["python3", "-c", "pass"])
-        base = median_time(burn)
-        throttled = median_time(burn, max_cpu=50)
-
-        base_compute = base - overhead
-        throttled_compute = throttled - overhead
-        assert base_compute > 0, (
-            f"workload too short to measure (base={base:.3f}s "
-            f"overhead={overhead:.3f}s)"
-        )
+        unthrottled = duty_cycle()
+        assert unthrottled > 0.8, f"unthrottled duty cycle {unthrottled:.2f}"
 
-        ratio = throttled_compute / base_compute
-        assert 1.5 <= ratio <= 3.0, (
-            f"ratio={ratio:.2f}, expected ~2.0 "
-            f"(base={base:.3f}s throttled={throttled:.3f}s overhead={overhead:.3f}s)"
+        throttled = duty_cycle(max_cpu=50)
+        assert 0.35 <= throttled <= 0.65, (
+            f"throttled duty cycle {throttled:.2f}, expected ~0.5"
         )
 
     def test_throttle_100_is_noop(self):