#!/usr/bin/env python3 """Summarize sampler closure, phase cost, and route census from L78 timing logs.""" from __future__ import annotations import csv import json import re import statistics from collections import defaultdict from pathlib import Path ROOT = Path(__file__).resolve().parents[4] TIMING = ROOT / "Saved/Linux/l78/timing" OUT = ROOT / "Saved/Linux/l78" SAMPLE = re.compile( r"^\[L77Sample\] pc=\S+ lod=(\d+) stage=(\S+) route=(\S+) " r"phase=(\S+) halo=(\d+) weight=(\d+)$", re.M) SAMPLER = re.compile( r"^\[L77Sampler\] action=stop .*?samples=(\d+) .*?lost=(\d+) " r"timer_overruns=(\d+).*?profiled_thread_cpu_s=([0-9.]+).*?" r"handler_cpu_ms=([0-9.]+) handler_share_pct=([0-9.]+) " r"observed_us_per_tick=([0-9.]+)", re.M) DISPATCH = re.compile( r"^\[WorldAssemblerL77Dispatch\] lod=(\d+) stage=(\S+) route=(\S+) " r"calls=(\d+) points=(\d+)$", re.M) LOD_COST = re.compile( r"^\[WorldAssemblerLODCostDetail\] lod=(\d+) tiles=(\d+).*?" r"mc_grid_ms=([0-9.]+).*?worker_ms_per_tile=([0-9.]+)", re.M) def analyze(row: dict) -> tuple[dict, list[dict], list[dict]]: path = ROOT / row["log"] content = path.read_text(encoding="utf-8", errors="replace") sampler_match = SAMPLER.search(content) if sampler_match is None: raise SystemExit(f"sampler summary missing from {path}") samples = [ (int(lod), stage, route, phase, int(halo), int(weight)) for lod, stage, route, phase, halo, weight in SAMPLE.findall(content) ] sample_count, lost, overruns = map(int, sampler_match.group(1, 2, 3)) profiled_cpu_s, handler_cpu_ms, handler_share_pct, period_us = map( float, sampler_match.group(4, 5, 6, 7)) period_s = period_us / 1_000_000.0 lod_costs = { int(lod): (int(tiles), float(grid_ms) / 1000.0, float(worker_ms) * int(tiles) / 1000.0) for lod, tiles, grid_ms, worker_ms in LOD_COST.findall(content) } dispatch = { (int(lod), stage, route): (int(calls), int(points)) for lod, stage, route, calls, points in DISPATCH.findall(content) } scope_metrics = [] for scope, predicate in (("LOD0-2", lambda lod: 0 <= lod <= 2), ("whole_view", lambda lod: 0 <= lod <= 8)): selected = [item for item in samples if predicate(item[0])] sampled_cpu_s = sum(item[5] for item in selected) * period_s sampled_grid_s = sum(item[5] for item in selected if item[1] == "grid_fill") * period_s sampled_passage_s = sum(item[5] for item in selected if item[3] == "passages") * period_s sampled_support_s = sum(item[5] for item in selected if item[3] == "support") * period_s direct_grid_s = sum(value[1] for lod, value in lod_costs.items() if predicate(lod)) direct_worker_s = sum(value[2] for lod, value in lod_costs.items() if predicate(lod)) scope_metrics.append({ "strate": row["strate"], "mode": row["mode"], "pair": row["pair"], "scope": scope, "sample_events": len(selected), "sampled_cpu_s": sampled_cpu_s, "sampled_grid_fill_s": sampled_grid_s, "direct_grid_fill_s": direct_grid_s, "grid_closure_pct": 100.0 * sampled_grid_s / direct_grid_s if direct_grid_s else 0.0, "sampled_passage_s": sampled_passage_s, "sampled_passage_pct": 100.0 * sampled_passage_s / sampled_cpu_s if sampled_cpu_s else 0.0, "sampled_support_s": sampled_support_s, "sampled_support_pct": 100.0 * sampled_support_s / sampled_cpu_s if sampled_cpu_s else 0.0, "direct_worker_cpu_s": direct_worker_s, }) route_metrics = [] scopes = (("LOD0-2", lambda lod: 0 <= lod <= 2), ("whole_view", lambda lod: 0 <= lod <= 8)) for scope, predicate in scopes: route_keys = { (item[1], item[2]) for item in samples if predicate(item[0]) } | { (stage, route) for (lod, stage, route) in dispatch if predicate(lod) } sampled_total = sum(item[5] for item in samples if predicate(item[0])) * period_s for stage, route in sorted(route_keys): weighted = sum(item[5] for item in samples if predicate(item[0]) and item[1] == stage and item[2] == route) calls = sum(value[0] for (lod, current_stage, current_route), value in dispatch.items() if predicate(lod) and current_stage == stage and current_route == route) points = sum(value[1] for (lod, current_stage, current_route), value in dispatch.items() if predicate(lod) and current_stage == stage and current_route == route) sampled_s = weighted * period_s route_metrics.append({ "strate": row["strate"], "mode": row["mode"], "pair": row["pair"], "scope": scope, "caller_stage": stage, "route": route, "calls": calls, "points": points, "sampled_cpu_s": sampled_s, "share_scope_sampled_cpu_pct": 100.0 * sampled_s / sampled_total if sampled_total else 0.0, }) summary = { "strate": row["strate"], "mode": row["mode"], "pair": row["pair"], "log": row["log"], "sampler_events": sample_count, "parsed_sampler_events": len(samples), "lost": lost, "timer_overruns": overruns, "profiled_thread_cpu_s": profiled_cpu_s, "handler_cpu_ms": handler_cpu_ms, "handler_share_pct": handler_share_pct, "observed_us_per_tick": period_us, } return summary, scope_metrics, route_metrics def write_csv(path: Path, rows: list[dict]) -> None: if not rows: return with path.open("w", encoding="utf-8", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) def main() -> int: timing = json.loads((TIMING / "summary.json").read_text(encoding="utf-8")) sampler_rows, scope_rows, route_rows = [], [], [] for row in timing["runs"]: sampler, scopes, routes = analyze(row) sampler_rows.append(sampler) scope_rows.extend(scopes) route_rows.extend(routes) write_csv(OUT / "sampler-census.csv", sampler_rows) write_csv(OUT / "closure-census.csv", scope_rows) write_csv(OUT / "route-census.csv", route_rows) averaged = [] for strate in range(4): for mode in ("before", "after"): samples = [row for row in scope_rows if row["strate"] == strate and row["mode"] == mode] for scope in ("LOD0-2", "whole_view"): values = [row for row in samples if row["scope"] == scope] averaged.append({ "strate": strate, "mode": mode, "scope": scope, "runs": len(values), "sampled_cpu_s_mean": statistics.mean(row["sampled_cpu_s"] for row in values), "sampled_grid_fill_s_mean": statistics.mean(row["sampled_grid_fill_s"] for row in values), "direct_grid_fill_s_mean": statistics.mean(row["direct_grid_fill_s"] for row in values), "grid_closure_pct_mean": statistics.mean(row["grid_closure_pct"] for row in values), "passage_s_mean": statistics.mean(row["sampled_passage_s"] for row in values), "passage_sample_pct_mean": statistics.mean( row["sampled_passage_pct"] for row in values), "support_s_mean": statistics.mean(row["sampled_support_s"] for row in values), "support_sample_pct_mean": statistics.mean( row["sampled_support_pct"] for row in values), "direct_worker_cpu_s_mean": statistics.mean(row["direct_worker_cpu_s"] for row in values), }) (OUT / "profile-summary.json").write_text( json.dumps({"sampler_runs": sampler_rows, "scope_summary": averaged}, indent=2) + "\n", encoding="utf-8") return 0 if __name__ == "__main__": raise SystemExit(main())