#!/usr/bin/env python3 """Constrained L21 snapshot, output-parity, crossing, and off-cost evidence.""" import os import re import signal import statistics import subprocess import time from pathlib import Path ROOT = Path(__file__).resolve().parents[4] OUT = ROOT / "Saved/Linux/linux21" OUT.mkdir(parents=True, exist_ok=True) if not os.sched_getaffinity(0).issubset(set(range(12))): raise SystemExit("run under taskset -c 0-11") BIN = { "clang-off": ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-linux21-clang", "gcc-default": ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-linux21-gcc", } OLD_CLANG = ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-linux19-clang" OLD_GCC = ROOT / "Standalone/bin/linux/VoxelForgeCoreHeadless-linux19-gcc" DATA = ROOT / "Standalone/data/owner_world.json" HASH_ROW = re.compile(r"^\[Pipeline(?:FullDetailHash|WholeViewHash|LODHash|ExactQueryHash)\].*$", re.M) LODS = [(1, 1), (8, 1), (12, 1), (12, 2), (12, 3), (12, 4), (12, 5)] RECIPES = { "surface-world": "0,-35,-487.94,0,-22", "floating-islands": "-22,-19,-22.8,0,-4", } results = [] def launch(name, argv, timeout=900): logfile = OUT / f"{name}.log" command = ["taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--", *map(str, argv)] started = time.monotonic() peak_tree = 0 peak_process = 0 timed_out = False with logfile.open("w", encoding="utf-8") as stream: process = subprocess.Popen(command, stdout=stream, stderr=subprocess.STDOUT, start_new_session=True) while process.poll() is None: pending, seen, total = [process.pid], set(), 0 while pending: pid = pending.pop() if pid in seen: continue seen.add(pid) try: pending.extend(int(x) for x in Path(f"/proc/{pid}/task/{pid}/children").read_text().split()) values = {} for line in Path(f"/proc/{pid}/status").read_text().splitlines(): if line.startswith(("VmRSS:", "VmHWM:")): key, amount, *_ = line.split() values[key] = int(amount) total += values.get("VmRSS:", 0) peak_process = max(peak_process, values.get("VmRSS:", 0), values.get("VmHWM:", 0)) except (FileNotFoundError, PermissionError, ProcessLookupError, ValueError): pass peak_tree = max(peak_tree, total) if time.monotonic() - started > timeout: os.killpg(process.pid, signal.SIGTERM) try: process.wait(timeout=5) except subprocess.TimeoutExpired: os.killpg(process.pid, signal.SIGKILL) process.wait() timed_out = True break time.sleep(0.1) code = process.wait() elapsed = time.monotonic() - started rows = HASH_ROW.findall(logfile.read_text(encoding="utf-8", errors="replace")) results.append((name, code, elapsed, peak_process, peak_tree, timed_out, len(rows))) print(f"[L21Run] {name} exit={code} seconds={elapsed:.3f} peak_process_rss_kib={peak_process} " f"peak_tree_rss_kib={peak_tree} hash_rows={len(rows)} timed_out={str(timed_out).lower()}", flush=True) return logfile def base_args(binary, recipe, workers, hash_output=False, crossing=False): args = [binary, DATA, "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline", "--force-generate", "--clip-radius=3", "--max-clip-level=8", f"--workers={workers}", f"--camera={RECIPES[recipe]}", f"--recipe={recipe}"] if hash_output: args.append("--stream-hash") if crossing: args += ["--camera-step=24,0,0", "--region-crossing"] return args # Plan-time identities and canonical serialization at W1/W8/W12, both field recipes. snapshots = {} for recipe in RECIPES: for workers, repeat in [(1, 1), (8, 1), (12, 1)]: path = OUT / f"snapshot-{recipe}-w{workers}-r{repeat}.json" args = [BIN["clang-off"], DATA, "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline", f"--recipe={recipe}", f"--workers={workers}", f"--resolved-layout-output={path}", "--resolved-layout-only"] log = launch(f"snapshot-{recipe}-w{workers}-r{repeat}", args) if results[-1][1] != 0: raise SystemExit(f"snapshot plan failed: {log}") snapshots[(recipe, workers)] = path.read_bytes() if len({snapshots[(recipe, workers)] for workers in (1, 8, 12)}) != 1: raise SystemExit(f"snapshot serialization changed with worker count: {recipe}") print(f"[L21SnapshotIdentity] recipe={recipe} worker_counts=1,8,12 canonical_bytes=identical status=PASS", flush=True) # Full matrix. Every row is compared against the L19 W12x5 reference. for recipe in RECIPES: for compiler, binary in BIN.items(): for workers, repeat in LODS: name = f"matrix-{recipe}-{compiler}-w{workers}-r{repeat}" log = launch(name, base_args(binary, recipe, workers, hash_output=True)) rows = HASH_ROW.findall(log.read_text(encoding="utf-8", errors="replace")) reference_name = f"{recipe}-{compiler}-w12-r5.log" reference = ROOT / "Saved/Linux/l19-matrix" / reference_name expected = HASH_ROW.findall(reference.read_text(encoding="utf-8", errors="replace")) if results[-1][1] != 0 or len(rows) != 12 or rows != expected: print(f"[L21HashMismatch] {name} expected_rows={len(expected)} actual_rows={len(rows)}", flush=True) raise SystemExit(1) # The crossing matrix retains the 24 m two-view query proof on both compilers. for compiler, binary in BIN.items(): for workers, repeat in LODS: name = f"crossing-{compiler}-w{workers}-r{repeat}" log = launch(name, base_args(binary, "surface-world", workers, hash_output=True, crossing=True)) rows = HASH_ROW.findall(log.read_text(encoding="utf-8", errors="replace")) reference = ROOT / "Saved/Linux/l19-crossing" / f"{compiler}-w12-r5.log" expected = HASH_ROW.findall(reference.read_text(encoding="utf-8", errors="replace")) if results[-1][1] != 0 or len(rows) != 12 or rows != expected: print(f"[L21CrossingHashMismatch] {name} expected_rows={len(expected)} actual_rows={len(rows)}", flush=True) raise SystemExit(1) # Five interleaved W12 pairs per recipe, hashing disabled. Alternate order by pair to reduce drift. off_pairs = {recipe: {"old": [], "new": []} for recipe in RECIPES} for recipe in RECIPES: for pair in range(1, 6): order = [("old", OLD_CLANG), ("new", BIN["clang-off"])] if pair % 2 else [("new", BIN["clang-off"]), ("old", OLD_CLANG)] for label, binary in order: name = f"offcost-{recipe}-pair{pair}-{label}" log = launch(name, base_args(binary, recipe, 12, hash_output=False), timeout=900) if results[-1][1] != 0: raise SystemExit(f"hashing-off cost run failed: {log}") off_pairs[recipe][label].append(results[-1][2]) print(f"[L21OffCostPair] recipe={recipe} pair={pair} old_seconds={off_pairs[recipe]['old'][-1]:.3f} " f"new_seconds={off_pairs[recipe]['new'][-1]:.3f}", flush=True) for recipe, pair in off_pairs.items(): old = statistics.median(pair["old"]) new = statistics.median(pair["new"]) print(f"[L21OffCostSummary] recipe={recipe} pairs=5 median_old_seconds={old:.3f} " f"median_new_seconds={new:.3f} delta_percent={(new-old)/old*100.0:.2f}", flush=True) # Query the real compiled Core field at every distinct recorded stratum/band endpoint. for recipe in RECIPES: name = f"boundary-probes-{recipe}" args = base_args(BIN["clang-off"], recipe, 12, hash_output=False) args += ["--audit-resolved-layout-boundaries"] log = launch(name, args) if results[-1][1] != 0: raise SystemExit(f"Core boundary probes failed: {log}") print(f"[L21EvidenceSummary] snapshots=6 matrix=28 crossing=14 off_cost_pairs=10 boundary_audits=2 status=PASS", flush=True) for name, code, elapsed, rss, tree, timed_out, rows in results: print(f"[L21RunDetail] {name} exit={code} seconds={elapsed:.3f} peak_process_rss_kib={rss} " f"peak_tree_rss_kib={tree} hash_rows={rows} timed_out={str(timed_out).lower()}", flush=True)