1"""Cell-queue job runner: runs every cell of an experiment script as a fresh subprocess (one solver thread each),2at most --workers concurrent, in phases (seed 0 for all cells first, then the remaining seeds) with cells ordered by3expected cost inside a phase, so that a --deadline cuts the number of seeds of the largest cells rather than whole4rows. Per-cell JSONs go to --cells-dir; cells already present are skipped (re-runnable). Finally --aggregate.56Usage: python cluster_driver.py --script preimage_sat.py --cells-dir results/cells_preimage \7 --out results/preimage_sat_scaling.json --workers 24 --deadline 7200 [--timeout 300]8"""9from __future__ import annotations10import argparse, json, os, pathlib, subprocess, sys, time11from concurrent.futures import ThreadPoolExecutor, as_completed1213HERE = pathlib.Path(__file__).resolve().parent141516def cell_name(c: dict) -> str:17 return "_".join(f"{k}-{c[k]}" for k in sorted(c))181920def cost(c: dict) -> tuple:21 return (c.get("placement", "first") == "spread", c["r"], c.get("k", 0), c.get("iv") == "free", int(c.get("feedforward", 0)))222324def main():25 ap = argparse.ArgumentParser()26 ap.add_argument("--script", required=True)27 ap.add_argument("--cells-dir", required=True)28 ap.add_argument("--out", required=True)29 ap.add_argument("--workers", type=int, default=24)30 ap.add_argument("--deadline", type=float, default=1e9, help="seconds after start; no new cell is launched after it")31 ap.add_argument("--timeout", type=float, default=None)32 ap.add_argument("--list-args", default="")33 ap.add_argument("--python", default=sys.executable)34 a = ap.parse_args()35 script = str(HERE / a.script)36 cells_dir = pathlib.Path(a.cells_dir); cells_dir.mkdir(parents=True, exist_ok=True)37 lst = subprocess.run([a.python, script, "--list-cells"] + a.list_args.split(), capture_output=True, text=True, check=True)38 cells = [json.loads(l) for l in lst.stdout.splitlines() if l.strip()]39 # phases: seed 0 first (all cells), then seeds 1, 2, ...; inside a phase by cost40 cells.sort(key=lambda c: (c["seed"], cost(c)))41 todo = [c for c in cells if not (cells_dir / (cell_name(c) + ".json")).exists()]42 print(f"{len(cells)} cells, {len(todo)} to run, workers={a.workers}, deadline={a.deadline}s", flush=True)43 t0 = time.time()44 log = open(cells_dir / "driver.log", "a")45 skipped = []4647 def run(c):48 outp = cells_dir / (cell_name(c) + ".json")49 cmd = [a.python, script, "--cell", json.dumps(c), "--out", str(outp)]50 if a.timeout:51 cmd += ["--timeout", str(a.timeout)]52 env = dict(os.environ, OMP_NUM_THREADS="1", PYTHONPATH=str(HERE.parents[1] / "src"))53 t = time.time()54 try:55 p = subprocess.run(cmd, capture_output=True, text=True, env=env, timeout=(a.timeout or 600) + 120)56 except subprocess.TimeoutExpired:57 outp.write_text(json.dumps({"cell": c, "status": "timeout", "killed_by_driver": True,58 "wall_s": time.time() - t, "hostname": os.uname().nodename}))59 p = subprocess.CompletedProcess(cmd, -9, "", "killed by driver guard")60 status = "?"61 if outp.exists():62 try:63 status = json.loads(outp.read_text()).get("status")64 except Exception:65 status = "bad-json"66 else:67 status = "failed: " + p.stderr[-300:].replace("\n", " | ")68 line = f"{time.strftime('%H:%M:%S')} {cell_name(c)} -> {status} ({time.time()-t:.1f}s) [elapsed {time.time()-t0:.0f}s]"69 print(line, flush=True); log.write(line + "\n"); log.flush()70 return c, status7172 with ThreadPoolExecutor(max_workers=a.workers) as ex:73 futs = []74 it = iter(todo)75 # submit lazily so the deadline can stop launches76 pending = set()77 for c in it:78 while len(pending) >= a.workers:79 done = next(as_completed(pending)); pending.discard(done)80 if time.time() - t0 > a.deadline:81 skipped.append(c)82 continue83 pending.add(ex.submit(run, c))84 for f in as_completed(pending):85 pass86 if skipped:87 (cells_dir / "skipped_cells.json").write_text(json.dumps(skipped, indent=1))88 print(f"deadline reached: {len(skipped)} cells not started (see skipped_cells.json)")89 subprocess.run([a.python, script, "--aggregate", str(cells_dir), "--out", a.out], check=True,90 env=dict(os.environ, PYTHONPATH=str(HERE.parents[1] / "src")))91 print("done in %.0f s" % (time.time() - t0))929394if __name__ == "__main__":95 main()96