"""Explicit solo-card leases for isolated TP2 adapter evaluation. No production pair.""" from __future__ import annotations import fcntl import os from pathlib import Path from .safety import GPU, SOLO_B, assert_gpu_lease def require_evaluation_lease(emulated: bool) -> None: if emulated: assert_gpu_lease() return if os.environ.get("CUDA_VISIBLE_DEVICES") != GPU+","+SOLO_B or os.environ.get("D2_ADAPTER_EVAL") != "1": raise RuntimeError("TP2 adapter evaluation permits only both named solos in order with explicit evaluation flag") for fd, suffix in ((8, "240876ad"), (9, "bc62df2a")): path = Path.home()/f"dsv41-offload/locks/gpu-{suffix}.lock" held, expected = os.fstat(fd), path.stat() if (held.st_dev, held.st_ino) != (expected.st_dev, expected.st_ino): raise RuntimeError("adapter evaluation did not inherit the correct solo flock") with path.open("a") as probe: try: fcntl.flock(probe, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: pass else: fcntl.flock(probe, fcntl.LOCK_UN) raise RuntimeError("adapter evaluation solo flock was not already held") fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)