VoxelForge/Standalone/Research/WorldAssemblerM3/Linux125/RunL125PassageCounts.py

#!/usr/bin/env python3
"""Measure passage sweep work by stratum and LOD."""
from __future__ import annotations

import json
import os
import re
import subprocess
import time
from collections import defaultdict
from pathlib import Path

ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / os.environ.get("L125_COUNTS_DIR", "Saved/Linux/l125/counts")
DATA = ROOT / "Standalone/data/owner_world.json"
BINARY = ROOT / os.environ.get("L125_COUNTS_BINARY",
    "Saved/Linux/l125/bin/VoxelForgeCoreHeadless-l125-clang")
CAMERAS = {
    0: (8.0, 16.0, -372.2626953125),
    1: (25.8369140625, 3.19921875, -737.0),
    2: (19.166015625, -14.2001953125, -753.3740234375),
    3: (-20.0625, -5.068359375, -900.353515625),
}
FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)")
CALLERS = (
    "Unknown", "DensityGrid", "GradientNormalFallback", "MidpointCorrection",
    "SkirtSurfaceVertex", "SkirtExtrudedVertex", "SkirtClippedVertex",
    "SkirtRefinementFace", "SkirtCoarseCentroid", "FarShellSurfaceQuery",
    "FarShellProjectionNormal", "FarShellStratumRefinement",
)


def records(path: Path, tag: str) -> dict[str, dict[str, str]]:
    result = {}
    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
        if line.startswith(tag + " "):
            row = dict(FIELDS.findall(line))
            result[row["key"]] = row
    return result


def lod_of(key: str) -> int:
    return int(key.rsplit("lod", 1)[1].rstrip(")"))


