VoxelForge/Standalone/Research/WorldAssemblerM3/Linux84/RunL84SerialFloor.py

#!/usr/bin/env python3
"""Interleaved W16 tile and shape-stage profiles before and after L84."""
from __future__ import annotations

import json
import re
import statistics
import subprocess
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[4]
OUT = ROOT / "Saved/Linux/l84/serial-floor-final"
DATA = ROOT / "Standalone/data/owner_world.json"
BINARIES = {
    "L83": ROOT / "Saved/Linux/l83/profile-bin/VoxelForgeCoreHeadless-l83-profile-clang19",
    "L84": ROOT / "Saved/Linux/l84/profile-bin/VoxelForgeCoreHeadless-l84-profile-clang19",
}
STAMPS = {
    "L83": "08af4e98c86a69477cc79e21463d8c93bcd11ebe0f077d7c11519031d9141fa9",
    "L84": "c433a039b0255ce3c4794bf20313b8a5a11442feeb07b6555f585688949741a5",
}
CAMERAS = {
    1: (25.8369140625, 3.19921875, -737.0),
    2: (19.166015625, -14.2001953125, -753.3740234375),
}
NAMES = {1: "FI", 2: "TN"}
EXPECTED_NEAR = {1: 1216, 2: 1344}
EXPECTED_WHOLE = {1: 2249, 2: 2233}
STAMP = re.compile(r"^\[VoxelForgeSourceStamp\] sha256=(\S+)$", re.M)
NEAR = re.compile(r"^\[WorldAssemblerNearLODReady\] lods=0-2 tiles=(\d+)/(\d+) cold_ready_s=([0-9.]+)$", re.M)
PIPELINE = re.compile(r"^\[StreamPipeline\] (.*)$", re.M)
FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)")
SHAPE_FIELDS = re.compile(r"([a-zA-Z0-9_+\-]+)=([0-9.]+)")
TIMELINE = re.compile(
    r"^\[CpuTileTimeline\] event=finish key=\(([-0-9]+),([-0-9]+),([-0-9]+),lod([0-9]+)\) (.*)$", re.M)
STAGES = re.compile(
    r"^\[CpuTileStages\] key=\(([-0-9]+),([-0-9]+),([-0-9]+),lod([0-9]+)\) (.*)$", re.M)
SHAPE_STAGE = re.compile(r"^\[ShapeStage\] (.*)$", re.M)
FAR_STAGE = re.compile(r"^\[FarShapeStage\] (.*)$", re.M)


