From a43cc0e01054357c2d33db1031e788f959b9ebcb Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Thu, 22 Oct 2026 12:04:40 -0500 Subject: [PATCH 1/3] test(real-cpu): pin portable peak RSS and an import that leaves thread env alone Fails on the parent: _peak_rss_mb (the old _vm_hwm_mb, renamed only) reads /proc/self/status and ignores ru_maxrss, and importing the module sets the four thread variables for every later test. Co-Authored-By: Claude Opus 5.5 --- backend/tests/test_real_cpu_upstream.py | 66 ++++++++++++++++++++++++- 1 file changed, 64 insertions(+), 2 deletions(-) diff --git a/backend/tests/test_real_cpu_upstream.py b/backend/tests/test_real_cpu_upstream.py index 68268ac1..2d81de06 100644 --- a/backend/tests/test_real_cpu_upstream.py +++ b/backend/tests/test_real_cpu_upstream.py @@ -30,10 +30,16 @@ import json import os +import resource +import subprocess +import sys import time from pathlib import Path +from types import SimpleNamespace from typing import Any +import pytest + os.environ.setdefault("OMP_NUM_THREADS", "2") os.environ.setdefault("OPENBLAS_NUM_THREADS", "2") os.environ.setdefault("MKL_NUM_THREADS", "2") @@ -54,6 +60,7 @@ WALL_BUDGET_S = 600.0 RSS_BUDGET_MB = 8 * 1024 +THREAD_ENV = ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS", "HSS_POSES_THREADS") STAGE_ENTRY = { "validate": validate_run, @@ -69,7 +76,7 @@ class _Settings: VISION_EXTERNAL_CALLS = False -def _vm_hwm_mb() -> float: +def _peak_rss_mb() -> float: """Process peak RSS so far. Monotonic, so a stage's entry is the high-water mark through it.""" for line in Path("/proc/self/status").read_text().splitlines(): if line.startswith("VmHWM"): @@ -167,7 +174,7 @@ def _run_chain( stored = _publish(_ctx(stage, work_dir, packet_dir, storage, capture_id, run_id, branch, tier), result) report[stage] = { "wall_s": round(wall_s, 2), - "vm_hwm_mb": round(_vm_hwm_mb(), 1), + "vm_hwm_mb": round(_peak_rss_mb(), 1), "artifacts": sorted(stored), "metrics": {k: round(float(v), 4) for k, v in result.metrics.items()}, "warnings": list(result.warnings), @@ -258,3 +265,58 @@ def test_mini_arkit_refuses_metric_scale_loudly(tmp_path) -> None: assert scale["reason"] == "scale_unresolved" assert scale["reason_detail"] == "meter_model_unread" assert scale["confidence"] == "weak" + + +def _fake_getrusage(ru_maxrss: int): + def getrusage(who: int) -> SimpleNamespace: + assert who == resource.RUSAGE_SELF + return SimpleNamespace(ru_maxrss=ru_maxrss) + + return getrusage + + +def test_peak_rss_mb_reads_bytes_on_darwin(monkeypatch) -> None: + """macOS has no /proc and reports ru_maxrss in bytes.""" + monkeypatch.setattr(sys, "platform", "darwin") + monkeypatch.setattr(resource, "getrusage", _fake_getrusage(2 * 1024 * 1024)) + assert _peak_rss_mb() == 2.0 + + +def test_peak_rss_mb_reads_kib_on_linux(monkeypatch) -> None: + monkeypatch.setattr(sys, "platform", "linux") + monkeypatch.setattr(resource, "getrusage", _fake_getrusage(2 * 1024 * 1024)) + assert _peak_rss_mb() == 2048.0 + + +def _child_env(request: pytest.FixtureRequest, threads: str | None) -> dict[str, str]: + """The environment for a child Python: pytest's `pythonpath` entries, and the + thread variables set to `threads`, or removed when it is None.""" + env = {k: v for k, v in os.environ.items() if k not in THREAD_ENV} + if threads is not None: + env.update(dict.fromkeys(THREAD_ENV, threads)) + paths = [str(p) for p in request.config.getini("pythonpath")] + if os.environ.get("PYTHONPATH"): + paths.append(os.environ["PYTHONPATH"]) + env["PYTHONPATH"] = os.pathsep.join(paths) + return env + + +def test_importing_this_module_leaves_thread_env_alone(request) -> None: + """Thread limits belong to the chain's own process. Collecting this module must + not set them for every test that runs after it.""" + script = ( + "import importlib.util, os, sys\n" + "spec = importlib.util.spec_from_file_location('realcpu_import_probe', sys.argv[1])\n" + "spec.loader.exec_module(importlib.util.module_from_spec(spec))\n" + f"print(sorted(k for k in {THREAD_ENV!r} if k in os.environ))\n" + ) + proc = subprocess.run( + [sys.executable, "-c", script, __file__], + cwd=ROOT, + env=_child_env(request, threads=None), + capture_output=True, + text=True, + check=False, + ) + assert proc.returncode == 0, proc.stderr[-4000:] + assert proc.stdout.strip().splitlines()[-1] == "[]" From 9237a5b86780773cdb34ac508ece0d6c94a679de Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Thu, 22 Oct 2026 12:10:23 -0500 Subject: [PATCH 2/3] fix(real-cpu): measure each chain in its own pinned process with ru_maxrss _peak_rss_mb reads resource.getrusage(RUSAGE_SELF).ru_maxrss, KiB on Linux and bytes on macOS, instead of /proc/self/status, which macOS lacks. Each chain now runs through tests/realcpu_launcher.py: pinned to two CPUs, the four thread variables set to 2 in that process only, and forked from a bare interpreter so its peak RSS is its own. A subprocess child inherits the parent's lifetime peak in ru_maxrss on Linux (vfork + execve); measured on WSL2 6.18 as 626 MB in a child whose VmHWM was 11 MB. The module no longer sets thread variables at import. The wall budget is asserted only when the chain actually ran on two CPUs or fewer, and skipped loudly on macOS. The scale assertions now pin the shipped claim: source arkit, status unverified, the refusal reason, ARKit poses as the only used estimate, a positive bound, and on the LiDAR fixture the unused LiDAR estimate with lidar_too_few_pairs. Co-Authored-By: Claude Opus 5.5 --- backend/tests/realcpu_launcher.py | 71 ++++++++ backend/tests/test_real_cpu_upstream.py | 212 ++++++++++++++++++------ 2 files changed, 233 insertions(+), 50 deletions(-) create mode 100644 backend/tests/realcpu_launcher.py diff --git a/backend/tests/realcpu_launcher.py b/backend/tests/realcpu_launcher.py new file mode 100644 index 00000000..585b2f53 --- /dev/null +++ b/backend/tests/realcpu_launcher.py @@ -0,0 +1,71 @@ +"""Run one function of a test module in a fresh process pinned to two CPUs. + +Usage: python realcpu_launcher.py MODULE_PATH FUNCTION [ARGS...] + +`test_real_cpu_upstream.py` measures the real stages' wall time and peak RSS +against the 2-CPU, 8 GB worker budget. Both numbers have to belong to the +stages, not to the pytest process that started them, which is why this +launcher exists and why it forks. + +A child started with `subprocess` reports its parent's lifetime peak as its own +ru_maxrss on Linux: CPython starts it with vfork, and execve folds the parent's +high-water mark into the child's counter. Measured on WSL2 (kernel 6.18): after +the parent touched 600 MB, a child whose VmHWM was 11 MB reported a ru_maxrss of +626 MB. A plain fork() starts a fresh counter at the forking process's current +RSS, so the launcher forks while it is still a bare interpreter and runs the +work in that fork. + +Pinning happens here for a related reason. sched_setaffinity on pid 0 applies +to the calling thread and the threads it creates afterwards, and OpenBLAS starts +its thread pool when numpy is imported. Pinning and forking before the module +under test is imported keeps every thread on the two CPUs and keeps the fork +single-threaded. macOS has no sched_setaffinity; the test reads the affinity +the work actually ran with and only claims the wall-time budget when it is two +CPUs or fewer. +""" + +from __future__ import annotations + +import importlib.util +import os +import sys +import traceback + +CPUS = 2 + + +def _pin() -> None: + if hasattr(os, "sched_setaffinity"): + os.sched_setaffinity(0, sorted(os.sched_getaffinity(0))[:CPUS]) + + +def _run(module_path: str, function: str, args: list[str]) -> int: + try: + spec = importlib.util.spec_from_file_location("realcpu_child", module_path) + if spec is None or spec.loader is None: + raise ImportError(f"cannot load {module_path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + getattr(module, function)(*args) + except BaseException: + traceback.print_exc() + return 1 + finally: + # os._exit skips interpreter shutdown, which is what would flush these. + sys.stdout.flush() + sys.stderr.flush() + return 0 + + +def main(argv: list[str]) -> int: + module_path, function, *args = argv + _pin() + pid = os.fork() + if pid == 0: + os._exit(_run(module_path, function, args)) + _, status = os.waitpid(pid, 0) + return os.waitstatus_to_exitcode(status) + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/backend/tests/test_real_cpu_upstream.py b/backend/tests/test_real_cpu_upstream.py index 2d81de06..0c0e6513 100644 --- a/backend/tests/test_real_cpu_upstream.py +++ b/backend/tests/test_real_cpu_upstream.py @@ -16,10 +16,14 @@ `captures/{capture_id}/runs/{run_id}/`, and each later stage's work dir is hydrated from `graph.STAGE_INPUTS` before its entry runs. -Resource budget: every measurement is taken with the process pinned to two -cores (`taskset -c 0,1`) and `OMP_NUM_THREADS=2 OPENBLAS_NUM_THREADS=2 -MKL_NUM_THREADS=2 HSS_POSES_THREADS=2`. A stage over 600 s or a process peak -RSS over 8 GiB fails the budget the target host (2 CPUs, 8 GB) imposes. +Resource budget: each chain runs in its own process, started through +`realcpu_launcher.py`, pinned to two CPUs, with `OMP_NUM_THREADS`, +`OPENBLAS_NUM_THREADS`, `MKL_NUM_THREADS` and `HSS_POSES_THREADS` set to 2 in +that process only. A stage over 600 s or a peak RSS over 8 GiB fails the budget +the target host (2 CPUs, 8 GB) imposes. The peak RSS is the chain process's +own, so earlier tests in the same pytest run cannot raise it. Where the OS +cannot pin CPUs (macOS), the correctness and RSS checks still run and the test +skips the wall-time budget instead of claiming it. No weights, no network, no GPU: the VLM readers stay off (`VISION_EXTERNAL_CALLS`), so `reads` runs only the local barcode decoder, and @@ -40,21 +44,17 @@ import pytest -os.environ.setdefault("OMP_NUM_THREADS", "2") -os.environ.setdefault("OPENBLAS_NUM_THREADS", "2") -os.environ.setdefault("MKL_NUM_THREADS", "2") -os.environ.setdefault("HSS_POSES_THREADS", "2") - -from server.contracts.jobs import StageContext, StageResult # noqa: E402 -from server.orchestrator.graph import STAGE_INPUTS, STAGES # noqa: E402 -from server.orchestrator.runner import _publish # noqa: E402 -from server.packet import run as validate_run # noqa: E402 -from server.poses import run as poses_run # noqa: E402 -from server.reads import run as reads_run # noqa: E402 -from server.reads.cache import MemoryStorage # noqa: E402 -from server.scale.run import execute as scale_run # noqa: E402 +from server.contracts.jobs import StageContext, StageResult +from server.orchestrator.graph import STAGE_INPUTS, STAGES +from server.orchestrator.runner import _publish +from server.packet import run as validate_run +from server.poses import run as poses_run +from server.reads import run as reads_run +from server.reads.cache import MemoryStorage +from server.scale.run import execute as scale_run ROOT = Path(__file__).resolve().parents[1] +LAUNCHER = Path(__file__).with_name("realcpu_launcher.py") LIDAR_PACKET = ROOT / "ios-kit" / "fixtures" / "mini-arkit-lidar" MINI_ARKIT_PACKET = ROOT / "data" / "fixtures" / "packets" / "mini-arkit" @@ -77,11 +77,18 @@ class _Settings: def _peak_rss_mb() -> float: - """Process peak RSS so far. Monotonic, so a stage's entry is the high-water mark through it.""" + """This process's peak RSS so far, in MiB. Monotonic, so a stage's entry is the + high-water mark through it. ru_maxrss counts KiB on Linux and bytes on macOS.""" + peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + return peak / (1024 * 1024) if sys.platform == "darwin" else peak / 1024 + + +def _proc_vm_hwm_mb() -> float: + """Linux's own peak-RSS figure for this process (VmHWM, in kB), to cross-check `_peak_rss_mb`.""" for line in Path("/proc/self/status").read_text().splitlines(): - if line.startswith("VmHWM"): + if line.startswith("VmHWM:"): return int(line.split()[1]) / 1024 - return -1.0 + raise AssertionError("/proc/self/status has no VmHWM line") def _ctx( @@ -152,7 +159,7 @@ def _hydrate(work_dir: Path, keys: tuple[str, ...], storage: MemoryStorage, pref def _run_chain( - tmp_path: Path, + work_root: Path, packet_dir: Path, capture_id: str, run_id: str, @@ -165,7 +172,7 @@ def _run_chain( _seed_stills(storage, capture_id, packet_dir) report: dict[str, dict[str, Any]] = {} for stage, fn in STAGE_ENTRY.items(): - work_dir = tmp_path / stage + work_dir = work_root / stage work_dir.mkdir(parents=True, exist_ok=True) _hydrate(work_dir, STAGE_INPUTS[stage], storage, prefix) started = time.perf_counter() @@ -174,20 +181,100 @@ def _run_chain( stored = _publish(_ctx(stage, work_dir, packet_dir, storage, capture_id, run_id, branch, tier), result) report[stage] = { "wall_s": round(wall_s, 2), - "vm_hwm_mb": round(_peak_rss_mb(), 1), + "peak_rss_mb": round(_peak_rss_mb(), 1), "artifacts": sorted(stored), "metrics": {k: round(float(v), 4) for k, v in result.metrics.items()}, "warnings": list(result.warnings), } - if os.environ.get("REALCPU_REPORT"): - print(f"REALCPU_REPORT {json.dumps({'packet': packet_dir.name, 'stages': report}, sort_keys=True)}") return report -def _assert_budget(report: dict[str, dict[str, Any]]) -> None: - for stage, entry in report.items(): +def chain_in_child( + out_json: str, packet_dir: str, work_root: str, capture_id: str, run_id: str, branch: str, tier: str +) -> None: + """Runs inside the launcher's process. Writes the stage report plus the CPUs + and thread settings the chain actually ran with.""" + stages = _run_chain(Path(work_root), Path(packet_dir), capture_id, run_id, branch, tier) + cpus = sorted(os.sched_getaffinity(0)) if hasattr(os, "sched_getaffinity") else None + threads = {key: os.environ.get(key) for key in THREAD_ENV} + Path(out_json).write_text(json.dumps({"stages": stages, "cpus": cpus, "threads": threads})) + + +def rss_probe_in_child(out_json: str) -> None: + """Runs inside the launcher's process. Touches 64 MiB, then reads both peak figures.""" + block = bytearray(64 * 1024 * 1024) + block[::4096] = b"\x01" * (len(block) // 4096) + Path(out_json).write_text(json.dumps({"peak_mb": _peak_rss_mb(), "vm_hwm_mb": _proc_vm_hwm_mb()})) + del block + + +def _child_env(request: pytest.FixtureRequest, threads: str | None) -> dict[str, str]: + """The environment for a child Python: pytest's `pythonpath` entries, and the + thread variables set to `threads`, or removed when it is None.""" + env = {k: v for k, v in os.environ.items() if k not in THREAD_ENV} + if threads is not None: + env.update(dict.fromkeys(THREAD_ENV, threads)) + paths = [str(p) for p in request.config.getini("pythonpath")] + if os.environ.get("PYTHONPATH"): + paths.append(os.environ["PYTHONPATH"]) + env["PYTHONPATH"] = os.pathsep.join(paths) + return env + + +def _in_fresh_process(request: pytest.FixtureRequest, out_json: Path, function: str, *args: str) -> dict[str, Any]: + """Run `function` from this module through `realcpu_launcher.py` (two CPUs, + two threads per numeric library, a peak-RSS counter of its own) and return + the JSON it wrote to `out_json`.""" + proc = subprocess.run( + [sys.executable, str(LAUNCHER), __file__, function, str(out_json), *args], + cwd=ROOT, + env=_child_env(request, threads="2"), + capture_output=True, + text=True, + check=False, + ) + assert proc.returncode == 0, ( + f"{function} failed in its own process (exit {proc.returncode}):\n{proc.stderr[-6000:]}" + ) + return json.loads(out_json.read_text()) + + +def _chain( + request: pytest.FixtureRequest, + work_root: Path, + packet_dir: Path, + capture_id: str, + run_id: str, + branch: str, + tier: str, +) -> dict[str, Any]: + run = _in_fresh_process( + request, + work_root / "run.json", + "chain_in_child", + str(packet_dir), + str(work_root), + capture_id, + run_id, + branch, + tier, + ) + if os.environ.get("REALCPU_REPORT"): + print(f"REALCPU_REPORT {json.dumps({'packet': packet_dir.name, **run}, sort_keys=True)}") + return run + + +def _assert_budget(run: dict[str, Any]) -> None: + """Called last in each chain test, because it skips where the wall budget cannot be measured.""" + assert run["threads"] == dict.fromkeys(THREAD_ENV, "2"), run["threads"] + for stage, entry in run["stages"].items(): + peak = entry["peak_rss_mb"] + assert 0 < peak < RSS_BUDGET_MB, f"{stage}: peak RSS {peak} MB is outside the 8 GB budget" + if run["cpus"] is None: + pytest.skip(f"checks passed, but the 2-CPU wall budget is unmeasured: {sys.platform} cannot pin CPUs") + assert 1 <= len(run["cpus"]) <= 2, f"the chain ran on CPUs {run['cpus']}, not two" + for stage, entry in run["stages"].items(): assert entry["wall_s"] < WALL_BUDGET_S, f"{stage} took {entry['wall_s']}s (budget {WALL_BUDGET_S}s)" - assert 0 < entry["vm_hwm_mb"] < RSS_BUDGET_MB, f"peak RSS {entry['vm_hwm_mb']}MB exceeds the 8 GB budget" def _covers(declared: tuple[str, ...], published: list[str]) -> bool: @@ -202,7 +289,11 @@ def _covers(declared: tuple[str, ...], published: list[str]) -> bool: return True -def test_real_validate_poses_reads_scale_on_mini_arkit_lidar(tmp_path) -> None: +def _estimates(scale: dict[str, Any]) -> dict[str, dict[str, Any]]: + return {row["kind"]: row for row in scale["estimates"]} + + +def test_real_validate_poses_reads_scale_on_mini_arkit_lidar(tmp_path, request) -> None: """The committed LiDAR packet walks all four real stages on CPU. The fixture has too few frames for free SfM (`source.notes`), so poses must @@ -211,8 +302,8 @@ def test_real_validate_poses_reads_scale_on_mini_arkit_lidar(tmp_path) -> None: poses (`arkit_fixed`). All six frames register with `pose_source` recorded. """ capture_id, run_id = "cap_realcpu_lidar", "run_realcpu_lidar" - report = _run_chain(tmp_path, LIDAR_PACKET, capture_id, run_id, branch="lidar", tier="arkit_lidar") - _assert_budget(report) + run = _chain(request, tmp_path, LIDAR_PACKET, capture_id, run_id, branch="lidar", tier="arkit_lidar") + report = run["stages"] # Published artifact keys match graph.py's declared artifacts for each stage. assert set(report["validate"]["artifacts"]) == set(STAGES["validate"].artifacts) @@ -233,15 +324,26 @@ def test_real_validate_poses_reads_scale_on_mini_arkit_lidar(tmp_path) -> None: assert reg["status"] == "fallback_arkit_fixed" assert "sfm_no_model" in reg["warnings"] - # Scale carries a source and a 95% relative bound. The LiDAR depth and the - # meter tap both feed it; `source` names what shipped. + # Six frames give the LiDAR-vs-sparse alignment no usable pairs, so LiDAR is + # recorded as an unused estimate and the ARKit poses alone set the scale, + # unverified, with a request for the meter close-up the packet lacks. scale = json.loads((tmp_path / "scale" / "scale" / "scale.json").read_text()) - assert scale["source"] - assert scale["source"] != "none" - assert isinstance(scale["rel_bound_95"], float) and scale["rel_bound_95"] >= 0.0 + assert scale["source"] == "arkit" + assert scale["status"] == "unverified" + assert scale["reason"] == "scale_unresolved" + assert scale["reason_detail"] == "meter_closeup_missing" + assert scale["confidence"] == "near_only" + assert scale["rel_bound_95"] > 0.0 + estimates = _estimates(scale) + assert [kind for kind, row in estimates.items() if row["used"]] == ["arkit_poses"] + assert estimates["lidar"]["used"] is False + assert estimates["lidar"]["reason"] == "lidar_too_few_pairs" + assert estimates["lidar"]["pairs"] == 0 + _assert_budget(run) -def test_mini_arkit_refuses_metric_scale_loudly(tmp_path) -> None: + +def test_mini_arkit_refuses_metric_scale_loudly(tmp_path, request) -> None: """The 64x48 mini-arkit packet (tier `arkit`, depth null) gets the same real chain and refuses to certify a scale instead of passing. @@ -250,8 +352,8 @@ def test_mini_arkit_refuses_metric_scale_loudly(tmp_path) -> None: `scale_unresolved` / `meter_model_unread`, confidence weak, ARKit-only source — rather than claiming a metric scale.""" capture_id, run_id = "cap_realcpu_arkit", "run_realcpu_arkit" - report = _run_chain(tmp_path, MINI_ARKIT_PACKET, capture_id, run_id, branch="image", tier="arkit") - _assert_budget(report) + run = _chain(request, tmp_path, MINI_ARKIT_PACKET, capture_id, run_id, branch="image", tier="arkit") + report = run["stages"] assert set(report["validate"]["artifacts"]) == set(STAGES["validate"].artifacts) assert _covers(STAGES["poses"].artifacts, report["poses"]["artifacts"]) @@ -261,10 +363,17 @@ def test_mini_arkit_refuses_metric_scale_loudly(tmp_path) -> None: assert reads["status"] == "disabled" scale = json.loads((tmp_path / "scale" / "scale" / "scale.json").read_text()) + assert scale["source"] == "arkit" assert scale["status"] == "unverified" assert scale["reason"] == "scale_unresolved" assert scale["reason_detail"] == "meter_model_unread" assert scale["confidence"] == "weak" + assert scale["rel_bound_95"] > 0.0 + estimates = _estimates(scale) + assert "lidar" not in estimates + assert [kind for kind, row in estimates.items() if row["used"]] == ["arkit_poses"] + + _assert_budget(run) def _fake_getrusage(ru_maxrss: int): @@ -288,17 +397,20 @@ def test_peak_rss_mb_reads_kib_on_linux(monkeypatch) -> None: assert _peak_rss_mb() == 2048.0 -def _child_env(request: pytest.FixtureRequest, threads: str | None) -> dict[str, str]: - """The environment for a child Python: pytest's `pythonpath` entries, and the - thread variables set to `threads`, or removed when it is None.""" - env = {k: v for k, v in os.environ.items() if k not in THREAD_ENV} - if threads is not None: - env.update(dict.fromkeys(THREAD_ENV, threads)) - paths = [str(p) for p in request.config.getini("pythonpath")] - if os.environ.get("PYTHONPATH"): - paths.append(os.environ["PYTHONPATH"]) - env["PYTHONPATH"] = os.pathsep.join(paths) - return env +@pytest.mark.skipif(not Path("/proc/self/status").is_file(), reason="VmHWM comes from Linux's /proc") +def test_peak_rss_mb_matches_vmhwm_in_the_chain_process(tmp_path, request) -> None: + """On Linux, ru_maxrss read through `_peak_rss_mb` equals the kernel's VmHWM + in the process the chains run in. + + This test first raises its own peak by 256 MiB. Without the launcher's fork, + the probe's ru_maxrss would carry that peak (see `realcpu_launcher.py`) and + sit well above the probe's VmHWM.""" + ballast = bytearray(256 * 1024 * 1024) + ballast[::4096] = b"\x01" * (len(ballast) // 4096) + del ballast + probe = _in_fresh_process(request, tmp_path / "probe.json", "rss_probe_in_child") + assert probe["peak_mb"] >= 64 + assert probe["peak_mb"] == pytest.approx(probe["vm_hwm_mb"], abs=1.0) def test_importing_this_module_leaves_thread_env_alone(request) -> None: From c716c377017e0d15f67ce4267b912124e3903265 Mon Sep 17 00:00:00 2001 From: SamGu-NRX Date: Fri, 23 Oct 2026 07:52:21 -0500 Subject: [PATCH 3/3] test(real-cpu): bound launcher and subprocess waits Co-Authored-By: Claude Code --- backend/tests/realcpu_launcher.py | 39 ++++++++++++++++++-- backend/tests/test_real_cpu_launcher.py | 47 +++++++++++++++++++++++++ backend/tests/test_real_cpu_upstream.py | 5 +++ 3 files changed, 89 insertions(+), 2 deletions(-) create mode 100644 backend/tests/test_real_cpu_launcher.py diff --git a/backend/tests/realcpu_launcher.py b/backend/tests/realcpu_launcher.py index 585b2f53..486f615e 100644 --- a/backend/tests/realcpu_launcher.py +++ b/backend/tests/realcpu_launcher.py @@ -28,10 +28,15 @@ import importlib.util import os +import signal import sys +import time import traceback CPUS = 2 +CHILD_TIMEOUT_S = 900.0 +_TERMINATE_GRACE_S = 1.0 +_POLL_INTERVAL_S = 0.02 def _pin() -> None: @@ -57,14 +62,44 @@ def _run(module_path: str, function: str, args: list[str]) -> int: return 0 +def _wait_for_child(pid: int, stage: str, timeout_s: float) -> int: + deadline = time.monotonic() + timeout_s + while True: + waited_pid, status = os.waitpid(pid, os.WNOHANG) + if waited_pid == pid: + return os.waitstatus_to_exitcode(status) + remaining = deadline - time.monotonic() + if remaining <= 0: + break + time.sleep(min(_POLL_INTERVAL_S, remaining)) + + try: + os.kill(pid, signal.SIGTERM) + except ProcessLookupError: + os.waitpid(pid, 0) + raise TimeoutError(f"{stage} exceeded launcher timeout of {timeout_s:g}s") from None + terminate_deadline = time.monotonic() + _TERMINATE_GRACE_S + while True: + waited_pid, status = os.waitpid(pid, os.WNOHANG) + if waited_pid == pid: + raise TimeoutError(f"{stage} exceeded launcher timeout of {timeout_s:g}s") + remaining = terminate_deadline - time.monotonic() + if remaining <= 0: + break + time.sleep(min(_POLL_INTERVAL_S, remaining)) + + os.kill(pid, signal.SIGKILL) + _, status = os.waitpid(pid, 0) + raise TimeoutError(f"{stage} exceeded launcher timeout of {timeout_s:g}s") + + def main(argv: list[str]) -> int: module_path, function, *args = argv _pin() pid = os.fork() if pid == 0: os._exit(_run(module_path, function, args)) - _, status = os.waitpid(pid, 0) - return os.waitstatus_to_exitcode(status) + return _wait_for_child(pid, function, CHILD_TIMEOUT_S) if __name__ == "__main__": diff --git a/backend/tests/test_real_cpu_launcher.py b/backend/tests/test_real_cpu_launcher.py new file mode 100644 index 00000000..b9421ad0 --- /dev/null +++ b/backend/tests/test_real_cpu_launcher.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import importlib.util +import os +import time +from pathlib import Path + +import pytest + +LAUNCHER = Path(__file__).with_name("realcpu_launcher.py") + + +def _load_launcher(): + spec = importlib.util.spec_from_file_location("realcpu_launcher_test", LAUNCHER) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def sleep_stage(pid_path: str) -> None: + Path(pid_path).write_text(str(os.getpid())) + time.sleep(30) + + +def test_launcher_terminates_and_reaps_a_timed_out_stage(tmp_path, monkeypatch) -> None: + launcher = _load_launcher() + monkeypatch.setattr(launcher, "CHILD_TIMEOUT_S", 0.05) + pid_path = tmp_path / "child.pid" + + started = time.monotonic() + with pytest.raises(TimeoutError, match=r"sleep_stage exceeded launcher timeout of 0\.05s"): + launcher.main([str(Path(__file__)), "sleep_stage", str(pid_path)]) + + assert time.monotonic() - started < 2 + pid = int(pid_path.read_text()) + with pytest.raises(ProcessLookupError): + os.kill(pid, 0) + + +def test_wait_for_child_preserves_a_normal_exit_status() -> None: + launcher = _load_launcher() + pid = os.fork() + if pid == 0: + os._exit(7) + + assert launcher._wait_for_child(pid, "exit_stage", 1) == 7 diff --git a/backend/tests/test_real_cpu_upstream.py b/backend/tests/test_real_cpu_upstream.py index 0c0e6513..86e30ac7 100644 --- a/backend/tests/test_real_cpu_upstream.py +++ b/backend/tests/test_real_cpu_upstream.py @@ -60,6 +60,9 @@ WALL_BUDGET_S = 600.0 RSS_BUDGET_MB = 8 * 1024 +# Chosen well above the expected chain runtime; this is a policy limit, not a measurement. +SUBPROCESS_TIMEOUT_S = 900 +IMPORT_SUBPROCESS_TIMEOUT_S = 30 THREAD_ENV = ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS", "HSS_POSES_THREADS") STAGE_ENTRY = { @@ -232,6 +235,7 @@ def _in_fresh_process(request: pytest.FixtureRequest, out_json: Path, function: capture_output=True, text=True, check=False, + timeout=SUBPROCESS_TIMEOUT_S, ) assert proc.returncode == 0, ( f"{function} failed in its own process (exit {proc.returncode}):\n{proc.stderr[-6000:]}" @@ -429,6 +433,7 @@ def test_importing_this_module_leaves_thread_env_alone(request) -> None: capture_output=True, text=True, check=False, + timeout=IMPORT_SUBPROCESS_TIMEOUT_S, ) assert proc.returncode == 0, proc.stderr[-4000:] assert proc.stdout.strip().splitlines()[-1] == "[]"