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 ap.add_argument("--filter", default="", help="python expression over the cell dict c, e.g. \"c['k']==16 or c['r']<=16\"")35 a = ap.parse_args()36 script = str(HERE / a.script)37 cells_dir = pathlib.Path(a.cells_dir); cells_dir.mkdir(parents=True, exist_ok=True)38 lst = subprocess.run([a.python, script, "--list-cells"] + a.list_args.split(), capture_output=True, text=True, check=True)39 cells = [json.loads(l) for l in lst.stdout.splitlines() if l.strip()]40 if a.filter:41 cells = [c for c in cells if eval(a.filter, {}, {"c": c})]42 # phases: seed 0 first (all cells), then seeds 1, 2, ...; inside a phase by cost43 cells.sort(key=lambda c: (c["seed"], cost(c)))44 todo = [c for c in cells if not (cells_dir / (cell_name(c) + ".json")).exists()]45 print(f"{len(cells)} cells, {len(todo)} to run, workers={a.workers}, deadline={a.deadline}s", flush=True)46 t0 = time.time()47 log = open(cells_dir / "driver.log", "a")48 skipped = []4950 def run(c):51 outp = cells_dir / (cell_name(c) + ".json")52 cmd = [a.python, script, "--cell", json.dumps(c), "--out", str(outp)]53 if a.timeout:54 cmd += ["--timeout", str(a.timeout)]55 env = dict(os.environ, OMP_NUM_THREADS="1", PYTHONPATH=str(HERE.parents[1] / "src"))56 t = time.time()57 try:58 p = subprocess.run(cmd, capture_output=True, text=True, env=env, timeout=(a.timeout or 600) + 120)59 except subprocess.TimeoutExpired:60 outp.write_text(json.dumps({"cell": c, "status": "timeout", "killed_by_driver": True,61 "wall_s": time.time() - t, "hostname": os.uname().nodename}))62 p = subprocess.CompletedProcess(cmd, -9, "", "killed by driver guard")63 status = "?"64 if outp.exists():65 try:66 status = json.loads(outp.read_text()).get("status")67 except Exception:68 status = "bad-json"69 else:70 status = "failed: " + p.stderr[-300:].replace("\n", " | ")71 line = f"{time.strftime('%H:%M:%S')} {cell_name(c)} -> {status} ({time.time()-t:.1f}s) [elapsed {time.time()-t0:.0f}s]"72 print(line, flush=True); log.write(line + "\n"); log.flush()73 return c, status7475 with ThreadPoolExecutor(max_workers=a.workers) as ex:76 futs = []77 it = iter(todo)78 # submit lazily so the deadline can stop launches79 pending = set()80 for c in it:81 while len(pending) >= a.workers:82 done = next(as_completed(pending)); pending.discard(done)83 if time.time() - t0 > a.deadline:84 skipped.append(c)85 continue86 pending.add(ex.submit(run, c))87 for f in as_completed(pending):88 pass89 if skipped:90 (cells_dir / "skipped_cells.json").write_text(json.dumps(skipped, indent=1))91 print(f"deadline reached: {len(skipped)} cells not started (see skipped_cells.json)")92 subprocess.run([a.python, script, "--aggregate", str(cells_dir), "--out", a.out], check=True,93 env=dict(os.environ, PYTHONPATH=str(HERE.parents[1] / "src")))94 print("done in %.0f s" % (time.time() - t0))959697if __name__ == "__main__":98 main()99