def run(strate: int, replicate: int) -> dict:
    OUT.mkdir(parents=True, exist_ok=True)
    label = f"s{strate}-w8-profile{replicate}"
    log = OUT / f"{label}.log"
    camera = ",".join(f"{value:.12f}" for value in CAMERAS[strate]) + ",0,0"
    command = [
        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
        "timeout", "1740s", str(BINARY), str(DATA),
        "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline",
        "--workers=8", "--clip-radius=3", "--max-clip-level=8", "--max-level=8",
        f"--camera={camera}", "--profile-lod-work", "--profile-passage-composer",
    ]
    started = time.monotonic()
    peak_tree_rss_kib = 0
    with log.open("w", encoding="utf-8") as output:
        process = subprocess.Popen(command, cwd=ROOT, stdout=output,
                                   stderr=subprocess.STDOUT)
        deadline = started + 1740
        page_kib = os.sysconf("SC_PAGE_SIZE") // 1024
        while process.poll() is None:
            pending = [process.pid]
            seen = set()
            rss_kib = 0
            while pending:
                pid = pending.pop()
                if pid in seen:
                    continue
                seen.add(pid)
                try:
                    with open(f"/proc/{pid}/statm", encoding="ascii") as statm:
                        rss_kib += int(statm.read().split()[1]) * page_kib
                    with open(f"/proc/{pid}/task/{pid}/children", encoding="ascii") as children:
                        pending.extend(int(value) for value in children.read().split())
                except (FileNotFoundError, ProcessLookupError, PermissionError,
                        IndexError, ValueError):
                    pass
            peak_tree_rss_kib = max(peak_tree_rss_kib, rss_kib)
            if time.monotonic() >= deadline:
                process.kill()
                process.wait()
                raise SystemExit(f"profile exceeded 30 minutes: {log}")
            time.sleep(0.01)
        return_code = process.wait()
    if return_code:
        raise SystemExit(f"profile failed: {log}; exit={return_code}")

    composer = records(log, "[PassageComposerTile]")
    stages = records(log, "[CpuTileStages]")
    counts = records(log, "[PassageSweepCount]")
    per_lod: dict[int, dict] = defaultdict(lambda: {
        "tiles": 0, "grid_samples": 0, "sweep_pairs": 0,
        "route_candidates": 0, "carve_segment_visits": 0,
        "sweep_prepare_ms": 0.0, "passage_worker_ms": 0.0,
        "classify_ms": 0.0, "column_ms": 0.0, "mc_ms": 0.0,
        "touched": 0, "touched_unique": 0, "body_band": 0,
        "stamped_outside": 0, "cache_valid": 0, "cache_valid_unique": 0,
        "shell_band": 0, "exact_band_unique": 0,
        "contributing_pairs": 0, "overlap_pairs": 0, "cross_passage_overlap": 0,
        "floor_profile_calls": 0,
        "cache_bytes": 0, "scratch_bytes": 0,
        "cache_bytes_max_tile": 0, "scratch_bytes_max_tile": 0,
        "lookups": 0, "cache_hits": 0, "missing": 0,
        "offgrid_miss": 0, "unprepared_miss": 0,
        "missing_by_caller": [0] * len(CALLERS),
        "fallback_by_caller": [0] * len(CALLERS),
    })
    for key, comp in composer.items():
        lod = lod_of(key)
        if lod > 2 or int(comp.get("features", "0")) == 0:
            continue
        stage = stages.get(key)
        count = counts.get(key)
        if stage is None or count is None:
            raise SystemExit(f"missing stage/count row for {label} {key}")
        row = per_lod[lod]
        row["tiles"] += 1
        row["grid_samples"] += int(comp["sweep_grid_samples"])
        for name in ("sweep_pairs", "route_candidates", "carve_segment_visits",
                     "lookups", "cache_hits", "missing"):
            source = {"route_candidates": "route_candidates",
                      "carve_segment_visits": "carve_segment_visits",
                      "lookups": "sweep_lookups", "cache_hits": "sweep_cache_hits",
                      "missing": "sweep_missing"}.get(name, name)
            row[name] += int(comp[source])
        row["sweep_prepare_ms"] += float(comp["sweep_prepare_ms"])
        row["passage_worker_ms"] += float(comp["worker_ms"])
        for name in ("classify_ms", "column_ms", "mc_ms"):
            row[name] += float(stage[name])
        for name in ("touched", "stamped_outside", "cache_valid", "cache_valid_unique",
                     "touched_unique", "body_band", "shell_band",
                     "exact_band_unique", "overlap_pairs", "cross_passage_overlap",
                     "contributing_pairs", "floor_profile_calls",
                     "cache_bytes", "scratch_bytes", "offgrid_miss",
                     "unprepared_miss"):
            row[name] += int(count[name])
        row["cache_bytes_max_tile"] = max(row["cache_bytes_max_tile"],
                                          int(count["cache_bytes"]))
        row["scratch_bytes_max_tile"] = max(row["scratch_bytes_max_tile"],
                                            int(count["scratch_bytes"]))
        for name, field in (("missing_by_caller", "sweep_missing_by_caller"),
                            ("fallback_by_caller", "sweep_exact_fallback_by_caller")):
            values = [int(value) for value in stage[field].split(",")]
            row[name] = [a + b for a, b in zip(row[name], values)]

    result_row = {
        "strate": strate, "workers": 8, "replicate": replicate,
        "elapsed_s": round(time.monotonic() - started, 3),
        "peak_tree_rss_kib": peak_tree_rss_kib,
        "by_lod": {str(lod): values for lod, values in sorted(per_lod.items())},
        "log": str(log.relative_to(ROOT)),
    }
    print("[L125PassageCounts] " + json.dumps(result_row, sort_keys=True), flush=True)
    return result_row


def main() -> int:
    if not DATA.is_file() or not BINARY.is_file():
        raise SystemExit("missing L125 passage-counter input")
    runs = [run(strate, replicate) for strate in range(4) for replicate in (1, 2)]
    (OUT / "summary.json").write_text(
        json.dumps({"runs": runs, "caller_names": CALLERS}, indent=2) + "\n",
        encoding="utf-8")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())