src/verify.py

"""Every statement of the paper, checked, with the time each check takes.

The checks are in three tiers, and the checks of a tier run concurrently. The
first establishes the statements: each one is an exhaustive or near-exhaustive
audit of a definition, a lemma or a proposition, and none of them runs the full
instance. The second executes every construction end to end: the instances on
the integer reference, four machines on the netlist that sigma denotes, the
organism on the interpreter, the hierarchy, the constructor inside the
interpreter, and the host and the organism on the evaluator in C, which runs
three generations of the self-reproducing instance.

The third runs the complete instance on the threshold evaluators and the
perturbation experiments. Its cost belongs to the object: one generation is
1,647,743 steps of a map of 87,080 units.
The rate of each evaluator is in paper/runs/paper_throughput.json, so the cost
of any one line below is a multiplication.

    python src/verify.py                # the statements
    python src/verify.py --tier 2       # and every construction, end to end
    python src/verify.py --tier 3       # and the full-scale runs
"""
from __future__ import annotations

import argparse
import concurrent.futures as cf
import json
import os
import subprocess
import sys
import time

REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SRC = os.path.join(REPO, "src")
PY = sys.executable

TIER1 = [
    ("check_sigma.py", [], "the canonical serialization, Definition 2.11"),
    ("check_device.py", [], "the device, Definition 2.8"),
    ("check_framing.py", [], "the framing, Definition 2.14"),
    ("check_host.py", [], "the machine, Definition 3.1"),
    ("check_lev.py", [], "the levelization, Lemma 2.6"),
    ("check_ternarity.py", [], "ternary depth, Proposition 3.5"),
    ("check_codec.py", [], "the recipe codec, Lemma 4.6"),
    ("check_optimal.py", [], "optimality of the bound, Proposition 4.4"),
    ("check_steps.py", [], "the step counts, Theorem 4.8 and Proposition 5.4"),
    ("check_total.py", [], "arbitrary tapes, Propositions 4.12 and 4.14"),
    ("check_environment.py", [], "the environment, Proposition 2.18"),
    ("check_settle.py", [], "order-independent settling, all 9! orders"),
    ("check_dependence.py", [], "the size hypothesis"),
    ("check_interp.py", [], "the one-record semantics of the interpreter"),
]

TIER2 = [
    ("organism.py", ["--net-generations", "1"],
     "universal construction and self-reproduction by the dynamics"),
    ("hosted_constructor.py", [], "the constructor inside the interpreter"),
    ("hierarchy.py", [], "a larger interpreter runs a smaller one"),
    ("check_family_gate.py", [], "four machines on the netlist of sigma"),
    ("check_c.py", [], "the host and the organism on the evaluator in C"),
    ("paper_runs.py", ["reference"], "the constructions on the integer reference"),
]

TIER3 = [
    ("paper_verify.py", [], "the codec over the whole artifact, 971 MB"),
    ("check_family.py", [], "all 29 machines on the integer reference"),
    ("paper_runs.py", ["lockstep", "--device", "cuda"],
     "four evaluators in lockstep"),
    ("paper_runs.py", ["net", "--device", "cuda"],
     "three generations on the netlist of sigma"),
    ("paper_runs.py", ["lev", "--device", "cuda", "--generations", "1"],
     "one generation on Lev(N_host)"),
    ("throughput.py", [], "the rate of each evaluator"),
    ("noise.py", ["margin"], "the margin along the trajectory, Theorem 6.2"),
    ("noise.py", ["curve"], "the deviation rate against the union bound"),
    ("noise.py", ["leak", "--steps", "10000"],
     "static weight error with the zeros leaking"),
    ("noise.py", ["omega"], "the instance under read noise, to completion"),
    ("noise.py", ["profile"], "the pre-activation profile and its bound"),
    ("organism.py", ["--generations", "8", "--net-generations", "2"],
     "the organism over eight generations, two on the netlist"),
]


def run(entry, quiet, threads=None):
    name, args, what = entry
    t0 = time.perf_counter()
    env = dict(os.environ)
    if threads:
        env["OMP_NUM_THREADS"] = str(threads)
    p = subprocess.run([PY, os.path.join(SRC, name)] + args,
                       capture_output=True, text=True, cwd=REPO, env=env)
    dt = time.perf_counter() - t0
    ok = p.returncode == 0
    label = f"{name} {' '.join(args)}".strip()
    print(f"  {'ok  ' if ok else 'FAIL'} {label:<34} {dt:7.1f} s   {what}",
          flush=True)
    if not ok and not quiet:
        print(p.stdout[-1500:])
        print(p.stderr[-1500:])
    return ok, dt


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--tier", type=int, default=1)
    ap.add_argument("--quiet", action="store_true")
    ap.add_argument("--jobs", type=int, default=os.cpu_count() or 1,
                    help="checks of a tier run concurrently")
    args = ap.parse_args()

    plan = [("statements", TIER1)]
    if args.tier >= 2:
        plan.append(("constructions", TIER2))
    if args.tier >= 3:
        plan.append(("full scale", TIER3))

    total = 0.0
    bad = 0
    record = {}
    for title, entries in plan:
        print(f"[{title}]", flush=True)
        t0 = time.perf_counter()
        jobs = max(1, min(args.jobs, len(entries)))
        threads = max(1, (os.cpu_count() or 1) // jobs)
        with cf.ThreadPoolExecutor(max_workers=jobs) as ex:
            futs = {ex.submit(run, e, args.quiet, threads): e for e in entries}
            for f in cf.as_completed(futs):
                e = futs[f]
                ok, dt = f.result()
                bad += not ok
                record[f"{e[0]} {' '.join(e[1])}".strip()] = {"ok": ok, "seconds": dt}
        sub = time.perf_counter() - t0
        total += sub
        print(f"  {title}: {sub / 60:.1f} min wall time", flush=True)
    print(f"total {total / 60:.1f} min; failures: {bad}")
    d = os.path.join(REPO, "paper", "runs")
    os.makedirs(d, exist_ok=True)
    json.dump({"tier": args.tier, "total_seconds": total, "failures": bad,
               "checks": record},
              open(os.path.join(d, "paper_verify_times.json"), "w"), indent=1)
    return 0 if bad == 0 else 1


if __name__ == "__main__":
    sys.exit(main())