Fr0zkaandClaude Opus 5.5
0fe7f371f7
L86's switch left ApplyResolvedStratePassageCarves running (near-tile service unchanged at ~21 worker-s), so its "no change" ceiling was wrong. Corrected (bands + carves off): FI LOD0-2 2.34 -> 0.66 s, whole 3.47 -> 1.49 s @W12; TN 2.27 -> 0.74 / 3.28 -> 1.52 s. Near-tile service 15.6-22.5 -> 2.6-3.8 worker-s. Critical chain: shape/base compile -> LOD0-1 -> far-parent build (107-153 ms, pipeline thread) -> LOD2. Workers 81-85% busy in the direct window. No production scheduling change; hashes equal L85. Fast tier PASS. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016dhxM43dnWvUu2B5Ph1iG1
272 lines
13 KiB
Python
272 lines
13 KiB
Python
#!/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())
|