diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bcfd543..87523e8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -71,10 +71,15 @@ jobs: # - 线程被 start() 了但迟迟跑不到第一行(等待被拉长管不到这种——它没有任何等待 # 可拉长)。这条轴真的抓到过东西:它让 TunnelManager.connect 里 stderr drain # 线程的竞态(读已被清空的 self.process)从"偶尔"变成"每次"。 + # - 有竞争者在真的烧 CPU。前两条轴说的都是"线程晚了",而且都只花墙钟——睡觉的线程 + # 不占谁的 CPU,所以"这段预算里该算完多少活"这一类断言能从它们下面走过去。第三条 + # 轴让工作线程每次 sleep/wait 都真的占一段 CPU,于是在 GIL 上留下真的竞争;代价与 + # 工作线程自己的活动成比例,而不是一条 24/7 的 hog 线程——那种做法实测把 pytest 的 + # 收集阶段从 0.43s 变成 30s,记录留在 tests/_injection.py 里。 # # 只跑一个 OS/版本:它的价值在确定性,不在覆盖率——版本/平台差异由上面的矩阵负责。 timing: - name: Timing guard (injected thread latency) + name: Timing guard (injected latency, CPU share) runs-on: ubuntu-latest steps: @@ -88,17 +93,23 @@ jobs: - name: Install project with dev deps run: python -m pip install --upgrade pip && pip install -e ".[dev]" - # 故意不加 `-q`:报告头里的那两行横幅就是"守卫真的开着"的证据,下面 grep 它们。 + # 故意不加 `-q`:报告头里的三行横幅就是"守卫真的开着"的证据,下面 grep 它们。 # 一条静默失效的注入会把这条腿变成空跑,而输出仍然是绿的。 - - name: Run tests with injected latency + # + # CPU 轴取 0.05s:默认的 GIL 切换间隔是 5ms,所以每次烧 CPU 都确定会被抢占至少 + # 一次——"竞争"是真的发生,而不只是一段更长的计算。开销与工作线程自己的活动成 + # 比例:套件里能被它触及的 worker sleep/wait 只有约 40 次,于是总共约 2 秒。 + - name: Run tests with injected latency and CPU contention env: PONTE_TEST_THREAD_DELAY: "0.15" PONTE_TEST_THREAD_START_DELAY: "0.15" + PONTE_TEST_THREAD_CPU: "0.05" run: | set -o pipefail pytest 2>&1 | tee slow-runner.log grep -q "thread latency: +0.15s" slow-runner.log grep -q "thread start delay: +0.15s" slow-runner.log + grep -q "worker CPU share: +0.05s" slow-runner.log # 一条腿覆盖三个"环境"维度(不是平台维度,平台由上面的矩阵负责)。合成一条是因为 # 它们都是同一件事的不同侧面:**这台机器不是开发者那台**。出错时 traceback 会指出 diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index a88f4d7..17ab826 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -37,16 +37,22 @@ Python 3.11–3.14, a job that re-runs it under a C locale / half-hour timezone with deprecations as errors, and a `build` job that installs the wheel and runs `ponte init`. Coverage is uploaded to Codecov. -One more job re-runs the suite with two independent injections (`tests/_injection.py`): -`PONTE_TEST_THREAD_DELAY=0.15` stretches every worker-thread sleep and wait, and +One more job re-runs the suite with three independent injections (`tests/_injection.py`): +`PONTE_TEST_THREAD_DELAY=0.15` stretches every worker-thread sleep and wait, `PONTE_TEST_THREAD_START_DELAY=0.15` holds a freshly started thread before it runs -its first line. The second one matters because it is unreachable by the first: a -thread that has not started executing yet has no waits to stretch. Either way, a -test that only passes because the machine is fast fails there **every time** -instead of flaking once in a while. Run it the same way before blaming a runner: +its first line, and `PONTE_TEST_THREAD_CPU=0.05` makes every worker sleep/wait also +burn real CPU. Each axis is unreachable by the others, which is why there are three +rather than one: a thread that has not started executing yet has no waits to +stretch, and a *sleeping* thread takes nobody's CPU — so an assertion about how +much work fits in a fixed budget walks straight past the first two and only fails +under the third. Either way, a test that only passes because the machine is fast +fails there **every time** instead of flaking once in a while. Run it the same way +before blaming a runner: ```bash -PONTE_TEST_THREAD_DELAY=0.15 PONTE_TEST_THREAD_START_DELAY=0.15 pytest +PONTE_TEST_THREAD_DELAY=0.15 \ + PONTE_TEST_THREAD_START_DELAY=0.15 \ + PONTE_TEST_THREAD_CPU=0.05 pytest ``` If you add an assertion that waits for something a background thread produces, wait @@ -169,15 +175,20 @@ Linux 上跑 lint + 类型检查,在 Windows / Linux / macOS × Python 3.11– 上跑同一套测试,另有一个任务在 C locale / 半时区偏移下、并把弃用告警当错误地跑一遍, 以及 `build` 任务会安装 wheel 并执行 `ponte init`。覆盖率上报到 Codecov。 -还有一个任务会用两个互相独立的注入(见 `tests/_injection.py`)再跑一遍: +还有一个任务会用三个互相独立的注入(见 `tests/_injection.py`)再跑一遍: `PONTE_TEST_THREAD_DELAY=0.15` 把工作线程的每次 sleep/wait 拉长; -`PONTE_TEST_THREAD_START_DELAY=0.15` 让刚 `start()` 的线程迟迟跑不到第一行。 -后者是前者够不到的:还没开始执行的线程没有任何等待可以被拉长——而它在空闲机器上 -永远通过、在忙机器上随机失败。两者共同的效果是:“只有机器够快才通过”的测试会 -**每次都**在那里失败,而不是偶发地红一次。怀疑是 runner 抽风之前,先这样在本地跑一遍: +`PONTE_TEST_THREAD_START_DELAY=0.15` 让刚 `start()` 的线程迟迟跑不到第一行; +`PONTE_TEST_THREAD_CPU=0.05` 让工作线程每次 sleep/wait 额外真的烧一段 CPU。 +为什么是三条而不是一条:每条都够不到另两条——还没开始执行的线程没有任何等待可以被 +拉长;而**正在睡觉**的线程不占任何人的 CPU,所以“这段预算里该算完多少活”这类断言 +从这两条下面直接走过去,只在第三条下才失败。它们共同的效果是:“只有机器够快才通过” +的测试会**每次都**在那里失败,而不是偶发地红一次。怀疑是 runner 抽风之前,先这样在 +本地跑一遍: ```bash -PONTE_TEST_THREAD_DELAY=0.15 PONTE_TEST_THREAD_START_DELAY=0.15 pytest +PONTE_TEST_THREAD_DELAY=0.15 \ + PONTE_TEST_THREAD_START_DELAY=0.15 \ + PONTE_TEST_THREAD_CPU=0.05 pytest ``` 如果你要写“等后台线程产出某件东西”的断言,请等截止时间(见 `tests/_waits.py`), diff --git a/README.md b/README.md index ca309bb..ca19dc4 100644 --- a/README.md +++ b/README.md @@ -326,12 +326,14 @@ python _smoke_test.py # zero-dependency quick check CI runs lint + types on Linux, and the test suite across Windows/Linux/macOS × Python 3.11–3.14, reporting coverage to [Codecov](https://codecov.io/gh/modusensus/ponte). One extra job re-runs the -suite with `PONTE_TEST_THREAD_DELAY=0.15` **and** -`PONTE_TEST_THREAD_START_DELAY=0.15`: the first stretches every worker-thread -sleep and wait, the second holds a freshly started thread before its first line -runs (a case no amount of wait-stretching can reach). A test that only passes on -a fast machine fails there *every* time instead of flaking once in a while. Set -the same variables locally to reproduce such a machine. Another job runs the +suite with three injections — `PONTE_TEST_THREAD_DELAY=0.15`, +`PONTE_TEST_THREAD_START_DELAY=0.15` and `PONTE_TEST_THREAD_CPU=0.05`: the first +stretches every worker-thread sleep and wait, the second holds a freshly started +thread before its first line runs (a case no amount of wait-stretching can +reach), and the third makes those same waits burn real CPU, so a test that +measures how much work fits in a budget is starved too. A test that only passes +on a fast machine fails there *every* time instead of flaking once in a while. +Set the same variables locally to reproduce such a machine. Another job runs the suite under a C locale (non-UTF-8 stdio) in a half-hour timezone with deprecation warnings as errors, and a `build` job installs the built wheel and runs `ponte init`, so a packaging regression cannot ship again. See @@ -634,11 +636,13 @@ python _smoke_test.py # 零依赖快速自检 CI 在 Linux 上跑 lint + 类型检查,在 Windows/Linux/macOS × Python 3.11–3.14 上跑测试,覆盖率上报到 -[Codecov](https://codecov.io/gh/modusensus/ponte)。另有一个任务会同时用 -`PONTE_TEST_THREAD_DELAY=0.15` 与 `PONTE_TEST_THREAD_START_DELAY=0.15` 再跑 -一遍:前者把工作线程的每次 sleep/wait 拉长,后者让刚 start() 的线程迟迟跑不到 -第一行(后半种情况没有任何等待可以被拉长)。于是"只有机器够快才通过"的测试会 -**每次都**在那里失败,而不是偶发地红一次。本地设同样两个变量即可复现这种机器。 +[Codecov](https://codecov.io/gh/modusensus/ponte)。另有一个任务会用三个注入再跑 +一遍:`PONTE_TEST_THREAD_DELAY=0.15`、`PONTE_TEST_THREAD_START_DELAY=0.15` +与 `PONTE_TEST_THREAD_CPU=0.05`。第一条把工作线程的每次 sleep/wait 拉长;第二条让刚 +start() 的线程迟迟跑不到第一行(没有任何等待可以被拉长的那种情况);第三条让这些 +等待额外真的烧一段 CPU,于是"这段预算里该算完多少活"的断言也会被饿着。于是"只有机器够快 +才通过"的测试会**每次都**在那里失败,而不是偶发地红一次。本地设同样三个变量即可 +复现这种机器。 还有一个任务在 C locale(非 UTF-8 的 stdio)、半时区偏移下跑,并把弃用告警当错误; `build` 任务会安装打好的 wheel 并执行 `ponte init`,避免打包问题再次溜进发布。详见 [CONTRIBUTING.md](CONTRIBUTING.md)。 diff --git a/tests/_injection.py b/tests/_injection.py index ab41ee1..6e24706 100644 --- a/tests/_injection.py +++ b/tests/_injection.py @@ -5,7 +5,7 @@ 够快"**。快机器上它永远绿,慢机器上它偶尔红,两种结果都不回答"被测代码对不对"。 于是这里做的事是:在需要的时候,把"慢"变成一个**开关**,让那类测试当场失败。它属于 -测试基础设施而不是产品代码,但它也是唯一能被"证明"的部分——两条轴都有对应的自检 +测试基础设施而不是产品代码,但它也是唯一能被"证明"的部分——三条轴都有对应的自检 (``tests/test_timing_guard.py``),因为一条静默失效的守卫比没有守卫更糟。 用法:由 ``tests/conftest.py`` 在会话开始时按环境变量装上(见 :func:`active`)。 @@ -22,13 +22,17 @@ THREAD_DELAY_ENV = "PONTE_TEST_THREAD_DELAY" #: Seconds a newly started thread waits *before running its body*. START_DELAY_ENV = "PONTE_TEST_THREAD_START_DELAY" +#: Seconds of *busy* work (GIL held) added to every worker sleep/wait. +THREAD_CPU_ENV = "PONTE_TEST_THREAD_CPU" # The unpatched primitives, captured at import time: without these the injection # would slow itself (and a thread start delay would be stretched by the wait -# delay, so the two axes would stop being independent). +# delay, so the axes would stop being independent). ``perf_counter`` is captured +# for the same reason — the burn must measure its own deadline, not a patched one. _ORIGINAL_SLEEP = time.sleep _ORIGINAL_WAIT = threading.Event.wait _ORIGINAL_THREAD_START = threading.Thread.start +_ORIGINAL_PERF_COUNTER = time.perf_counter def _number_env(name: str) -> float: @@ -49,15 +53,31 @@ def _number_env(name: str) -> float: return value +def _burn(seconds: float) -> None: + """Hold the GIL for *seconds* of real work — the thing a sleeping thread never does. + + A busy loop rather than another ``sleep``: the point of this axis is that the + worker *uses* the CPU, so it competes for the GIL with whatever else is + running. Sleeping releases the GIL and therefore takes nobody's CPU, which is + exactly why the delay axis cannot emulate this. + """ + if seconds <= 0: + return + deadline = _ORIGINAL_PERF_COUNTER() + seconds + while _ORIGINAL_PERF_COUNTER() < deadline: + pass + + @dataclass(frozen=True) class Injection: """The load this process is asked to emulate. Zero means "off".""" wait_delay: float = 0.0 start_delay: float = 0.0 + thread_cpu: float = 0.0 def enabled(self) -> bool: - return bool(self.wait_delay or self.start_delay) + return bool(self.wait_delay or self.start_delay or self.thread_cpu) def banner(self) -> str: """One line per enabled axis, so a green run can prove the guard was on.""" @@ -66,6 +86,10 @@ def banner(self) -> str: lines.append( f"thread latency: +{self.wait_delay:g}s injected into every worker sleep/wait" ) + if self.thread_cpu: + lines.append( + f"worker CPU share: +{self.thread_cpu:g}s of busy work per worker sleep/wait" + ) if self.start_delay: lines.append( f"thread start delay: +{self.start_delay:g}s before every thread body runs" @@ -84,7 +108,11 @@ def active() -> Injection: """ global _INSTALLED if _INSTALLED is None: - _INSTALLED = Injection(_number_env(THREAD_DELAY_ENV), _number_env(START_DELAY_ENV)) + _INSTALLED = Injection( + _number_env(THREAD_DELAY_ENV), + _number_env(START_DELAY_ENV), + _number_env(THREAD_CPU_ENV), + ) if _INSTALLED.enabled(): _apply(_INSTALLED) return _INSTALLED @@ -98,7 +126,7 @@ def pytest_configure(config) -> None: # noqa: ARG001 - 只为让 `-p _injection def _apply(injection: Injection) -> None: """Make the code under test slower than the test that watches it. - A slow machine fails timing-dependent tests in two distinct ways, and each + A slow machine fails timing-dependent tests in three distinct ways, and each one needs its own instrument: **1. the worker's own waits are stretched** (``wait_delay``). "Slow machine" @@ -120,29 +148,40 @@ def _apply(injection: Injection) -> None: race in ``TunnelManager.connect`` (the drain thread read ``self.process``, which the session's ``finally`` had already cleared). - **...and deliberately no third axis for "the worker got no CPU at all".** - That was tried and measured — a busy Python thread competing for the GIL, - plus a duty-cycled variant — and it is both worse at finding bugs and far - more expensive. On this suite, one such thread turned pytest's *collection* - phase from 0.43s into 30s (two threads: 68s): GIL contention adds up to - ``sys.getswitchinterval()`` of latency to every reacquisition, and a suite - performs thousands of those. And it did not fail the very test this whole - mechanism exists for — a counter-scheduled worker still runs inside a 50ms - nap, because that nap releases the GIL. What a hog adds on top of the two - axes above is wall-clock cost, not coverage. The genuine "no spare CPU" - failure — a worker that never reaches its first bytecode in time — is axis 2. + **3. the worker runs, but it is not alone on the CPU** (``thread_cpu``). + The first two axes are both "a worker that is *late*", and both of them cost + nothing but wall clock — which is the point, and also the limit: a thread + that is merely asleep takes nobody's CPU, so a test that measures **how much + work fits in a fixed budget** (rather than "did this happen yet") sails + through both. Real contention looks different: something else is holding the + CPU, so whoever else wants it waits for the GIL. ``thread_cpu`` adds real + busy work *inside the worker's own sleep/wait*, which means two things at + once — the worker's iteration takes longer, and while it burns, every other + thread waits up to ``sys.getswitchinterval()`` to get the GIL back. + + Cost is what made this axis worth designing carefully. The obvious + implementation — a free-running busy thread for the whole process — was tried + and measured, and it is unusable: one such thread turned pytest's *collection* + phase from 0.43s into 30s (two threads: 68s), because collection performs + thousands of tiny GIL reacquisitions and each one can cost up to + ``sys.getswitchinterval()``. Worse, it did not even fail the tests this + mechanism exists for, since their workers sleep (releasing the GIL). Burning + *proportionally to the worker's own activity* fixes both halves: the cost is + bounded by the number of worker sleep/waits rather than by wall time (nothing + burns during collection, or in the test body, or while injection is off), and + the contention lands exactly on the threads that are supposedly competing. + + Scope, honestly: axis 3 emulates *another thread in this process* holding the + CPU, which is what GIL contention is. It cannot express "this container got a + fraction of a core" for a worker that never sleeps or waits — there is no + Python-level hook for "every bytecode of that thread" that would not itself + distort the measurement — nor OS-level effects such as a stalled filesystem + or a cold CPU. Don't read a green run here as "no timing assumptions left". Two primitives cover how a background thread paces itself: ``Event.wait`` and ``time.sleep``. Clocks (``time.monotonic``) are deliberately left alone — moving those would corrupt deadlines instead of emulating load. A ``sleep(0)`` is a yield rather than a wait, and load does not stretch it, so it stays. - - Scope, honestly: both axes emulate *a worker that is late*, which is the - failure mode behind every flake this exists for. What they cannot express is a - main-thread budget measured in work rather than in time — "this loop should - have run 10_000 times in 0.2s", which is the one thing the rejected hog above - *would* have caught, and only by an unpredictable factor. Don't read a green - run here as "no timing assumptions left". """ if injection.start_delay: # Instance-level ``run`` override: the delay happens *inside* the new @@ -160,17 +199,19 @@ def delayed_run() -> None: threading.Thread.start = slow_start # type: ignore[method-assign] - if injection.wait_delay: + if injection.wait_delay or injection.thread_cpu: main_thread = threading.main_thread() def slow_sleep(seconds: float) -> None: if seconds > 0 and threading.current_thread() is not main_thread: seconds += injection.wait_delay + _burn(injection.thread_cpu) _ORIGINAL_SLEEP(seconds) def slow_wait(event: threading.Event, timeout: float | None = None) -> bool: if timeout is not None and threading.current_thread() is not main_thread: timeout += injection.wait_delay + _burn(injection.thread_cpu) return _ORIGINAL_WAIT(event, timeout) time.sleep = slow_sleep diff --git a/tests/test_timing_guard.py b/tests/test_timing_guard.py index e1be461..1aece70 100644 --- a/tests/test_timing_guard.py +++ b/tests/test_timing_guard.py @@ -19,7 +19,7 @@ import pytest -from _injection import START_DELAY_ENV, THREAD_DELAY_ENV, active +from _injection import START_DELAY_ENV, THREAD_CPU_ENV, THREAD_DELAY_ENV, active _INJECTION = active() @@ -62,6 +62,73 @@ def worker(): ''' +#: 第三条轴的探针:"这段固定的 CPU 工作量应该在这个预算内跑完"——这正是头两条轴 +#: 够不到的那类断言(它们模拟的是"谁什么时候跑到",不是"这段时间能算多少")。 +#: 预算是按机器标定出来的**时长**(秒),不是固定的循环次数:固定次数在快机器上短到 +#: 装不下几个竞争周期,效果就被抹平(实测在一个 CI runner 上只有 1.79,而本机 3.1)。 +_PROBE_SECONDS = "0.3" +_PROBE_COMPETITORS = "3" +#: 同一档里重复几轮取最快的一次:外部负载只会把时间**拖长**,取最小值就是各档真实 +#: 水平的一致估计,而不会把"机器正好忙"算成"这条轴起了作用"。 +_PROBE_REPEATS = "2" +#: 刻意比 CI 用的 0.05s 夸张(同 ``_PROBE_START_DELAY``):这条自检要在任何机器上 +#: 确定性地分出高下,而不是复现 CI 的取值。 +_PROBE_CPU_SHARE = "0.10" + +#: 判据:被饿着的那次必须比两个基线都快这么多。实测的比值:本机 3.1,最坏一次 1.79 +#: (CI 的 3.13/ubuntu 腿);而"关掉注入"与"只拉长等待"几乎一样(约 1.0,最坏 1.2)。 +#: 取 1.3 是**实测最坏值的一半以下**——这条自检自己也不能变成"只有机器够快才通过"的 +#: 断言,否则它就是在重犯它要防的错。把 ``_burn`` 禁掉后比值落到 0.94,仍稳稳地在 +#: 判据之下(已复现),所以放宽容度不等于失去灵敏度。 +_MIN_SLOWDOWN = 1.3 + +#: 探针源码刻意全 ASCII:C locale 下(见 ci.yml 的 env-edges 腿)非 ASCII 连 +#: ``-c`` 的 argv 都传不进子进程(``os.posix_spawn`` 抛 UnicodeEncodeError)。 +_CPU_CONTENTION_PROBE = ''' +import os +import threading +import time + +import _injection + +_injection.active() + +unit = 200000 +began = time.perf_counter() +total = 0 +for i in range(unit): + total += i * i +unit_seconds = max(time.perf_counter() - began, 1e-6) +work = max(unit, int(unit * float(os.environ["PROBE_SECONDS"]) / unit_seconds)) + +stop = threading.Event() + + +def worker(): + event = threading.Event() + while not stop.is_set(): + event.wait(0.001) + + +for _ in range(int(os.environ["PROBE_COMPETITORS"])): + threading.Thread(target=worker, daemon=True).start() + +time.sleep(0.2) + +best = None +for _ in range(int(os.environ["PROBE_REPEATS"])): + began = time.perf_counter() + total = 0 + for i in range(work): + total += i * i + elapsed = time.perf_counter() - began + best = elapsed if best is None else min(best, elapsed) + +stop.set() +print(best) +''' + + def _wait_in_worker() -> float: """Seconds a *worker* thread actually spends in ``Event.wait(0.01)``.""" elapsed: list[float] = [] @@ -118,6 +185,96 @@ def test_start_delay_holds_a_new_thread_before_its_body_runs() -> None: assert latency < 0.25, latency +def _contention_probe(*, delay: str, cpu: str) -> float: + """在子进程里跑一次固定 CPU 预算,返回它花掉的秒数。 + + 子进程自带开关(而不是继承本进程的),所以这条自检在**任何**配置下都有效——包括 + 注入全关的普通矩阵腿里:它验证的是机制本身,而不是"CI 那一步恰好开着"。 + """ + env = os.environ.copy() + for name in ( + "COV_CORE_SOURCE", + "COV_CORE_CONFIG", + "COV_CORE_DATAFILE", + "PYTEST_CURRENT_TEST", + THREAD_DELAY_ENV, + START_DELAY_ENV, + THREAD_CPU_ENV, + ): + env.pop(name, None) + env[THREAD_DELAY_ENV] = delay + env[THREAD_CPU_ENV] = cpu + env["PROBE_SECONDS"] = _PROBE_SECONDS + env["PROBE_COMPETITORS"] = _PROBE_COMPETITORS + env["PROBE_REPEATS"] = _PROBE_REPEATS + tests_dir = str(Path(__file__).parent) + env["PYTHONPATH"] = tests_dir + os.pathsep + env.get("PYTHONPATH", "") + done = subprocess.run( + [sys.executable, "-c", _CPU_CONTENTION_PROBE], + capture_output=True, + text=True, + env=env, + timeout=120, + ) + assert done.returncode == 0, done.stdout + done.stderr + return float(done.stdout.strip()) + + +def test_cpu_share_slows_a_competing_thread_where_delays_cannot() -> None: + """第三条轴的**独有**覆盖:同一段 CPU 工作量,在竞争者真的烧 CPU 时明显变慢。 + + 两个基线缺一不可,因为一个比值本身说明不了"是谁干的":关掉注入那次给出这台机器 + 本来多快;``delay`` 那次才是重点——**拉长等待做不到这件事**。睡觉的线程不占 CPU, + 所以"工作线程缺 CPU"这类断言在 delay 轴下照样通过,这正是第三条轴存在的理由。 + """ + quiet = _contention_probe(delay="0", cpu="0") + delayed = _contention_probe(delay="0.15", cpu="0") + starved = _contention_probe(delay="0", cpu=_PROBE_CPU_SHARE) + assert starved >= max(quiet, delayed) * _MIN_SLOWDOWN, ( + f"quiet={quiet:.3f}s delayed={delayed:.3f}s starved={starved:.3f}s " + f"(ratio={starved / max(quiet, delayed):.2f}, need {_MIN_SLOWDOWN})" + ) + + +def _thread_cpu_spent_in_a_wait(seconds: float, *, in_worker: bool) -> float: + """一次 ``Event.wait(seconds)`` 花掉**本线程**多少 CPU 秒。 + + ``time.thread_time()`` 而不是墙钟:这样断言是"花/不花 CPU"的类别差别,不随机器 + 快慢漂移,也不会因为 runner 忙而误报。 + """ + spent: list[float] = [] + + def run() -> None: + before = time.thread_time() + threading.Event().wait(seconds) + spent.append(time.thread_time() - before) + + if in_worker: + thread = threading.Thread(target=run, daemon=True) + thread.start() + thread.join(timeout=5) + else: + run() + assert spent, "the wait never completed" + return spent[0] + + +def test_cpu_share_burns_worker_threads_and_not_the_main_thread() -> None: + """第三条轴只让**工作线程**花 CPU;主线程那次等待一毫秒都不多。 + + 主线程这一半与第一条轴同理(见 ``_injection``):把测试自己的时间也拿去竞争,会 + 让"测试的预算"和"被测代码的预算"一起变,比例不变、什么都测不出来。 + """ + worker_cpu = _thread_cpu_spent_in_a_wait(0.02, in_worker=True) + main_cpu = _thread_cpu_spent_in_a_wait(0.02, in_worker=False) + if _INJECTION.thread_cpu: + assert worker_cpu >= _INJECTION.thread_cpu * 0.6, worker_cpu + assert main_cpu < _INJECTION.thread_cpu * 0.3, main_cpu + else: + assert worker_cpu < 0.02, worker_cpu + assert main_cpu < 0.02, main_cpu + + def _conftest_header() -> str: """The header our own conftest contributes, via the module pytest loaded.""" # 不调 pytestconfig.hook:那个钩子是 firstresult,返回的是**别的插件**的结果; @@ -141,6 +298,7 @@ def test_injection_is_announced_in_the_report_header() -> None: assert _INJECTION.banner() in header, header else: assert "thread latency" not in header, header + assert "worker CPU share" not in header, header assert "thread start delay" not in header, header