mirror/psxrecomp
1
0
Fork 0
mirror of https://github.com/RetroPortingToolKit/psxrecomp synced 2026-10-26 10:19:47 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #430 from RetroPortingToolKit/feat/fp-identity-tool

Browse Source 
tools: generic A/B guest-identity check (fp_identity.py)
This commit is contained in:
Matt Stanley
2026-09-29 11:24:04 -07:00
committed by GitHub
parent 6fa0dd3c71 c4e50d92e0
commit 470f03b782
6 changed files with 1643 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/TCP_COMMANDS.md
+32
View File
@@ -242,6 +242,38 @@ For A/B identity, set `PSX_DEBUG_FMV_QUIET=0` in both runs.
the opposite order, even though the final RAM differs. Such a fork shows up in
later writes, `cyc`, or `read_ram`.
### A/B identity with `tools/fp_identity.py`
`tools/fp_identity.py` applies these rules to two seeded runs of any title.
`run` starts the runtime from a command template or from `--runtime`,
`--game` and `--disc`. It turbos with no input to frame N and saves this ring
and the dispatch-miss count. It sets `PSX_DEBUG_FMV_QUIET=0` and
`PSX_OVERLAY_AUTOCOMPILE_OFF=1` unless told otherwise. `--seed` installs a
saved overlay state first, so a run can start from a warm or a cold cache.
`compare` judges the two runs on the judge columns and reports `wr` and `pc`
only as the frame where the runs part. It tolerates the one-write straddle
above, and the `wc + qc` shift when FMV-quiet was on. Every tolerated frame is
counted and listed, and `--strict` disables both tolerances. It exits
`0` IDENTICAL, `1` MISMATCH, `3` INCOMPLETE (frame window, wrapped ring,
missing columns from an older runtime, an unconfirmable last-frame
straddle, or different FMV-quiet settings once a quiet frame occurred,
which leaves the runs not comparable), `4` MISSES (dispatch misses), or `2`
on a usage error. `run` exits `3` when it writes no dump. On Windows, a
`--launch` backslash is a path separator, not an escape.
```sh
L='tools/run_game.sh {build} --debug-port {port} {headless}'
python3 psxrecomp/tools/fp_identity.py snapshot build /tmp/seed-warm
mkdir -p /tmp/seed-cold
python3 psxrecomp/tools/fp_identity.py run warm.json --launch "$L" --port 4781 --frames 12000 --seed /tmp/seed-warm
python3 psxrecomp/tools/fp_identity.py run cold.json --launch "$L" --port 4781 --frames 12000 --seed /tmp/seed-cold
python3 psxrecomp/tools/fp_identity.py compare warm.json cold.json
```
Run an A/A pair first to show that the run is deterministic. The tool's
docstring (`--help`) lists every option. The self-test is
`tools/tests/test_fp_identity.py` (ctest `fp_identity`).
## `bios_info` — linked recompiled-BIOS identity (native only)
Reports which BIOS image this build's recompiled C was generated from
docs/internal/ACCURACY_BURNDOWN.md
+4
View File
@@ -206,6 +206,10 @@ Status: STRONG (most project effort lives here).
targets) — resolve all dispatch misses each run (Tomba2Recomp CLAUDE.md).
- [ ] Call/return contract + stack fidelity — the blue-screen/wedge class.
- [ ] Backend equivalence (compiled == interp) — necessary, not sufficient.
Measured with `tools/fp_identity.py` (2026-09-29): seeded warm vs cold
overlay-cache runs, judged on the `frame_fingerprint` guest-fact columns. R4,
12000 frames on #417+#418+#420: IDENTICAL with 497 tolerated one-write VBlank
straddles.
## Axis 7 — Determinism
docs/internal/FAITHFUL_TIMING_PLAN.md
+13
View File
@@ -213,6 +213,19 @@ on a fixed region -> next.
## 5. Status / Log (update every session)
- **2026-09-29 (generic A/B identity tool, `feat/fp-identity-tool` on #420):**
`tools/fp_identity.py` moves R4's warm/cold check into the framework for
any title (launch template or `--runtime/--game/--disc`, `--seed` overlay
state, FMV-quiet and autocompile off by default). It judges on
cyc/mmio/mc/sp/sc/wc/ws/qc. `wr`/`pc` only locate a fork. The one-write VBlank
straddle and the FMV-quiet `wc+qc` shift are tolerated, counted and listed,
and `--strict` rejects them. Measurement only; no runtime or codegen change.
R4 on #420+#417+#418, 12000 seeded frames: warm/warm and cold/cold are
IDENTICAL with no tolerance, even on `wr`/`pc`. Warm/cold is IDENTICAL with
497 one-write straddles, the same count #420 reported. With FMV-quiet on,
warm/cold is IDENTICAL over 8076 frames, with 497 carried shifts. Self-test:
ctest `fp_identity`.
- **2026-09-13 (SIO card hack removal — branch-only review checkpoint):**
Reproduced fixed-Ape-RAM IRQ7/mask injection after an absent-card probe,
plus SELECT-time ACK fabrication and INTC-pending ACK requeueing with the
recompiler/CMakeLists.txt
+1
View File
@@ -810,6 +810,7 @@ if(BUILD_TESTING)
cursor_writers
disc_companion
duckstation_oracle
fp_identity
gpu_frame
menu_scan
mips_tagged_relocations
tools/fp_identity.py
+820
View File
@@ -0,0 +1,820 @@
#!/usr/bin/env python3
"""fp_identity.py - A/B check that two runs executed the guest identically.
`run` starts a runtime with no pad input (turbo, headless by default), lets it
reach guest frame N, and saves the debug server's per-frame guest-write
fingerprint ring (`frame_fingerprint`, docs/TCP_COMMANDS.md) with the
dispatch-miss count. `compare` judges two such dumps on the guest-fact
columns. Any title works: give the launch as a command template or as
runtime + game.toml + disc. The runtime needs PSX_DEBUG_TOOLS=ON.
Run a build against itself first (A/A) to show the run is deterministic, then
the candidate against the base (A/B). Warm vs cold overlay cache on R4, from
the game repo's root:
T=psxrecomp/tools/fp_identity.py
L='tools/run_r4.sh {build} --debug-port {port} {headless}'
python3 $T snapshot build /tmp/seed-warm # after a warm-up run
mkdir -p /tmp/seed-cold # empty seed = cold cache
python3 $T run warm.json --launch "$L" --port 4781 --frames 12000 --seed /tmp/seed-warm
python3 $T run cold.json --launch "$L" --port 4781 --frames 12000 --seed /tmp/seed-cold
python3 $T compare warm.json cold.json
python3 $T run out.json --runtime build/game-runtime --game build/game.toml \\
--disc disc/game.cue --port 4781 --frames 6000
--launch placeholders: {build} (--build, default "build"), {port}, {headless}
("--headless", or nothing with --windowed) and {state_dir}. The template is
split like a shell command line but not run through a shell; on Windows a
backslash is a path separator, not an escape, so C:\\psx\\game.exe works
(quote paths with spaces). Arguments after `--` are appended unchanged.
--runtime launches
`RUNTIME --game TOML --disc DISC --no-launcher --debug-port PORT [--headless]`.
Environment: runs set PSX_DEBUG_FMV_QUIET=0 (FMV-quiet would keep MDEC-era
writes out of the judge columns) and PSX_OVERLAY_AUTOCOMPILE_OFF=1 (the
overlay state stays what --seed made it). --fmv-quiet and --autocompile opt
back in; --env K=V adds more. The dump records what the run used.
Overlay state is part of the input: a cold overlay cache runs code in the
interpreter that a warm one runs natively. --seed DIR replaces the state
directory's cache/ and overlay_captures.json(.d) with DIR's before launching
(an empty DIR gives a cold cache). Use the same DIR for both runs of an A/A
pair. The state directory is --state-dir, else the --runtime executable's
directory, else --build under --cwd when the template uses {build}.
Judging (docs/TCP_COMMANDS.md, `frame_fingerprint`):
judge cyc mmio mc sp sc wc ws qc must agree at every shared frame
locator wr pc reported as where the runs part, never
a failure (host device-service order)
Tolerances, each counted and listed in the output, never silent:
VBlank straddle a device write lands on the other side of a snapshot in one
run: at one frame ws/wc differ by at most --straddle-writes
(default 1) writes and every other judge column agrees; the
next shared frame must agree again. A straddle on the last
shared frame cannot be confirmed and makes the result
INCOMPLETE.
FMV-quiet shift only when FMV-quiet was on (or not recorded) in a run: a
straddled write that one run counted in qc and the other
in wc. wc+qc must still agree; wc, qc and
ws are judged against the offsets carried from the shift.
--strict (= --straddle-writes 0) turns both off.
`compare` prints one verdict and exits with its code:
0 IDENTICAL same frame set, every judge column equal (tolerances listed),
0 dispatch misses in both
1 MISMATCH a shared frame differs on a judge column, or one run
fingerprinted frames inside the other's range that the other
did not
3 INCOMPLETE the shared frames agree but identity is not proven: different
--frames, frames missing at either end of one run, a wrapped
fingerprint ring, an empty run, no dispatch-miss count, a
judge column the runtime did not report (runtimes before
psxrecomp #420 lack ws and qc), or a straddle on the last
shared frame. Also different FMV-quiet settings once a quiet
frame occurred (qc > 0, or the frame sets differ): quiet
frames are not fingerprinted and their writes count only in
qc, so the frames and columns are not judged at all
4 MISSES everything else holds but a run had dispatch misses (resolve
them before trusting an identity result)
2 usage error
Every problem found is listed, not only the first. `run` exits 0 once the
dump is written, 3 when it did not write one (the runtime did not start, the
port was taken, the runtime exited, stalled, timed out or dropped the
connection, or seeding or writing failed), and 2 on usage errors. `snapshot`
exits 3 when the copy fails.
"""
import argparse
import json
import os
import shlex
import shutil
import signal
import socket
import subprocess
import sys
import time
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from debug_client import query # noqa: E402
EXIT_IDENTICAL, EXIT_MISMATCH, EXIT_USAGE, EXIT_INCOMPLETE, EXIT_MISSES = 0, 1, 2, 3, 4
HOST = "127.0.0.1"
EXACT = ("cyc", "mmio", "mc", "sp", "sc") # judged with no tolerance
WRITES = ("wc", "ws", "qc") # judged, straddle-aware
JUDGE = EXACT + WRITES
LOCATORS = ("wr", "pc")
HASHES = ("mmio", "sp", "ws", "wr", "pc")
MASK64 = (1 << 64) - 1
RING_CAP = 32768
OVERLAY_STATE = ("cache", "overlay_captures.json", "overlay_captures.json.d")
DEFAULT_ENV = {"PSX_DEBUG_FMV_QUIET": "0", "PSX_OVERLAY_AUTOCOMPILE_OFF": "1"}
SCHEMA = 2
class UsageError(Exception):
pass
class RunError(Exception):
pass
# ---- run -------------------------------------------------------------------
def copy_overlay_state(src, dst):
for name in OVERLAY_STATE:
target = os.path.join(dst, name)
if os.path.isdir(target) and not os.path.islink(target):
shutil.rmtree(target)
elif os.path.lexists(target):
os.remove(target)
source = os.path.join(src, name)
if os.path.isdir(source):
shutil.copytree(source, target, symlinks=True)
elif os.path.exists(source):
shutil.copy2(source, target)
def resolve_state_dir(args, cwd):
if args.state_dir:
return os.path.abspath(os.path.join(cwd, args.state_dir))
if args.runtime:
return os.path.dirname(os.path.abspath(os.path.join(cwd, args.runtime)))
if args.launch and "{build}" in args.launch:
return os.path.abspath(os.path.join(cwd, args.build))
return None
def split_template(text, windows=None):
"""Split a --launch template into arguments, without a shell.
POSIX: shell quoting and backslash escapes (shlex.split). Windows:
backslashes are path separators, not escapes, so C:\\psx\\game.exe stays
as written; quote a path with spaces as "C:\\Program Files\\...".
"""
if windows is None:
windows = os.name == "nt"
if not windows:
return shlex.split(text)
lex = shlex.shlex(text, posix=True)
lex.whitespace_split = True
lex.commenters = ""
lex.escape = ""
return list(lex)
def launch_argv(args, state_dir, windows=None):
headless = [] if args.windowed else ["--headless"]
if not args.launch:
return [args.runtime, "--game", args.game, "--disc", args.disc,
"--no-launcher", "--debug-port", str(args.port),
*headless, *args.extra]
try:
tokens = split_template(args.launch, windows)
except ValueError as e:
raise UsageError(f"--launch: {e}")
if args.windowed and "--headless" in tokens:
raise UsageError("--windowed, but the --launch template passes "
"--headless itself; write {headless} instead")
values = {"build": args.build, "port": str(args.port),
"headless": " ".join(headless), "state_dir": state_dir or ""}
argv = []
for token in tokens:
try:
value = token.format(**values)
except (KeyError, IndexError, ValueError) as e:
raise UsageError(f"--launch: bad placeholder in {token!r} ({e!r}); "
f"known: {{build}} {{port}} {{headless}} "
f"{{state_dir}}")
if value:
argv.append(value)
if not argv:
raise UsageError("--launch is empty")
return argv + args.extra
def port_in_use(port):
try:
with socket.create_connection((HOST, port), timeout=0.5):
return True
except OSError:
return False
def stop(proc):
"""Stop the launched process group: a launcher script may not exec.
Returns None, or why the runtime could not be stopped."""
try:
if os.name == "nt":
subprocess.run(["taskkill", "/T", "/F", "/PID", str(proc.pid)],
capture_output=True)
else:
os.killpg(proc.pid, signal.SIGTERM)
except OSError:
pass
try:
proc.wait(timeout=15)
return None
except subprocess.TimeoutExpired:
pass
try:
if os.name == "nt":
proc.kill()
else:
os.killpg(proc.pid, signal.SIGKILL)
except OSError:
pass
try:
proc.wait(timeout=10)
return None
except subprocess.TimeoutExpired:
return (f"runtime pid {proc.pid} did not exit after SIGKILL; it may "
f"still hold its port")
def ask(port, obj, timeout):
return query(HOST, port, obj, timeout=timeout)
def drive(proc, args, log_path):
"""Turbo to frame N; return (fingerprint reply, dispatch_stats reply)."""
port = args.port
deadline = time.time() + args.boot_timeout
while True:
try:
ask(port, {"cmd": "turbo", "enabled": 1}, 2)
break
except (OSError, ValueError):
if proc.poll() is not None:
raise RunError(f"runtime exited (status {proc.returncode}) "
f"before its debug server answered on port "
f"{port} (see {log_path})")
if time.time() > deadline:
raise RunError(f"no debug server on port {port} after "
f"{args.boot_timeout}s (built with "
f"PSX_DEBUG_TOOLS=ON? does the launch pass the "
f"port? see {log_path})")
time.sleep(0.3)
start = last_move = time.time()
last = -1
while True:
if proc.poll() is not None:
raise RunError(f"runtime exited (status {proc.returncode}) at "
f"frame {last} of {args.frames} (see {log_path})")
try:
frame = int(ask(port, {"cmd": "frame"}, 120).get("frame", 0))
except (OSError, ValueError):
frame = last
if frame >= args.frames:
break
now = time.time()
if frame != last:
last, last_move = frame, now
elif now - last_move > args.stall:
raise RunError(f"stalled: frame {frame} did not advance for "
f"{args.stall}s (see {log_path})")
if now - start > args.timeout:
raise RunError(f"reached only frame {frame} of {args.frames} in "
f"{args.timeout}s (see {log_path})")
time.sleep(0.2)
replies = []
for req, timeout in (({"cmd": "frame_fingerprint",
"count": min(args.frames + 1, RING_CAP),
"frame_lo": 0, "frame_hi": args.frames}, 300),
({"cmd": "dispatch_stats"}, 120)):
try:
reply = ask(port, req, timeout)
except (OSError, ValueError) as e:
# A crash, a dropped connection or a socket timeout after frame
# N: no dump, so INCOMPLETE rather than a traceback.
try:
proc.wait(timeout=2) # a crash drops the socket first
except subprocess.TimeoutExpired:
pass
state = ("exited (status %s)" % proc.returncode
if proc.returncode is not None else "did not answer")
raise RunError(f"runtime {state} on {req['cmd']} after reaching "
f"frame {frame} ({e!r}; see {log_path})")
if not isinstance(reply, dict):
raise RunError(f"{req['cmd']} replied {reply!r}, not an object")
if reply.get("ok") is False:
raise RunError(f"{req['cmd']} failed: {reply.get('error', reply)}")
replies.append(reply)
fp, stats = replies
if "entries" not in fp:
raise RunError(f"frame_fingerprint returned no entries: {fp}")
return fp, stats
def cmd_run(args):
cwd = os.path.abspath(args.cwd)
if not args.launch and not (args.runtime and args.game and args.disc):
raise UsageError("run needs --launch TEMPLATE, or --runtime, --game "
"and --disc")
if args.launch and (args.runtime or args.game or args.disc):
raise UsageError("give --launch or --runtime/--game/--disc, not both")
state_dir = resolve_state_dir(args, cwd)
if args.seed:
if not state_dir:
raise UsageError("--seed needs --state-dir (or --runtime, or "
"{build} in --launch)")
if not os.path.isdir(args.seed):
raise UsageError(f"--seed {args.seed} is not a directory")
if not os.path.isdir(state_dir):
raise UsageError(f"state directory {state_dir} does not exist")
out = os.path.abspath(args.out)
if not os.path.isdir(os.path.dirname(out)):
raise UsageError(f"output directory {os.path.dirname(out)} does not "
f"exist")
argv = launch_argv(args, state_dir)
env = dict(os.environ)
record = dict(DEFAULT_ENV)
if args.fmv_quiet:
record["PSX_DEBUG_FMV_QUIET"] = "1"
if args.autocompile:
record["PSX_OVERLAY_AUTOCOMPILE_OFF"] = "0"
for kv in args.env:
if "=" not in kv:
raise UsageError(f"--env {kv!r}: expected KEY=VALUE")
key, value = kv.split("=", 1)
record[key] = value
env.update(record)
if port_in_use(args.port):
raise RunError(f"port {args.port} already has a listener; pick a free "
f"--port so the run cannot query another runtime")
if args.seed:
copy_overlay_state(args.seed, state_dir)
log_path = out + ".log"
label = args.label or (args.build if args.launch else args.runtime)
with open(log_path, "w") as log:
try:
proc = subprocess.Popen(argv, cwd=cwd, env=env, stdout=log,
stderr=subprocess.STDOUT,
start_new_session=(os.name != "nt"))
except OSError as e:
raise RunError(f"cannot launch {argv[0]}: {e}")
failure = None
try:
fp, stats = drive(proc, args, log_path)
except RunError as e:
failure = str(e)
except BaseException:
stop(proc)
raise
stuck = stop(proc)
if stuck:
failure = f"{failure}; {stuck}" if failure else stuck
if failure:
raise RunError(failure)
entries = fp.get("entries", [])
# Write then rename: an interrupted write never leaves a partial dump.
tmp = out + ".tmp"
try:
with open(tmp, "w") as f:
json.dump({"schema": SCHEMA, "build": label,
"frames": args.frames, "launch": argv, "cwd": cwd,
"state_dir": state_dir,
"seed": os.path.abspath(args.seed) if args.seed
else None,
"env": record, "port": args.port,
"miss_total": stats.get("miss_total"),
"miss_unique": stats.get("miss_unique"),
"ring_total": fp.get("total"),
"ring_available": fp.get("available"),
"entries": entries}, f)
os.replace(tmp, out)
except BaseException:
try:
os.remove(tmp)
except OSError:
pass
raise
print(f"{label}: {len(entries)} frames fingerprinted to frame "
f"{args.frames}, dispatch misses total={stats.get('miss_total')} "
f"unique={stats.get('miss_unique')} -> {out}")
return EXIT_IDENTICAL
def cmd_snapshot(args):
if not os.path.isdir(args.state_dir):
raise UsageError(f"{args.state_dir} is not a directory")
os.makedirs(args.seed_dir, exist_ok=True)
copy_overlay_state(args.state_dir, args.seed_dir)
kept = [n for n in OVERLAY_STATE
if os.path.lexists(os.path.join(args.seed_dir, n))]
print(f"overlay state of {args.state_dir} -> {args.seed_dir} "
f"({', '.join(kept) if kept else 'empty: a cold seed'})")
return EXIT_IDENTICAL
# ---- compare ---------------------------------------------------------------
def num(value):
return int(value, 0) if isinstance(value, str) else int(value)
def load_run(path):
try:
with open(path) as f:
run = json.load(f)
except (OSError, ValueError) as e:
raise UsageError(f"cannot read {path}: {e}")
if not isinstance(run, dict) or not isinstance(run.get("entries", []), list):
raise UsageError(f"{path} is not an fp_identity dump")
frames, present = {}, None
try:
for e in run.get("entries", []):
cols = {k: num(e[k]) for k in JUDGE + LOCATORS if k in e}
frames[num(e["frame"])] = cols
present = set(cols) if present is None else present & set(cols)
except (KeyError, TypeError, ValueError) as e:
raise UsageError(f"{path}: malformed fingerprint entry ({e!r})")
return run, frames, present or set()
def show(cols):
"""One entry's columns, hashes in hex as the server prints them."""
return " ".join(f"{k}={v:#018x}" if k in HASHES else f"{k}={v}"
for k, v in cols.items())
def fmv_quiet(run):
"""True/False as the run set it; None when the dump did not record it."""
value = (run.get("env") or {}).get("PSX_DEBUG_FMV_QUIET")
return None if value is None else value[:1] != "0"
def quiet_text(setting):
return {True: "on", False: "off", None: "not recorded (runtime default: on)"}[setting]
def frame_ranges(frames, limit=8):
"""Sorted frame numbers as compact runs: '1..100, 205, 300..310'."""
runs = []
for frame in frames:
if runs and frame == runs[-1][1] + 1:
runs[-1][1] = frame
else:
runs.append([frame, frame])
text = [f"{lo}" if lo == hi else f"{lo}..{hi}" for lo, hi in runs[:limit]]
if len(runs) > limit:
text.append(f"... (+{len(runs) - limit} more runs)")
return ", ".join(text)
def scan(common, fa, fb, judged, quiet, limit):
"""Walk the shared frames in order and judge them; see the module doc."""
exact = [c for c in EXACT if c in judged]
def offset(f):
a, b = fa[f], fb[f]
return (a["wc"] - b["wc"] if "wc" in judged else 0,
a["qc"] - b["qc"] if "qc" in judged else 0,
(a["ws"] - b["ws"]) & MASK64 if "ws" in judged else 0)
def exact_diff(f):
return [c for c in exact if fa[f][c] != fb[f][c]]
def moved(d, ref):
"""Writes a straddle moved across the snapshot (0 = none)."""
if quiet:
return abs((d[0] + d[1]) - (ref[0] + ref[1]))
return abs(d[0] - ref[0]) if d[1] == ref[1] else 0
def shift(d, ref, cap):
return (quiet and d != ref and d[0] + d[1] == ref[0] + ref[1]
and 1 <= abs(d[0] - ref[0]) <= cap)
base = (0, 0, 0)
out = {"straddles": [], "shifts": [], "diverge": None, "unconfirmed": None}
for i, f in enumerate(common):
xd, d = exact_diff(f), offset(f)
if not xd and d == base:
continue
note = None
last = i + 1 == len(common)
if not xd and limit and shift(d, base, limit):
if last:
out["unconfirmed"] = (f, "an FMV-quiet shift")
break
out["shifts"].append((f, abs(d[0] - base[0])))
base = d
continue
m = 0 if xd else moved(d, base)
if m and limit and m <= limit:
if last:
out["unconfirmed"] = (f, f"a {m}-write VBlank straddle")
break
g = common[i + 1]
dn = offset(g)
if not exact_diff(g) and (dn == base or shift(dn, base, limit)):
out["straddles"].append((f, m))
continue
note = (f"a {m}-write straddle pattern, but frame {g} did not "
f"re-converge")
elif m and limit:
note = f"{m} writes moved, more than --straddle-writes {limit}"
elif not xd and not limit and (m == 1 or shift(d, base, 1)):
note = ("the one-write VBlank straddle pattern; --strict rejects "
"it")
cols = xd + [c for c, x, y in zip(("wc", "qc", "ws"), d, base)
if x != y and c in judged]
out["diverge"] = (f, cols, note)
break
return out
def compare(a_path, b_path, limit):
loaded = {"A": load_run(a_path), "B": load_run(b_path)}
runs = {tag: v[0] for tag, v in loaded.items()}
fps = {tag: v[1] for tag, v in loaded.items()}
present = {tag: v[2] for tag, v in loaded.items()}
fa, fb = fps["A"], fps["B"]
mismatch, incomplete, misses, notes = [], [], [], []
quiet_set = {tag: fmv_quiet(runs[tag]) for tag in runs}
for tag, path in (("A", a_path), ("B", b_path)):
run, frames = runs[tag], fps[tag]
span = f"{min(frames)}..{max(frames)}" if frames else "none"
env = run.get("env") or {}
print(f"{tag}: {path} build={run.get('build')} "
f"frames={run.get('frames')} fingerprinted={len(frames)} "
f"({span}) misses={run.get('miss_total')} "
f"fmv_quiet={quiet_text(quiet_set[tag])} "
f"autocompile_off={env.get('PSX_OVERLAY_AUTOCOMPILE_OFF', '?')} "
f"seed={run.get('seed', '?')}")
if not frames:
incomplete.append(f"{tag} has no fingerprint entries")
total, kept = run.get("ring_total"), run.get("ring_available")
if total is not None and kept is not None and total > kept:
incomplete.append(f"{tag}'s fingerprint ring wrapped ({total} "
f"snapshots, {kept} kept): its earliest frames "
f"were overwritten")
if run.get("miss_total") is None:
incomplete.append(f"{tag} has no dispatch-miss count")
elif run["miss_total"]:
misses.append(f"{tag} had {run['miss_total']} dispatch misses "
f"({run.get('miss_unique')} unique)")
lacking = [c for c in JUDGE if c not in present[tag]]
if frames and lacking:
incomplete.append(f"{tag} lacks judge column(s) "
f"{','.join(lacking)} (runtimes before psxrecomp "
f"#420 report no ws/qc): not judged")
if runs["A"].get("frames") != runs["B"].get("frames"):
incomplete.append(f"the runs stopped at different frames: A at "
f"{runs['A'].get('frames')}, B at "
f"{runs['B'].get('frames')}")
judged = [c for c in JUDGE if c in present["A"] and c in present["B"]]
common = sorted(set(fa) & set(fb))
quiet = any(q is not False for q in quiet_set.values())
effective = {t: quiet_set[t] is not False for t in quiet_set}
# FMV-quiet frames are not fingerprinted and their writes count only in
# qc. With the setting different in the two runs, once a quiet frame has
# happened the frame sets and every write column part for that reason
# alone: a DIVERGE or a missing frame would be an artefact, not a fork.
comparable = True
if effective["A"] != effective["B"]:
settings = (f"A {quiet_text(quiet_set['A'])}, "
f"B {quiet_text(quiet_set['B'])}")
signs = [f"{t} counted quiet writes (qc > 0)" for t in fps
if "qc" in judged and any(fps[t][f]["qc"] for f in fps[t])]
if set(fa) != set(fb):
signs.append(f"the frame sets differ ({len(fa)} and {len(fb)} "
f"frames fingerprinted)")
if "qc" not in judged:
signs.append("qc is not reported, so quiet frames cannot be "
"ruled out")
if signs:
comparable = False
incomplete.append(
f"the runs used different FMV-quiet settings ({settings}) "
f"and {'; '.join(signs)}: quiet frames are not fingerprinted "
f"and their writes count only in qc, so frames and columns "
f"were not judged. Rerun both with the same setting (the "
f"default is off)")
else:
notes.append(f"FMV-quiet settings differ ({settings}), but no "
f"quiet frame occurred (qc = 0, same frame set): "
f"judged as usual")
if comparable:
for tag, other_tag in (("A", "B"), ("B", "A")):
only = sorted(set(fps[tag]) - set(fps[other_tag]))
other = fps[other_tag]
if not only:
continue
if other:
lo, hi = min(other), max(other)
inside = [f for f in only if lo <= f <= hi]
outside = [f for f in only if not lo <= f <= hi]
where = f"outside its range {lo}..{hi}"
else:
inside, outside, where = [], only, "(it has none)"
if inside:
mismatch.append(f"{len(inside)} frames fingerprinted by {tag} "
f"are missing inside {other_tag}'s range: "
f"{frame_ranges(inside)}")
if outside:
incomplete.append(f"{other_tag} lacks {len(outside)} of "
f"{tag}'s frames {where}: "
f"{frame_ranges(outside)}")
if quiet:
why = ", ".join(f"{t} {quiet_text(q)}"
for t, q in quiet_set.items())
notes.append(f"FMV-quiet ({why}): "
f"wc+qc must agree; wc, qc and ws are judged against "
f"the offsets carried from any FMV-quiet shift. Rerun "
f"without --fmv-quiet for the strict rule")
else:
notes.append("FMV-quiet off in both runs: wc, qc and ws judged "
"directly")
result = scan(common, fa, fb, judged, quiet, limit)
else:
result = {"straddles": [], "shifts": [], "diverge": None,
"unconfirmed": None}
if result["diverge"]:
f, cols, note = result["diverge"]
text = f"DIVERGE at frame {f}: judge columns {cols}"
if note:
text += f" ({note})"
if result["shifts"]:
text += (f" [after {len(result['shifts'])} FMV-quiet shift(s); "
f"wc/qc/ws compared against the carried offsets]")
text += (f"\n A {show(fa[f])}\n B {show(fb[f])}"
f"\n next: arm record_frame {f} in both runs and diff the "
f"ordered logs")
mismatch.insert(0, text)
if result["unconfirmed"]:
f, what = result["unconfirmed"]
incomplete.append(f"frame {f}, the last shared frame, differs on "
f"ws/wc/qc in the pattern of {what}, but no later "
f"frame shows the runs re-converge; rerun with a "
f"larger --frames")
tolerated = []
if result["straddles"]:
n = len(result["straddles"])
most = max(m for _, m in result["straddles"])
tolerated.append(f"{n} VBlank straddle frame(s) (ws/wc off by <= "
f"{most} write(s), re-converged at the next frame): "
f"{frame_ranges([f for f, _ in result['straddles']])}")
if result["shifts"]:
tolerated.append(f"{len(result['shifts'])} FMV-quiet shift(s) (wc+qc "
f"agreed, offsets carried forward): "
f"{frame_ranges([f for f, _ in result['shifts']])}")
notes += [f"tolerated: {t}" for t in tolerated]
for loc in LOCATORS:
if not comparable:
notes.append(f"locator {loc}: not compared (FMV-quiet settings "
f"differ)")
continue
if loc not in present["A"] or loc not in present["B"]:
notes.append(f"locator {loc}: not reported by both runs")
continue
parted = [f for f in common if fa[f][loc] != fb[f][loc]]
notes.append(f"locator {loc}: " + (
f"first differs at frame {parted[0]} ({len(parted)} of "
f"{len(common)} frames); not a failure: device writes are "
f"recorded in host service order" if parted
else "equal at every shared frame"))
shared = (f"{len(common)} shared frames ({common[0]}..{common[-1]})"
if common else "no shared frames")
if mismatch:
verdict, code = "MISMATCH", EXIT_MISMATCH
elif incomplete:
verdict, code = "INCOMPLETE", EXIT_INCOMPLETE
elif misses:
verdict, code = "MISSES", EXIT_MISSES
else:
verdict, code = "IDENTICAL", EXIT_IDENTICAL
if code == EXIT_IDENTICAL:
tol = ("; tolerated " + " and ".join(
f"{len(result[k])} {what}" for k, what in
(("straddles", "VBlank straddle(s)"),
("shifts", "FMV-quiet shift(s)")) if result[k])
if tolerated else "; no tolerance applied")
print(f"IDENTICAL: {len(common)} frames ({common[0]}..{common[-1]}) "
f"agree on {','.join(judged)}{tol}")
elif comparable:
print(f"{verdict}: {shared}, judged on {','.join(judged) or 'nothing'}")
else:
print(f"{verdict}: {shared}, not judged (FMV-quiet settings differ)")
for problem in mismatch + incomplete + misses:
print(f" {problem}")
for note in notes:
print(f" {note}")
return code
def cmd_compare(args):
limit = 0 if args.strict else args.straddle_writes
if limit < 0:
raise UsageError("--straddle-writes must be >= 0")
return compare(args.a, args.b, limit)
# ---- command line ------------------------------------------------------------
class Parser(argparse.ArgumentParser):
def error(self, message):
self.print_usage(sys.stderr)
raise UsageError(message)
def parser():
top = Parser(prog="fp_identity.py", description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
sub = top.add_subparsers(dest="command", parser_class=Parser)
snap = sub.add_parser("snapshot", help="copy a state dir's overlay state "
"into a seed dir")
snap.add_argument("state_dir")
snap.add_argument("seed_dir")
snap.set_defaults(func=cmd_snapshot)
run = sub.add_parser("run", help="launch, turbo to frame N, dump the "
"fingerprint ring")
run.add_argument("out", help="dump to write (OUT.log gets the runtime log)")
run.add_argument("--launch", help="command template, e.g. "
"'tools/run_game.sh {build} --debug-port {port} {headless}'")
run.add_argument("--build", default="build", help="value of {build}")
run.add_argument("--runtime", help="runtime executable (instead of "
"--launch)")
run.add_argument("--game", help="game.toml for --runtime")
run.add_argument("--disc", help="disc image for --runtime")
run.add_argument("--cwd", default=".", help="working directory of the "
"launch (default: current)")
run.add_argument("--state-dir", help="directory holding cache/ and "
"overlay_captures.json")
run.add_argument("--seed", help="overlay state to install before launching")
run.add_argument("--port", type=int, default=4370)
run.add_argument("--frames", type=int, default=6000)
run.add_argument("--label", help="name recorded in the dump")
run.add_argument("--windowed", action="store_true",
help="no --headless (renderer-dependent checks)")
run.add_argument("--fmv-quiet", action="store_true",
help="leave FMV-quiet on (PSX_DEBUG_FMV_QUIET=1)")
run.add_argument("--autocompile", action="store_true",
help="allow overlay autocompile "
"(PSX_OVERLAY_AUTOCOMPILE_OFF=0)")
run.add_argument("--env", action="append", default=[],
metavar="KEY=VALUE", help="extra environment (repeatable)")
run.add_argument("--boot-timeout", type=float, default=180,
help="seconds for the debug server to answer")
run.add_argument("--stall", type=float, default=120,
help="seconds the frame counter may stand still")
run.add_argument("--timeout", type=float, default=3600,
help="seconds to reach frame N")
run.set_defaults(func=cmd_run)
cmp_ = sub.add_parser("compare", help="judge two dumps")
cmp_.add_argument("a")
cmp_.add_argument("b")
cmp_.add_argument("--straddle-writes", type=int, default=1,
help="most writes one VBlank straddle may move "
"(default 1; 0 = strict)")
cmp_.add_argument("--strict", action="store_true",
help="no tolerance: same as --straddle-writes 0")
cmp_.set_defaults(func=cmd_compare)
return top
def main(argv):
extra = []
if "--" in argv:
i = argv.index("--")
argv, extra = argv[:i], argv[i + 1:]
top = parser()
args = None
try:
args = top.parse_args(argv)
if not getattr(args, "func", None):
top.print_help()
return EXIT_USAGE
args.extra = extra
if extra and args.command != "run":
raise UsageError("arguments after -- only apply to run")
return args.func(args)
except UsageError as e:
print(f"fp_identity.py: error: {e}", file=sys.stderr)
return EXIT_USAGE
except RunError as e:
print(f"fp_identity.py: run failed: {e}", file=sys.stderr)
return EXIT_INCOMPLETE
except (OSError, subprocess.SubprocessError) as e:
# Seeding, the log or the dump could not be written: no result, and
# never exit 1, which means MISMATCH.
what = getattr(args, "command", None) or "command"
print(f"fp_identity.py: {what} failed: {e}", file=sys.stderr)
return EXIT_INCOMPLETE
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))
tools/tests/test_fp_identity.py
+773
View File
@@ -0,0 +1,773 @@
#!/usr/bin/env python3
"""Self-test for tools/fp_identity.py (no runtime, disc or build needed).
`compare` runs on synthetic fingerprint dumps: column judging, the VBlank
straddle and FMV-quiet tolerances, truncated and mismatched frame sets, dumps
from a runtime that predates the ws/qc columns, and the exit codes. `run`
drives a fake runtime (this file, started with --fake-runtime) through the
real launch path: command templates, --runtime arguments, seeding, the
environment defaults and a taken port.
python3 tools/tests/test_fp_identity.py (or ctest -R fp_identity)
"""
import argparse
import importlib.util
import json
import os
import shlex
import signal
import socket
import subprocess
import sys
import tempfile
import time
import unittest
from unittest import mock
HERE = os.path.dirname(os.path.abspath(__file__))
TOOL = os.path.join(os.path.dirname(HERE), "fp_identity.py")
def load_tool():
spec = importlib.util.spec_from_file_location("fp_identity", TOOL)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
# Sparse like a real run: the ring holds one entry per guest frame that
# snapshotted, not one per vblank.
FRAMES = [f for f in range(1, 2001) if f % 4 != 3]
LAST = 2000
MASK64 = (1 << 64) - 1
def term(frame):
return (frame * 0x9E3779B97F4A7C15) & MASK64
def entry(frame, quiet=False, **override):
"""One ring entry. Cumulative columns grow with the frame number."""
e = {"frame": frame, "wr": f"0x{frame * 7919:016x}",
"pc": f"0x{frame * 104729:016x}", "wc": frame * 11,
"ws": f"0x{(frame * 0x2545F4914F6CDD1D) & MASK64:016x}",
"mmio": f"0x{frame * 31:016x}", "mc": frame * 3,
"sp": f"0x{frame * 17:016x}", "sc": frame * 5,
"qc": frame * 2 if quiet else 0, "cyc": frame * 564480}
e.update(override)
return e
def run_dump(frames=None, last=LAST, misses=0, quiet=False, env=True,
**extra):
dump = {"schema": 2, "build": "build", "frames": last,
"miss_total": misses, "miss_unique": 1 if misses else 0,
"ring_total": len(FRAMES) + 40, "ring_available": len(FRAMES) + 40,
"entries": [entry(f, quiet) for f in
(FRAMES if frames is None else frames)]}
if env:
dump["env"] = {"PSX_DEBUG_FMV_QUIET": "1" if quiet else "0",
"PSX_OVERLAY_AUTOCOMPILE_OFF": "1"}
dump.update(extra)
return dump
def bump(dump, index, stop=None, **delta):
"""Add delta to wc/qc/ws of entries[index:stop] (ws modulo 2^64)."""
for e in dump["entries"][index:stop]:
for k, v in delta.items():
if k == "ws":
e[k] = f"0x{(int(e[k], 16) + v) & MASK64:016x}"
else:
e[k] += v
return dump
def at(i):
return FRAMES[i]
class Base(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.dir = self.tmp.name
def tearDown(self):
self.tmp.cleanup()
def tool(self, *args, timeout=120):
proc = subprocess.run([sys.executable, TOOL, *args],
capture_output=True, text=True, timeout=timeout)
return proc.returncode, proc.stdout + proc.stderr
def compare(self, a, b, *flags):
paths = []
for name, dump in (("a.json", a), ("b.json", b)):
path = os.path.join(self.dir, name)
with open(path, "w") as f:
json.dump(dump, f)
paths.append(path)
return self.tool("compare", *paths, *flags)
def assertVerdict(self, result, code, verdict, *details):
status, out = result
self.assertEqual(status, code, out)
verdicts = [line.split(":")[0] for line in out.splitlines()
if line.split(":")[0] in
("IDENTICAL", "MISMATCH", "INCOMPLETE", "MISSES")]
self.assertEqual(verdicts, [verdict], out)
for detail in details:
self.assertIn(detail, out)
return out
class FrameSetTest(Base):
"""The frame-set and miss checks carried over from the R4 tool."""
def test_identical(self):
out = self.assertVerdict(
self.compare(run_dump(), run_dump()), 0, "IDENTICAL",
f"IDENTICAL: {len(FRAMES)} frames (1..2000) agree on "
f"cyc,mmio,mc,sp,sc,wc,ws,qc; no tolerance applied",
"locator wr: equal at every shared frame",
"FMV-quiet off in both runs")
self.assertNotIn("tolerated:", out)
def test_truncated_candidate_with_misses_is_not_identical(self):
# A candidate cut to its first 100 entries, with miss_total=5, once
# reported IDENTICAL over the shared frames.
cut = run_dump(misses=5)
cut["entries"] = cut["entries"][:100]
self.assertVerdict(self.compare(run_dump(), cut), 3, "INCOMPLETE",
"B lacks", "outside its range 1..133",
"B had 5 dispatch misses")
def test_truncated_either_side(self):
short = FRAMES[:500]
self.assertVerdict(self.compare(run_dump(), run_dump(short)), 3,
"INCOMPLETE", "B lacks 1000 of A's frames")
self.assertVerdict(self.compare(run_dump(short), run_dump()), 3,
"INCOMPLETE", "A lacks 1000 of B's frames")
def test_missing_head(self):
self.assertVerdict(self.compare(run_dump(), run_dump(FRAMES[10:])), 3,
"INCOMPLETE", "B lacks 10 of A's frames",
"1..2, 4..6")
def test_different_window(self):
self.assertVerdict(self.compare(run_dump(), run_dump(last=LAST + 1)), 3,
"INCOMPLETE", "A at 2000, B at 2001")
def test_interior_frame_set_difference_is_a_mismatch(self):
holed = [f for f in FRAMES if f != 1000]
self.assertVerdict(self.compare(run_dump(), run_dump(holed)), 1,
"MISMATCH", "fingerprinted by A are missing inside "
"B's range: 1000")
def test_divergence_and_truncation_report_both(self):
b = run_dump(FRAMES[:500])
b["entries"][10]["mmio"] = "0x0"
self.assertVerdict(self.compare(run_dump(), b), 1, "MISMATCH",
f"DIVERGE at frame {at(10)}",
"B lacks 1000 of A's frames")
def test_dispatch_misses(self):
self.assertVerdict(self.compare(run_dump(misses=2), run_dump()), 4,
"MISSES", "A had 2 dispatch misses")
def test_unknown_miss_count(self):
self.assertVerdict(self.compare(run_dump(), run_dump(miss_total=None)),
3, "INCOMPLETE", "B has no dispatch-miss count")
def test_wrapped_ring(self):
wrapped = run_dump(ring_total=40000, ring_available=32768)
self.assertVerdict(self.compare(run_dump(), wrapped), 3, "INCOMPLETE",
"B's fingerprint ring wrapped")
def test_dump_without_ring_counts_still_compares(self):
old = run_dump()
del old["ring_total"], old["ring_available"]
self.assertVerdict(self.compare(old, run_dump()), 0, "IDENTICAL")
def test_empty_run(self):
self.assertVerdict(self.compare(run_dump(), run_dump([])), 3,
"INCOMPLETE", "B has no fingerprint entries",
"no shared frames")
class ColumnTest(Base):
"""Judge columns fail a run; locator columns only say where runs part."""
def test_every_judge_column_is_judged(self):
for col, value in (("cyc", 1), ("mmio", "0x1"), ("mc", 1),
("sp", "0x1"), ("sc", 1), ("qc", 99),
("ws", "0x1")):
with self.subTest(col=col):
b = run_dump()
b["entries"][300][col] = value
self.assertVerdict(self.compare(run_dump(), b), 1, "MISMATCH",
f"DIVERGE at frame {at(300)}: judge "
f"columns ['{col}']",
f"record_frame {at(300)}")
def test_ws_value_fork_with_equal_count_is_not_a_straddle(self):
# Same number of writes, different values: a real fork.
b = bump(run_dump(), 300, 301, ws=12345)
self.assertVerdict(self.compare(run_dump(), b), 1, "MISMATCH",
f"DIVERGE at frame {at(300)}: judge columns ['ws']")
def test_locators_are_reported_not_judged(self):
b = run_dump()
for e in b["entries"][700:]:
e["wr"], e["pc"] = "0x1", "0x2"
self.assertVerdict(self.compare(run_dump(), b), 0, "IDENTICAL",
f"locator wr: first differs at frame {at(700)} "
f"({len(FRAMES) - 700} of {len(FRAMES)} frames); "
f"not a failure",
f"locator pc: first differs at frame {at(700)}")
def test_mismatch_reports_locators_too(self):
b = run_dump()
b["entries"][500]["wr"] = "0x1"
b["entries"][600]["sc"] = 0
self.assertVerdict(self.compare(run_dump(), b), 1, "MISMATCH",
f"DIVERGE at frame {at(600)}: judge columns ['sc']",
f"locator wr: first differs at frame {at(500)}")
def test_hex_and_integer_spellings_compare_equal(self):
b = run_dump()
for e in b["entries"]:
e["mmio"] = int(e["mmio"], 16)
e["wc"] = str(e["wc"])
self.assertVerdict(self.compare(run_dump(), b), 0, "IDENTICAL")
def test_older_runtime_without_ws_qc(self):
# Dumps from a runtime before psxrecomp #420: no ws/qc, no env.
old = run_dump(env=False)
for e in old["entries"]:
del e["ws"], e["qc"]
self.assertVerdict(self.compare(old, json.loads(json.dumps(old))), 3,
"INCOMPLETE", "A lacks judge column(s) ws,qc",
"B lacks judge column(s) ws,qc",
"judged on cyc,mmio,mc,sp,sc,wc")
def test_older_runtime_on_one_side_only(self):
old = run_dump()
for e in old["entries"]:
del e["ws"]
self.assertVerdict(self.compare(run_dump(), old), 3, "INCOMPLETE",
"B lacks judge column(s) ws")
def test_older_runtime_still_reports_divergence(self):
old = run_dump(env=False)
for e in old["entries"]:
del e["ws"], e["qc"]
b = json.loads(json.dumps(old))
b["entries"][40]["mc"] = 0
self.assertVerdict(self.compare(old, b), 1, "MISMATCH",
f"DIVERGE at frame {at(40)}: judge columns ['mc']",
"lacks judge column(s) ws,qc")
class StraddleTest(Base):
"""The one-write VBlank straddle of docs/TCP_COMMANDS.md."""
def straddled(self, index, writes=1, sign=1):
return bump(run_dump(), index, index + 1, wc=sign * writes,
ws=sign * term(index))
def test_one_write_straddle_is_tolerated_and_reported(self):
b = self.straddled(300)
b = bump(b, 900, 901, wc=-1, ws=-term(900))
self.assertVerdict(self.compare(run_dump(), b), 0, "IDENTICAL",
"tolerated 2 VBlank straddle(s)",
"tolerated: 2 VBlank straddle frame(s) (ws/wc off "
f"by <= 1 write(s), re-converged at the next "
f"frame): {at(300)}, {at(900)}")
def test_strict_rejects_a_straddle(self):
self.assertVerdict(self.compare(run_dump(), self.straddled(300),
"--strict"), 1, "MISMATCH",
f"DIVERGE at frame {at(300)}: judge columns "
f"['wc', 'ws'] (the one-write VBlank straddle "
f"pattern; --strict rejects it)")
def test_straddle_that_does_not_reconverge(self):
b = bump(run_dump(), 300, None, wc=1, ws=term(300))
self.assertVerdict(self.compare(run_dump(), b), 1, "MISMATCH",
f"DIVERGE at frame {at(300)}: judge columns "
f"['wc', 'ws'] (a 1-write straddle pattern, but "
f"frame {at(301)} did not re-converge)")
def test_two_write_straddle_needs_a_wider_limit(self):
b = self.straddled(300, writes=2)
self.assertVerdict(self.compare(run_dump(), b), 1, "MISMATCH",
"2 writes moved, more than --straddle-writes 1")
self.assertVerdict(self.compare(run_dump(), b, "--straddle-writes",
"2"), 0, "IDENTICAL",
"ws/wc off by <= 2 write(s)")
def test_straddle_with_another_column_is_a_divergence(self):
b = self.straddled(300)
b["entries"][300]["mmio"] = "0x5"
self.assertVerdict(self.compare(run_dump(), b), 1, "MISMATCH",
f"DIVERGE at frame {at(300)}: judge columns "
f"['mmio', 'wc', 'ws']")
def test_next_frame_diverging_is_not_a_straddle(self):
b = self.straddled(300)
b["entries"][301]["cyc"] = 1
self.assertVerdict(self.compare(run_dump(), b), 1, "MISMATCH",
f"DIVERGE at frame {at(300)}",
f"frame {at(301)} did not re-converge")
def test_straddle_on_the_last_frame_is_unconfirmed(self):
b = self.straddled(len(FRAMES) - 1)
self.assertVerdict(self.compare(run_dump(), b), 3, "INCOMPLETE",
f"frame {FRAMES[-1]}, the last shared frame",
"a 1-write VBlank straddle",
"rerun with a larger --frames")
def test_bad_straddle_limit_is_a_usage_error(self):
status, out = self.compare(run_dump(), run_dump(),
"--straddle-writes", "-1")
self.assertEqual(status, 2, out)
class FmvQuietTest(Base):
"""With FMV-quiet on, a straddle can move a write between wc and qc."""
def shifted(self, index, **extra):
# From `index` on, B counted one write in wc that A counted in qc.
return bump(run_dump(quiet=True, **extra), index, None, wc=1, qc=-1,
ws=term(index))
def test_quiet_shift_is_tolerated_with_carried_offsets(self):
self.assertVerdict(self.compare(run_dump(quiet=True),
self.shifted(300)), 0, "IDENTICAL",
"tolerated 1 FMV-quiet shift(s)",
f"FMV-quiet shift(s) (wc+qc agreed, offsets "
f"carried forward): {at(300)}",
"wc+qc must agree")
def test_quiet_shift_after_a_quiet_straddle(self):
# A's write lands in a quiet frame (qc) one frame before B's, which
# lands in the next, non-quiet frame (wc).
b = run_dump(quiet=True)
bump(b, 300, 301, qc=-1)
bump(b, 301, None, wc=1, qc=-1, ws=term(301))
self.assertVerdict(self.compare(run_dump(quiet=True), b), 0,
"IDENTICAL", "1 VBlank straddle(s) and 1 FMV-quiet "
"shift(s)")
def test_ws_drift_after_a_shift_is_a_divergence(self):
b = bump(self.shifted(300), 800, None, ws=77)
self.assertVerdict(self.compare(run_dump(quiet=True), b), 1,
"MISMATCH", f"DIVERGE at frame {at(800)}: judge "
f"columns ['ws']", "after 1 FMV-quiet shift(s)")
def test_wc_plus_qc_must_agree(self):
b = bump(run_dump(quiet=True), 300, None, qc=1)
self.assertVerdict(self.compare(run_dump(quiet=True), b), 1,
"MISMATCH", f"DIVERGE at frame {at(300)}: judge "
f"columns ['qc']")
def test_shift_is_not_tolerated_with_quiet_off(self):
# The same pattern with FMV-quiet off in both runs is a fork.
a = run_dump()
b = bump(run_dump(), 300, None, wc=1, qc=-1, ws=term(300))
self.assertVerdict(self.compare(a, b), 1, "MISMATCH",
f"DIVERGE at frame {at(300)}: judge columns "
f"['wc', 'qc', 'ws']")
def test_different_quiet_settings_are_incomplete(self):
a = run_dump(quiet=True)
b = run_dump(quiet=True)
b["env"]["PSX_DEBUG_FMV_QUIET"] = "0"
self.assertVerdict(self.compare(a, b), 3, "INCOMPLETE",
"different FMV-quiet settings (A on, B off)")
def test_mixed_settings_with_quiet_frames_are_not_judged(self):
# As on R4: the quiet run skips the fingerprint on quiet frames and
# counts their writes (MMIO too) in qc, so frames and columns part
# from the first quiet frame. That is the setting, not a fork.
a = run_dump()
quiet_frames = set(FRAMES[300:400])
b = run_dump([f for f in FRAMES if f not in quiet_frames])
b["env"]["PSX_DEBUG_FMV_QUIET"] = "1"
for e in b["entries"]:
if e["frame"] > FRAMES[300]:
e["qc"], e["mmio"], e["mc"] = 50, "0x1", 0
out = self.assertVerdict(
self.compare(a, b), 3, "INCOMPLETE",
"not judged (FMV-quiet settings differ)",
"different FMV-quiet settings (A off, B on)",
"B counted quiet writes (qc > 0)",
f"the frame sets differ ({len(FRAMES)} and {len(FRAMES) - 100} "
f"frames fingerprinted)",
"Rerun both with the same setting",
"locator wr: not compared")
for artefact in ("DIVERGE", "record_frame", "missing inside",
"tolerated:"):
self.assertNotIn(artefact, out)
def test_mixed_settings_without_qc_column_are_not_judged(self):
a, b = run_dump(), run_dump(env=False)
for dump in (a, b):
for e in dump["entries"]:
del e["qc"]
self.assertVerdict(self.compare(a, b), 3, "INCOMPLETE",
"qc is not reported, so quiet frames cannot be "
"ruled out")
def test_mixed_settings_without_a_quiet_frame_are_judged(self):
# FMV-quiet on in B, but no MDEC decode: nothing was kept quiet.
b = run_dump()
b["env"]["PSX_DEBUG_FMV_QUIET"] = "1"
self.assertVerdict(self.compare(run_dump(), b), 0, "IDENTICAL",
"FMV-quiet settings differ (A off, B on), but no "
"quiet frame occurred")
b["entries"][300]["cyc"] = 1
self.assertVerdict(self.compare(run_dump(), b), 1, "MISMATCH",
f"DIVERGE at frame {at(300)}: judge columns "
f"['cyc']")
def test_unrecorded_quiet_setting_uses_the_quiet_rule(self):
a, b = run_dump(quiet=True, env=False), self.shifted(300, env=False)
self.assertVerdict(self.compare(a, b), 0, "IDENTICAL",
"not recorded (runtime default: on)",
"tolerated 1 FMV-quiet shift(s)")
class UsageTest(Base):
def test_compare_needs_two_dumps(self):
status, out = self.tool("compare", "only-one")
self.assertEqual(status, 2, out)
def test_unreadable_dump(self):
bad = os.path.join(self.dir, "bad.json")
with open(bad, "w") as f:
f.write("{not json")
status, out = self.tool("compare", bad, bad)
self.assertEqual(status, 2, out)
self.assertIn("cannot read", out)
def test_no_command(self):
status, _ = self.tool()
self.assertEqual(status, 2)
def test_run_needs_a_launcher(self):
status, out = self.tool("run", os.path.join(self.dir, "x.json"))
self.assertEqual(status, 2, out)
self.assertIn("--launch", out)
def test_windowed_with_a_literal_headless(self):
status, out = self.tool("run", "x.json", "--launch",
"game --headless", "--windowed")
self.assertEqual(status, 2, out)
self.assertIn("{headless}", out)
def test_unknown_placeholder(self):
status, out = self.tool("run", "x.json", "--launch", "game {bogus}")
self.assertEqual(status, 2, out)
self.assertIn("bad placeholder", out)
def test_seed_without_state_dir(self):
status, out = self.tool("run", "x.json", "--launch", "game {port}",
"--seed", self.dir)
self.assertEqual(status, 2, out)
self.assertIn("--seed needs --state-dir", out)
class LaunchTemplateTest(unittest.TestCase):
"""How --launch is split into arguments (no runtime needed)."""
@classmethod
def setUpClass(cls):
cls.fp = load_tool()
def argv(self, template, windows, **extra):
args = argparse.Namespace(launch=template, build="build", port=4781,
windowed=False, extra=[], **extra)
return self.fp.launch_argv(args, None, windows=windows)
def test_windows_paths_keep_their_backslashes(self):
self.assertEqual(
self.argv(r"C:\psx\build\game-runtime.exe --game "
r"build\game.toml --debug-port {port} {headless}", True),
[r"C:\psx\build\game-runtime.exe", "--game", r"build\game.toml",
"--debug-port", "4781", "--headless"])
def test_windows_quoted_path_with_spaces(self):
self.assertEqual(
self.argv(r'"C:\Program Files\psx\game.exe" --debug-port {port}',
True),
[r"C:\Program Files\psx\game.exe", "--debug-port", "4781"])
def test_posix_keeps_shell_escapes(self):
self.assertEqual(
self.argv(r"tools/run\ game.sh '{build}' --debug-port {port}",
False),
["tools/run game.sh", "build", "--debug-port", "4781"])
def test_unbalanced_quote_is_a_usage_error(self):
for windows in (False, True):
with self.subTest(windows=windows):
with self.assertRaises(self.fp.UsageError):
self.argv('"C:\\psx\\game.exe --debug-port {port}',
windows)
# ---- fake runtime ------------------------------------------------------------
def fake_runtime(argv):
"""A debug server that answers the four commands `run` sends."""
port = int(argv[argv.index("--debug-port") + 1])
report = argv[argv.index("--report") + 1]
with open(report, "w") as f:
json.dump({"argv": argv, "env": {
k: os.environ.get(k) for k in
("PSX_DEBUG_FMV_QUIET", "PSX_OVERLAY_AUTOCOMPILE_OFF", "FAKE_X")},
"cwd": os.getcwd()}, f)
if "--die" in argv:
sys.exit(7)
srv = socket.socket()
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind(("127.0.0.1", port))
srv.listen(8)
start = time.time()
signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
while True:
conn, _ = srv.accept()
with conn:
line = b""
while not line.endswith(b"\n"):
chunk = conn.recv(65536)
if not chunk:
break
line += chunk
req = json.loads(line or b"{}")
frame = int((time.time() - start) * 40000)
cmd = req.get("cmd")
if "--crash-on" in argv and cmd == argv[argv.index("--crash-on")
+ 1]:
os._exit(9) # dies mid-request, no reply
if "--hang-up-on" in argv and cmd == argv[
argv.index("--hang-up-on") + 1]:
continue # closes the connection, no reply
if cmd == "frame_fingerprint":
lo, hi = req.get("frame_lo", 0), req.get("frame_hi", frame)
entries = [entry(f) for f in FRAMES if lo <= f <= hi]
reply = {"ok": True, "total": len(FRAMES),
"available": len(FRAMES), "entries": entries}
elif cmd == "dispatch_stats":
reply = {"ok": True, "miss_total": 0, "miss_unique": 0}
else:
reply = {"ok": True, "frame": frame}
conn.sendall(json.dumps(reply).encode() + b"\n")
def free_port():
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
@unittest.skipIf(os.name == "nt", "the fake runtime is a POSIX launch")
class RunTest(Base):
def launch(self, out, *args):
return self.tool("run", os.path.join(self.dir, out), "--frames",
str(LAST), "--boot-timeout", "30", "--stall", "30",
"--timeout", "60", *args)
def template(self, report, extra=""):
return (f"{shlex.quote(sys.executable)} {shlex.quote(__file__)} "
f"--fake-runtime {{headless}} --debug-port {{port}} "
f"--report {shlex.quote(report)} {extra}")
def report(self, name):
with open(os.path.join(self.dir, name)) as f:
return json.load(f)
def test_template_run_then_compare(self):
rep = os.path.join(self.dir, "rep.json")
port = str(free_port())
for out in ("a.json", "b.json"):
status, text = self.launch(out, "--launch", self.template(rep),
"--port", port, "--env", "FAKE_X=7",
"--", "--extra-arg")
self.assertEqual(status, 0, text)
seen = self.report("rep.json")
self.assertEqual(seen["env"], {"PSX_DEBUG_FMV_QUIET": "0",
"PSX_OVERLAY_AUTOCOMPILE_OFF": "1",
"FAKE_X": "7"})
self.assertIn("--headless", seen["argv"])
self.assertEqual(seen["argv"][-1], "--extra-arg")
with open(os.path.join(self.dir, "a.json")) as f:
dump = json.load(f)
self.assertEqual(dump["frames"], LAST)
self.assertEqual(dump["miss_total"], 0)
self.assertEqual(dump["env"]["PSX_DEBUG_FMV_QUIET"], "0")
self.assertEqual(len(dump["entries"]), len(FRAMES))
status, text = self.tool("compare", os.path.join(self.dir, "a.json"),
os.path.join(self.dir, "b.json"))
self.assertEqual(status, 0, text)
self.assertIn("IDENTICAL", text)
def test_windowed_and_opt_ins(self):
rep = os.path.join(self.dir, "rep.json")
status, text = self.launch("a.json", "--launch", self.template(rep),
"--port", str(free_port()), "--windowed",
"--fmv-quiet", "--autocompile")
self.assertEqual(status, 0, text)
seen = self.report("rep.json")
self.assertNotIn("--headless", seen["argv"])
self.assertEqual(seen["env"]["PSX_DEBUG_FMV_QUIET"], "1")
self.assertEqual(seen["env"]["PSX_OVERLAY_AUTOCOMPILE_OFF"], "0")
def test_runtime_arguments_and_seed(self):
# --runtime mode: the state dir defaults to the runtime's directory.
bindir = os.path.join(self.dir, "build")
os.makedirs(os.path.join(bindir, "cache", "stale"))
with open(os.path.join(bindir, "overlay_captures.json"), "w") as f:
f.write("old")
runtime = os.path.join(bindir, "fake-runtime")
rep = os.path.join(self.dir, "rep.json")
with open(runtime, "w") as f:
f.write(f"#!/bin/sh\nexec {shlex.quote(sys.executable)} "
f"{shlex.quote(__file__)} --fake-runtime --report "
f"{shlex.quote(rep)} \"$@\"\n")
os.chmod(runtime, 0o755)
seed = os.path.join(self.dir, "seed")
os.makedirs(os.path.join(seed, "cache", "shard"))
status, text = self.launch("a.json", "--runtime", "build/fake-runtime",
"--game", "g.toml", "--disc", "d.cue",
"--cwd", self.dir, "--seed", seed,
"--port", str(free_port()))
self.assertEqual(status, 0, text)
seen = self.report("rep.json")
self.assertEqual(seen["argv"][seen["argv"].index("--game") + 1],
"g.toml")
self.assertIn("--no-launcher", seen["argv"])
self.assertEqual(os.path.realpath(seen["cwd"]),
os.path.realpath(self.dir))
self.assertTrue(os.path.isdir(os.path.join(bindir, "cache", "shard")))
self.assertFalse(os.path.exists(os.path.join(bindir, "cache",
"stale")))
self.assertFalse(os.path.exists(os.path.join(
bindir, "overlay_captures.json")))
def test_snapshot_copies_overlay_state(self):
state = os.path.join(self.dir, "state")
os.makedirs(os.path.join(state, "cache", "shard"))
with open(os.path.join(state, "overlay_captures.json"), "w") as f:
f.write("{}")
seed = os.path.join(self.dir, "seed")
status, text = self.tool("snapshot", state, seed)
self.assertEqual(status, 0, text)
self.assertTrue(os.path.isdir(os.path.join(seed, "cache", "shard")))
self.assertTrue(os.path.isfile(os.path.join(
seed, "overlay_captures.json")))
def test_taken_port_is_refused(self):
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
s.listen(1)
port = str(s.getsockname()[1])
status, text = self.launch(
"a.json", "--launch",
self.template(os.path.join(self.dir, "rep.json")),
"--port", port)
self.assertEqual(status, 3, text)
self.assertIn("already has a listener", text)
self.assertFalse(os.path.exists(os.path.join(self.dir, "rep.json")))
def test_runtime_that_fails_while_dumping(self):
# Reaches frame N, then crashes or drops the connection on one of
# the final queries: exit 3 with a reason, never 1 (MISMATCH) with a
# traceback, and no dump.
for how, cmd, reason in (
("--crash-on", "frame_fingerprint",
"runtime exited (status 9) on frame_fingerprint after "
"reaching frame"),
("--crash-on", "dispatch_stats",
"runtime exited (status 9) on dispatch_stats"),
("--hang-up-on", "frame_fingerprint",
"runtime did not answer on frame_fingerprint"),
("--hang-up-on", "dispatch_stats",
"runtime did not answer on dispatch_stats")):
with self.subTest(how=how, cmd=cmd):
rep = os.path.join(self.dir, "rep.json")
status, text = self.launch(
"a.json", "--launch", self.template(rep, f"{how} {cmd}"),
"--port", str(free_port()))
self.assertEqual(status, 3, text)
self.assertIn(reason, text)
self.assertNotIn("Traceback", text)
for name in ("a.json", "a.json.tmp"):
self.assertFalse(os.path.exists(os.path.join(self.dir,
name)))
@unittest.skipIf(hasattr(os, "geteuid") and os.geteuid() == 0,
"root reads a mode-000 file")
def test_io_errors_exit_incomplete(self):
seed = os.path.join(self.dir, "seed")
os.makedirs(os.path.join(seed, "cache"))
locked = os.path.join(seed, "cache", "shard.bin")
with open(locked, "w") as f:
f.write("x")
os.chmod(locked, 0)
state = os.path.join(self.dir, "state")
os.makedirs(state)
rep = os.path.join(self.dir, "rep.json")
try:
status, text = self.launch("a.json", "--launch",
self.template(rep), "--port",
str(free_port()), "--state-dir", state,
"--seed", seed)
self.assertEqual(status, 3, text)
self.assertIn("run failed", text)
self.assertNotIn("Traceback", text)
self.assertFalse(os.path.exists(rep)) # never launched
status, text = self.tool("snapshot", seed,
os.path.join(self.dir, "copy"))
self.assertEqual(status, 3, text)
self.assertIn("snapshot failed", text)
finally:
os.chmod(locked, 0o644)
def test_missing_output_directory_is_a_usage_error(self):
status, text = self.launch(os.path.join("no", "such", "a.json"),
"--launch", self.template("rep.json"),
"--port", str(free_port()))
self.assertEqual(status, 2, text)
self.assertIn("output directory", text)
def test_stop_reports_a_process_that_will_not_exit(self):
fp = load_tool()
class Stuck:
pid = 999999999
def wait(self, timeout):
raise subprocess.TimeoutExpired("runtime", timeout)
with mock.patch.object(fp.os, "killpg") as killpg:
self.assertIn("did not exit after SIGKILL", fp.stop(Stuck()))
self.assertEqual([c.args[1] for c in killpg.call_args_list],
[signal.SIGTERM, signal.SIGKILL])
def test_runtime_that_exits_early(self):
rep = os.path.join(self.dir, "rep.json")
status, text = self.launch("a.json", "--launch",
self.template(rep, "--die"),
"--port", str(free_port()))
self.assertEqual(status, 3, text)
self.assertIn("runtime exited (status 7)", text)
self.assertFalse(os.path.exists(os.path.join(self.dir, "a.json")))
if __name__ == "__main__":
if "--fake-runtime" in sys.argv:
fake_runtime(sys.argv)
else:
unittest.main()