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 <[email protected]>

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()

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)

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
 
 

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)

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)