mirror of
https://github.com/RetroPortingToolKit/psxrecomp
synced 2026-10-25 18:05:12 -04:00
Shokunin
fa05858e24
docs/SEGMENT_AWARE_CODE.md §5.7. compile_overlays.py splits each capture
into one view per segment with entries (capture_segment_views). A v2
capture has no dispatch_entry_segments and comes back unchanged, KSEG0 only;
in v3, a dispatch entry that no segment list names is a KSEG0 entry. A view
spells load_addr and every PC in its segment, holds only that segment's
dispatch entries, and carries all the execution evidence, which has no
segment. A malformed segment record fails that capture loudly.
Two kinds of record also get a view where they saw no dispatch entry:
- a record with no dispatch entry in any segment (execution evidence only,
which the runtime writes) is one KSEG0 view, exactly as a v2 reader reads
it, so classification, forced interiors, declared entries and prior
manifests apply as before;
- declared_view_segments(): a --force-interior PC inside the record (KSEG0)
asks for the KSEG0 view, and a game.toml [[overlays]] table asks for the
segment its load address, or else each of its entries inside the record,
is spelled in.
Each view compiles like any capture, region and fragment passes alike:
- the image is compiled at `segment | phys`, so the recompiler names and
bakes it in that segment;
- helpers that spell PCs without an image address use the view's segment
(image_segment(); KSEG0 by default, so a KSEG0 compile is unchanged);
- the shard lands in segment_cache_dir(): the tag directory for KSEG0,
seg-kuseg/ and seg-kseg1/ otherwise.
The manifest writer adds `S <segment>` for KUSEG/KSEG1. The runtime-contract
parser keeps F entries' full VA, applies the S rules, and refuses a
manifest whose segment differs from its directory's. Exports are looked up
by their full VA.
Capacity accounting counts each leaf's segment directories in the leaf's
namespace, since the runtime's candidate table is shared, and tells shards
apart by segment and name. The static generator emits the exact PC per row,
so KUSEG, KSEG0 and KSEG1 rows of one word coexist, with a segment gate
over the segments it compiled. A KSEG0-only table emits the same text as
before. Non-KSEG0 static namespaces are ov_s0_ (KUSEG) and ov_s5_ (KSEG1).
The static isolated-fragment pass keys its images by segment too, so each
(entry, image) demand compiles against its own segment's view and one entry
dispatched in two segments gets a fragment in each. --force-interior PCs
stay KSEG0 and force KSEG0 views.
coverage_vault.py unions dispatch_entry_segments when it merges variants,
making a v2 record's entries explicit KSEG0 entries, and crops it when it
compacts regions.
Output for v2 captures is unchanged: the 52 R4 shard sources and manifests
compiled from D's race captures are byte-identical to D's.
Tests:
- new overlay_segment_keys (CTest). It runs capture views, manifests and the
cache layout, then the real compile_overlays.py driver, in process, on a v3
capture of an overlay entered at KUSEG and KSEG1, and checks:
- each shard lands in its directory with S and full-VA F records;
- every baked PC is in its segment, and KSEG1 charges a fetch per
instruction;
- a second run is served from each segment's cache;
- a KSEG0 view compiles byte-identically to the same bytes captured as v2;
- --static emits per-segment rows and namespaces, and an isolated
fragment per segment for an entry with no callable boundary;
- a dispatch-less v3 record with --force-interior compiles byte-identically
to its v2 reading, and a KUSEG-only record still gets its forced KSEG0
fragment;
- a KUSEG/KSEG1 view compiled into a cache holding a KSEG0 manifest of the
same bytes merges none of it, and the empty-primary reconciliation counts
only the view's own segment's fragments;
- the real loader runs the shards: links, store-PC stamps and fetch tags in
each segment, fetch cycles equal to Beetle's model, KSEG0 interpreted.
- coverage_vault: segment unions and crops.
- compile_overlays_static_split: a KUSEG row beside a KSEG0 row dispatches
by exact PC; unmapped-segment entries are refused. This replaces the
KSEG0-only rejection.
- aot_overlay_discovery: the capacity namespace lists the segment
directories.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
253 lines
12 KiB
Python
253 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
import importlib.util
|
|
import base64
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parents[2]
|
|
SPEC = importlib.util.spec_from_file_location(
|
|
"coverage_vault", ROOT / "tools" / "coverage_vault.py")
|
|
MOD = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader is not None
|
|
SPEC.loader.exec_module(MOD)
|
|
|
|
|
|
class CoverageVaultHistoryTests(unittest.TestCase):
|
|
def test_streaming_array_reader_handles_chunk_boundaries(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
source = os.path.join(tmp, "large.json")
|
|
records = [{"value": "x" * 100}, {"value": "y" * 137}]
|
|
with open(source, "w", encoding="utf-8") as out:
|
|
json.dump(records, out)
|
|
self.assertEqual(list(MOD._iter_json_array(source, chunk_size=7)),
|
|
records)
|
|
|
|
def test_compaction_splits_execution_pages_and_collapses_mutable_gap(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
source = os.path.join(tmp, "overlay_captures.json")
|
|
output = os.path.join(tmp, "compacted.json")
|
|
first = bytearray(3 * 4096)
|
|
second = bytearray(first)
|
|
first[0:4] = second[0:4] = b"CODE"
|
|
first[8192:8196] = second[8192:8196] = b"TAIL"
|
|
first[4096 + 100] = 0x11
|
|
second[4096 + 100] = 0x22 # mutable, never-executed middle page
|
|
records = []
|
|
for image, dispatch in ((first, "0x80010000"),
|
|
(second, "0x80012000")):
|
|
records.append({
|
|
"schema": "psxrecomp overlay capture v2",
|
|
"load_addr": "0x80010000", "size": len(image),
|
|
"bytes_b64": base64.b64encode(image).decode("ascii"),
|
|
"executed_pcs": ["0x80010000", "0x80012000"],
|
|
"dispatch_entry_pcs": [dispatch],
|
|
"function_entry_pcs": [], "seeds": [dispatch],
|
|
})
|
|
with open(source, "w", encoding="utf-8") as out:
|
|
json.dump(records, out)
|
|
compacted, stats = MOD.compact_capture_manifest(source, output)
|
|
self.assertEqual(stats["source_regions"], 2)
|
|
self.assertEqual(stats["output_variants"], 2)
|
|
self.assertEqual(stats["max_output_region"], 4100)
|
|
self.assertEqual([r["load_addr"] for r in compacted],
|
|
["0x80010000", "0x80012000"])
|
|
self.assertEqual(compacted[0]["size"], 4100)
|
|
self.assertEqual(compacted[1]["size"], 4096)
|
|
self.assertEqual(compacted[0]["dispatch_entry_pcs"],
|
|
["0x80010000"])
|
|
self.assertEqual(compacted[1]["dispatch_entry_pcs"],
|
|
["0x80012000"])
|
|
self.assertEqual(base64.b64decode(compacted[0]["bytes_b64"])[:4],
|
|
b"CODE")
|
|
self.assertEqual(base64.b64decode(compacted[1]["bytes_b64"])[:4],
|
|
b"TAIL")
|
|
with open(output, encoding="utf-8") as saved:
|
|
self.assertEqual(json.load(saved), compacted)
|
|
|
|
def test_compaction_fails_closed_on_out_of_region_evidence(self):
|
|
image = bytes(4096)
|
|
region = {
|
|
"load_addr": "0x80010000", "size": len(image),
|
|
"bytes_b64": base64.b64encode(image).decode("ascii"),
|
|
"executed_pcs": ["0x80012000"],
|
|
}
|
|
with self.assertRaisesRegex(ValueError, "outside/alignment-invalid"):
|
|
MOD._compact_region(region)
|
|
compacted, evidence, invalid = MOD._compact_region(
|
|
region, drop_invalid_evidence=True)
|
|
self.assertEqual(compacted, [])
|
|
self.assertEqual(evidence, 0)
|
|
self.assertEqual(invalid, 1)
|
|
|
|
def test_capture_merge_preserves_static_dispatch_provenance(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
vault = os.path.join(tmp, 'captures.json')
|
|
base = {
|
|
'load_addr': '0x80010000', 'size': 4,
|
|
'bytes_b64': 'AAAAAA==',
|
|
'dispatch_entry_pcs': ['0x80010000'],
|
|
}
|
|
with open(vault, 'w', encoding='utf-8') as out:
|
|
json.dump([base], out)
|
|
enriched = dict(
|
|
base,
|
|
dispatch_entry_pcs=['0x80010000', '0x80010004'],
|
|
static_dispatch_entry_pcs=['0x80010004'])
|
|
self.assertEqual(
|
|
MOD.merge_capture_regions(vault, [enriched]), (0, 0))
|
|
with open(vault, encoding='utf-8') as source:
|
|
merged = json.load(source)[0]
|
|
self.assertEqual(merged['dispatch_entry_pcs'],
|
|
['0x80010000', '0x80010004'])
|
|
self.assertEqual(merged['static_dispatch_entry_pcs'],
|
|
['0x80010004'])
|
|
|
|
def test_capture_merge_keeps_dispatch_segments(self):
|
|
# docs/SEGMENT_AWARE_CODE.md §5.7: a v3 record names the segments each
|
|
# dispatch entry entered through; a v2 record's entries are KSEG0.
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
vault = os.path.join(tmp, 'captures.json')
|
|
v2 = {
|
|
'load_addr': '0x80010000', 'size': 8,
|
|
'bytes_b64': 'AAAAAAAAAAA=',
|
|
'dispatch_entry_pcs': ['0x80010000'],
|
|
}
|
|
with open(vault, 'w', encoding='utf-8') as out:
|
|
json.dump([v2], out)
|
|
v3 = dict(v2, dispatch_entry_pcs=['0x80010000', '0x80010004'],
|
|
dispatch_entry_segments={
|
|
'kuseg': ['0x00010000', '0x00010004'],
|
|
'kseg0': [], 'kseg1': ['0xA0010004']})
|
|
self.assertEqual(MOD.merge_capture_regions(vault, [v3]), (0, 0))
|
|
with open(vault, encoding='utf-8') as source:
|
|
merged = json.load(source)[0]
|
|
self.assertEqual(merged['dispatch_entry_pcs'],
|
|
['0x80010000', '0x80010004'])
|
|
self.assertEqual(merged['dispatch_entry_segments'], {
|
|
'kuseg': ['0x00010000', '0x00010004'],
|
|
'kseg0': ['0x80010000'],
|
|
'kseg1': ['0xA0010004']})
|
|
# v2 into v2 stays v2.
|
|
with open(vault, 'w', encoding='utf-8') as out:
|
|
json.dump([v2], out)
|
|
MOD.merge_capture_regions(vault, [dict(v2)])
|
|
with open(vault, encoding='utf-8') as source:
|
|
self.assertNotIn('dispatch_entry_segments', json.load(source)[0])
|
|
|
|
def test_compaction_crops_dispatch_segments(self):
|
|
image = bytes(0x3004)
|
|
region = {
|
|
'load_addr': '0x80010000', 'size': len(image),
|
|
'bytes_b64': base64.b64encode(image).decode('ascii'),
|
|
'executed_pcs': ['0x80010000', '0x80012000'],
|
|
'dispatch_entry_pcs': ['0x80010000', '0x80012000'],
|
|
'dispatch_entry_segments': {
|
|
'kuseg': ['0x00010000'], 'kseg0': ['0x80012000'],
|
|
'kseg1': ['0xA0012000']},
|
|
}
|
|
runs, _entries, _invalid = MOD._compact_region(region)
|
|
self.assertEqual([r['dispatch_entry_segments'] for r in runs], [
|
|
{'kuseg': ['0x00010000'], 'kseg0': [], 'kseg1': []},
|
|
{'kuseg': [], 'kseg0': ['0x80012000'], 'kseg1': ['0xA0012000']}])
|
|
|
|
def test_cache_merge_preserves_resident_sidecar(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
src = os.path.join(tmp, 'source')
|
|
dst = os.path.join(tmp, 'vault')
|
|
rel = os.path.join('gcc', 'win-x64', 'cg5_12345678')
|
|
os.makedirs(os.path.join(src, rel))
|
|
stem = os.path.join(src, rel, '0000DF80_4EE6AC69')
|
|
for ext, contents in (('.dll', b'dll'), ('.ranges', b'ranges'),
|
|
('.resident', b'resident')):
|
|
with open(stem + ext, 'wb') as out:
|
|
out.write(contents)
|
|
self.assertEqual(MOD.merge_cache(dst, src), 1)
|
|
for ext in ('.dll', '.ranges', '.resident'):
|
|
self.assertTrue(os.path.exists(os.path.join(
|
|
dst, rel, '0000DF80_4EE6AC69' + ext)))
|
|
|
|
def test_v1_and_verified_v2_records_survive_a_torn_tail(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
snapshot = os.path.join(tmp, "immutable.json")
|
|
addendum = os.path.join(tmp, "history.jsonl")
|
|
v1_region = {"load_addr": "0x80010000", "bytes_b64": "AA=="}
|
|
v2_region = {"load_addr": "0x80100000", "bytes_b64": "AQ=="}
|
|
with open(snapshot, "w", encoding="utf-8", newline="\n") as out:
|
|
json.dump([v2_region], out)
|
|
signature = "%016X" % MOD._fnv64_file(snapshot)
|
|
records = [
|
|
{"schema": "psxrecomp overlay capture addendum v1",
|
|
"captures": [v1_region]},
|
|
{"schema": "psxrecomp overlay capture addendum v2",
|
|
"snapshot": snapshot, "fnv64": signature},
|
|
# A repeated reference is idempotent and should not inflate RAM.
|
|
{"schema": "psxrecomp overlay capture addendum v2",
|
|
"snapshot": snapshot, "fnv64": signature},
|
|
# A bad signature must fail closed instead of ingesting corruption.
|
|
{"schema": "psxrecomp overlay capture addendum v2",
|
|
"snapshot": snapshot, "fnv64": "0000000000000000"},
|
|
]
|
|
with open(addendum, "w", encoding="utf-8", newline="\n") as out:
|
|
for record in records:
|
|
out.write(json.dumps(record) + "\n")
|
|
out.write('{"schema":"torn')
|
|
self.assertEqual(MOD._load_addendum(addendum),
|
|
[v1_region, v2_region])
|
|
|
|
def test_compact_addendum_converts_v1_and_is_idempotent(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
persist = os.path.join(tmp, "immutable")
|
|
os.mkdir(persist)
|
|
addendum = os.path.join(tmp, "history.jsonl")
|
|
snapshot = os.path.join(
|
|
persist, "GAME_session_0001_0000000000000000.json")
|
|
region = {"load_addr": "0x80010000", "bytes_b64": "AA=="}
|
|
with open(snapshot, "w", encoding="utf-8", newline="\n") as out:
|
|
json.dump([region], out)
|
|
signature = "%016X" % MOD._fnv64_file(snapshot)
|
|
corrected = os.path.join(
|
|
persist, "GAME_session_0001_%s.json" % signature)
|
|
os.replace(snapshot, corrected)
|
|
v1 = {"schema": "psxrecomp overlay capture addendum v1",
|
|
"game": "GAME", "session": "session", "sequence": 1,
|
|
"reason": "autocap", "fnv64": signature,
|
|
"captures": [region]}
|
|
with open(addendum, "w", encoding="utf-8", newline="\n") as out:
|
|
out.write(json.dumps(v1) + "\n")
|
|
out.write('{"schema":"torn')
|
|
self.assertEqual(MOD.compact_addendum(addendum, persist),
|
|
(1, 0, 0, 1))
|
|
with open(addendum, encoding="utf-8") as source:
|
|
compacted = json.loads(source.readline())
|
|
self.assertNotIn("captures", compacted)
|
|
self.assertEqual(compacted["snapshot"], corrected)
|
|
self.assertEqual(MOD._load_addendum(addendum), [region])
|
|
self.assertEqual(MOD.compact_addendum(addendum, persist),
|
|
(0, 1, 0, 0))
|
|
self.assertEqual(MOD._load_addendum(addendum), [region])
|
|
|
|
def test_compact_addendum_failure_preserves_source(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
persist = os.path.join(tmp, "immutable")
|
|
os.mkdir(persist)
|
|
addendum = os.path.join(tmp, "history.jsonl")
|
|
v1 = {"schema": "psxrecomp overlay capture addendum v1",
|
|
"game": "GAME", "session": "missing", "sequence": 1,
|
|
"reason": "autocap", "fnv64": "1234567890ABCDEF",
|
|
"captures": []}
|
|
original = (json.dumps(v1) + "\n").encode()
|
|
with open(addendum, "wb") as out:
|
|
out.write(original)
|
|
with self.assertRaisesRegex(ValueError, "snapshot missing"):
|
|
MOD.compact_addendum(addendum, persist)
|
|
with open(addendum, "rb") as source:
|
|
self.assertEqual(source.read(), original)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|