Fr0zkaandClaude Opus 5.5
5389abed2e
Unreal per-tile timeline: per-tile Core work is not slower than headless (FI 11.2 vs 16.2 ms per shared key); the gap was tick-paced admission with a small queue (FI ~5.2 of 8 slots busy). voxel.StreamingReadyQueueDepth default 16 -> 64 (effective 32 with 8 slots, total cap 40); result backlog cap 256 unchanged. First admission -> LOD0-2 settled: FI 2.72 -> 2.10 s, SW 2.37 -> 1.31 s (3 of 4 runs; one 6.9 s outlier re-admitting 3 LOD0 keys, cause open). apply_ms max 4.2 ms (14 ms budget). Frame-interval max 118-145 ms both before and after (open). World unchanged. Fast tier PASS; Linux UE host builds. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_016dhxM43dnWvUu2B5Ph1iG1
380 lines
19 KiB
Python
380 lines
19 KiB
Python
#!/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()
|