#!/usr/bin/env python3 """Export per-tile and per-frame L95 Unreal streaming timelines.""" 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] SAVED = ROOT / "Saved/Linux/l95" BEFORE = SAVED / "before/unreal-arrival" AFTER = SAVED / "unreal-arrival" HEADLESS = SAVED / "headless" OUT = SAVED / "timelines" MOVE_START_S = 20.0 HEADLESS_EVENT = re.compile(r"^\[CpuTileTimeline\] event=(\S+) (.*)$", re.M) HEADLESS_FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)") HEADLESS_KEY = re.compile(r"key=\((-?\d+),(-?\d+),(-?\d+),lod(-?\d+)\)") def read_jsonl(path: Path) -> list[dict]: rows = [] for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if line.startswith("{"): try: rows.append(json.loads(line)) except json.JSONDecodeError: continue return rows def sum3(event: dict, field: str) -> int: return sum(event.get(field, [0] * 9)[:3]) def first_settled_sample(events: list[dict]) -> dict: for event in events: if event.get("type") != "streaming_sample" or event.get("t_s", 0) >= MOVE_START_S: continue desired = sum3(event, "desired_by_lod") if desired and sum3(event, "settled_by_lod") >= desired: return event raise RuntimeError("no initial LOD0-2 settled sample") def write_csv(path: Path, rows: list[dict]) -> None: if not rows: return fields = list(rows[0]) with path.open("w", newline="", encoding="utf-8") as stream: writer = csv.DictWriter(stream, fieldnames=fields) writer.writeheader() writer.writerows(rows) def headless_tiles(path: Path) -> dict[tuple[int, int, int, int], dict]: content = path.read_text(encoding="utf-8", errors="replace") tiles: dict[tuple[int, int, int, int], dict] = {} for match in HEADLESS_EVENT.finditer(content): event, payload = match.groups() if event != "finished": continue fields = dict(HEADLESS_FIELDS.findall(payload)) key_match = HEADLESS_KEY.search(payload) if not key_match: continue x, y, z, lod = (int(value) for value in key_match.groups()) if lod > 2: continue key = (x, y, z, lod) tiles[key] = {"worker_id": int(fields["worker_id"]), "worker_ms": float(fields["worker_ms"]), "queued_ms": float(fields["queued_ms"]), "dispatch_ms": float(fields["dispatch_ms"]), "finish_ms": float(fields["finish_ms"])} return tiles def headless_lifecycle(path: Path) -> list[dict]: content = path.read_text(encoding="utf-8", errors="replace") tiles: dict[tuple[int, int, int, int], dict] = defaultdict(dict) for match in HEADLESS_EVENT.finditer(content): event, payload = match.groups() fields = dict(HEADLESS_FIELDS.findall(payload)) key_match = HEADLESS_KEY.search(payload) if not key_match: continue x, y, z, lod = (int(value) for value in key_match.groups()) if lod > 2 or event not in { "requested", "admitted", "started", "finished", "published", "drained" }: continue key = (x, y, z, lod) tiles[key][event] = fields rows = [] for key, events in sorted(tiles.items(), key=lambda item: (item[0][3], item[0][2], item[0][1], item[0][0])): at = {name: (float(events[name]["at_ms"]) if name in events else None) for name in ("requested", "admitted", "started", "finished", "published", "drained")} started = events.get("started", {}) finished = events.get("finished", {}) rows.append({ "run": path.stem, "x": key[0], "y": key[1], "z": key[2], "lod": key[3], **{f"{name}_ms": value for name, value in at.items()}, "request_to_admit_ms": round(at["admitted"] - at["requested"], 3) if at["admitted"] is not None and at["requested"] is not None else None, "admit_to_start_ms": round(at["started"] - at["admitted"], 3) if at["started"] is not None and at["admitted"] is not None else None, "worker_service_ms": float(finished.get("worker_ms", "nan")), "finish_to_publish_ms": round(at["published"] - at["finished"], 3) if at["published"] is not None and at["finished"] is not None else None, "publish_to_drain_ms": round(at["drained"] - at["published"], 3) if at["drained"] is not None and at["published"] is not None else None, "worker_id": int(started.get("worker_id", -1)), }) return rows def merge_seconds(intervals: list[tuple[float, float]], lo: float, hi: float) -> float: clipped = sorted((max(lo, a), min(hi, b)) for a, b in intervals if b > lo and a < hi) if not clipped: return 0.0 total = 0.0 start, end = clipped[0] for a, b in clipped[1:]: if a <= end: end = max(end, b) else: total += end - start start, end = a, b return total + end - start def analyze_trace(path: Path, phase: str, stratum: str) -> tuple[dict, list[dict], list[dict], dict]: events = read_jsonl(path) desired = next(e for e in events if e.get("type") == "desired_set") settled = first_settled_sample(events) first_admit = next(e for e in events if e.get("type") == "scheduler_admission" and e.get("desired_rebuilt") == 1) start, finish = float(desired["t_s"]), float(settled["t_s"]) cfg = next(e for e in events if e.get("type") == "worker_backend_config") requests = [e for e in events if e.get("type") == "request" and e.get("epoch") == 0 and e.get("level", 9) <= 2 and start <= e.get("t_s", -1) <= finish] tiles = [e for e in events if e.get("type") == "tile" and e.get("level", 9) <= 2 and e.get("t_s", 1e99) <= finish] by_ordinal = {e.get("request_ordinal"): e for e in tiles} collision_by_key: dict[tuple[int, int, int, int], list[dict]] = defaultdict(list) for e in events: if e.get("type") == "collision_ready": key = tuple(e.get("tile", [])) + (e.get("level", 0),) collision_by_key[key].append(e) elif e.get("type") == "collision_empty_classification": key = tuple(e.get("key", [])) + (e.get("lod", 0),) collision_by_key[key].append(e) used_collision: dict[tuple[int, int, int, int], float] = defaultdict(float) timeline = [] intervals_by_worker: dict[int, list[tuple[float, float]]] = defaultdict(list) admitted_keys = set() near_service = core_service = 0.0 for request in requests: tile = by_ordinal.get(request.get("request_ordinal")) if not tile: continue key = tuple(request["tile"]) + (request["level"],) admitted_keys.add(key) worker_start = tile.get("worker_start_t_s") worker_end = tile.get("worker_end_t_s") apply_t = tile.get("t_s") collision = None if apply_t is not None: for candidate in collision_by_key.get(key, []): candidate_t = float(candidate.get("t_s", -1)) if candidate_t >= float(apply_t) and candidate_t > used_collision[key]: collision = candidate used_collision[key] = candidate_t break queue_at = worker_end event_type = collision.get("type") if collision else ( "no_cook" if request["level"] > 0 else "pending_or_unobserved") timeline.append({ "phase": phase, "stratum": stratum, "run": path.stem, "x": key[0], "y": key[1], "z": key[2], "lod": key[3], "request_ordinal": request.get("request_ordinal"), "cohort_desired_t_s": round(start, 6), "admitted_t_s": request.get("t_s"), "worker_start_t_s": worker_start, "worker_end_t_s": worker_end, "result_queued_approx_t_s": queue_at, "applied_mesh_t_s": apply_t, "collision_event": event_type, "collision_ready_t_s": (collision.get("t_s") if collision and collision.get("type") == "collision_ready" else None), "collision_classified_t_s": (collision.get("t_s") if collision and collision.get("type") == "collision_empty_classification" else None), "worker_queue_s": tile.get("worker_queue_s"), "worker_task_s": tile.get("worker_task_s"), "result_queue_s": tile.get("result_queue_s"), "core_generate_s": tile.get("core_generate_s"), "mesh_copy_s": tile.get("mesh_copy_s"), "rmc_stream_build_s": tile.get("rmc_stream_build_s"), "apply_s": tile.get("apply_s"), "worker_thread_id": tile.get("worker_thread_id"), "empty": tile.get("empty"), }) if worker_start is not None and worker_end is not None: tid = int(tile.get("worker_thread_id", -1)) if tid >= 0: intervals_by_worker[tid].append((float(worker_start), float(worker_end))) near_service += float(tile.get("worker_task_s", 0.0)) core_service += float(tile.get("core_generate_s", 0.0)) sched_by_frame: dict[int, list[dict]] = defaultdict(list) apply_by_frame: dict[int, list[dict]] = defaultdict(list) queue_by_frame: dict[int, list[dict]] = defaultdict(list) for e in events: if not float(first_admit["t_s"]) <= e.get("t_s", -1) <= finish: continue if e.get("type") == "scheduler_admission": sched_by_frame[int(e.get("frame", -1))].append(e) elif e.get("type") == "apply_frame": apply_by_frame[int(e.get("frame", -1))].append(e) elif e.get("type") == "result_queue_frame": queue_by_frame[int(e.get("frame", -1))].append(e) frames = [] for frame in sorted(set(sched_by_frame) | set(apply_by_frame) | set(queue_by_frame)): sched, applies, queues = sched_by_frame[frame], apply_by_frame[frame], queue_by_frame[frame] submitted = [sum(e.get("submitted_by_lod", [0] * 9)[lod] for e in sched) for lod in range(9)] apply_results = [sum(e.get("apply_results_by_lod", [0] * 9)[lod] for e in applies) for lod in range(9)] meshes = [sum(e.get("meshes_by_lod", [0] * 9)[lod] for e in applies) for lod in range(9)] t_values = [e.get("t_s") for e in sched + applies + queues if e.get("t_s") is not None] frames.append({ "phase": phase, "stratum": stratum, "run": path.stem, "frame": frame, "t_s_first_event": min(t_values) if t_values else None, "scheduler_calls": len(sched), "admitted_lod0_2": sum(submitted[:3]), "admitted_by_lod": json.dumps(submitted, separators=(",", ":")), "running_workers_avg": (round(statistics.mean(e.get("running_workers", 0) for e in sched), 3) if sched else None), "apply_results_lod0_2": sum(apply_results[:3]), "meshes_applied_lod0_2": sum(meshes[:3]), "apply_ms": round(sum(e.get("apply_ms", 0.0) for e in applies), 4), "frame_ms": max((e.get("frame_ms", 0.0) for e in applies), default=None), "apply_budget_ms": max((e.get("apply_budget_ms", 0.0) for e in applies), default=None), "queued_results_max": max((e.get("queued_results", 0) for e in queues), default=None), "backlog_max": max((e.get("backlog_count", 0) for e in queues), default=None), }) worker_rows = [] for thread_id, intervals in sorted(intervals_by_worker.items()): busy = merge_seconds(intervals, float(first_admit["t_s"]), finish) window = max(0.0, finish - float(first_admit["t_s"])) worker_rows.append({ "phase": phase, "stratum": stratum, "run": path.stem, "worker_thread_id": thread_id, "near_task_busy_s": round(busy, 6), "no_recorded_near_task_s": round(max(0.0, window - busy), 6), "admission_to_settle_window_s": round(window, 6), "near_task_busy_pct": round(100.0 * busy / window, 2) if window else 0.0, "worker_task_count": sum(1 for e in tiles if int(e.get("worker_thread_id", -1)) == thread_id), }) summary = { "phase": phase, "stratum": stratum, "run": path.stem, "desired_t_s": start, "first_admission_t_s": float(first_admit["t_s"]), "settled_sample_t_s": finish, "admission_to_settle_s": finish - float(first_admit["t_s"]), "desired_counts_lod0_2": settled.get("desired_by_lod", [0] * 9)[:3], "settled_counts_lod0_2": settled.get("settled_by_lod", [0] * 9)[:3], "unique_admitted_keys": len(admitted_keys), "tile_events": len(timeline), "near_worker_s": near_service, "near_core_s": core_service, "effective_workers": cfg.get("effective_tile_slots"), "taskgraph_background_workers": cfg.get("taskgraph_background_threads"), "ready_queue_depth_cvar": cfg.get("ready_queue_depth_cvar"), "effective_ready_queue_depth": cfg.get("ready_queue_depth"), "total_admission_cap": next(e.get("total_admission_cap") for e in events if e.get("type") == "scheduler_admission"), "result_backlog_cap": cfg.get("streaming_result_backlog_cap"), "max_result_queue": max((r["queued_results_max"] or 0 for r in frames), default=0), "max_result_backlog": max((r["backlog_max"] or 0 for r in frames), default=0), "frame_count": len(frames), "max_admit_per_engine_frame": max((r["admitted_lod0_2"] for r in frames), default=0), "max_apply_meshes_per_engine_frame": max((r["meshes_applied_lod0_2"] for r in frames), default=0), "max_apply_ms": max((r["apply_ms"] for r in frames), default=0), "max_apply_budget_ms": max((r["apply_budget_ms"] or 0 for r in frames), default=0), "p95_frame_ms": (sorted(r["frame_ms"] for r in frames if r["frame_ms"] is not None) [int(.95 * (len([r for r in frames if r["frame_ms"] is not None]) - 1))] if any(r["frame_ms"] is not None for r in frames) else None), "max_frame_ms": max((r["frame_ms"] or 0 for r in frames), default=0), } return summary, timeline, frames, worker_rows def matched_worker_rows(trace: Path, phase: str, stratum: str, headless_runs: list[Path]) -> list[dict]: events = read_jsonl(trace) settled = first_settled_sample(events) finish = float(settled["t_s"]) requests = [e for e in events if e.get("type") == "request" and e.get("epoch") == 0 and e.get("level", 9) <= 2 and e.get("t_s", 1e99) <= finish] tile_events = [e for e in events if e.get("type") == "tile" and e.get("level", 9) <= 2 and e.get("t_s", 1e99) <= finish] tile_by_ordinal = {e.get("request_ordinal"): e for e in tile_events} request_keys = {tuple(e["tile"]) + (e["level"],) for e in requests} headless = [(path, headless_tiles(path)) for path in headless_runs] reference_keys = set(headless[0][1]) common = reference_keys & request_keys headless_s = [sum(row["worker_ms"] for key, row in tiles.items() if key in common) / 1000.0 for _, tiles in headless] matched = [tile_by_ordinal[e.get("request_ordinal")] for e in requests if e.get("request_ordinal") in tile_by_ordinal and tuple(e["tile"]) + (e["level"],) in common] total_worker = sum(float(e.get("worker_task_s", 0.0)) for e in matched) total_core = sum(float(e.get("core_generate_s", 0.0)) for e in matched) total_copy = sum(float(e.get("mesh_copy_s", 0.0)) for e in matched) total_rmc = sum(float(e.get("rmc_stream_build_s", 0.0)) for e in matched) return [{ "phase": phase, "stratum": stratum, "run": trace.stem, "headless_key_count": len(reference_keys), "common_key_count": len(common), "unreal_only_key_count": len(request_keys - reference_keys), "headless_only_key_count": len(reference_keys - request_keys), "unreal_task_count_on_common_keys": len(matched), "unreal_worker_seconds_on_common_keys": round(total_worker, 6), "unreal_core_seconds_on_common_keys": round(total_core, 6), "unreal_mesh_copy_seconds_on_common_keys": round(total_copy, 6), "unreal_rmc_stream_build_seconds_on_common_keys": round(total_rmc, 6), "unreal_worker_ms_per_common_key": round(total_worker * 1000.0 / len(common), 6) if common else None, "headless_worker_seconds_median_on_common_keys": round(statistics.median(headless_s), 6), "headless_worker_ms_per_common_key": round(statistics.median(headless_s) * 1000.0 / len(common), 6) if common else None, "headless_ready_s_median": round(statistics.median( float(re.search(r"cold_ready_s=([0-9.]+)", path.read_text( encoding="utf-8", errors="replace")).group(1)) for path in headless_runs), 6), }] def main() -> None: OUT.mkdir(parents=True, exist_ok=True) all_summaries, all_tiles, all_frames, all_workers = [], [], [], [] all_headless_tiles, all_matched_workers = [], [] headless_paths = { "FI": sorted(HEADLESS.glob("fi-plain-w12-r*.log")), "SW": sorted(HEADLESS.glob("sw-plain-w12-r*.log")), } for stratum, paths in headless_paths.items(): for path in paths: all_headless_tiles.extend(headless_lifecycle(path)) for phase, directory, patterns in ( ("before", BEFORE, ("after-fi-on-*.trace.jsonl", "after-sw-on-*.trace.jsonl")), ("after", AFTER, ("after-fi-rd64-bg1-boost1-*.trace.jsonl", "after-sw-rd64-bg1-boost1-*.trace.jsonl")), ): traces = sorted({p for pattern in patterns for p in directory.glob(pattern)}) for path in traces: stratum = "FI" if "fi-" in path.name else "SW" summary, tiles, frames, workers = analyze_trace(path, phase, stratum) all_summaries.append(summary) all_tiles.extend(tiles) all_frames.extend(frames) all_workers.extend(workers) all_matched_workers.extend(matched_worker_rows( path, phase, stratum, headless_paths[stratum])) print("[L95Timeline] " + json.dumps(summary, sort_keys=True), flush=True) write_csv(OUT / "unreal-tile-lifecycle.csv", all_tiles) write_csv(OUT / "unreal-per-frame.csv", all_frames) write_csv(OUT / "unreal-worker-occupancy.csv", all_workers) write_csv(OUT / "headless-tile-lifecycle.csv", all_headless_tiles) write_csv(OUT / "matched-worker-seconds.csv", all_matched_workers) (OUT / "unreal-run-summary.json").write_text( json.dumps(all_summaries, indent=2) + "\n", encoding="utf-8") if __name__ == "__main__": main()