From fabd431adb4f49f0661adb7afa09edeeb9afc4f8 Mon Sep 17 00:00:00 2001 From: Abdessamad Derraz <3028866+Abdess@users.noreply.github.com> Date: Tue, 27 Oct 2026 21:02:03 +0100 Subject: [PATCH] feat: take the host probe lock per run core_probe takes probe.lock itself around one execution, load check included, and releases it when the run ends. A campaign pass never holds it between runs, so two passes interleave. RETROBIOS_PROBE_LOCK names another file, RETROBIOS_PROBE_LOCK_HELD says the caller holds it. Tests use their own lock file. --- scripts/campaign.py | 4 ++-- scripts/core_probe.py | 41 ++++++++++++++++++++++++++++++++++- tests/__init__.py | 5 +++++ tests/test_campaign.py | 46 ++++++++++++++++++++++++++++++++++++++++ tests/test_core_probe.py | 26 +++++++++++++++++++++++ wiki/tools.md | 10 +++++++++ 6 files changed, 129 insertions(+), 3 deletions(-) diff --git a/scripts/campaign.py b/scripts/campaign.py index ae0cd8eb4..c8cbd69da 100644 --- a/scripts/campaign.py +++ b/scripts/campaign.py @@ -303,9 +303,9 @@ def child_env(romfetch: bool = False) -> dict[str, str]: """What a program the campaign starts inherits: autorun's allowlist, never a token or a password. romfetch also gets its own settings (ROMFETCH_*), its account staying in its config file.""" - names: list[str] = [] + names: list[str] = [core_probe.PROBE_LOCK_ENV, core_probe.PROBE_LOCK_HELD_ENV] if romfetch: - names = [k for k in os.environ if k.startswith("ROMFETCH_")] + list(ROMFETCH_PASSED) + names += [k for k in os.environ if k.startswith("ROMFETCH_")] + list(ROMFETCH_PASSED) return autorun.safe_env(names) diff --git a/scripts/core_probe.py b/scripts/core_probe.py index bb35bf43e..1f979430e 100644 --- a/scripts/core_probe.py +++ b/scripts/core_probe.py @@ -21,6 +21,7 @@ import argparse import contextlib import ctypes import datetime as dt +import fcntl import hashlib import json import os @@ -35,7 +36,7 @@ import sys import tempfile import time import zlib -from collections.abc import Iterable, Sequence +from collections.abc import Iterable, Iterator, Sequence from dataclasses import dataclass, field from pathlib import Path @@ -73,6 +74,39 @@ PRINTF_SPEC = re.compile(r"%(?P[-+ #0]*)(?P\*|\d+)?(?:\.(?P\ r"(?Phh|h|ll|l|j|z|t|L)?(?P[diouxXeEfFgGaAcspn%])") +# One measurement at a time on the host: each core_probe execution takes this +# lock itself, its load check included, and releases it when it ends. A pass +# of many runs never holds it between them, so two passes interleave. +PROBE_LOCK_ENV = "RETROBIOS_PROBE_LOCK" +PROBE_LOCK_HELD_ENV = "RETROBIOS_PROBE_LOCK_HELD" +DEFAULT_PROBE_LOCK = Path.home() / ".local" / "state" / "retrobios" / "probe.lock" + + +def probe_lock_path() -> Path: + return Path(os.environ.get(PROBE_LOCK_ENV) or DEFAULT_PROBE_LOCK) + + +@contextlib.contextmanager +def host_probe_lock() -> Iterator[None]: + """The host's probe lock for one execution. A caller that already holds + it for this run says so with RETROBIOS_PROBE_LOCK_HELD.""" + if os.environ.get(PROBE_LOCK_HELD_ENV): + yield + return + path = probe_lock_path() + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("a") as handle: + try: + fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + print(f"waiting for {path}: another measurement is running", file=sys.stderr) + fcntl.flock(handle, fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(handle, fcntl.LOCK_UN) + + class ProbeError(RuntimeError): """A run that cannot produce a trustworthy result.""" @@ -1355,6 +1389,11 @@ def probe(args: argparse.Namespace) -> int: for problem in problems: print(f"ERROR: {problem}", file=sys.stderr) return 2 + with host_probe_lock(): + return _probe(args) + + +def _probe(args: argparse.Namespace) -> int: max_load = args.max_load if args.max_load is not None else (os.cpu_count() or 1) / 4 load_before = load_average() if args.record and load_before and load_before[0] > max_load: diff --git a/tests/__init__.py b/tests/__init__.py index 0190e539f..625337b3c 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -16,6 +16,11 @@ import os from pathlib import Path os.environ.setdefault("RETROBIOS_OFFLINE", "1") +# A test never waits on, nor blocks, the measurements running on the host. +os.environ.setdefault( + "RETROBIOS_PROBE_LOCK", + str(Path(__file__).resolve().parent.parent / "tmp" / "tests" / "probe.lock"), +) (Path(__file__).resolve().parent.parent / "tmp" / "tests").mkdir( parents=True, exist_ok=True diff --git a/tests/test_campaign.py b/tests/test_campaign.py index ec5753227..071200d8b 100644 --- a/tests/test_campaign.py +++ b/tests/test_campaign.py @@ -1126,6 +1126,52 @@ class FakeCorePasses(Workspace): self.assertEqual({rep["verdict_evidence"]["rule"] for r in records for v in r["variants"] for rep in v["repetitions"]}, {"timeout"}) + def test_two_passes_interleave_their_runs(self) -> None: + """Each core_probe takes the host lock for its own run: a pass never + holds it between runs, so a second pass runs between them.""" + import threading + import time + + spans: dict[str, list[tuple[float, float]]] = {"a": [], "b": []} + other = Workspace(methodName="setUp") + other.setUp() + self.addCleanup(other.tearDown) + passes = [] + for name, workspace in (("a", self), ("b", other)): + c = FakeCorePasses.passing(workspace, "hle") if workspace is self else self.passing_in(workspace) + real = c.host.run + + def timed(argv: list[str], *args: object, _name: str = name, _real=real, **kwargs: object): + start = time.monotonic() + try: + return _real(argv, *args, **kwargs) + finally: + spans[_name].append((start, time.monotonic())) + c.host.run = timed + passes.append(c) + threads = [threading.Thread(target=c.run, args=(set(), None)) for c in passes] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + ends = {name: sorted(end for _, end in runs) for name, runs in spans.items()} + self.assertGreater(len(ends["a"]), 2) + self.assertGreater(len(ends["b"]), 2) + # Neither pass finished all its runs before the other started finishing its own. + self.assertLess(ends["b"][0], ends["a"][-1]) + self.assertLess(ends["a"][0], ends["b"][-1]) + + def passing_in(self, workspace: Workspace) -> campaign.Campaign: + workspace.put_game() + host = campaign.Host(loadavg=lambda: (0.0, 0.0, 0.0), free_gb=lambda path: 500.0, + cpu_count=1_000_000) + c = campaign.Campaign(workspace.paths, campaign.Settings( + check_repeat=2, probe_timeout=30, delays=(1,), echo=False), host, + lock_path=workspace.tmp / "auto.lock") + so = self.cores["hle"] + c.unit_cores = lambda *a, **k: [campaign.Core("fake", so, campaign.file_sha256(so))] + return c + def test_missing_content_leaves_the_checks_pending(self) -> None: c = self.passing(romfetch=str(self.tmp / "no-romfetch")) shutil.rmtree(self.game) diff --git a/tests/test_core_probe.py b/tests/test_core_probe.py index e927cbe2d..ead050b66 100644 --- a/tests/test_core_probe.py +++ b/tests/test_core_probe.py @@ -504,6 +504,32 @@ class FakeCoreTests(unittest.TestCase): sealed = measurement_records.seal({**record, "emulator": "fake", "subject": "s"}) self.assertEqual([e.message for e in Draft202012Validator(schema).iter_errors(sealed)], []) + def test_a_run_waits_for_the_host_lock_and_frees_it(self) -> None: + import fcntl + import time + + lock = self.tmp / "host.lock" + argv = [sys.executable, str(REPO_ROOT / "scripts" / "core_probe.py"), "run", "--core", + str(self.core), "--frames", "1", "--no-trace", "--work-dir", str(self.tmp / "work")] + env = {**os.environ, "RETROBIOS_PROBE_LOCK": str(lock)} + with lock.open("a") as held: + fcntl.flock(held, fcntl.LOCK_EX) + waiting = subprocess.Popen(argv, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, + text=True) + time.sleep(1.5) + self.assertIsNone(waiting.poll(), "the run went ahead while another held the lock") + # A caller that holds the lock for this run says so. + held_run = subprocess.run(argv, env={**env, "RETROBIOS_PROBE_LOCK_HELD": "1"}, + capture_output=True, text=True, timeout=60, check=False) + self.assertEqual(held_run.returncode, 0, held_run.stderr) + fcntl.flock(held, fcntl.LOCK_UN) + _, err = waiting.communicate(timeout=60) + self.assertEqual(waiting.returncode, 0, err) + self.assertIn("another measurement is running", err) + # Released at the end of the run: the lock is free again. + with lock.open("a") as again: + fcntl.flock(again, fcntl.LOCK_EX | fcntl.LOCK_NB) + def test_second_run_is_appended_beside_the_first(self) -> None: record_dir = checkout(self.tmp / "appended") argv = ["run", "--core", str(self.core), "--frames", "1", "--no-trace", "--max-load", "1000", diff --git a/wiki/tools.md b/wiki/tools.md index 622c37414..942ae6ca6 100644 --- a/wiki/tools.md +++ b/wiki/tools.md @@ -855,6 +855,15 @@ python scripts/core_probe.py check --core core.so --content disc.cue \ --observable frame_hashes --claim "..." --method "..." --repeat 2 --record ``` +One measurement runs at a time on the host. core_probe takes the host lock +itself, per run: `~/.local/state/retrobios/probe.lock` (`RETROBIOS_PROBE_LOCK` +names another file), from its load check to its record, released when the run +ends. A run that finds it held waits and says so. Never wrap a whole pass in +`flock`: a pass of many runs then holds the host for hours, while called +directly its runs interleave with everyone else's. A caller that holds the +lock around one core_probe call sets `RETROBIOS_PROBE_LOCK_HELD=1`. The tests +use their own lock file. + Each repetition runs in a fresh process under `strace -f -e trace=openat,open` when strace is installed. The worker starts with an empty environment and clears its own before it loads the core, keeping only the `--env` values. @@ -1046,6 +1055,7 @@ checks. The campaign's own `evidence:` writes do not count as a profile change. | Guard | Rule | |-------|------| | Lock | The host-wide autorun lock. Under the `campaign` job, autorun holds it and says so with `RETROBIOS_AUTO_LOCK_HELD` | +| Probe lock | Not held by the pass: each core_probe it starts takes the host probe lock for its own run, so two passes, or a pass and another agent's runs, interleave. Do not wrap `campaign.py run` in `flock` | | Environment | `core_probe.py`, git and `make` get autorun's allowlist, romfetch the same plus its `ROMFETCH_*` settings. None sees a token or a password | | Sandbox | Fails closed: without a working bubblewrap the pass is `refused` (exit 3) and nothing third-party runs, every probe is started with `--require-sandbox`, and a build is refused. `--allow-unsandboxed` overrides. Under autorun the refusal raises the `job:campaign` alert | | Load | Before each run, waits while the 1-minute load is above a quarter of the CPUs, up to `--max-load-wait` seconds, then the pass stops as `paused` |