def run(strate: int, version: str, rep: int) -> dict:
    out_dir = OUT / version.lower()
    out_dir.mkdir(parents=True, exist_ok=True)
    log = out_dir / f"{NAMES[strate].lower()}-w16-r{rep}.log"
    camera = ",".join(f"{v:.12f}" for v in CAMERAS[strate]) + ",0,0"
    command = [
        "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--",
        "env", "VOXELFORGE_L81_PASSAGE_BOUND_PADDING_VOXELS=0", "timeout", "1740s",
        str(BINARIES[version]), str(DATA), "--world-assembler-prototype",
        "--stream-benchmark", "--stream-pipeline", "--workers=16", "--clip-radius=3",
        "--max-clip-level=8", "--max-level=8", f"--camera={camera}",
        "--profile-lod-work", "--instrument-tile-drain", "--stream-scheduler-policy=3",
    ]
    started = time.monotonic()
    with log.open("w", encoding="utf-8") as stream:
        result = subprocess.run(command, cwd=ROOT, stdout=stream,
                                stderr=subprocess.STDOUT, check=False, timeout=1740)
    output = log.read_text(encoding="utf-8", errors="replace")
    source = STAMP.search(output)
    near = NEAR.search(output)
    pipeline = PIPELINE.search(output)
    if result.returncode or source is None or near is None or pipeline is None:
        raise RuntimeError(f"profile run failed: exit={result.returncode} {log}")
    if source.group(1) != STAMPS[version]:
        raise RuntimeError(f"source stamp mismatch in {log}: {source.group(1)}")
    near_done, near_total = int(near.group(1)), int(near.group(2))
    p = dict(FIELDS.findall(pipeline.group(1)))
    whole_done, whole_total = (int(value) for value in p["whole_tiles"].split("/"))
    if near_total != EXPECTED_NEAR[strate] or near_done != near_total:
        raise RuntimeError(f"incomplete near tiles in {log}: {near_done}/{near_total}")
    if whole_total != EXPECTED_WHOLE[strate] or whole_done != whole_total:
        raise RuntimeError(f"incomplete whole view in {log}: {whole_done}/{whole_total}")

    stage_rows = {}
    for match in STAGES.finditer(output):
        key = tuple(int(value) for value in match.group(1, 2, 3, 4))
        stage_rows[key] = dict(FIELDS.findall(match.group(5)))
    tiles = []
    for match in TIMELINE.finditer(output):
        key = tuple(int(value) for value in match.group(1, 2, 3, 4))
        fields = dict(FIELDS.findall(match.group(5)))
        if key[3] > 2:
            continue
        stage = stage_rows.get(key, {})
        tiles.append({
            "key": key, "lod": key[3], "service_ms": float(fields["service_ms"]),
            "queued_ms": float(fields["queued_ms"]), "dispatch_ms": float(fields["dispatch_ms"]),
            "finish_ms": float(fields["finish_ms"]), "priority_rank": int(fields["priority_rank"]),
            "queue_depth": int(fields["queue_depth"]), "active_workers": int(fields["active_workers"]),
            "samples": int(stage.get("samples", "0")),
            "candidate_cells": int(stage.get("candidate_cells", "0")),
            "surface_cells": int(stage.get("surface_cells", "0")),
            "tile_class": stage.get("tile_class", "unknown"),
            "mc_grid_ms": float(stage.get("mc_grid_ms", "0")),
            "mc_cell_ms": float(stage.get("mc_cell_ms", "0")),
            "mc_gradient_ms": float(stage.get("mc_gradient_ms", "0")),
            "mc_skirt_ms": float(stage.get("mc_skirt_ms", "0")),
        })
    if len(tiles) != near_total:
        raise RuntimeError(f"incomplete near tile timeline in {log}: {len(tiles)}/{near_total}")
    by_lod = {}
    for lod in range(3):
        subset = [tile for tile in tiles if tile["lod"] == lod]
        services = [tile["service_ms"] for tile in subset]
        by_lod[lod] = {
            "tiles": len(subset), "p50_service_ms": statistics.median(services),
            "p90_service_ms": sorted(services)[int((len(services) - 1) * 0.90)],
            "p95_service_ms": sorted(services)[int((len(services) - 1) * 0.95)],
            "max_service_ms": max(services),
        }
    slow = sorted(tiles, key=lambda tile: tile["service_ms"], reverse=True)[:12]
    stages = [dict(SHAPE_FIELDS.findall(match.group(1))) for match in SHAPE_STAGE.finditer(output)]
    far_stages = [dict(FIELDS.findall(match.group(1))) for match in FAR_STAGE.finditer(output)]
    far_shape_seconds = (float(p["far_shape"].removesuffix("s"))
                         if "far_shape" in p else
                         (float(far_stages[-1]["total"]) / 1000.0 if far_stages else 0.0))
    row = {
        "version": version, "strate": strate, "name": NAMES[strate], "replicate": rep,
        "workers": 16, "exit_code": result.returncode,
        "wall_s": round(time.monotonic() - started, 3),
        "near_ready_s": float(near.group(3)), "pipeline_shape_s": float(p["shape"].removesuffix("s")),
        "pipeline_far_shape_s": far_shape_seconds,
        "shape_stages": stages, "far_shape_stages": far_stages,
        "tile_count": len(tiles), "by_lod": by_lod, "slow_tiles": slow,
        "source_stamp": source.group(1), "log": str(log.relative_to(ROOT)),
    }
    print("[L84SerialFloorRun] " + json.dumps(row, sort_keys=True), flush=True)
    return row


def main() -> int:
    for binary in BINARIES.values():
        if not binary.is_file():
            raise SystemExit(f"missing binary: {binary}")
    rows = []
    orders = ((1, "L83"), (1, "L84"), (2, "L84"), (2, "L83"))
    for rep in range(1, 4):
        order = orders if rep % 2 else tuple(reversed(orders))
        for strate, version in order:
            rows.append(run(strate, version, rep))
            (OUT / "runs-partial.json").write_text(
                json.dumps(rows, indent=2) + "\n", encoding="utf-8")
    summary = []
    for strate in (1, 2):
        for version in ("L83", "L84"):
            subset = [row for row in rows
                      if row["strate"] == strate and row["version"] == version]
            item = {"strate": strate, "name": NAMES[strate], "version": version,
                    "runs": [row["log"] for row in subset]}
            for metric in ("near_ready_s", "pipeline_shape_s", "pipeline_far_shape_s"):
                values = [row[metric] for row in subset]
                item[f"{metric}_runs"] = values
                item[f"{metric}_median"] = statistics.median(values)
                item[f"{metric}_min"] = min(values)
                item[f"{metric}_max"] = max(values)
            item["shape_stage_medians_ms"] = {}
            for stage, key in (("local_shapes", "local-shapes"),
                               ("intervals_signs", "intervals+signs"),
                               ("surface_band", "surface-band"),
                               ("far_parents", "far-parents")):
                values = [float(shape_row[key]) for row in subset
                          for shape_row in row["shape_stages"] if key in shape_row]
                item["shape_stage_medians_ms"][stage] = (
                    statistics.median(values) if values else None)
            slow_per_run = [tile for row in subset for tile in row["slow_tiles"]]
            item["slowest_tiles_all_runs"] = slow_per_run
            summary.append(item)
    (OUT / "summary.json").write_text(
        json.dumps({"runs": rows, "summary": summary}, indent=2) + "\n",
        encoding="utf-8")
    return 0


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