goodboy
/
tractor
1
0
Fork 2
Code Issues 5 Pull Requests 6 Packages Projects Releases Wiki Activity

Merge pull request #468 from goodboy/test_cpu_throttling 

Test cpu throttling
wkt/start_or_cancel_tests_474
Bd 2026-06-29 19:18:34 -04:00 committed by GitHub
parent f8b7233b3e 51326f4b7a
commit 65bf9df5ba
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 441 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

170
scripts/cpu-perf-check 100755
View File

@ -0,0 +1,170 @@
#!/usr/bin/env python3
# tractor: distributed structured concurrency.
# Copyright 2018-eternity Tyler Goodlet.
#
# SPDX-License-Identifier: AGPL-3.0-or-later
'''
`cpu-perf-check` — sustained-load CPU throttle detector.
Idle freq snapshots LIE. A laptop can read
`governor=performance`, `EPP=performance`,
`platform_profile=performance`, `scaling_max_freq=<full>`
and momentarily clock a P-core at 5GHz — while a
firmware/EC power cap (AMD PPT/STAPM and friends) clamps
the whole package to ~1.5GHz the instant a sustained
multi-core load lands. That throttle masquerades as a
`trio`-backend test *regression*: a wave of `fail_after` /
`TooSlowError` / `Cancelled(source='deadline')` deadline
misses on spawn-heavy tests, on byte-identical code that
was green yesterday.
The existing `tests/conftest.py:cpu_scaling_factor()` only
reads STATIC `scaling_max_freq` vs `*_pstate_max_freq`, so
it returns `1.0` (no throttle) during exactly this failure
— it can't see the cap. This script complements it by
BURNING every core for a few seconds and sampling the
ACHIEVED `scaling_cur_freq`, which is the only thing that
exposes the clamp.
Exit code: `0` if sustained perf looks restored, `1` if
throttled — so it gates a test run:
py313/bin/python scripts/cpu-perf-check && pytest tests/ ...
Tunables (env-overridable):
CPU_PERF_SECS load duration (default 4.0)
CPU_PERF_HEALTHY_FRAC sustained/max floor (default 0.45)
'''
from __future__ import annotations
import glob
import os
import time
import multiprocessing as mp
def _read(path: str) -> str | None:
try:
with open(path) as f:
return f.read().strip()
except OSError:
return None
def _cur_freqs_mhz() -> list[int]:
out: list[int] = []
for f in glob.glob(
'/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_cur_freq'
):
if (v := _read(f)):
out.append(int(v) // 1000)
return out
def _pkg_max_mhz() -> int:
'''
Highest per-core ceiling across the package — the
P-core max on hybrid parts.
'''
mxs: list[int] = []
for f in glob.glob(
'/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_max_freq'
):
if (v := _read(f)):
mxs.append(int(v) // 1000)
return max(mxs) if mxs else 0
def _burn(stop: float) -> None:
# fixed-width (64-bit-masked) arithmetic keeps this a steady
# ALU load; an unmasked `x` grows ~x**2/iter into a huge bigint,
# degenerating into noisy alloc/mul-bound work (and needless
# memory) across N procs.
x: int = 1
while time.perf_counter() < stop:
x = (x + (x * x ^ 0x5)) & 0xFFFF_FFFF_FFFF_FFFF
def main(
secs: float = float(os.environ.get('CPU_PERF_SECS', 4.0)),
# sustained aggregate must clear this fraction of the
# package max-freq ceiling. Throttled (~1.5GHz of 5.1GHz)
# ~= 0.29; a healthy all-core load easily clears 0.5.
healthy_frac: float = float(
os.environ.get('CPU_PERF_HEALTHY_FRAC', 0.45)
),
) -> int:
if not glob.glob('/sys/devices/system/cpu/cpu0/cpufreq'):
print('no cpufreq sysfs (non-linux?) — skipping, assume OK')
return 0
b: str = '/sys/devices/system/cpu/cpu0/cpufreq/'
pkg_max: int = _pkg_max_mhz()
print('=== static knobs (ALL can read fine while throttled) ===')
print(f' governor : {_read(b + "scaling_governor")}')
print(f' EPP : {_read(b + "energy_performance_preference")}')
print(f' platform_profile : '
f'{_read("/sys/firmware/acpi/platform_profile")}')
print(f' pkg max freq : {pkg_max} MHz')
ncpu: int = os.cpu_count() or 1
print(f'\n=== sustained {ncpu}-core load ({secs:.0f}s) — the real test ===')
stop: float = time.perf_counter() + secs
procs = [
mp.Process(target=_burn, args=(stop,), daemon=True)
for _ in range(ncpu)
]
for p in procs:
p.start()
# skip the initial ~0.6s ramp, then sample steady-state.
# `try/finally` so a Ctrl-C / sampling error still reaps the
# burners instead of leaving stray CPU hogs (daemon=True is the
# backstop should we exit abnormally before the join).
samples: list[int] = []
try:
time.sleep(0.6)
while time.perf_counter() < stop - 0.2:
if (fr := _cur_freqs_mhz()):
samples.append(sum(fr) // len(fr))
time.sleep(0.3)
finally:
for p in procs:
p.terminate()
for p in procs:
p.join()
if not (samples and pkg_max):
print(' could not sample cur freq — assume OK')
return 0
sustained: int = sum(samples) // len(samples)
frac: float = sustained / pkg_max
print(f' aggregate cur-freq samples: {samples}')
print(f' sustained avg : {sustained} MHz '
f'({frac * 100:.0f}% of {pkg_max} MHz max)')
if frac < healthy_frac:
print(
f'\n ❌ THROTTLED — sustained {sustained}MHz is only '
f'{frac * 100:.0f}% of max (< {healthy_frac * 100:.0f}%).\n'
f' Power cap (PPT/STAPM) still engaged. Fixes:\n'
f' - bounce /sys/firmware/acpi/platform_profile\n'
f' (balanced -> performance)\n'
f' - unplug/replug USB-C to re-negotiate PD\n'
f' - ryzenadj to lift STAPM/PPT\n'
f' - else reboot\n'
f' Do NOT bump test budgets — the box is slow, not the code.'
)
return 1
print(
f'\n ✅ PERF OK — sustained {sustained}MHz holds '
f'{frac * 100:.0f}% of max. Cap looks lifted; safe to run tests.'
)
return 0
if __name__ == '__main__':
raise SystemExit(main())

144
tests/conftest.py
View File

@ -134,6 +134,150 @@ def cpu_scaling_factor() -> float:
return factor
# session-cached sustained-load throttle multiplier — measured
# once (lazily) on the first `cpu_perf_headroom()` call. `None`
# = not-yet-measured.
_sustained_headroom: float|None = None
def _measure_sustained_headroom(
secs: float = 0.9,
# a healthy all-core sustained clock holds AT/ABOVE this
# fraction of the package single-core max ceiling (boost sags
# under full multi-core load even un-throttled, but not far);
# at/above it we assume no throttle and return 1.0.
throttle_gate: float = 0.6,
max_headroom: float = 4.,
) -> float:
'''
One-shot all-core burn returning a latency multiplier
(>= 1.0) that reflects *sustained-load* CPU throttle.
Catches the firmware/EC power-cap clamp (AMD PPT/STAPM &
friends) that pins achieved `scaling_cur_freq` to a fraction
of the ceiling under multi-core load while EVERY static knob
(`governor`, `scaling_max_freq`, `EPP`, `platform_profile`)
still reads "full performance". That cap is INVISIBLE to
`cpu_scaling_factor()` and is the gremlin behind mass `trio`
deadline-miss failures on byte-identical code — see
`scripts/cpu-perf-check`.
Best-effort: returns 1.0 on non-linux / missing sysfs / any
error so it can never break a test run.
'''
import glob
import multiprocessing as mp
def _read_mhz(path: str) -> int|None:
try:
with open(path) as f:
return int(f.read()) // 1000
except OSError:
return None
try:
maxs: list[int] = [
v for f in glob.glob(
'/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_max_freq'
)
if (v := _read_mhz(f)) is not None
]
pkg_max: int = max(maxs) if maxs else 0
if not pkg_max:
return 1.
def _burn(stop: float) -> None:
# fixed-width (64-bit-masked) arithmetic keeps this a
# steady ALU load; an unmasked `x` grows ~x**2/iter into
# a huge bigint, degenerating into noisy alloc/mul-bound
# work (and needless memory) across N procs.
x: int = 1
while time.perf_counter() < stop:
x = (x + (x * x ^ 0x5)) & 0xFFFF_FFFF_FFFF_FFFF
# explicit `fork` ctx so we're immune to whatever global
# mp start-method tractor/the suite may have set (`spawn`
# would re-exec + re-import 24x — slow and pointless here).
ctx = mp.get_context('fork')
ncpu: int = os.cpu_count() or 1
stop: float = time.perf_counter() + secs
procs = [
ctx.Process(target=_burn, args=(stop,), daemon=True)
for _ in range(ncpu)
]
for p in procs:
p.start()
# skip the ~0.4s boost window so we sample the steady
# state AFTER any power-cap has engaged.
samples: list[int] = []
time.sleep(0.4)
while time.perf_counter() < stop - 0.1:
curs: list[int] = [
v for f in glob.glob(
'/sys/devices/system/cpu/cpu[0-9]*/cpufreq/scaling_cur_freq'
)
if (v := _read_mhz(f)) is not None
]
if curs:
samples.append(sum(curs) // len(curs))
time.sleep(0.15)
for p in procs:
p.join()
if not samples:
return 1.
frac: float = (sum(samples) // len(samples)) / pkg_max
# below the gate we read it as a power-cap throttle. The
# spawn/IPC/fork-bound work these budgets guard slows ~1:1
# with the achieved-vs-max freq ratio, so compensate by the
# FULL inverse fraction (a boost-discounted factor
# under-shoots and still trips the marginal cases).
if frac >= throttle_gate:
return 1.
# a 0/parked-core freq read would `ZeroDivisionError` the
# inverse below — silently swallowed by the outer `except`
# into a 1.0 (no-throttle), defeating the probe on exactly
# the broken box it should flag; read 0 as max throttle.
if frac <= 0:
return max_headroom
return min(max_headroom, 1. / frac)
except Exception:
return 1.
def cpu_perf_headroom() -> float:
'''
Latency-headroom multiplier (>= 1.0) covering BOTH cpu-perf
throttle classes — multiply a test's deadline by it, e.g.
`timeout *= cpu_perf_headroom()`:
- static cpu-freq scaling — via `cpu_scaling_factor()`
(governor/policy lowered the `scaling_max_freq` ceiling).
- sustained-load power-cap throttle — via
`_measure_sustained_headroom()` (firmware/EC PPT/STAPM
clamps achieved freq under load while every static knob
reads "performance"; INVISIBLE to the static check). This
is the gremlin behind mass `trio` deadline-miss failures
on unchanged code — see
`ai/conc-anal/trio_033_cancel_cascade_slowdown_depth3_issue.md`.
The sustained probe runs ONCE per session (cached); the cost
is a ~0.9s all-core burn on first call only.
'''
global _sustained_headroom
static: float = cpu_scaling_factor()
if _non_linux:
return static
if _sustained_headroom is None:
_sustained_headroom = _measure_sustained_headroom()
return max(static, _sustained_headroom)
# NOTE, the `--ll`/`--tl` CLI flags + the `loglevel`, `test_log`
# and `testing_pkg_name` fixtures have been factored into the
# `tractor._testing.pytest` plugin (loaded via the `-p` entry in

11
tests/devx/test_debugger.py
View File

@ -794,6 +794,14 @@ def test_multi_nested_subactors_error_through_nurseries(
loglevel='pdb',
)
last_send_char: str|None = None
# inflate pexpect waits under CPU throttle — incl. the
# sustained-load power-cap invisible to static freq reads — so
# a slow-to-boot child REPL doesn't trip a false `TIMEOUT`.
# See `scripts/cpu-perf-check`.
from ..conftest import cpu_perf_headroom
headroom: float = cpu_perf_headroom()
for (
i,
send_char,
@ -817,6 +825,9 @@ def test_multi_nested_subactors_error_through_nurseries(
if is_forking_spawner:
timeout += 4
if headroom != 1.:
timeout *= headroom
try:
child.expect(
PROMPT,

9
tests/test_advanced_streaming.py
View File

@ -189,11 +189,18 @@ def test_dynamic_pub_sub(
# sits forever until external SIGINT. The `afk_alarm_w_trace`
# outer guard below is the AFK-safety counterpart (SIGALRM
# raises in the main thread regardless of trio scope state).
fail_after_s: int = (
fail_after_s: float = (
8
if is_forking_spawner
else 20
)
# inflate under CPU throttle — incl. the sustained-load
# power-cap invisible to static freq reads — so a slow box
# doesn't trip the deadline. See `scripts/cpu-perf-check`.
from .conftest import cpu_perf_headroom
headroom: float = cpu_perf_headroom()
if headroom != 1.:
fail_after_s *= headroom
async def main():
# bug-class-3 breadcrumb: tag each level of the cancel path

67
tests/test_cancellation.py
View File

@ -596,6 +596,15 @@ async def test_nested_multierrors(
# depth=3, BOTH variants will reliably `xpass` and
# pytest will yell — our signal to drop the marker. See
# `ai/conc-anal/cancel_cascade_too_slow_under_main_thread_forkserver_issue.md`.
#
# Probe CPU throttle ONCE up-front (folds in the sustained-load
# power-cap that static freq reads miss): used BOTH to inflate
# the deadline budget below AND to xfail depth=3, whose failure
# mode under throttle is a runtime-internal reap deadline — not
# a test-budget miss. See `scripts/cpu-perf-check`.
from .conftest import cpu_perf_headroom
headroom: float = cpu_perf_headroom()
if start_method == 'main_thread_forkserver':
request.node.add_marker(
pytest.mark.xfail(
@ -609,6 +618,34 @@ async def test_nested_multierrors(
)
)
# Under CPU throttle (incl. the sustained-load power-cap that
# static freq reads miss) the DEEP depth=3 tree trips tractor's
# INTERNAL reap deadlines (`soft_kill`/`hard_kill`
# `move_on_after`/`terminate_after=1.6`) before slow subprocs
# exit, injecting a `Cancelled(source='deadline')` into the BEG
# — the SAME shape-mismatch class as the MTF xfail above, and
# NOT fixable by inflating the test-level budget (the Cancelled
# is minted inside the runtime, not by our `fail_after`).
# xfail(strict=False) so it auto-clears the moment the box is
# un-throttled (`headroom == 1.`); depth=1's shallow tree stays
# under those internal deadlines so it just rides the budget
# inflation below. See `scripts/cpu-perf-check`.
elif (
depth == 3
and
headroom != 1.
):
request.node.add_marker(
pytest.mark.xfail(
strict=False,
reason=(
'CPU throttled — tractor reap deadline injects '
'Cancelled into BEG; see conc-anal/'
'trio_033_cancel_cascade_slowdown_depth3_issue.md'
),
)
)
# Per-backend/-depth budgets: in the non-hang case the
# whole spawn + cancel-cascade should complete in well
# under these. On the borderline hang case the
@ -636,11 +673,14 @@ async def test_nested_multierrors(
case ('main_thread_forkserver', 3):
timeout = 30
# headroom for CPU-freq scaling AND/OR slow CI so the inner
# snapshot-capturing budget doesn't fire spuriously on a
# sluggish runner; see `cpu_scaling_factor()`.
from .conftest import cpu_scaling_factor
timeout *= cpu_scaling_factor()
# inflate the budget by the throttle headroom probed above so
# a slow box doesn't masquerade as a deadline regression.
# NOTE, `headroom = cpu_perf_headroom()` (set above) is the
# SUPERSET of `cpu_scaling_factor()` — it folds in the static
# cpu-freq scaling + slow-CI bump AND the sustained-load
# throttle probe this depth-3 cascade was the poster child for.
if headroom != 1.:
timeout *= headroom
async with fail_after_w_trace(timeout):
try:
@ -748,7 +788,7 @@ def test_cancel_via_SIGINT_other_task(
started from a seperate ``trio`` child task.
'''
from .conftest import cpu_scaling_factor
from .conftest import cpu_perf_headroom
pid: int = os.getpid()
timeout: float = (
@ -758,8 +798,9 @@ def test_cancel_via_SIGINT_other_task(
if _friggin_windows: # smh
timeout += 1
# add latency headroom for CPU freq scaling (auto-cpufreq et al.)
headroom: float = cpu_scaling_factor()
# latency headroom for static cpu-freq scaling + sustained-load
# throttle + CI (auto-cpufreq et al.); see `cpu_perf_headroom()`.
headroom: float = cpu_perf_headroom()
if headroom != 1.:
timeout *= headroom
@ -962,11 +1003,11 @@ def test_fast_graceful_cancel_when_spawn_task_in_soft_proc_wait_for_daemon(
if _friggin_windows: # smh
timeout += 1
# CPU-scaling / CI latency headroom — macOS CI especially is
# slow for this graceful-vs-hard-reap timing race; see
# `cpu_scaling_factor()`.
from .conftest import cpu_scaling_factor
timeout *= cpu_scaling_factor()
# CPU-scaling / sustained-throttle / CI latency headroom — macOS
# CI especially is slow for this graceful-vs-hard-reap timing
# race; see `cpu_perf_headroom()`.
from .conftest import cpu_perf_headroom
timeout *= cpu_perf_headroom()
async def main():
start = time.time()

15
tests/test_clustering.py
View File

@ -24,8 +24,14 @@ def test_empty_mngrs_input_raises(
'actor-cluster teardown hangs intermittently on UDS'
)
# inflate under CPU throttle — incl. the sustained-load
# power-cap invisible to static freq reads. See
# `scripts/cpu-perf-check`.
from .conftest import cpu_perf_headroom
fail_after_s: float = 3 * cpu_perf_headroom()
async def main():
with trio.fail_after(3):
with trio.fail_after(fail_after_s):
async with (
open_actor_cluster(
modules=[__name__],
@ -93,6 +99,13 @@ async def test_streaming_to_actor_cluster(
10 if is_forking_spawner
else 6
)
# inflate under CPU throttle — incl. the sustained-load
# power-cap invisible to static freq reads. See
# `scripts/cpu-perf-check`.
from .conftest import cpu_perf_headroom
headroom: float = cpu_perf_headroom()
if headroom != 1.:
delay *= headroom
with trio.fail_after(delay):
async with (
open_actor_cluster(modules=[__name__]) as portals,

7
tests/test_docs_examples.py
View File

@ -160,7 +160,7 @@ def test_example(
'root-causing. Passes on Linux.'
)
from .conftest import cpu_scaling_factor
from .conftest import cpu_perf_headroom
timeout: float = (
60
@ -168,8 +168,9 @@ def test_example(
else 16
)
# add latency headroom for CPU freq scaling (auto-cpufreq et al.)
headroom: float = cpu_scaling_factor()
# add latency headroom for CPU freq scaling/throttle
# (auto-cpufreq et al.)
headroom: float = cpu_perf_headroom()
if headroom != 1.:
timeout *= headroom

6
tests/test_inter_peer_cancellation.py
View File

@ -24,7 +24,7 @@ from tractor._testing import (
expect_ctxc,
)
from .conftest import cpu_scaling_factor
from .conftest import cpu_perf_headroom
pytestmark = [
pytest.mark.skipon_spawn_backend(
@ -1280,12 +1280,12 @@ def test_peer_spawns_and_cancels_service_subactor(
async def _main():
headroom: float = cpu_scaling_factor()
headroom: float = cpu_perf_headroom()
this_fast_on_linux: float = 3
this_fast = this_fast_on_linux * headroom
if headroom != 1.:
test_log.warning(
f'Adding latency headroom on linux bc CPU scaling,\n'
f'Adding latency headroom on linux bc CPU perf scaling/throttle,\n'
f'headroom: {headroom}\n'
f'this_fast_on_linux: {this_fast_on_linux} -> {this_fast}\n'
)

14
tests/test_legacy_one_way_streaming.py
View File

@ -326,11 +326,11 @@ def time_quad_ex(
):
timeout += 1
# inflate the cancel-deadline for CPU-freq scaling AND/OR CI
# latency (see `cpu_scaling_factor()`) so the example isn't
# cancelled mid-stream on a throttled/CI runner.
from .conftest import cpu_scaling_factor
timeout *= cpu_scaling_factor()
# inflate the cancel-deadline for static cpu-freq scaling +
# sustained-load throttle + CI latency (see `cpu_perf_headroom()`)
# so the example isn't cancelled mid-stream on a throttled/CI box.
from .conftest import cpu_perf_headroom
timeout *= cpu_perf_headroom()
start: float = time.time()
results: list[int] = trio.run(partial(
@ -379,8 +379,8 @@ def test_a_quadruple_example(
# https://github.com/AdnanHodzic/auto-cpufreq?tab=readme-ov-file#example-config-file-contents
#
# HENCE this below latency-headroom compensation logic..
from .conftest import cpu_scaling_factor
headroom: float = cpu_scaling_factor()
from .conftest import cpu_perf_headroom
headroom: float = cpu_perf_headroom()
if headroom != 1.:
this_fast = this_fast_on_linux * headroom
test_log.warning(

4
tests/test_resource_cache.py
View File

@ -213,12 +213,12 @@ def test_open_local_sub_to_stream(
N local tasks using `trionics.maybe_open_context()`.
'''
from .conftest import cpu_scaling_factor
from .conftest import cpu_perf_headroom
timeout: float = (
4
if not platform.system() == "Windows"
else 10
) * cpu_scaling_factor()
) * cpu_perf_headroom()
if debug_mode:
timeout = 999

6
tests/test_spawning.py
View File

@ -153,9 +153,9 @@ async def test_most_beautiful_word(
# actor spawn + IPC round-trip is comfortably sub-second on a
# warm box, but slow/noisy CI runners (esp. macOS) blow a flat
# 1s deadline. Scale for CI/CPU-throttle headroom — `== 1s`
# locally where `cpu_scaling_factor()` is `1.0`.
from .conftest import cpu_scaling_factor
with trio.fail_after(1 * cpu_scaling_factor()):
# locally where `cpu_perf_headroom()` is `1.0`.
from .conftest import cpu_perf_headroom
with trio.fail_after(1 * cpu_perf_headroom()):
async with tractor.open_nursery(
debug_mode=debug_mode,
) as an:

11
tractor/_testing/addr.py
View File

@ -50,9 +50,14 @@ def get_rando_addr(
fight over the bind, cascading into a chain of
"Address already in use" failures.
For UDS this concern doesn't apply: `UDSAddress.get_random()`
already builds socket paths from `os.getpid()` so each
pytest process gets its own socket-path namespace.
For UDS the *cross*-process concern is handled by keying
socket paths off `os.getpid()` (each pytest process gets its
own namespace). *Within* a process — where `get_random()` has
no live runtime to name from — it also appends a per-call
`uuid4` token, so two calls (e.g. a `reg_addr` + a distinct
bind addr in one test body) yield distinct sockpaths rather
than aliasing to the same `name@pid.sock` and tripping
`EADDRINUSE`.
'''
addr_type: Type[_addr.Addres] = _addr._address_types[tpt_proto]

14
tractor/ipc/_uds.py
View File

@ -36,6 +36,7 @@ from typing import (
TYPE_CHECKING,
ClassVar,
)
from uuid import uuid4
import msgspec
import trio
@ -204,7 +205,18 @@ class UDSAddress(
else:
prefix: str = 'no_runtime_actor'
sockname: str = f'{prefix}@{pid}'
# XXX, no live actor -> no `Aid` to key off, so mix a
# per-CALL token into the NAME part for uniqueness:
# w/o a runtime the sockname is otherwise a pure fn of
# `(prefix, pid)`, so two `get_random()` calls in one
# proc alias to the SAME sockpath and the 2nd `.bind()`
# trips `EADDRINUSE`. Token goes BEFORE `@{pid}` so the
# canonical `...@{pid}.sock` suffix stays intact for the
# reapers keyed off it: `._testing._reap`'s
# `(?P<name>.+)@(?P<pid>\d+)\.sock` regex, and the
# `spawn._reap` `{name}@{pid}.sock` reconstruction.
token: str = uuid4().hex[:8]
sockname: str = f'{prefix}.{token}@{pid}'
sockpath: Path = Path(f'{sockname}.sock')
return UDSAddress(