#!/usr/bin/env python3 """Three-run L80 movement protocol for headless and Unreal, policy 3 only.""" from __future__ import annotations import argparse import csv import json import math import re import statistics import subprocess import time from pathlib import Path ROOT = Path(__file__).resolve().parents[4] HOST = Path("/home/jahni/VFHost") EDITOR = Path("/home/jahni/UnrealEngine/Engine/Binaries/Linux/UnrealEditor-Cmd") DATA = ROOT / "Standalone/data/owner_world.json" HEADLESS = { "baseline": ROOT / "Saved/Linux/l108/bin/VoxelForgeCoreHeadless-l108-baseline-clang", "after": ROOT / "Saved/Linux/l108/bin/VoxelForgeCoreHeadless-l108-after-clang", } OUT = ROOT / "Saved/Linux/l108/movement" HOST_OUT = HOST / "Plugins/VoxelForge/Saved/Linux/l108/movement" SPAWN_M = (19.166015625, -14.2001953125, -753.3740234375) SPEEDS = {"walk": 1.4, "fast": 30.0} SPEED_CM_S = {"walk": 140.0, "fast": 3000.0} TRACE_EVENT = re.compile(r"^\{.*\}$") HEADLESS_SAMPLE = re.compile(r"^\[L79MovingSample\] (.*)$", re.M) HEADLESS_SUMMARY = re.compile(r"^\[L79MovingSummary\] (.*)$", re.M) FIELDS = re.compile(r"([a-zA-Z0-9_]+)=([^ ]+)") def p99(values: list[float]) -> float | None: if not values: return None values.sort() return values[max(0, math.ceil(0.99 * len(values)) - 1)] def mean(values: list[float]) -> float | None: return sum(values) / len(values) if values else None def parse_events(path: Path) -> list[dict]: events = [] if not path.is_file(): return events for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if not TRACE_EVENT.match(line): continue try: events.append(json.loads(line)) except json.JSONDecodeError: pass return events def summarize_rows(rows: list[dict], moving_key: str = "move") -> dict: moving = [r for r in rows if r["phase"] == moving_key] dwell = [r for r in rows if r["phase"] == "dwell"] dwell_end = dwell[-1] if dwell else {} full_dwell = next((r for r in dwell if r["near_ready_pct"] >= 99.999 and r["far_hole_pct"] <= 0.001), None) return { "moving_samples": len(moving), "dwell_samples": len(dwell), "near_ready_moving_mean_pct": mean([r["near_ready_pct"] for r in moving]), "lod0_ready_moving_mean_pct": mean([r["lod0_ready_pct"] for r in moving if "lod0_ready_pct" in r]), "lod1_ready_moving_mean_pct": mean([r["lod1_ready_pct"] for r in moving if "lod1_ready_pct" in r]), "lod2_ready_moving_mean_pct": mean([r["lod2_ready_pct"] for r in moving if "lod2_ready_pct" in r]), "walkable_ring_moving_mean_pct": mean([r["walkable_ring_ready_pct"] for r in moving]), "far_holes_moving_mean_pct": mean([r["far_hole_pct"] for r in moving]), "update_work_p99_ms": p99([float(r["queue_update_ms"]) for r in moving if "queue_update_ms" in r]), "update_work_max_ms": max([float(r["queue_update_ms"]) for r in moving if "queue_update_ms" in r], default=None), "near_ready_dwell_end_pct": dwell_end.get("near_ready_pct"), "walkable_ring_dwell_end_pct": dwell_end.get("walkable_ring_ready_pct"), "far_holes_dwell_end_pct": dwell_end.get("far_hole_pct"), "dwell_complete_after_stop_s": full_dwell.get("since_stop_s") if full_dwell else None, } def run_headless(phase: str, speed_name: str, repetition: int) -> dict: label = f"{speed_name}-r{repetition}" folder = OUT / phase / "headless" folder.mkdir(parents=True, exist_ok=True) log = folder / f"{label}.log" camera = ",".join(f"{v:.12f}" for v in SPAWN_M) + ",0,0" command = [ "taskset", "-c", "0-11", "prlimit", "--as=6442450944", "--", "timeout", "1740s", str(HEADLESS[phase]), str(DATA), "--world-assembler-prototype", "--stream-benchmark", "--stream-pipeline", "--workers=8", "--clip-radius=3", "--max-clip-level=8", "--max-level=8", f"--camera={camera}", "--profile-lod-work", "--stream-scheduler-policy=3", f"--moving-camera-speed-mps={SPEEDS[speed_name]:.1f}", "--moving-camera-duration-s=12", "--moving-camera-dwell-s=5", ] started = time.monotonic() with log.open("w", encoding="utf-8") as stream: result = subprocess.run(command, cwd=ROOT, stdout=stream, stderr=subprocess.STDOUT, check=False, timeout=1740) content = log.read_text(encoding="utf-8", errors="replace") rows = [] for match in HEADLESS_SAMPLE.finditer(content): row = dict(FIELDS.findall(match.group(1))) row.update({key: float(row[key]) for key in ( "t_s", "near_ready_pct", "walkable_ring_ready_pct", "far_hole_pct", "queue_update_ms")}) row["phase"] = row["phase"] row["walkable_ring_ready_pct"] = float(row.get("walkable_ring_ready_pct", 0.0)) row["since_stop_s"] = max(0.0, float(row["t_s"]) - 12.0) rows.append(row) summary_match = HEADLESS_SUMMARY.search(content) engine_summary = dict(FIELDS.findall(summary_match.group(1))) if summary_match else {} if result.returncode or not rows or int(engine_summary.get("generation_failures", "-1")) != 0: raise RuntimeError(f"headless movement run failed: {log}; exit={result.returncode}") info = { "phase": phase, "platform": "headless", "speed": speed_name, "speed_mps": SPEEDS[speed_name], "repetition": repetition, "exit_code": result.returncode, "wall_s": round(time.monotonic() - started, 3), "log": str(log.relative_to(ROOT)), **summarize_rows(rows), } return info def run_unreal(phase: str, speed_name: str, repetition: int) -> tuple[dict, list[dict], list[dict]]: label = f"{speed_name}-r{repetition}" folder = OUT / phase / "unreal" host_folder = HOST_OUT / phase / "unreal" folder.mkdir(parents=True, exist_ok=True) trace = folder / f"{label}.trace.jsonl" abslog = folder / f"{label}.log" stdout = folder / f"{label}.stdout.log" export = HOST / "Plugins/VoxelForge/Saved/Linux/l108/output" / f"{phase}-{label}" userdir = HOST / "Saved/Linux/l108/user" / f"{phase}-{label}" zendata = HOST / "Saved/Linux/l108/zen" / f"{phase}-{label}" ddc = HOST / "Saved/Linux/l108/DDC" for path in (export, userdir, zendata, ddc): path.mkdir(parents=True, exist_ok=True) spawn_cm = ",".join(f"{value * 100.0:.3f}" for value in SPAWN_M) command = [ str(EDITOR), str(HOST / "VFHost.uproject"), "-nullrhi", "-unattended", "-nop4", "-nosplash", "-NoSound", "-corelimit=12", "-NoZenAutoLaunch", "-notraceserver", "-nocrashreports", "-DisablePlugins=AndroidFileServer", "-ddc=NoZenLocalFallback", f"-userdir={userdir}", f"-ZenDataPath={zendata}", f"-LocalDataCachePath={ddc}", "-run=VoxelForgeExplore", "-seed=0", "-archetype=TunnelNetwork", "-slot=0", "-modes=export", "-gameconfig=1", "-coreworldtest=1", "-ownerfourstrateplan=1", "-corevoxelsizecm=25", "-coreislandtest=0", "-corefallbacktest=0", "-coreofftest=0", "-coreplantest=0", "-settingsref=/Game/VoxelForge/DA_Settings.DA_Settings", "-corehybridmesher=0", f"-out={export}", "-voxel.StreamingCoarseFirst=1", "-voxel.StreamingDynamicNearReserve=1", "-voxel.StreamingLegacyPendingResultCap=0", "-voxel.CoreRegionConcurrentAdmission=1", "-voxel.StreamingResultBacklogCap=256", "-voxel.StreamingReadyQueueDepth=8", "-voxel.StreamingBackgroundWorkerCap=1", "-voxel.CoreColumnCacheDirectMapped=0", "-voxel.CoreMCGradientMemoization=1", "-voxel.StreamingCollisionLookahead=1", "-voxel.TestMove=1", f"-voxel.TestMoveSpeedCmPerSecond={SPEED_CM_S[speed_name]:.1f}", "-voxel.TestMoveStartSeconds=6.0", "-voxel.TestMoveDurationSeconds=12.0", "-voxel.TestExitSeconds=23.0", "-voxel.TestMoveDirection=1,0,0", "-voxel.StreamingSchedulerPolicy=3", "-voxel.TestExitOnSteady=0", "-coreworldstreamtrace=1", f"-voxel.StartupTraceSpawnCm={spawn_cm}", "-voxel.StartupTraceUntilEndPlay=1", "-voxel.StartupTraceThroughCrossing=1", "-voxel.StartupTraceMaxTiles=16384", f"-voxel.StartupTraceFile={trace}", f"-abslog={abslog}", ] wrapped = ["taskset", "-c", "0-11", "timeout", "1740s", *command] started = time.monotonic() with stdout.open("w", encoding="utf-8") as stream: result = subprocess.run(wrapped, cwd=ROOT, stdout=stream, stderr=subprocess.STDOUT, check=False, timeout=1740) events = parse_events(trace) samples = [event for event in events if event.get("type") == "streaming_sample"] move_samples = [event for event in samples if float(event.get("speed_cm_s", 0.0)) >= 50.0] move_start_t = min((float(event.get("t_s", 0.0)) for event in move_samples), default=float("inf")) stop = next((event for event in events if event.get("type") == "streaming_test_move_stopped"), None) stop_t = float(stop.get("t_s", 0.0)) if stop else float("inf") dwell_samples = [event for event in samples if float(event.get("t_s", 0.0)) >= stop_t and float(event.get("speed_cm_s", 0.0)) < 50.0] rows = [] for event in samples: desired = event.get("desired_by_lod", [0] * 9) resident = event.get("resident_by_lod", [0] * 9) covered = event.get("covered_by_lod", resident) desired_near = sum(desired[:3]) resident_near = sum(resident[:3]) desired_far = sum(desired[3:]) covered_far = sum(covered[3:]) event_time = float(event.get("t_s", 0.0)) speed = float(event.get("speed_cm_s", 0.0)) row_phase = "move" if speed >= 50.0 else ( "dwell" if event_time >= stop_t else "arrival") ring_pct = 100.0 * float(event.get("walkable_ready_tiles", 0)) / max( 1.0, float(event.get("walkable_required_tiles", 9))) rows.append({ "t_s": event_time, "speed_cm_s": speed, "phase": row_phase, "since_stop_s": max(0.0, event_time - stop_t) if row_phase == "dwell" else None, "near_ready_pct": 100.0 * resident_near / max(1, desired_near), "lod0_ready_pct": 100.0 * resident[0] / max(1, desired[0]), "lod1_ready_pct": 100.0 * resident[1] / max(1, desired[1]), "lod2_ready_pct": 100.0 * resident[2] / max(1, desired[2]), "walkable_ring_ready_pct": ring_pct, "far_hole_pct": 100.0 * (desired_far - covered_far) / max(1, desired_far), "near_desired": desired_near, "near_resident": resident_near, "walkable_ready_tiles": event.get("walkable_ready_tiles"), "epoch": event.get("epoch"), }) if len(move_samples) < 40 or len(dwell_samples) < 16: raise RuntimeError(f"incomplete Unreal movement trace: {trace}; " f"move={len(move_samples)} dwell={len(dwell_samples)} " f"exit={result.returncode}") frame_samples = [event for event in events if event.get("type") == "apply_frame" and move_start_t <= float(event.get("t_s", 0.0)) < stop_t] tick_samples = [event for event in events if event.get("type") == "game_thread_work_frame" and move_start_t <= float(event.get("t_s", 0.0)) < stop_t] tick_ms = [float(event.get("tick_ms", 0.0)) for event in tick_samples] update_ms = [float(event.get("update_chunks_ms", 0.0)) for event in tick_samples] drain_ms = [float(event.get("pending_drain_ms", 0.0)) for event in tick_samples] apply_ms = [float(event.get("apply_ms", 0.0)) for event in frame_samples] gate_epochs = [event for event in events if event.get("type") == "walkable_gate" and move_start_t <= float(event.get("t_s", 0.0)) < stop_t] summary = summarize_rows(rows) summary.update({ "game_thread_tick_p99_ms": p99(tick_ms), "game_thread_tick_max_ms": max(tick_ms, default=None), "stream_update_p99_ms": p99(update_ms), "stream_update_max_ms": max(update_ms, default=None), "pending_drain_p99_ms": p99(drain_ms), "pending_drain_max_ms": max(drain_ms, default=None), "tile_apply_p99_ms": p99(apply_ms), "tile_apply_max_ms": max(apply_ms, default=None), "apply_budget_ms": max((float(event.get("apply_budget_ms", 0.0)) for event in frame_samples), default=None), "apply_budget_hit_frames": sum(int(event.get("time_limit_reached", 0)) for event in frame_samples), "desired_epochs_during_move": len({event.get("desired_epoch") for event in gate_epochs}), "desired_centers_during_move": len({tuple(event.get("support_center", [])) for event in gate_epochs}), }) info = { "phase": phase, "platform": "unreal", "speed": speed_name, "speed_mps": SPEEDS[speed_name], "repetition": repetition, "exit_code": result.returncode, "wall_s": round(time.monotonic() - started, 3), "moving_samples": len(move_samples), "dwell_samples": len(dwell_samples), "trace": str(trace.relative_to(ROOT)), "stdout": str(stdout.relative_to(ROOT)), **summary, } return info, events, rows def write_tile_timelines(phase: str, unreal_runs: list[tuple[dict, list[dict]]]) -> dict: path = OUT / phase / "tile-timelines.csv" path.parent.mkdir(parents=True, exist_ok=True) columns = ["speed", "repetition", "ordinal", "tile", "level", "priority", "epoch", "desired_epoch", "in_move_window", "request_s", "cancel_s", "worker_end_s", "generated", "aborted", "obsolete", "applied_s", "request_to_apply_s", "worker_queue_s", "worker_task_s", "result_queue_s", "discard_s", "discard_reason", "ahead_m_at_request", "request_to_worker_end_s", "cancel_before_worker_end", "out_of_range_at_worker_end"] stale_by_run = {} with path.open("w", newline="", encoding="utf-8") as stream: writer = csv.DictWriter(stream, fieldnames=columns) writer.writeheader() for run, events in unreal_runs: requests, cancelled, finished, applied, discarded = {}, {}, {}, {}, {} samples = sorted((event for event in events if event.get("type") == "streaming_sample"), key=lambda event: float(event.get("t_s", 0.0))) move_samples = [event for event in samples if float(event.get("speed_cm_s", 0.0)) >= 50.0] move_start_s = min((float(event.get("t_s", 0.0)) for event in move_samples), default=float("inf")) move_stop = next((event for event in events if event.get("type") == "streaming_test_move_stopped"), None) move_stop_s = float(move_stop.get("t_s", float("inf"))) if move_stop else float("inf") def pawn_x_at(t_s: float) -> float | None: if not samples: return None before = None for sample in samples: sample_t = float(sample.get("t_s", 0.0)) if sample_t >= t_s: if before is None: return float(sample.get("position_cm", [0.0])[0]) / 100.0 t0 = float(before.get("t_s", 0.0)) t1 = sample_t x0 = float(before.get("position_cm", [0.0])[0]) / 100.0 x1 = float(sample.get("position_cm", [0.0])[0]) / 100.0 alpha = 0.0 if t1 <= t0 else (t_s - t0) / (t1 - t0) return x0 + alpha * (x1 - x0) before = sample return float(samples[-1].get("position_cm", [0.0])[0]) / 100.0 for event in events: ordinal = event.get("request_ordinal") if ordinal is None: continue ordinal = int(ordinal) kind = event.get("type") if kind == "request": requests[ordinal] = event elif kind == "tile_cancelled": cancelled[ordinal] = event elif kind == "tile_worker_end": finished[ordinal] = event elif kind == "tile": applied[ordinal] = event elif kind == "tile_discard": discarded[ordinal] = event generated_l0 = 0 stale_l0 = 0 out_of_range_l0 = 0 l0_cancelled_while_running = 0 ahead_distances = [] gate_epochs = sorted((event for event in events if event.get("type") == "walkable_gate"), key=lambda event: float(event.get("t_s", 0.0))) def desired_epoch_at(t_s: float) -> int | None: previous = [event for event in gate_epochs if float(event.get("t_s", 0.0)) <= t_s] return int(previous[-1]["desired_epoch"]) if previous else None for ordinal, request in requests.items(): key = request.get("tile", [None, None, None]) level = int(request.get("level", -1)) cancel = cancelled.get(ordinal) end = finished.get(ordinal) tile = applied.get(ordinal) discard = discarded.get(ordinal) request_t = float(request.get("t_s", 0.0)) end_t = float(end.get("t_s", 0.0)) if end else None cancel_t = float(cancel.get("t_s", 0.0)) if cancel else None # LOD0 clip tiles are 8 m across; the live x-range is [center-4, center+3]. if end and int(end.get("generated", 0)) == 1 and end_t is not None: finish_x = pawn_x_at(end_t) center_x = math.floor(finish_x / 8.0) if finish_x is not None else 0 physically_out = bool(level == 0 and finish_x is not None and key[0] is not None and (int(key[0]) < center_x - 4 or int(key[0]) > center_x + 3)) else: physically_out = False in_move_window = bool(end_t is not None and move_start_s <= end_t < move_stop_s) if level == 0 and end and int(end.get("generated", 0)) == 1 and in_move_window: generated_l0 += 1 stale_l0 += int(cancel_t is not None and cancel_t < end_t) out_of_range_l0 += int(physically_out) if level == 0 and end and int(end.get("aborted", 0)) == 1 \ and cancel_t is not None and end_t is not None and cancel_t < end_t: l0_cancelled_while_running += 1 pawn_x = pawn_x_at(request_t) extent_m = 8.0 * (1 << max(0, level)) tile_center_m = (float(key[0]) + 0.5) * extent_m if key[0] is not None else None ahead_m = (tile_center_m - pawn_x) if tile_center_m is not None and pawn_x is not None else None if level == 0 and move_start_s <= request_t < move_stop_s and ahead_m is not None: ahead_distances.append(ahead_m) writer.writerow({ "speed": run["speed"], "repetition": run["repetition"], "ordinal": ordinal, "tile": ",".join(str(v) for v in key), "level": level, "priority": request.get("priority"), "epoch": request.get("epoch"), "desired_epoch": request.get("desired_epoch", desired_epoch_at(request_t)), "in_move_window": int(move_start_s <= request_t < move_stop_s), "request_s": request_t, "cancel_s": cancel_t, "worker_end_s": end_t, "generated": end.get("generated") if end else None, "aborted": end.get("aborted") if end else None, "obsolete": end.get("obsolete") if end else None, "applied_s": float(tile.get("t_s", 0.0)) if tile else None, "request_to_apply_s": tile.get("request_to_apply_s") if tile else None, "worker_queue_s": tile.get("worker_queue_s") if tile else None, "worker_task_s": tile.get("worker_task_s") if tile else ( end.get("worker_task_s") if end else None), "result_queue_s": tile.get("result_queue_s") if tile else None, "discard_s": float(discard.get("t_s", 0.0)) if discard else None, "discard_reason": discard.get("reason") if discard else None, "ahead_m_at_request": ahead_m, "request_to_worker_end_s": end_t - request_t if end_t is not None else None, "cancel_before_worker_end": int(bool(cancel_t is not None and end_t is not None and cancel_t < end_t)), "out_of_range_at_worker_end": int(physically_out), }) key = f"{run['speed']}-r{run['repetition']}" stale_by_run[key] = { "generated_lod0_tiles_finished_during_move": generated_l0, "generated_lod0_out_of_range_at_finish": out_of_range_l0, "out_of_range_fraction_pct": 100.0 * out_of_range_l0 / max(1, generated_l0), "generated_lod0_finished_after_cancel": stale_l0, "cancel_before_finish_fraction_pct": 100.0 * stale_l0 / max(1, generated_l0), "lod0_cancelled_while_worker_active": l0_cancelled_while_running, "lod0_admissions_in_move": len(ahead_distances), "lod0_ahead_at_admission_pct": 100.0 * sum(value > 0.0 for value in ahead_distances) / max(1, len(ahead_distances)), "lod0_ahead_median_m": statistics.median(ahead_distances) if ahead_distances else None, } return {"tile_timelines_csv": str(path.relative_to(ROOT)), "stale_by_run": stale_by_run} def run_phase(phase: str) -> None: partial_path = OUT / phase / "runs-partial.json" results = [] if partial_path.is_file(): try: stored = json.loads(partial_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): stored = [] for item in stored: if item.get("phase") != phase or int(item.get("exit_code", -1)) != 0: continue if item.get("platform") == "unreal": trace = ROOT / item.get("trace", "") if not trace.is_file() or not Path(str(trace) + ".done").is_file(): continue elif item.get("platform") == "headless": log = ROOT / item.get("log", "") if not log.is_file() or "[L79MovingSummary]" not in log.read_text( encoding="utf-8", errors="replace"): continue content = log.read_text(encoding="utf-8", errors="replace") update_ms = [] for match in HEADLESS_SAMPLE.finditer(content): row = dict(FIELDS.findall(match.group(1))) if row.get("phase") == "move" and "queue_update_ms" in row: update_ms.append(float(row["queue_update_ms"])) item["update_work_p99_ms"] = p99(update_ms) item["update_work_max_ms"] = max(update_ms, default=None) results.append(item) completed = {(item["platform"], item["speed"], item["repetition"]) for item in results} unreal_runs = [] for item in results: if item["platform"] == "unreal": trace = ROOT / item["trace"] unreal_runs.append((item, parse_events(trace))) def save_partial() -> None: partial_path.parent.mkdir(parents=True, exist_ok=True) partial_path.write_text(json.dumps(results, indent=2) + "\n", encoding="utf-8") for repetition in range(1, 4): speed_order = ("walk", "fast") if repetition % 2 else ("fast", "walk") for speed_name in speed_order: headless_key = ("headless", speed_name, repetition) if headless_key not in completed: headless_info = run_headless(phase, speed_name, repetition) results.append(headless_info) completed.add(headless_key) save_partial() print("[L108MovementRun] " + json.dumps(headless_info, sort_keys=True), flush=True) unreal_key = ("unreal", speed_name, repetition) if unreal_key not in completed: run_info, events, _rows = run_unreal(phase, speed_name, repetition) results.append(run_info) unreal_runs.append((run_info, events)) completed.add(unreal_key) save_partial() print("[L108MovementRun] " + json.dumps(run_info, sort_keys=True), flush=True) timeline = write_tile_timelines(phase, unreal_runs) (OUT / phase / "summary.json").write_text( json.dumps({"phase": phase, "runs": results, "timeline": timeline}, indent=2) + "\n", encoding="utf-8") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("phase", choices=("baseline", "after")) args = parser.parse_args() if not DATA.is_file() or not HEADLESS[args.phase].is_file() or not EDITOR.is_file(): raise SystemExit("missing movement runner input, headless executable or Unreal editor") run_phase(args.phase) return 0 if __name__ == "__main__": raise SystemExit(main())