mirror of
https://github.com/RetroPortingToolKit/psxrecomp
synced 2026-10-25 18:05:12 -04:00
tools: find install-slot ranges and A/B what they buy, for any title
The install-slot range work is only trustworthy if a second title can reproduce it, and the two tools that produced its evidence lived in one game repo with that game's paths baked in. Moved into the framework and generalised: every path defaults off --project-root (cwd) and the BIOS profile's out_stem, and the exe is found by globbing the build tree, because its name is derived from the window title and so differs per title. Run either from a game project root. kernel_patch_diff.py boots headless and diffs live kernel RAM against the ROM copy it came from, joining each differing word to the compiled kernel bodies. It reads the emitted PsxKernelPatchRange table, so it separates a DECLARED patch (the verifier skips it, the body still runs native) from one that genuinely unblesses a body. Without that split its verdict line calls every declared range "these interpret", which is exactly backwards once the ranges work. This is the tool you use to find ranges for a new image or a new Psy-Q release. kernel_patch_ab.py answers "what did declaring them buy". It boots the same binary twice to the same frame, once with the ranges live and once with PSX_KERNEL_PATCH_RANGES=0, which drops them and restores the whole-body bless memcmp. Holding the binary constant holds the overlays, the codegen and the disc constant too, so only the bless verdict moves. Headless runs uncapped, so it normalises per frame rather than comparing raw totals at different frame counts. Two things it gets right that cost real time to learn: - It does NOT discount the A0/B0/C0 vectors from the headline. An earlier version did, reasoning that they interpret on both sides and would mask the result. True on one title, false in general: on Mega Man X6 with OpenBIOS the B0 vector burns 2,361,470 instructions over 1,180,735 entries with the ranges off and vanishes entirely with them on, because a clean bless verdict lets the vector body dispatch native. Discounting it hid a 130 insn/frame win. Kernel and all-interp are the headline; the bodies/vectors split is printed underneath as diagnosis. - It fails fast on a wedge, detected as a frame counter that stops advancing while the debug server still answers, instead of burning the whole timeout. That is how a dispatch key inside a declared range was caught spinning the dispatch loop at frame 0. debug_client.query grew an optional timeout so these can use it instead of carrying their own transport; the 10 s default is too short for dirty_ram_stats and a whole-window read_ram. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -143,9 +143,14 @@ def send_cmd(sock, cmd_dict):
|
||||
return json.loads(buf.decode().strip())
|
||||
|
||||
|
||||
def query(host, port, cmd_dict):
|
||||
"""One-shot: connect, send, receive, close."""
|
||||
s = connect(host, port)
|
||||
def query(host, port, cmd_dict, timeout=10.0):
|
||||
"""One-shot: connect, send, receive, close.
|
||||
|
||||
`timeout` is per-socket-operation. The 10 s default suits the small
|
||||
commands; bulk ones (dirty_ram_stats walks a large PC table, read_ram of a
|
||||
whole window) need more, so callers that ask for those must raise it.
|
||||
"""
|
||||
s = connect(host, port, timeout)
|
||||
try:
|
||||
return send_cmd(s, cmd_dict)
|
||||
finally:
|
||||
|
||||
@@ -0,0 +1,195 @@
|
||||
#!/usr/bin/env python3
|
||||
"""A/B the declared kernel install-slot ranges: what do they buy this game?
|
||||
|
||||
Boots the game twice against the SAME binary, to the same frame, once with the
|
||||
profile's [[recompiler.install_slots]] ranges live and once with
|
||||
PSX_KERNEL_PATCH_RANGES=0, which drops them and restores the whole-body
|
||||
kernel-bless memcmp - the behaviour before the ranges existed. One binary
|
||||
measures both sides, so the overlays, the codegen and the disc are held
|
||||
constant and only the bless verdict moves.
|
||||
|
||||
Run from a game project root:
|
||||
|
||||
python psxrecomp/tools/kernel_patch_ab.py
|
||||
python psxrecomp/tools/kernel_patch_ab.py --bios psxrecomp/bios/SCPH1001.BIN
|
||||
python psxrecomp/tools/kernel_patch_ab.py --frames 20000
|
||||
|
||||
Set PSX_BIOS_HLE=0 to isolate this change from the kernel-call HLE tier. That
|
||||
matters on an image exporting a DeliverEvent anchor (retail SCPH-1001);
|
||||
OpenBIOS has none, so its kernel calls are LLE either way.
|
||||
|
||||
Headless runs uncapped, so a frame is a unit of guest work rather than wall
|
||||
time, and the two sides reach slightly different frames in the same wall
|
||||
budget - the report normalises per frame.
|
||||
|
||||
Read "kernel" and "all" as the result. The bodies/vectors split underneath is
|
||||
diagnosis: whether the A0/B0/C0 vectors interpret on both sides is
|
||||
title-dependent (see kernel_patch_common.VECTOR_STUBS), so they are shown but
|
||||
never discounted.
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
import kernel_patch_common as kp
|
||||
|
||||
|
||||
def wait_port(host, port, secs=180):
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < secs:
|
||||
try:
|
||||
r = kp.q("frame", host, port, timeout=30.0)
|
||||
if r.get("ok") or "frame" in r:
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(1)
|
||||
raise SystemExit("debug port never came up")
|
||||
|
||||
|
||||
def run_side(a, root, ranges_on):
|
||||
env = dict(os.environ, PSX_STARVATION_TIMEOUT_US="0")
|
||||
env.pop("PSX_LOAD_SLOT", None)
|
||||
if ranges_on:
|
||||
env.pop("PSX_KERNEL_PATCH_RANGES", None)
|
||||
else:
|
||||
env["PSX_KERNEL_PATCH_RANGES"] = "0"
|
||||
argv = [os.path.join(root, a.exe), "--game", a.game, "--no-launcher",
|
||||
"--headless", "--debug-port", str(a.port)]
|
||||
if a.bios:
|
||||
argv += ["--bios", os.path.join(root, a.bios)]
|
||||
proc = subprocess.Popen(argv, cwd=root, env=env,
|
||||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
|
||||
try:
|
||||
wait_port(a.host, a.port)
|
||||
t0 = time.time()
|
||||
# A wedge shows up as a frame counter that stops advancing while the
|
||||
# debug server still answers, so watch for STALL as well as the overall
|
||||
# timeout - a boot that hangs at frame 0 should not burn the whole
|
||||
# budget before saying so. It did once: a dispatch key inside a
|
||||
# declared range spun the dispatch loop forever.
|
||||
last_fr, last_move = -1, time.time()
|
||||
while True:
|
||||
try:
|
||||
fr = kp.q("frame", a.host, a.port, timeout=30.0).get("frame") or 0
|
||||
except Exception:
|
||||
fr = last_fr # transient; the stall check still applies
|
||||
if fr >= a.frames:
|
||||
break
|
||||
if fr != last_fr:
|
||||
last_fr, last_move = fr, time.time()
|
||||
elif time.time() - last_move > a.stall:
|
||||
raise SystemExit("WEDGED: frame stuck at %s for %ds (target %d)"
|
||||
% (fr, a.stall, a.frames))
|
||||
if time.time() - t0 > a.timeout:
|
||||
raise SystemExit("only reached frame %s of %d in %ds"
|
||||
% (fr, a.frames, a.timeout))
|
||||
time.sleep(1.0)
|
||||
kb = kp.q("kernel_bless", a.host, a.port)
|
||||
st = kp.q("dirty_ram_stats", a.host, a.port)
|
||||
rows = [r for r in st.get("per_pc", [])
|
||||
if int(r["pc"], 16) < kp.KERNEL_WINDOW_END]
|
||||
kernel = sum(r["insns"] for r in rows)
|
||||
tramp = sum(r["insns"] for r in rows
|
||||
if int(r["pc"], 16) in kp.VECTOR_STUBS)
|
||||
top = sorted(rows, key=lambda r: -r["insns"])[:12]
|
||||
return {
|
||||
"ranges_on": ranges_on,
|
||||
"frame": fr,
|
||||
"wall_s": round(time.time() - t0, 1),
|
||||
"bless_entries": kb.get("entries"),
|
||||
"bless_clean": kb.get("clean"),
|
||||
"bless_mismatch": kb.get("mismatch"),
|
||||
"bless_native_hits": kb.get("native_hits"),
|
||||
"patch_ranges": kb.get("patch_ranges"),
|
||||
"patch_skips": kb.get("patch_skips"),
|
||||
"interp_insns_total": st.get("insns_run"),
|
||||
"interp_insns_kernel": kernel,
|
||||
"interp_insns_vectors": tramp,
|
||||
"interp_insns_kernel_bodies": kernel - tramp,
|
||||
"native_handoffs": st.get("native_handoffs"),
|
||||
"top_kernel_pcs": [{"pc": r["pc"], "insns": r["insns"],
|
||||
"entries": r["entries"]} for r in top],
|
||||
}
|
||||
finally:
|
||||
proc.kill()
|
||||
time.sleep(2)
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description=__doc__.split(chr(10) + chr(10), 1)[0])
|
||||
kp.add_project_args(ap, default_port=4372)
|
||||
ap.add_argument("--frames", type=int, default=11000,
|
||||
help="frame target for both sides (default 11000)")
|
||||
ap.add_argument("--timeout", type=int, default=900)
|
||||
ap.add_argument("--stall", type=int, default=90,
|
||||
help="fail if the frame counter does not advance for this long")
|
||||
ap.add_argument("--out", default="analysis/kernel_patch_ab.json")
|
||||
a = ap.parse_args()
|
||||
root = kp.resolve(a)
|
||||
|
||||
pranges = kp.parse_patch_ranges(os.path.join(root, a.dispatch))
|
||||
print("%s declares %d install-slot range(s): %s"
|
||||
% (os.path.basename(a.dispatch), len(pranges),
|
||||
", ".join("0x%04X..0x%04X" % r for r in pranges) or "none"), flush=True)
|
||||
if not pranges:
|
||||
raise SystemExit("nothing to A/B: this backend declares no ranges. Add "
|
||||
"[[recompiler.install_slots]] to the BIOS profile and "
|
||||
"regenerate (see docs/dynamic_handler_install.md).")
|
||||
|
||||
rows = []
|
||||
for ranges_on in (False, True):
|
||||
print("=== %s" % ("ranges ON (declared install slots)" if ranges_on else
|
||||
"ranges OFF (PSX_KERNEL_PATCH_RANGES=0, pre-change "
|
||||
"behaviour)"), flush=True)
|
||||
r = run_side(a, root, ranges_on)
|
||||
rows.append(r)
|
||||
print(" frame %(frame)s in %(wall_s)ss bless: %(bless_clean)s clean / "
|
||||
"%(bless_mismatch)s mismatch of %(bless_entries)s "
|
||||
"ranges=%(patch_ranges)s skips=%(patch_skips)s" % r, flush=True)
|
||||
print(" interpreted: %(interp_insns_total)s total, "
|
||||
"%(interp_insns_kernel)s kernel RAM" % r, flush=True)
|
||||
for t in r["top_kernel_pcs"][:6]:
|
||||
print(" %s insns=%-10s entries=%s"
|
||||
% (t["pc"], t["insns"], t["entries"]), flush=True)
|
||||
|
||||
off, on = rows[0], rows[1]
|
||||
print()
|
||||
print("frames: OFF %s, ON %s; counts below are PER FRAME, since headless "
|
||||
"runs uncapped" % (off["frame"], on["frame"]))
|
||||
print("%-30s %14s %14s %9s" % ("", "ranges OFF", "ranges ON", "change"))
|
||||
|
||||
def line(label, key, per_frame=True):
|
||||
o, n = off.get(key) or 0, on.get(key) or 0
|
||||
if per_frame:
|
||||
o = o / float(off["frame"] or 1)
|
||||
n = n / float(on["frame"] or 1)
|
||||
os_, ns_ = "%.1f" % o, "%.1f" % n
|
||||
else:
|
||||
os_, ns_ = str(o), str(n)
|
||||
ch = ("%+.1f%%" % (100.0 * (n - o) / o)) if o else "n/a"
|
||||
print("%-30s %14s %14s %9s" % (label, os_, ns_, ch))
|
||||
|
||||
line("kernel-bless mismatch", "bless_mismatch", False)
|
||||
line("kernel-bless clean", "bless_clean", False)
|
||||
line("interp insns / frame, kernel", "interp_insns_kernel")
|
||||
line("interp insns / frame, all", "interp_insns_total")
|
||||
print(" of which (diagnostic, not discounted):")
|
||||
line(" kernel bodies", "interp_insns_kernel_bodies")
|
||||
line(" A0/B0/C0 vectors", "interp_insns_vectors")
|
||||
|
||||
out = os.path.join(root, a.out)
|
||||
os.makedirs(os.path.dirname(out), exist_ok=True)
|
||||
json.dump({"frames": a.frames, "bios": a.bios, "dispatch": a.dispatch,
|
||||
"declared_ranges": ["0x%04X..0x%04X" % r for r in pranges],
|
||||
"sides": rows},
|
||||
open(out, "w", encoding="utf-8"), indent=1)
|
||||
print(chr(10) + "wrote " + a.out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,160 @@
|
||||
#!/usr/bin/env python3
|
||||
"""kernel_patch_common.py - shared plumbing for the kernel install-slot tools.
|
||||
|
||||
Both kernel_patch_diff.py (which kernel-RAM words does this game patch?) and
|
||||
kernel_patch_ab.py (what do the declared ranges buy?) run a GAME project
|
||||
headless and talk to its debug server. This module holds the three things they
|
||||
would otherwise each get subtly wrong: locating the project's exe, reading the
|
||||
BIOS profile, and reading the emitted tables back out of the generated C.
|
||||
|
||||
Run both tools from a game project root (the directory holding game.toml and
|
||||
the framework checkout), not from the framework:
|
||||
|
||||
python psxrecomp/tools/kernel_patch_diff.py
|
||||
python psxrecomp/tools/kernel_patch_ab.py --bios psxrecomp/bios/SCPH1001.BIN
|
||||
|
||||
See docs/dynamic_handler_install.md for what the ranges are and how to find
|
||||
them for a new image.
|
||||
"""
|
||||
import glob
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
import debug_client
|
||||
|
||||
# Kernel RAM below this is the relocated BIOS copy (dirty_ram_interp.h's
|
||||
# DIRTY_RAM_KERNEL_WINDOW_END).
|
||||
KERNEL_WINDOW_END = 0x10000
|
||||
|
||||
# The four 16-byte kernel call vectors, reported on their own line by the A/B.
|
||||
#
|
||||
# They are NOT discounted from its headline. An earlier version subtracted them
|
||||
# on the grounds that they interpret on both sides and would mask the result.
|
||||
# That is true of some titles and false in general: on Mega Man X6 with
|
||||
# OpenBIOS the B0 vector burns 2,361,470 instructions over 1,180,735 entries
|
||||
# with the ranges off and vanishes entirely with them on, because once the
|
||||
# bless verdict is clean the vector body dispatches native. Subtracting it hid
|
||||
# a 130 insn/frame win.
|
||||
VECTOR_STUBS = (0x80, 0xA0, 0xB0, 0xC0)
|
||||
|
||||
|
||||
def add_project_args(ap, default_port):
|
||||
"""The arguments every kernel-patch tool takes."""
|
||||
ap.add_argument("--project-root", default=".",
|
||||
help="game project root (holds game.toml); default cwd")
|
||||
ap.add_argument("--framework", default="psxrecomp",
|
||||
help="framework checkout, relative to --project-root")
|
||||
ap.add_argument("--build-dir", default="build-relprof",
|
||||
help="build tree to launch from, relative to --project-root")
|
||||
ap.add_argument("--exe", default=None,
|
||||
help="runtime exe; default: the single .exe in --build-dir")
|
||||
ap.add_argument("--game", default="game.toml",
|
||||
help="game config passed to --game")
|
||||
ap.add_argument("--profile", default=None,
|
||||
help="BIOS profile toml; default <framework>/bios/OpenBIOS.toml")
|
||||
ap.add_argument("--dispatch", default=None,
|
||||
help="generated <stem>_dispatch.c; default derived from --profile")
|
||||
ap.add_argument("--bios", default=None,
|
||||
help="BIOS image to launch with; default = the runtime's own pick")
|
||||
ap.add_argument("--port", type=int, default=default_port)
|
||||
ap.add_argument("--host", default="127.0.0.1")
|
||||
|
||||
|
||||
def resolve(a):
|
||||
"""Fill in the derived paths on a parsed args namespace. Returns the root."""
|
||||
root = os.path.abspath(a.project_root)
|
||||
fw = os.path.join(root, a.framework)
|
||||
if not os.path.isdir(fw):
|
||||
raise SystemExit("no framework checkout at %s (pass --framework)" % fw)
|
||||
if a.profile is None:
|
||||
a.profile = os.path.join(a.framework, "bios", "OpenBIOS.toml")
|
||||
if a.dispatch is None:
|
||||
stem = read_out_stem(os.path.join(root, a.profile))
|
||||
a.dispatch = os.path.join(a.framework, "generated", stem + "_dispatch.c")
|
||||
if a.exe is None:
|
||||
a.exe = find_exe(os.path.join(root, a.build_dir))
|
||||
for key in ("profile", "dispatch", "exe"):
|
||||
p = os.path.join(root, getattr(a, key))
|
||||
if not os.path.exists(p):
|
||||
raise SystemExit("--%s not found: %s" % (key, p))
|
||||
return root
|
||||
|
||||
|
||||
def find_exe(build_dir):
|
||||
"""The game's runtime exe. Its name is derived from the window title, so it
|
||||
differs per title; glob rather than guess, and refuse to choose."""
|
||||
if not os.path.isdir(build_dir):
|
||||
raise SystemExit("no build dir at %s (pass --build-dir or --exe)" % build_dir)
|
||||
found = [p for p in glob.glob(os.path.join(build_dir, "*.exe"))
|
||||
if "_oracle" not in os.path.basename(p)]
|
||||
if not found:
|
||||
raise SystemExit("no .exe in %s - build psx-runtime first" % build_dir)
|
||||
if len(found) > 1:
|
||||
names = "\n ".join(os.path.basename(p) for p in found)
|
||||
raise SystemExit("several exes in %s; pass --exe:\n %s" % (build_dir, names))
|
||||
return found[0]
|
||||
|
||||
|
||||
def read_out_stem(profile_path):
|
||||
"""[recompiler] out_stem from a BIOS profile ("OpenBIOS", "SCPH1001")."""
|
||||
txt = open(profile_path, encoding="utf-8").read()
|
||||
m = re.search(r'^\s*out_stem\s*=\s*"([^"]+)"', txt, re.M)
|
||||
if not m:
|
||||
raise SystemExit("%s has no [recompiler] out_stem" % profile_path)
|
||||
return m.group(1)
|
||||
|
||||
|
||||
def parse_profile(path):
|
||||
"""(rom_path, [(name, rom_lo, ram_lo, length, kernel_bless)]) from a profile."""
|
||||
txt = open(path, encoding="utf-8").read()
|
||||
rom = re.search(r'^rom\s*=\s*"([^"]+)"', txt, re.M).group(1)
|
||||
copies = []
|
||||
for m in re.finditer(r'\[\[recompiler\.address_model\.copy\]\](.*?)(?=\n\[|\Z)',
|
||||
txt, re.S):
|
||||
blk = m.group(1)
|
||||
g = lambda k: re.search(r'^\s*%s\s*=\s*"?([^"\n]+)"?' % k,
|
||||
blk, re.M).group(1).strip()
|
||||
rom_lo = int(g("rom_lo"), 16)
|
||||
rom_hi = int(g("rom_hi"), 16)
|
||||
copies.append((g("name"), rom_lo, int(g("ram_lo"), 16),
|
||||
rom_hi - rom_lo, "true" in g("kernel_bless")))
|
||||
return rom, copies
|
||||
|
||||
|
||||
def parse_bodies(dispatch_c):
|
||||
"""[(key, body_lo, body_hi)] from the emitted PsxKernelBody table."""
|
||||
txt = open(dispatch_c, encoding="utf-8", errors="replace").read()
|
||||
m = re.search(r'psx_bios_kernel_bodies\[\d+\]\s*=\s*\{(.*?)\};', txt, re.S)
|
||||
if not m:
|
||||
return []
|
||||
return [(int(a, 16), int(b, 16), int(c, 16)) for a, b, c in
|
||||
re.findall(r'\{\s*0x([0-9A-Fa-f]+)u,\s*0x([0-9A-Fa-f]+)u,'
|
||||
r'\s*0x([0-9A-Fa-f]+)u\s*\}', m.group(1))]
|
||||
|
||||
|
||||
def parse_patch_ranges(dispatch_c):
|
||||
"""[(lo, hi)] from the emitted PsxKernelPatchRange table.
|
||||
|
||||
The profile's [[recompiler.install_slots]], couriered into the generated C.
|
||||
A differing word INSIDE one of these is a declared patch: the bless
|
||||
verifier skips it and the body still runs native. A differing word outside
|
||||
every range is what actually unblesses a body. Empty for a backend
|
||||
generated before install-slot ranges existed.
|
||||
"""
|
||||
txt = open(dispatch_c, encoding="utf-8", errors="replace").read()
|
||||
m = re.search(r'psx_bios_kernel_patch_ranges\[\d+\]\s*=\s*\{(.*?)\};', txt, re.S)
|
||||
if not m:
|
||||
return []
|
||||
pairs = [(int(a, 16), int(b, 16)) for a, b in
|
||||
re.findall(r'\{\s*0x([0-9A-Fa-f]+)u,\s*0x([0-9A-Fa-f]+)u\s*\}',
|
||||
m.group(1))]
|
||||
# An image with no slots emits one inert { 0, 0 } entry and a count of 0.
|
||||
cnt = re.search(r'psx_bios_kernel_patch_range_count\s*=\s*(\d+)u', txt)
|
||||
return pairs[:int(cnt.group(1))] if cnt else pairs
|
||||
|
||||
|
||||
def q(cmd, host, port, timeout=120.0, **kw):
|
||||
"""One debug-server command. Bulk commands need the long default."""
|
||||
return debug_client.query(host, port, dict(cmd=cmd, **kw), timeout=timeout)
|
||||
@@ -0,0 +1,231 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Which kernel-RAM words does this game patch at runtime?
|
||||
|
||||
Boots the game headless, then diffs live kernel RAM against the BIOS image's
|
||||
boot-time copy (the [[recompiler.address_model.copy]] window with
|
||||
kernel_bless = true) and reports every differing word, joined to the compiled
|
||||
kernel bodies from the generated <stem>_dispatch.c. The output says which
|
||||
FUNCTIONS carry a patch, and which of those still fail the bless check and so
|
||||
run interpreted.
|
||||
|
||||
This is how you find [[recompiler.install_slots]] ranges for a new image or a
|
||||
new SDK version. Run from a game project root:
|
||||
|
||||
python psxrecomp/tools/kernel_patch_diff.py # OpenBIOS
|
||||
python psxrecomp/tools/kernel_patch_diff.py --at 5 --at 30 # two snapshots
|
||||
python psxrecomp/tools/kernel_patch_diff.py --no-launch --port 4370
|
||||
python psxrecomp/tools/kernel_patch_diff.py \
|
||||
--bios psxrecomp/bios/SCPH1001.BIN \
|
||||
--profile psxrecomp/bios/SCPH1001.toml
|
||||
|
||||
Snapshot twice. Some patches land at boot and some on the first card access,
|
||||
and a range that only appears late is still a range.
|
||||
|
||||
Reading the output: a differing word inside a compiled body and OUTSIDE every
|
||||
declared range is what unblesses that body. A word inside a declared range is
|
||||
already handled - the verifier skips it and the body runs native anyway. A
|
||||
word in no body at all is data (event tables, TCB saves, pad tables); it
|
||||
unblesses nothing, so do not declare it. A word whose writer you cannot
|
||||
identify is a word you should not declare either: the declaration tells the
|
||||
verifier to stop checking it (CLAUDE.md rule 14).
|
||||
|
||||
See docs/dynamic_handler_install.md for the patch shapes, the resume kinds,
|
||||
and how the three halves of the mechanism compose.
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import struct
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
import kernel_patch_common as kp
|
||||
|
||||
|
||||
def wait_port(host, port, secs=180):
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < secs:
|
||||
try:
|
||||
r = kp.q("frame", host, port, timeout=30.0)
|
||||
if r.get("ok") or "frame" in r:
|
||||
return time.time() - t0
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(1)
|
||||
raise SystemExit("debug port never came up")
|
||||
|
||||
|
||||
def parse_seeds(path):
|
||||
"""[(ram_addr, label)] for seeds, accepting the ROM-LMA form too."""
|
||||
d = json.load(open(path, encoding="utf-8"))
|
||||
rows = d if isinstance(d, list) else (d.get("functions") or d.get("seeds") or [])
|
||||
return [(int(r["address"], 16) & 0x1FFFFFFF, r.get("label", "?"))
|
||||
for r in rows if isinstance(r, dict)]
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser(description=__doc__.split(chr(10) + chr(10), 1)[0])
|
||||
kp.add_project_args(ap, default_port=4371)
|
||||
ap.add_argument("--seeds", default=None,
|
||||
help="seeds json for function names; default from the profile")
|
||||
ap.add_argument("--at", type=float, action="append", default=None,
|
||||
help="seconds after the port is up to snapshot "
|
||||
"(repeatable; default 8 and 30)")
|
||||
ap.add_argument("--no-launch", action="store_true",
|
||||
help="diff against an already-running game on --port")
|
||||
ap.add_argument("--out", default="analysis/kernel_patch_diff.json")
|
||||
a = ap.parse_args()
|
||||
root = kp.resolve(a)
|
||||
ats = a.at or [8.0, 30.0]
|
||||
|
||||
rom_rel, copies = kp.parse_profile(os.path.join(root, a.profile))
|
||||
rom = open(os.path.join(root, a.framework, rom_rel), "rb").read()
|
||||
bodies = kp.parse_bodies(os.path.join(root, a.dispatch))
|
||||
pranges = kp.parse_patch_ranges(os.path.join(root, a.dispatch))
|
||||
declared = lambda ram: any(lo <= ram < hi for lo, hi in pranges)
|
||||
if pranges:
|
||||
print("declared install-slot ranges (%s): %s"
|
||||
% (os.path.basename(a.dispatch),
|
||||
", ".join("0x%04X..0x%04X" % r for r in pranges)), flush=True)
|
||||
else:
|
||||
print("no install-slot ranges declared in %s - every patched word "
|
||||
"unblesses its body" % os.path.basename(a.dispatch), flush=True)
|
||||
|
||||
if a.seeds is None:
|
||||
import re
|
||||
txt = open(os.path.join(root, a.profile), encoding="utf-8").read()
|
||||
m = re.search(r'^\s*seeds\s*=\s*"([^"]+)"', txt, re.M)
|
||||
a.seeds = os.path.join(a.framework, m.group(1)) if m else None
|
||||
seeds = parse_seeds(os.path.join(root, a.seeds)) if a.seeds else []
|
||||
|
||||
# Map ROM-LMA seeds into RAM for each copy so a patched word can be named.
|
||||
ram_names = {}
|
||||
for name, rom_lo, ram_lo, ln, bless in copies:
|
||||
for addr, label in seeds:
|
||||
if rom_lo <= addr < rom_lo + ln:
|
||||
ram_names[addr - rom_lo + ram_lo] = label
|
||||
elif ram_lo <= addr < ram_lo + ln:
|
||||
ram_names[addr] = label
|
||||
|
||||
def fn_of(ram):
|
||||
best = [k for k in ram_names if k <= ram]
|
||||
return ("%s+0x%X" % (ram_names[max(best)], ram - max(best))) if best else "?"
|
||||
|
||||
proc = None
|
||||
if not a.no_launch:
|
||||
env = dict(os.environ, PSX_STARVATION_TIMEOUT_US="0")
|
||||
env.pop("PSX_LOAD_SLOT", None)
|
||||
argv = [os.path.join(root, a.exe), "--game", a.game, "--no-launcher",
|
||||
"--headless", "--debug-port", str(a.port)]
|
||||
if a.bios:
|
||||
argv += ["--bios", os.path.join(root, a.bios)]
|
||||
proc = subprocess.Popen(argv, cwd=root, env=env,
|
||||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
|
||||
print("launched pid %d" % proc.pid, flush=True)
|
||||
t_up = wait_port(a.host, a.port)
|
||||
print("port up after %.1fs" % t_up, flush=True)
|
||||
t0 = time.time()
|
||||
|
||||
report = {"copies": [], "snapshots": [],
|
||||
"declared_ranges": ["0x%04X..0x%04X" % r for r in pranges]}
|
||||
try:
|
||||
for at in sorted(ats):
|
||||
while time.time() - t0 < at:
|
||||
time.sleep(0.5)
|
||||
fr = kp.q("frame", a.host, a.port)
|
||||
kb = kp.q("kernel_bless", a.host, a.port)
|
||||
snap = {"t": round(time.time() - t0, 1), "frame": fr.get("frame"),
|
||||
"kernel_bless": kb, "diffs": []}
|
||||
print("\n=== t+%.0fs frame %s kernel_bless: %s"
|
||||
% (at, fr.get("frame"),
|
||||
{k: v for k, v in kb.items() if k not in ("id", "ok")}),
|
||||
flush=True)
|
||||
for name, rom_lo, ram_lo, ln, bless in copies:
|
||||
if not bless:
|
||||
continue # only the bless window gates native dispatch
|
||||
live = bytes.fromhex(kp.q("read_ram", a.host, a.port,
|
||||
addr="0x%08X" % ram_lo, len=ln)["hex"])
|
||||
src = rom[rom_lo - 0x1FC00000: rom_lo - 0x1FC00000 + ln]
|
||||
words = []
|
||||
for off in range(0, ln - 3, 4):
|
||||
if live[off:off + 4] != src[off:off + 4]:
|
||||
ram = ram_lo + off
|
||||
inb = [b for b in bodies if b[1] <= ram < b[2]]
|
||||
words.append({
|
||||
"ram": "0x%04X" % ram,
|
||||
"rom": struct.unpack("<I", src[off:off + 4])[0],
|
||||
"live": struct.unpack("<I", live[off:off + 4])[0],
|
||||
"in_code_body": bool(inb),
|
||||
"declared": declared(ram),
|
||||
"fn": fn_of(ram)})
|
||||
runs = []
|
||||
for w in words:
|
||||
r = int(w["ram"], 16)
|
||||
if runs and r == runs[-1]["end"] and \
|
||||
runs[-1]["in_code_body"] == w["in_code_body"]:
|
||||
runs[-1]["end"] = r + 4
|
||||
runs[-1]["n"] += 1
|
||||
else:
|
||||
runs.append({"start": r, "end": r + 4, "n": 1,
|
||||
"in_code_body": w["in_code_body"],
|
||||
"fn": w["fn"]})
|
||||
code_words = sum(1 for w in words if w["in_code_body"])
|
||||
undecl = sum(1 for w in words
|
||||
if w["in_code_body"] and not w["declared"])
|
||||
print("copy %-16s RAM 0x%04X..0x%04X: %d differing words, %d inside "
|
||||
"compiled code bodies (%d declared, %d NOT declared), %d runs"
|
||||
% (name, ram_lo, ram_lo + ln, len(words), code_words,
|
||||
code_words - undecl, undecl, len(runs)), flush=True)
|
||||
for r in runs:
|
||||
tag = "CODE" if r["in_code_body"] else "data"
|
||||
print(" %s 0x%04X..0x%04X (%d words) %s"
|
||||
% (tag, r["start"], r["end"], r["n"], r["fn"]), flush=True)
|
||||
if r["in_code_body"]:
|
||||
for w in words:
|
||||
if r["start"] <= int(w["ram"], 16) < r["end"]:
|
||||
print(" %s rom %08X -> live %08X %s"
|
||||
% (w["ram"], w["rom"], w["live"],
|
||||
"declared" if w["declared"]
|
||||
else "NOT DECLARED"), flush=True)
|
||||
snap["diffs"].append({"copy": name, "words": words, "runs": runs})
|
||||
report["snapshots"].append(snap)
|
||||
|
||||
# Only an UNDECLARED patched word unblesses a body.
|
||||
dirty, saved = set(), set()
|
||||
for sd in snap["diffs"]:
|
||||
for w in sd["words"]:
|
||||
r = int(w["ram"], 16)
|
||||
for key, lo, hi in bodies:
|
||||
if lo <= r < hi:
|
||||
(saved if w["declared"] else dirty).add((lo, hi))
|
||||
saved -= dirty # one undeclared word is enough to unbless
|
||||
if saved:
|
||||
print("compiled kernel bodies patched INSIDE a declared range "
|
||||
"(these still run native):", flush=True)
|
||||
for lo, hi in sorted(saved):
|
||||
print(" 0x%04X..0x%04X %s" % (lo, hi, fn_of(lo)), flush=True)
|
||||
if dirty:
|
||||
print("compiled kernel bodies containing an UNDECLARED patched "
|
||||
"word (these interpret):", flush=True)
|
||||
for lo, hi in sorted(dirty):
|
||||
print(" 0x%04X..0x%04X %s" % (lo, hi, fn_of(lo)), flush=True)
|
||||
else:
|
||||
print("no compiled kernel body carries an undeclared patched word",
|
||||
flush=True)
|
||||
snap["dirty_bodies"] = ["0x%04X..0x%04X %s" % (lo, hi, fn_of(lo))
|
||||
for lo, hi in sorted(dirty)]
|
||||
snap["blessed_patched_bodies"] = ["0x%04X..0x%04X %s" % (lo, hi, fn_of(lo))
|
||||
for lo, hi in sorted(saved)]
|
||||
finally:
|
||||
if proc:
|
||||
proc.kill()
|
||||
out = os.path.join(root, a.out)
|
||||
os.makedirs(os.path.dirname(out), exist_ok=True)
|
||||
json.dump(report, open(out, "w", encoding="utf-8"), indent=1)
|
||||
print("\nwrote " + a.out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user