From 5ebf42f7bfc2d81b2beb4328dc125bd2a9d3f681 Mon Sep 17 00:00:00 2001 From: Evrim Cicek Date: Sun, 1 Nov 2026 23:12:54 +0300 Subject: [PATCH] parity.py runs harnesses on Windows: file locks, process trees, no peak or shim there parity.py started on Windows since #3428 but stopped before any harness ran: machine_slot locked its slots with fcntl, which Windows lacks. The rest of the run path was POSIX too: a timed-out harness was killed with os.killpg, trace_main read the peak memory with resource, and the oracle shim went on PATH as a `java` symlink to a Python script. - tools/file_lock.py: an exclusive lock, flock or msvcrt's byte lock; used by machine_slot, integration_env.suite_lock and test_datafix_parity. - kill_tree: the process group where there is one, taskkill /T on Windows. - Without resource, the trace has no peak_bytes and the scheduler assumes DEFAULT_MEMORY, as for a harness that never ran. - No oracle shim on Windows: a program started without a shell is found there only as an .exe, and the shim's execv returns before the program. Harnesses run their oracles fresh. Fixes #6391 Co-Authored-By: Claude Opus 5.5 --- tests/test_file_lock.py | 25 +++++++++++++++++ tools/file_lock.py | 28 +++++++++++++++++++ tools/integration_env.py | 9 ++++--- tools/parity.py | 52 +++++++++++++++++++++++------------- tools/test_datafix_parity.py | 6 ++--- 5 files changed, 95 insertions(+), 25 deletions(-) create mode 100644 tests/test_file_lock.py create mode 100644 tools/file_lock.py diff --git a/tests/test_file_lock.py b/tests/test_file_lock.py new file mode 100644 index 0000000000..c1d1848c02 --- /dev/null +++ b/tests/test_file_lock.py @@ -0,0 +1,25 @@ +from pathlib import Path +import tempfile +import unittest + +from tools import file_lock + + +class FileLockTests(unittest.TestCase): + def test_a_held_lock_refuses_another_until_closed(self): + with tempfile.TemporaryDirectory() as temp: + path = Path(temp) / "slot.lock" + first = open(path, "w") + second = open(path, "a") + try: + self.assertTrue(file_lock.lock(first, wait=False)) + self.assertFalse(file_lock.lock(second, wait=False)) + first.close() + self.assertTrue(file_lock.lock(second, wait=False)) + finally: + first.close() + second.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/file_lock.py b/tools/file_lock.py new file mode 100644 index 0000000000..3b1b69278d --- /dev/null +++ b/tools/file_lock.py @@ -0,0 +1,28 @@ +"""An exclusive lock on an open file, held until it is closed: flock where it exists, msvcrt's byte lock on Windows.""" + +import os +import time + + +def lock(handle, wait=True): + """Lock `handle`; with wait=False, False at once when another process holds it.""" + if os.name != "nt": + import fcntl + try: + fcntl.flock(handle, fcntl.LOCK_EX | (0 if wait else fcntl.LOCK_NB)) + return True + except OSError: + if wait: + raise + return False + import msvcrt + handle.flush() + os.lseek(handle.fileno(), 0, os.SEEK_SET) # the lock covers the first byte, wherever the file was opened + while True: + try: + msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) + return True + except OSError: + if not wait: + return False + time.sleep(0.1) diff --git a/tools/integration_env.py b/tools/integration_env.py index 2f2b243969..a6659bfcbc 100644 --- a/tools/integration_env.py +++ b/tools/integration_env.py @@ -36,9 +36,12 @@ def run_game(command, **kwargs): @contextmanager def suite_lock(): """Real-client CI suites share a GPU, even when parity.py runs harnesses in parallel.""" - import fcntl - with (Path(tempfile.gettempdir()) / 'horizon-integration-gpu.lock').open('a') as lock: - fcntl.flock(lock, fcntl.LOCK_EX) + try: + from . import file_lock + except ImportError: + import file_lock + with (Path(tempfile.gettempdir()) / 'horizon-integration-gpu.lock').open('a') as handle: + file_lock.lock(handle) yield diff --git a/tools/parity.py b/tools/parity.py index 5bf9a6903f..96e8a6d33c 100644 --- a/tools/parity.py +++ b/tools/parity.py @@ -56,8 +56,10 @@ import time try: from . import parity_reach + from . import file_lock except ImportError: import parity_reach + import file_lock ROOT = Path(__file__).resolve().parents[1] LOCAL = ROOT / ".local" @@ -594,8 +596,9 @@ def java_sources(arguments): def run_environment(reuse_oracles=False): environment = dict(os.environ, CARGO_TARGET_DIR=str(TARGET), PYTHONDONTWRITEBYTECODE="1", HORIZON_NO_AUDIO="1", **PROFILE) environment.pop("CARGO_BUILD_JOBS", None) - if reuse_oracles: - # tools/parity_shim.py stands in for java and javac and replays unchanged oracle runs. + # tools/parity_shim.py stands in for java and javac and replays unchanged oracle runs. Not on Windows: a + # program started without a shell is found there only as an .exe, and the shim's execv returns before the program. + if reuse_oracles and os.name != "nt": SHIM.mkdir(parents=True, exist_ok=True) for name in ("java", "javac"): link = SHIM / name @@ -736,16 +739,13 @@ def publish(name, record): def machine_slot(): """One of MACHINE_JOBS lock files shared by every runner on this machine, or None if all are taken.""" - import fcntl slots = SHARED / "slots" slots.mkdir(parents=True, exist_ok=True) for index in range(MACHINE_JOBS): handle = open(slots / f"{index}.lock", "w") - try: - fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) + if file_lock.lock(handle, wait=False): return handle - except OSError: - handle.close() + handle.close() return None @@ -792,7 +792,8 @@ def outcome(name, script, passed, elapsed, trace_path, hasher=None): trace = None # killed before writing it, or the disk was full: the pass can't be keyed if trace: record["dependencies"], record["artifacts"] = dependencies(trace, harness_settings(script).get("PARITY_RUST_SCOPE")) - record["peak_bytes"] = trace["peak_bytes"] + if "peak_bytes" in trace: + record["peak_bytes"] = trace["peak_bytes"] if passed: clear_large_outputs(name, trace) if hasher is not None: @@ -873,7 +874,7 @@ class LocalExecutor: if process.poll() is None and elapsed < self.timeout: continue if process.poll() is None: - os.killpg(process.pid, 9) + kill_tree(process) process.wait() print(f"--- {name}: timed out after {self.timeout}s", file=item["log"]) item["log"].close() @@ -884,10 +885,18 @@ class LocalExecutor: def stop(self): for item in self.running.values(): - try: - os.killpg(item["process"].pid, 9) - except ProcessLookupError: - pass + kill_tree(item["process"]) + + +def kill_tree(process): + """Kill a harness and what it started: its process group (start_new_session), or its tree on Windows.""" + if os.name == "nt": + subprocess.run(["taskkill", "/T", "/F", "/PID", str(process.pid)], capture_output=True) + return + try: + os.killpg(process.pid, 9) + except ProcessLookupError: + pass def host_config(): @@ -1258,8 +1267,11 @@ def stale(name, state, hasher, environment): def trace_main(output, script, arguments): """Run one harness and record what it read and executed.""" import atexit - import resource import runpy + try: + import resource + except ImportError: # Windows: no peak to report, the scheduler assumes DEFAULT_MEMORY + resource = None read, commands = set(), [] def audit(event, details): @@ -1274,14 +1286,16 @@ def trace_main(output, script, arguments): def write(): modules = [getattr(module, "__file__", None) for module in list(sys.modules.values())] - peak = max(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss) - peak //= 1024 if sys.platform == "darwin" else 1 # bytes on macOS, KiB elsewhere - Path(output).write_text(json.dumps({ + trace = { "modules": sorted({m for m in modules if m and inside(m, ROOT)}), "read": sorted(path for path in read if os.path.isabs(path) or not path.startswith("<")), "commands": commands, - "peak_bytes": peak * 1024, - })) + } + if resource: + peak = max(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, resource.getrusage(resource.RUSAGE_CHILDREN).ru_maxrss) + peak //= 1024 if sys.platform == "darwin" else 1 # bytes on macOS, KiB elsewhere + trace["peak_bytes"] = peak * 1024 + Path(output).write_text(json.dumps(trace)) atexit.register(write) sys.addaudithook(audit) diff --git a/tools/test_datafix_parity.py b/tools/test_datafix_parity.py index 0e35143be1..a4ee306edc 100644 --- a/tools/test_datafix_parity.py +++ b/tools/test_datafix_parity.py @@ -212,10 +212,10 @@ def main(): 'file fixes), on the sample worlds saved before it, for developing the fixes of a range of versions') args = p.parse_args() import datafix_worlds - import fcntl + import file_lock WORLDS.parent.mkdir(parents=True, exist_ok=True) - with open(WORLDS.parent / 'generate.lock', 'w') as lock: - fcntl.flock(lock, fcntl.LOCK_EX) # another worktree may be generating them + with open(WORLDS.parent / 'generate.lock', 'w') as handle: + file_lock.lock(handle) # another worktree may be generating them missing = [v for v in [*datafix_worlds.VERSIONS, *datafix_worlds.VARIANTS] if not (datafix_worlds.worlds_dir(v) / v / 'level.dat').exists()] if missing: print('generating sample worlds:', ' '.join(missing), flush=True)