scripts/run_rgbd_long_data_experiment.py
#!/usr/bin/env python3
"""Execute the frozen V18 experiment on hit (CPU only, AVX2 kernels), never opening fresh data.
Derived from the V14 runner. Every child runs through the hash-locked path-mapping launcher
(records from zonghan_o store /home/zonghan/... paths) with the frozen thread and ISA environment.
Phases: the three training runs (one core each), the built-in V14 C1 control (stops on FAIL),
the EVAL seal of each cohort (concurrently), score, analyze and the report. A signal-terminated
run is archived and rerun from scratch within one total budget per run; any other error stops.
The process must be pinned to the frozen CPU cores (taskset -c 8-23); children inherit it.
"""
import argparse
import concurrent.futures
import os
import shutil
import subprocess
import sys
import threading
import time
from pathlib import Path
from mcss.mechanism_pilot.geometry_carrier_experiment import read, verify_lock
from mcss.mechanism_pilot.small_training import sha, write_json
_LOCK = threading.Lock()
def archive_attempt(root, label, attempt, returncode, archive):
root = Path(root)
failed = root / "audit" / "failed_attempts" / f"{label}_attempt{attempt}"
failed.mkdir(parents=True, exist_ok=False)
moved = []
for path in (root / "audit" / f"{label}.log", *(Path(p) for p in archive)):
if path.exists():
shutil.move(str(path), str(failed / path.name))
moved.append(str(path))
with _LOCK:
record = root / "audit" / "incidents.json"
incidents = read(record) if record.exists() else []
incidents.append(
{
"label": label,
"attempt": attempt,
"kind": "native_crash",
"returncode": returncode,
"archived_to": str(failed),
"moved": moved,
"unix_time": time.time(),
"policy": "signal-terminated: archive, rerun identical command from scratch",
}
)
write_json(record, incidents)
def execute(root, label, args, *, env, launcher, times, archive=(), retries=1):
"""Run one frozen phase through the launcher; retry only signal-terminated runs."""
root = Path(root)
failed = root / "audit" / "failed_attempts"
prior = len(list(failed.glob(f"{label}_attempt*"))) if failed.exists() else 0
if prior > retries:
raise PermissionError(f"{label}: retry budget already spent")
for attempt in range(prior + 1, retries + 2):
print(label, "START attempt", attempt, flush=True)
tick = time.perf_counter()
log = root / "audit" / f"{label}.log"
with log.open("x") as handle:
process = subprocess.run(
[sys.executable, launcher, *args], stdout=handle, stderr=subprocess.STDOUT, env=env
)
if process.returncode == 0:
times[label] = time.perf_counter() - tick
print(label, "PASS", times[label], flush=True)
return
if process.returncode >= 0:
raise subprocess.CalledProcessError(process.returncode, args)
archive_attempt(root, label, attempt, process.returncode, archive)
print(label, "NATIVE_CRASH", process.returncode, "archived", flush=True)
raise subprocess.CalledProcessError(process.returncode, args)
def parallel(root, jobs, *, env, launcher, times, retries, workers):
"""Run {label: (args, archive)} concurrently; any failure stops the experiment."""
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
futures = {
pool.submit(
execute,
root,
label,
args,
env=env,
launcher=launcher,
times=times,
archive=archive,
retries=retries,
): label
for label, (args, archive) in jobs.items()
}
concurrent.futures.wait(futures)
failures = {futures[f]: repr(f.exception()) for f in futures if f.exception() is not None}
if failures:
record = root / "audit" / "phase_failures.json"
write_json(record, {**(read(record) if record.exists() else {}), **failures})
raise RuntimeError(f"Phase failed; experiment stops: {sorted(failures)}")
def run(root):
root = Path(root).resolve()
_, config = verify_lock(root)
if read(root / "tests.json")["status"] != "PASS":
raise PermissionError("Full suite must PASS before training")
if not os.sched_getaffinity(0) <= set(config["cpu_affinity"]):
raise PermissionError("Run must be pinned to the frozen CPU cores (taskset)")
launcher = config["host_launcher"]["path"]
if sha(Path(launcher)) != config["host_launcher"]["sha256"]:
raise PermissionError("The path-mapping launcher changed after the freeze")
if not os.environ.get("CV_ROOT"):
raise PermissionError("CV_ROOT (the mirror root) must be set")
env = {
**os.environ,
"PYTHONPATH": "src",
"CUDA_VISIBLE_DEVICES": "",
**config["thread_environment"],
}
retries = config["infrastructure_retries"]
times = {}
start = time.perf_counter()
training = {
f"train_L_{seed}": (
[
"scripts/train_rgbd_long_data_carrier.py",
"--root",
str(root),
"--variant",
"L",
"--seed",
str(seed),
"--device",
"cpu",
],
[root / "checkpoints" / f"L_{seed}"],
)
for seed in config["seeds"]
if not (root / "checkpoints" / f"L_{seed}" / "summary.json").exists()
}
(root / "checkpoints").mkdir(exist_ok=True)
parallel(
root,
training,
env=env,
launcher=launcher,
times=times,
retries=retries,
workers=config["parallel_workers"],
)
if not (root / "audit/v14_prefix_control.json").exists():
execute(
root,
"v14_prefix_control",
["scripts/check_rgbd_long_data_prefix.py", "--root", str(root)],
env=env,
launcher=launcher,
times=times,
retries=0,
)
if read(root / "audit/v14_prefix_control.json")["status"] != "PASS":
raise RuntimeError("V18 does not reproduce V14 C1 for steps 1-6000; experiment stops")
seals = {
f"eval_seal_{name}": (
[
"scripts/evaluate_rgbd_long_data_eval.py",
"--root",
str(root),
"--stage",
"seal",
"--cohort",
name,
],
[root / "raw" / "eval" / name],
)
for name in config["eval"]["cohorts"]
if not (root / "raw" / "eval" / name / "prediction_seal.json").exists()
}
parallel(
root,
seals,
env=env,
launcher=launcher,
times=times,
retries=retries,
workers=len(seals) or 1,
)
for stage, done in (
("score", root / "raw/eval/query_rows.json"),
("analyze", root / "eval_results.json"),
):
if not done.exists():
execute(
root,
f"eval_{stage}",
["scripts/evaluate_rgbd_long_data_eval.py", "--root", str(root), "--stage", stage],
env=env,
launcher=launcher,
times=times,
retries=retries,
)
execute(
root,
"report",
["scripts/report_rgbd_long_data_carrier.py", "--root", str(root)],
env=env,
launcher=launcher,
times=times,
retries=0,
)
wall = root / "audit/phase_wall_times.json"
previous = read(wall) if wall.exists() else {"phases": {}, "invocations": []}
write_json(
wall,
{
"phases": {**previous["phases"], **times},
"invocations": [*previous.get("invocations", []), time.perf_counter() - start],
"parallel_workers": config["parallel_workers"],
"cpu_affinity": sorted(os.sched_getaffinity(0)),
},
)
print("ALL_PHASES_COMPLETE_FRESH_REMAINS_CLOSED", flush=True)
if __name__ == "__main__":
p = argparse.ArgumentParser()
p.add_argument("--root", required=True)
run(p.parse_args().root)