diff --git a/backend/tests/realcpu_launcher.py b/backend/tests/realcpu_launcher.py new file mode 100644 index 00000000..486f615e --- /dev/null +++ b/backend/tests/realcpu_launcher.py @@ -0,0 +1,106 @@ +"""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 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: + 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 _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)) + return _wait_for_child(pid, function, CHILD_TIMEOUT_S) + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) 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 68268ac1..86e30ac7 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 @@ -30,30 +34,36 @@ import json import os +import resource +import subprocess +import sys import time from pathlib import Path +from types import SimpleNamespace from typing import Any -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") +import pytest -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" 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 = { "validate": validate_run, @@ -69,12 +79,19 @@ class _Settings: VISION_EXTERNAL_CALLS = False -def _vm_hwm_mb() -> float: - """Process peak RSS so far. Monotonic, so a stage's entry is the high-water mark through it.""" +def _peak_rss_mb() -> float: + """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( @@ -145,7 +162,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, @@ -158,7 +175,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() @@ -167,20 +184,101 @@ 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), + "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, + timeout=SUBPROCESS_TIMEOUT_S, + ) + 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: @@ -195,7 +293,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 @@ -204,8 +306,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) @@ -226,15 +328,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. @@ -243,8 +356,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"]) @@ -254,7 +367,73 @@ 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): + 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 + + +@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: + """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, + timeout=IMPORT_SUBPROCESS_TIMEOUT_S, + ) + assert proc.returncode == 0, proc.stderr[-4000:] + assert proc.stdout.strip().splitlines()[-1] == "[]"