#!/usr/bin/env python3 """Measure passage and boundary-band ceilings independently under the L80 protocol.""" from __future__ import annotations import argparse import json import re import statistics import subprocess import time from pathlib import Path ROOT = Path(__file__).resolve().parents[4] OUT = ROOT / "Saved/Linux/l88/ceiling-split" DATA = ROOT / "Standalone/data/owner_world.json" BINARY = ROOT / "Saved/Linux/l88/bin/VoxelForgeCoreHeadless-l88-clang19" STRATA = { "S0": {"id": 0, "camera": (8.0, 16.0, -372.2626953125), "near": 1408, "whole": 2850}, "FI": {"id": 1, "camera": (25.8369140625, 3.19921875, -737.0), "near": 1216, "whole": 2249}, "TN": {"id": 2, "camera": (19.166015625, -14.2001953125, -753.3740234375), "near": 1344, "whole": 2233}, "S3": {"id": 3, "camera": (-20.0625, -5.068359375, -900.353515625), "near": 1344, "whole": 2329}, } WORKERS = (8, 12) MODES = ("direct", "passages_off", "bands_off") PIPELINE = re.compile(r"^\[StreamPipeline\] (.*)$", re.M) FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)") NEAR = re.compile( r"^\[WorldAssemblerNearLODReady\] lods=0-2 tiles=(\d+)/(\d+) cold_ready_s=([0-9.]+)$", re.M) PASSAGE_TILE = re.compile(r"^\[PassageComposerTile\] (.*)$", re.M) STAMP = re.compile(r"^\[VoxelForgeSourceStamp\] sha256=(\S+)$", re.M) def base_command(stratum: str, workers: int, mode: str, profile: bool = False) -> list[str]: spec = STRATA[stratum] camera = ",".join(f"{value:.12f}" for value in spec["camera"]) + ",0,0" child = [ str(BINARY), str(DATA), "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline", f"--workers={workers}", "--clip-radius=3", "--max-clip-level=8", "--max-level=8", f"--camera={camera}", "--profile-lod-work", "--stream-scheduler-policy=3", ] if profile: child.append("--profile-passage-composer") env = ["VOXELFORGE_L81_PASSAGE_BOUND_PADDING_VOXELS=0"] if mode == "passages_off": env.extend(("VOXELFORGE_L88_DISABLE_PASSAGES=1", "VOXELFORGE_L88_ALLOW_PASSAGE_ABLATION_WALKABILITY_FAILURE=1")) elif mode == "bands_off": env.append("VOXELFORGE_L88_DISABLE_BOUNDARY_BANDS=1") return [ "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--", "env", *env, "timeout", "1740s", *child, ] def execute(stratum: str, workers: int, mode: str, phase: str, replicate: int = 0, profile: bool = False) -> dict: OUT.mkdir(parents=True, exist_ok=True) label = f"{stratum.lower()}-w{workers}-{mode}-{phase}" if replicate: label += f"-r{replicate}" if profile: label += "-profile" log = OUT / f"{label}.log" started = time.monotonic() with log.open("w", encoding="utf-8") as stream: result = subprocess.run(base_command(stratum, workers, mode, profile), cwd=ROOT, stdout=stream, stderr=subprocess.STDOUT, check=False, timeout=1750) output = log.read_text(encoding="utf-8", errors="replace") pipeline, near, stamp = PIPELINE.search(output), NEAR.search(output), STAMP.search(output) if result.returncode or pipeline is None or near is None or stamp is None: raise RuntimeError(f"run failed or required output is missing: {log}; exit={result.returncode}") spec = STRATA[stratum] near_counts = (int(near.group(1)), int(near.group(2))) if near_counts != (spec["near"], spec["near"]): raise RuntimeError(f"incomplete near set in {log}: {near_counts}") fields = dict(FIELDS.findall(pipeline.group(1))) whole_counts = tuple(int(value) for value in fields["whole_tiles"].split("/")) if whole_counts != (spec["whole"], spec["whole"]): raise RuntimeError(f"incomplete whole view in {log}: {whole_counts}") ablation_walkability = "[L88PassageAblation]" in output if (mode == "passages_off") != ablation_walkability: raise RuntimeError(f"passage ablation walkability marker unexpected in {log}") row = { "stratum": stratum, "workers": workers, "mode": mode, "phase": phase, "replicate": replicate, "profile": profile, "exit_code": result.returncode, "wall_s": round(time.monotonic() - started, 3), "source_stamp": stamp.group(1), "lod02_ready_s": float(near.group(3)), "whole_view_ready_s": float(fields["whole_view_ready"].removesuffix("s")), "shape_s": float(fields["shape"].removesuffix("s")), "far_shape_s": float(fields["far_shape"].removesuffix("s")), "ablation_walkability_failed": ablation_walkability, "near_tiles": spec["near"], "whole_tiles": spec["whole"], "log": str(log.relative_to(ROOT)), } if profile: total = {name: 0 for name in ( "boundary_band_queries", "queries", "passage_iterations", "carve_segment_visits", "support_segment_visits", "relief_field_calls", )} profiled_tiles = 0 for match in PASSAGE_TILE.finditer(output): fields = dict(FIELDS.findall(match.group(1))) if int(fields["key"].rsplit("lod", 1)[1].rstrip(")")) > 2: continue profiled_tiles += 1 for name in total: total[name] += int(fields[name]) if not profiled_tiles: raise RuntimeError(f"passage/band profile rows missing in {log}") row.update({f"profile_{name}": value for name, value in total.items()}) row["profiled_near_tiles"] = profiled_tiles print("[L88CeilingSplitRun] " + json.dumps(row, sort_keys=True), flush=True) return row def profile_modes() -> list[dict]: rows = [execute(stratum, 8, mode, "profile", profile=True) for stratum in ("FI", "TN") for mode in MODES] for stratum in ("FI", "TN"): subset = {row["mode"]: row for row in rows if row["stratum"] == stratum} direct, passages, bands = (subset[name] for name in MODES) if direct["profile_boundary_band_queries"] == 0 or direct["profile_queries"] == 0: raise RuntimeError(f"direct profile did not exercise both paths for {stratum}") if passages["profile_boundary_band_queries"] == 0 or any( passages[f"profile_{name}"] != 0 for name in ("queries", "passage_iterations", "carve_segment_visits", "support_segment_visits", "relief_field_calls") ): raise RuntimeError(f"passages-off profile did not isolate the band path for {stratum}") if bands["profile_boundary_band_queries"] != 0 or any( bands[f"profile_{name}"] == 0 for name in ("queries", "passage_iterations", "carve_segment_visits") ): raise RuntimeError(f"bands-off profile did not preserve passage work for {stratum}") (OUT / "profile-summary.json").write_text(json.dumps(rows, indent=2) + "\n", encoding="utf-8") return rows def timing_matrix() -> list[dict]: # One warmup for each complete stratum/worker/mode window. for stratum in STRATA: for workers in WORKERS: for mode in MODES: execute(stratum, workers, mode, "warmup") orders = ( ((8, "direct"), (8, "passages_off"), (8, "bands_off"), (12, "bands_off"), (12, "passages_off"), (12, "direct")), ((12, "passages_off"), (12, "direct"), (12, "bands_off"), (8, "bands_off"), (8, "direct"), (8, "passages_off")), ((8, "bands_off"), (8, "direct"), (8, "passages_off"), (12, "direct"), (12, "passages_off"), (12, "bands_off")), ) measured = [] for stratum in STRATA: for replicate, order in enumerate(orders, 1): for workers, mode in order: measured.append(execute(stratum, workers, mode, "measured", replicate)) (OUT / "runs-partial.json").write_text( json.dumps(measured, indent=2) + "\n", encoding="utf-8") summary = [] for stratum in STRATA: for workers in WORKERS: for mode in MODES: subset = [r for r in measured if r["stratum"] == stratum and r["workers"] == workers and r["mode"] == mode] if len(subset) != 3: raise RuntimeError(f"expected 3 measured runs for {stratum} W{workers} {mode}") item = {"stratum": stratum, "workers": workers, "mode": mode, "measured_runs": len(subset)} for metric in ("lod02_ready_s", "whole_view_ready_s", "shape_s", "far_shape_s"): values = [r[metric] for r in subset] item[metric] = values item[f"{metric}_median"] = statistics.median(values) summary.append(item) (OUT / "summary.json").write_text( json.dumps({"runs": measured, "summary": summary}, indent=2) + "\n", encoding="utf-8") return measured def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--profiles-only", action="store_true") parser.add_argument("--timing-only", action="store_true") args = parser.parse_args() if not BINARY.is_file() or not DATA.is_file(): raise SystemExit(f"missing input: {BINARY} or {DATA}") if not args.timing_only: profile_modes() if not args.profiles_only: timing_matrix() return 0 if __name__ == "__main__": raise SystemExit(main())