#!/usr/bin/env python3 """Five interleaved L168 30 m/s UE pairs, using source-default streaming CVars.""" from __future__ import annotations import json import math import shutil 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") OUT = ROOT / "Saved/Linux/l168/ue/movement" BIN_DIR = ROOT / "Saved/Linux/l168/bin" HOST_PLUGIN = HOST / "Binaries/Linux/libUnrealEditor-VoxelForge.so" SPAWN_M = (19.166015625, -14.2001953125, -753.3740234375) MOVE_START_S = 6.0 MOVE_DURATION_S = 12.0 CONDITIONS = { "assembler_world": { "binary": BIN_DIR / "assembler-world/libUnrealEditor-VoxelForge.so", "incremental": 0, }, "incremental_default": { "binary": BIN_DIR / "incremental-default/libUnrealEditor-VoxelForge.so", "incremental": 1, }, } ORDER = { 1: ("assembler_world", "incremental_default"), 2: ("incremental_default", "assembler_world"), 3: ("assembler_world", "incremental_default"), 4: ("incremental_default", "assembler_world"), 5: ("assembler_world", "incremental_default"), } def percentile(values: list[float], fraction: float = 0.99) -> float | None: if not values: return None ordered = sorted(values) return ordered[max(0, min(len(ordered) - 1, math.ceil(len(ordered) * fraction) - 1))] def percent(numerator: int | float, denominator: int | float) -> float | None: return 100.0 * numerator / denominator if denominator > 0 else None def read_events(path: Path) -> list[dict]: rows = [] if path.is_file(): for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if not line.startswith("{"): continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue return rows def summarize_run(condition: str, run_name: str, trace: Path, log: Path, exit_code: int, wall_s: float, audit: bool = False) -> dict: expected_incremental = CONDITIONS[condition]["incremental"] events = read_events(trace) backend = next((e for e in events if e.get("type") == "worker_backend_config"), {}) scheduler = [e for e in events if e.get("type") == "scheduler_admission"] if not scheduler: raise RuntimeError(f"missing scheduler readback: {trace}") if any(int(e.get("incremental_desired_set", -1)) != expected_incremental for e in scheduler): raise RuntimeError(f"incremental default readback mismatch: {trace}") if any(int(e.get("forward_first", -1)) != 0 or int(e.get("rebuild_on_core_region_completion", -1)) != 1 or int(e.get("core_desired_cache_by_center", -1)) != 0 for e in scheduler): raise RuntimeError(f"L160 control default readback mismatch: {trace}") if any(int(e.get("incremental_desired_audit_every", -1)) != (1 if audit else 0) or int(e.get("incremental_desired_materialization_audit_every", -1)) != (1 if audit else 0) or int(e.get("incremental_desired_audit_inject_mismatch", -1)) != 0 for e in scheduler): raise RuntimeError(f"audit CVar readback mismatch: {trace}") if (int(backend.get("streaming_fast_travel_wait_for_all_critical_tiles", -1)) != 1 or abs(float(backend.get("streaming_fast_travel_apply_budget_ms", -1.0))) > 0.01 or int(backend.get("effective_tile_slots", -1)) != 8): raise RuntimeError(f"L157/default worker readback mismatch: {trace}; {backend}") move_stop = next((e for e in events if e.get("type") == "streaming_test_move_stopped"), None) if move_stop is None: raise RuntimeError(f"missing move-stop event: {trace}") stop_s = float(move_stop.get("t_s", 0.0)) samples = [e for e in events if e.get("type") == "streaming_sample"] moving = [e for e in samples if MOVE_START_S <= float(e.get("t_s", 0.0)) < stop_s and float(e.get("speed_cm_s", 0.0)) >= 50.0] dwell = [e for e in samples if float(e.get("t_s", 0.0)) >= stop_s and float(e.get("speed_cm_s", 0.0)) < 50.0] if exit_code != 0 or len(moving) < 40 or len(dwell) < 16: raise RuntimeError(f"incomplete run {run_name}: exit={exit_code}, move={len(moving)}, dwell={len(dwell)}") move_start = min(float(e.get("t_s", 0.0)) for e in moving) work = [e for e in events if e.get("type") == "game_thread_work_frame" and move_start <= float(e.get("t_s", 0.0)) < stop_s] ticks = [float(e.get("tick_ms", 0.0)) for e in work] rebuilds = [e for e in events if e.get("type") == "desired_rebuild_breakdown" and move_start <= float(e.get("t_s", 0.0)) < stop_s] core_events = [e for e in events if e.get("type") == "core_desired_cache" and move_start <= float(e.get("t_s", 0.0)) < stop_s] audits = [e for e in events if e.get("type") == "incremental_desired_audit" and move_start <= float(e.get("t_s", 0.0)) < stop_s] order_audits = [e for e in events if e.get("type") == "incremental_desired_order_audit" and move_start <= float(e.get("t_s", 0.0)) < stop_s] materialization_audits = [e for e in audits if int(e.get("materialization_audit", 0)) != 0] if audit: if (not audits or not order_audits or not materialization_audits or any(int(e.get("pass", 0)) != 1 for e in audits + order_audits)): raise RuntimeError( f"equality audit missing/failed: {trace}; set={len(audits)} " f"order={len(order_audits)} materialization={len(materialization_audits)}") near_values = [] far_holes = [] for event in moving: desired = event.get("desired_by_lod", [0] * 9) resident = event.get("resident_by_lod", [0] * 9) covered = event.get("covered_by_lod", resident) near_values.append(percent(sum(resident[:3]), sum(desired[:3])) or 0.0) far_holes.append(percent(sum(desired[3:]) - sum(covered[3:]), sum(desired[3:])) or 0.0) settled = next((e for e in samples if float(e.get("t_s", 0.0)) < move_start and float(e.get("speed_cm_s", 0.0)) < 50.0 and sum(e.get("settled_by_lod", [0] * 9)[:3]) >= sum(e.get("desired_by_lod", [0] * 9)[:3]) > 0), None) row = { "condition": condition, "run": run_name, "audit_run": audit, "source_default_incremental": expected_incremental, "speed_mps": 30, "effective_tile_slots": backend.get("effective_tile_slots"), "moving_samples": len(moving), "dwell_samples": len(dwell), "game_thread_tick_p99_ms": percentile(ticks), "game_thread_tick_worst_ms": max(ticks, default=None), "near_readiness_moving_mean_pct": statistics.mean(near_values) if near_values else None, "far_holes_moving_mean_pct": statistics.mean(far_holes) if far_holes else None, "first_settled_lod02_s": float(settled.get("t_s", 0.0)) if settled else None, "desired_rebuild_count": len(rebuilds), "incremental_materialization_update_count": sum( int(e.get("materialization_update", 0)) != 0 for e in rebuilds), "priority_order_update_count": sum( int(e.get("priority_order_updated", 0)) != 0 for e in rebuilds), "priority_order_reuse_count": sum( int(e.get("priority_order_reused", 0)) != 0 for e in rebuilds), "core_incremental_delta_count": sum(int(e.get("incremental_delta", 0)) != 0 for e in core_events), "core_cache_hit_count": sum(int(e.get("hit", 0)) != 0 for e in core_events), "set_audits": len(audits), "order_audits": len(order_audits), "materialization_audits": len(materialization_audits), "audit_failures": sum(int(e.get("pass", 1)) != 1 for e in audits + order_audits), "trace": str(trace.relative_to(ROOT)), "log": str(log.relative_to(ROOT)), "wall_s": round(wall_s, 3), } if expected_incremental and not audit and ( row["incremental_materialization_update_count"] == 0 or row["core_incremental_delta_count"] == 0): raise RuntimeError(f"default incremental trace counter stayed zero: {trace}; {row}") return row def run(condition: str, run_name: str, audit: bool = False) -> dict: binary = CONDITIONS[condition]["binary"] if not binary.is_file(): raise FileNotFoundError(binary) shutil.copy2(binary, HOST_PLUGIN) out = OUT / condition export = HOST / "Plugins/VoxelForge/Saved/Linux/l168" / condition / run_name userdir = HOST / "Saved/Linux/l168/user" / condition / run_name zendata = HOST / "Saved/Linux/l168/zen" / condition / run_name ddc = HOST / "Saved/Linux/l168/DDC" trace = out / f"{run_name}.trace.jsonl" log = out / f"{run_name}.log" stdout = out / f"{run_name}.stdout.log" for directory in (out, export, userdir, zendata, ddc): directory.mkdir(parents=True, exist_ok=True) trace.unlink(missing_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.TestMove=1", "-voxel.TestMoveSpeedCmPerSecond=3000.0", f"-voxel.TestMoveStartSeconds={MOVE_START_S:.1f}", f"-voxel.TestMoveDurationSeconds={MOVE_DURATION_S:.1f}", "-voxel.TestExitSeconds=0.0", "-voxel.TestMoveDirection=1,0,0", "-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={log}", ] if audit: command += ["-voxel.StreamingIncrementalDesiredAuditEvery=1", "-voxel.StreamingIncrementalDesiredMaterializationAuditEvery=1"] if any("StreamingIncrementalDesiredSet=" in arg for arg in command): raise RuntimeError("timing/default proof must not override incremental desired-set CVar") started = time.monotonic() with stdout.open("w", encoding="utf-8") as stream: process = subprocess.run(["taskset", "-c", "0-11", "timeout", "1740s", *command], cwd=ROOT, stdout=stream, stderr=subprocess.STDOUT, check=False, timeout=1740) return summarize_run(condition, run_name, trace, log, process.returncode, time.monotonic() - started, audit) def main() -> int: missing = [p for p in (EDITOR, HOST / "VFHost.uproject", HOST_PLUGIN) if not p.exists()] if missing: raise SystemExit(f"missing UE runner input: {missing}") OUT.mkdir(parents=True, exist_ok=True) BIN_DIR.mkdir(parents=True, exist_ok=True) # One warm-up for each binary, excluded from the paired summary. The candidate warm-up # is also the no-CVar-override proof that the compiled default executes incremental updates. warmups = [run("assembler_world", "warmup"), run("incremental_default", "warmup-default")] rows = [] partial = OUT / "runs-partial.json" for pair, sequence in ORDER.items(): for condition in sequence: rows.append(run(condition, f"pair-{pair}")) partial.write_text(json.dumps(rows, indent=2, sort_keys=True) + "\n", encoding="utf-8") # One audited candidate run, outside the timing matrix. audit_row = run("incremental_default", "equality-audit", audit=True) (OUT / "warmups.json").write_text(json.dumps(warmups, indent=2, sort_keys=True) + "\n", encoding="utf-8") (OUT / "equality-audit.json").write_text( json.dumps(audit_row, indent=2, sort_keys=True) + "\n", encoding="utf-8") metrics = ("game_thread_tick_p99_ms", "game_thread_tick_worst_ms", "near_readiness_moving_mean_pct", "far_holes_moving_mean_pct", "first_settled_lod02_s", "incremental_materialization_update_count", "core_incremental_delta_count") by_pair = {pair: {row["condition"]: row for row in rows if row["run"] == f"pair-{pair}"} for pair in ORDER} medians = {} for condition in CONDITIONS: selected = [row for row in rows if row["condition"] == condition] medians[condition] = {metric: statistics.median( row[metric] for row in selected if row[metric] is not None) for metric in metrics} deltas = [] for pair in ORDER: base = by_pair[pair]["assembler_world"] candidate = by_pair[pair]["incremental_default"] deltas.append({"pair": pair, **{f"{m}_delta": candidate[m] - base[m] for m in metrics if candidate[m] is not None and base[m] is not None}}) summary = { "protocol": "Five interleaved 30 m/s UE pairs; 12 s move plus 5 s dwell; 8 effective tile slots; four-strata owner plan. No streaming configuration CVar overrides during warm-up or timing runs. One warm-up per binary excluded. Equality audits run once afterward.", "default_readback": { "incremental_desired_set": 1, "incremental_desired_audit_every": 0, "incremental_desired_materialization_audit_every": 0, "incremental_desired_audit_inject_mismatch": 0, "fast_travel_wait_for_all_critical_tiles": 1, "fast_travel_apply_budget_ms": 0, "forward_first": 0, "rebuild_on_core_region_completion": 1, "core_desired_cache_by_center": 0, }, "warmups": warmups, "movement_medians": medians, "paired_deltas_candidate_minus_baseline": deltas, "equality_audit": audit_row, "runs": rows, } (OUT / "summary.json").write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n", encoding="utf-8") print("[L168Summary] " + json.dumps({"movement_medians": medians, "audit": audit_row, "pairs": len(ORDER)}, sort_keys=True), flush=True) return 0 if __name__ == "__main__": raise SystemExit(main())