#!/usr/bin/env python3 """Turn L87 worker traces into tile and worker interval tables.""" from __future__ import annotations import argparse import csv import json import re import statistics from collections import defaultdict from pathlib import Path ROOT = Path(__file__).resolve().parents[4] DEFAULT_DIR = ROOT / "Saved/Linux/l87/timeline/final-phase" EVENT = re.compile(r"^\[CpuTileTimeline\] event=(\S+) (.*)$", re.M) COLD = re.compile(r"^\[ColdTimeline\] event=(\S+) edge=(\S+) at_ms=([0-9.]+) cpu_ms=([0-9.]+)$", re.M) NEAR = re.compile(r"^\[WorldAssemblerNearLODReady\] lods=0-2 .* cold_ready_s=([0-9.]+)$", re.M) PIPELINE = re.compile(r"^\[WorldAssemblerNearLODPipelineReady\] lods=0-2 .* cold_ready_s=([0-9.]+)$", re.M) WORKER_COUNT = re.compile(r"^\[HeadlessConfig\] .*\bworkers=(\d+)", re.M) FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)") KEY = re.compile(r"key=\((-?\d+),(-?\d+),(-?\d+),lod(-?\d+)\)") TILE_EVENTS = {"requested", "admitted", "started", "finished", "published", "drained"} def parse_event(match: re.Match[str]) -> dict: event, payload = match.groups() fields = dict(FIELDS.findall(payload)) key_match = KEY.search(payload) key = None if key_match: x, y, z, lod = (int(value) for value in key_match.groups()) key = (x, y, z, lod) fields["event"] = event fields["key"] = key fields["at_ms"] = float(fields["at_ms"]) fields["worker_id"] = int(fields["worker_id"]) return fields def clip_interval(start: float, end: float, window_start: float, window_end: float) -> tuple[float, float] | None: start, end = max(start, window_start), min(end, window_end) return (start, end) if end > start else None def merge_duration(intervals: list[tuple[float, float]]) -> float: if not intervals: return 0.0 intervals = sorted(intervals) total, lo, hi = 0.0, intervals[0][0], intervals[0][1] for start, end in intervals[1:]: if start <= hi: hi = max(hi, end) else: total += hi - lo lo, hi = start, end return total + hi - lo def analyze(log: Path) -> tuple[dict, list[dict], list[dict], list[dict]]: content = log.read_text(encoding="utf-8", errors="replace") events = [parse_event(match) for match in EVENT.finditer(content)] if not events: raise RuntimeError(f"no worker timeline events in {log}") cold = {(event, edge): (float(at), float(cpu)) for event, edge, at, cpu in COLD.findall(content)} requested = cold[("desired_tile_set", "requested")] done = cold[("lod0_2_done", "pipeline_drain")] window_start, window_end = requested[0], done[0] window_ms = window_end - window_start workers_match = WORKER_COUNT.search(content) if workers_match is None: raise RuntimeError(f"worker count missing in {log}") worker_count = int(workers_match.group(1)) near_ready = NEAR.search(content) pipeline_ready = PIPELINE.search(content) if near_ready is None or pipeline_ready is None: raise RuntimeError(f"near-ready marks missing in {log}") tiles: dict[tuple[int, int, int, int], dict] = defaultdict(dict) for event in events: key = event["key"] if key is not None and key[3] <= 2 and event["event"] in TILE_EVENTS: tiles[key][event["event"]] = event tile_rows = [] for key in sorted(tiles, key=lambda item: (item[3], item[2], item[1], item[0])): row = tiles[key] timestamps = {name: row.get(name, {}).get("at_ms") for name in ("requested", "admitted", "started", "finished", "published", "drained")} if any(timestamps[name] is None for name in ("requested", "admitted", "started", "finished", "published")): raise RuntimeError(f"incomplete per-tile lifecycle for {key} in {log}") tile_rows.append({ "log": log.name, "x": key[0], "y": key[1], "z": key[2], "lod": key[3], **timestamps, "request_to_admit_ms": round(timestamps["admitted"] - timestamps["requested"], 3), "admit_to_start_ms": round(timestamps["started"] - timestamps["admitted"], 3), "start_to_finish_ms": round(timestamps["finished"] - timestamps["started"], 3), "finish_to_publish_ms": round(timestamps["published"] - timestamps["finished"], 3), "request_to_publish_ms": round(timestamps["published"] - timestamps["requested"], 3), "worker_id": row["started"]["worker_id"], }) by_worker_busy: dict[int, list[tuple[float, float]]] = defaultdict(list) busy_rows = [] for key, row in tiles.items(): start = row.get("started") end = row.get("published") if start is None or end is None: continue clipped = clip_interval(start["at_ms"], end["at_ms"], window_start, window_end) if clipped is None: continue wid = start["worker_id"] by_worker_busy[wid].append(clipped) busy_rows.append({ "log": log.name, "interval": "busy", "worker_id": wid, "tile": f"({key[0]},{key[1]},{key[2]},lod{key[3]})", "start_ms": round(clipped[0], 3), "end_ms": round(clipped[1], 3), "duration_ms": round(clipped[1] - clipped[0], 3), }) idle_starts: dict[int, list[float]] = defaultdict(list) idle_intervals: dict[int, list[tuple[float, float]]] = defaultdict(list) idle_rows = [] for event in sorted(events, key=lambda item: item["at_ms"]): wid = event["worker_id"] if event["event"] == "idle_begin": idle_starts[wid].append(event["at_ms"]) elif event["event"] == "idle_end" and idle_starts[wid]: start = idle_starts[wid].pop(0) clipped = clip_interval(start, event["at_ms"], window_start, window_end) if clipped is not None: idle_intervals[wid].append(clipped) idle_rows.append({ "log": log.name, "interval": "idle", "worker_id": wid, "tile": "", "start_ms": round(clipped[0], 3), "end_ms": round(clipped[1], 3), "duration_ms": round(clipped[1] - clipped[0], 3), }) worker_summary_rows = [] busy_total = idle_total = 0.0 for wid in range(worker_count): busy_ms = merge_duration(by_worker_busy[wid]) idle_ms = merge_duration(idle_intervals[wid]) busy_total += busy_ms idle_total += idle_ms worker_summary_rows.append({ "log": log.name, "worker_id": wid, "window_ms": round(window_ms, 3), "busy_ms": round(busy_ms, 3), "idle_ms": round(idle_ms, 3), "other_ms": round(max(0.0, window_ms - busy_ms - idle_ms), 3), "busy_pct": round(100.0 * busy_ms / window_ms, 2), "idle_pct": round(100.0 * idle_ms / window_ms, 2), }) worker_interval_rows = busy_rows + idle_rows phase_names = ( ("shape_compile", "begin"), ("base_compile", "begin"), ("base_compile", "end"), ("shape_compile", "end"), ("lod0_1_admission", "begin"), ("lod0_1_admission", "end"), ("lod0_1_done", "publish_drained"), ("far_parent_build", "begin"), ("far_parent_build", "end"), ("lod2_queue", "begin"), ("lod2_queue", "end"), ("lod2_done", "pipeline_drain"), ("lod0_2_done", "pipeline_drain"), ) phases = {} for name, edge in phase_names: if (name, edge) in cold: at, cpu = cold[(name, edge)] phases[f"{name}_{edge}"] = {"wall_ms": round(at, 3), "cpu_ms": round(cpu, 3)} cpu_start = requested[1] cpu_end = done[1] lod01_end = cold[("lod0_1_done", "publish_drained")] lod2_end = cold[("lod2_done", "pipeline_drain")] last = {} for name in ("finished", "published", "drained"): latest = max((row.get(name) for row in tiles.values() if row.get(name)), key=lambda item: item["at_ms"], default=None) last[name] = ({"at_ms": round(latest["at_ms"], 3), "tile": f"{latest['key'][0]},{latest['key'][1]},{latest['key'][2]},lod{latest['key'][3]}"} if latest else None) lod01_done_ms = cold[("lod0_1_done", "publish_drained")][0] lod2_queue_begin_ms = cold[("lod2_queue", "begin")][0] lod2_done_ms = lod2_end[0] far_begin_ms = cold[("far_parent_build", "begin")][0] far_end_ms = cold[("far_parent_build", "end")][0] tile_latency = {} for lod in range(3): rows = [row for row in tile_rows if row["lod"] == lod] tile_latency[str(lod)] = {"tiles": len(rows)} for metric in ("request_to_admit_ms", "admit_to_start_ms", "start_to_finish_ms", "finish_to_publish_ms", "request_to_publish_ms"): values = sorted(row[metric] for row in rows) if values: tile_latency[str(lod)][metric] = { "median": round(statistics.median(values), 3), "p90": round(values[min(len(values) - 1, int(0.90 * len(values)))], 3), "p99": round(values[min(len(values) - 1, int(0.99 * len(values)))], 3), } summary = { "log": str(log.resolve().relative_to(ROOT)), "worker_count": worker_count, "requested_near_tiles": len(tile_rows), "lod0_tiles": sum(1 for key in tiles if key[3] == 0), "lod1_tiles": sum(1 for key in tiles if key[3] == 1), "lod2_tiles": sum(1 for key in tiles if key[3] == 2), "window_start_ms": round(window_start, 3), "window_end_ms": round(window_end, 3), "window_ms": round(window_ms, 3), "worker_pool_busy_pct": round(100.0 * busy_total / (worker_count * window_ms), 2), "worker_pool_explicit_idle_pct": round(100.0 * idle_total / (worker_count * window_ms), 2), "mean_busy_workers": round(busy_total / window_ms, 2), "mean_idle_workers": round(idle_total / window_ms, 2), "unaccounted_worker_pct": round(100.0 * max( 0.0, worker_count * window_ms - busy_total - idle_total) / (worker_count * window_ms), 2), "process_cpu_ms": round(cpu_end - cpu_start, 3), "process_cpu_core_equivalent": round((cpu_end - cpu_start) / window_ms, 2), "lod02_ready_s": float(near_ready.group(1)), "lod02_pipeline_ready_s": float(pipeline_ready.group(1)), "finish_to_pipeline_drain_ms": round( (float(pipeline_ready.group(1)) - float(near_ready.group(1))) * 1000.0, 3), "lod01_done_ms": round(lod01_done_ms, 3), "lod2_queue_begin_ms": round(lod2_queue_begin_ms, 3), "lod2_done_ms": round(lod2_done_ms, 3), "far_parent_build_ms": round(far_end_ms - far_begin_ms, 3), "far_parent_start_minus_lod02_done_ms": round(far_begin_ms - lod2_done_ms, 3), "lod2_queue_start_minus_far_parent_end_ms": round(lod2_queue_begin_ms - far_end_ms, 3), "lod2_queue_after_lod01_ms": round(lod2_queue_begin_ms - lod01_done_ms, 3), "lod2_queue_to_done_ms": round(lod2_done_ms - cold[("lod2_queue", "end")][0], 3), "tile_latency_by_lod": tile_latency, "last_tile_events": last, "phases": phases, } return summary, tile_rows, worker_interval_rows, worker_summary_rows def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--directory", type=Path, default=DEFAULT_DIR) args = parser.parse_args() logs = sorted(args.directory.glob("*-measured-r1.log")) summaries, all_tiles, all_worker_rows, all_worker_summary_rows = [], [], [], [] for log in logs: summary, tile_rows, worker_rows, worker_summary_rows = analyze(log) summaries.append(summary) all_tiles.extend(tile_rows) all_worker_rows.extend(worker_rows) all_worker_summary_rows.extend(worker_summary_rows) print("[L87TimelineAnalysis] " + json.dumps(summary, sort_keys=True), flush=True) if not summaries: raise SystemExit(f"no timeline traces found in {args.directory}") with (args.directory / "tile-lifecycle.csv").open("w", newline="", encoding="utf-8") as stream: fields = list(all_tiles[0]) writer = csv.DictWriter(stream, fieldnames=fields) writer.writeheader() writer.writerows(all_tiles) with (args.directory / "worker-intervals.csv").open("w", newline="", encoding="utf-8") as stream: fields = list(all_worker_rows[0]) writer = csv.DictWriter(stream, fieldnames=fields) writer.writeheader() writer.writerows(all_worker_rows) with (args.directory / "worker-utilization.csv").open("w", newline="", encoding="utf-8") as stream: fields = list(all_worker_summary_rows[0]) writer = csv.DictWriter(stream, fieldnames=fields) writer.writeheader() writer.writerows(all_worker_summary_rows) (args.directory / "timeline-analysis.json").write_text( json.dumps({"summaries": summaries}, indent=2) + "\n", encoding="utf-8") return 0 if __name__ == "__main__": raise SystemExit(main())