diff options
| author | Arthur Hanson Rasmusson <arthur@vgpu.io> | 2026-10-06 19:28:39 -0400 |
|---|---|---|
| committer | Arthur Hanson Rasmusson <arthur@vgpu.io> | 2026-10-06 19:28:39 -0400 |
| commit | cd1f2528e8243fed3ce80d5170ef93928e4e7284 (patch) | |
| tree | 208c326a109347baf6c04917dca23ef60c73eabc /debug/risc-v-gsp-rm-vm/sim | |
| parent | 595dcded851dc45d78cc76280137100ea8b3d0ad (diff) | |
| download | open-iov-cd1f2528e8243fed3ce80d5170ef93928e4e7284.tar.gz open-iov-cd1f2528e8243fed3ce80d5170ef93928e4e7284.zip | |
risc-v-gsp-rm-debugger: land docs 45-53 + harness/sim tool set with tests
rm_capture (chunked decode), launch_audit, state_seed/state_decode,
valprobe, rehost, contracts+run_contracts, host_traffic, gmmu_walk,
msgq_attach, libos_port, log_decode, server_snapshot; doc index updates
in the tool overview, operations pages and the repo README.
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/sim')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/sim/gmmu_walk.py | 488 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/sim/host_traffic.py | 1437 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/sim/libos_port.py | 386 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/sim/msgq_attach.py | 1122 |
4 files changed, 3433 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/sim/gmmu_walk.py b/debug/risc-v-gsp-rm-vm/sim/gmmu_walk.py new file mode 100644 index 0000000..50960ab --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/sim/gmmu_walk.py @@ -0,0 +1,488 @@ +#!/usr/bin/env python3 +"""Functional GMMU page-table walker (Turing VER2 geometry). + +The decoder in :mod:`sim.gmmu_pte` decodes one captured entry. This module +follows those entries: given a page-directory base and a sparse physical-memory +view of the table pages, it walks a virtual address to its leaf mapping and +reports every level, the page size, the aperture and the access flags -- or the +exact fault (missing table, invalid aperture, invalid leaf). + +Why it exists: the engine sees memory through the GMMU, not through the host's +bookkeeping. When a forwarded submission goes silent (an engine-side write +never appears), the first question is *where would the engine read/write this +address?* A capture of the channel's page tables plus this walker answers it +offline, deterministically, without hardware. + +Evidence and honesty: + +* Geometry, field positions and the entry encodings come from the same + evidence base as :mod:`sim.gmmu_pte` (see ``docs/43`` and ``docs/48``). +* The 4K and 2M paths are **validated against a live captured channel PDB** + (see :func:`channel_pdb_fixture`); the 64K dual path is modelled from the + format definitions and reported with ``evidence: "modeled"`` until a capture + validates it. +* Unsupported encodings fail closed (``E_UNSUPPORTED``); a video-memory leaf + address is reported with ``pa_complete: false`` because the decoder cannot + claim the high address bits there (never guessed). +* No timing claim; this is the engine's *address* view (L1), not its behaviour. +""" +from __future__ import annotations + +import argparse +import json +import os +import struct +import sys + +TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if TOOL_ROOT not in sys.path: + sys.path.insert(0, TOOL_ROOT) + +from sim.gmmu_pte import ( # noqa: E402 + GmmuDecodeError, + KIND_VER2, + KIND_VER2_DUAL, + LEVELS, + SUPPORTED_KINDS, + decode_dual_pde, + decode_entry, + decode_pde, + decode_pte, +) + +#: Schema tag of a walk report. +WALK_FORMAT = "greeneye-gmmu-walk/1" + +#: Page modes the walker accepts. ``auto`` prefers the 64K dual half for a +#: 64K-aligned VA when that half is valid, else the 4K half. +PAGE_MODES = ("4k", "64k", "auto") + +VA_MASK = (1 << 49) - 1 + +#: Fault codes (never a guess; each names a concrete missing/invalid piece). +FAULT_TABLE_MISSING = "table_missing" +FAULT_INVALID_APERTURE = "invalid_aperture" +FAULT_PTE_INVALID = "pte_invalid" +FAULT_UNSUPPORTED = "unsupported" +FAULT_OUT_OF_RANGE = "va_out_of_range" + + +class WalkError(ValueError): + """A walk request the evidence cannot answer.""" + + +class MemoryView: + """Sparse physical memory: ``pa -> bytes`` blobs (table pages, targets).""" + + def __init__(self, blobs: dict[int, bytes] | None = None): + self._blobs = dict(blobs or {}) + + def add(self, pa: int, data: bytes) -> None: + if pa < 0 or not isinstance(data, (bytes, bytearray)): + raise WalkError("add() needs a non-negative pa and bytes") + self._blobs[pa] = bytes(data) + + def has(self, pa: int) -> bool: + return pa in self._blobs + + def read(self, pa: int, size: int) -> bytes | None: + for base, blob in self._blobs.items(): + if base <= pa and pa + size <= base + len(blob): + off = pa - base + return blob[off:off + size] + return None + + def spans(self) -> list[dict]: + return [{"pa": f"{pa:#x}", "size": len(b)} for pa, b in sorted(self._blobs.items())] + + +# --------------------------------------------------------------------------- +# Encoders (fixtures/tests; mirror of the decoder's field positions) +# --------------------------------------------------------------------------- + +def encode_pde_pointer(pa: int, *, volatile: bool = True) -> int: + """Encode a VER2 directory PDE: bit0=0, aperture 3 (sysmem non-coherent).""" + if pa & 0xFFF: + raise WalkError(f"table PA must be 4K aligned, got {pa:#x}") + return ((pa >> 12) << 8) | (3 << 1) | (8 if volatile else 0) + + +def encode_pte(pa: int, *, page_size: str = "4k", read_only: bool = False, + volatile: bool = True) -> int: + """Encode a VER2 leaf PTE: bit0=1, aperture 3 (sysmem non-coherent).""" + if page_size == "4k": + align = 0xFFF + elif page_size == "64k": + align = 0xFFFF + elif page_size == "2m": + align = 0x1F_FFFF + else: + raise WalkError(f"unknown page_size {page_size!r}") + if pa & align: + raise WalkError(f"PA {pa:#x} is not {page_size}-aligned") + return ((pa >> 12) << 8) | (3 << 1) | 1 | (8 if volatile else 0) | (0x40 if read_only else 0) + + +def encode_dual_slot(big_pt_pa: int | None, small_pt_pa: int) -> bytes: + """Encode a VER2 dual PDE0 slot; BIG = 64K table, SMALL = 4K table.""" + big = 0 + if big_pt_pa is not None: + if big_pt_pa & 0xFFF: + raise WalkError(f"big table PA must be 4K aligned, got {big_pt_pa:#x}") + big = (((big_pt_pa >> 8) & ~1) << 4) | (3 << 1) | 8 + small = ((small_pt_pa >> 12) << 8) | (3 << 1) | 8 + return struct.pack("<QQ", big, small) + + +def _fault(code: str, message: str, steps: list, **extra) -> dict: + report = { + "format": WALK_FORMAT, "status": "fault", "fault": code, + "fault_message": message, "steps": steps, + } + report.update(extra) + return report + + +def walk( + va: int, + *, + root_pa: int, + mem: MemoryView, + kind: str = KIND_VER2, + page_mode: str = "4k", +) -> dict: + """Walk ``va`` through the page tables rooted at ``root_pa``. + + Returns a ``greeneye-gmmu-walk/1`` report. ``status`` is ``ok`` with + ``pa``/``page_size``/``writable`` on success, otherwise ``fault`` with a + concrete ``fault`` code. Every level's raw entry and decode is kept in + ``steps`` so the result is auditable. + """ + if kind not in SUPPORTED_KINDS: + return _fault(FAULT_UNSUPPORTED, f"kind {kind!r} not in {SUPPORTED_KINDS}", + [], va=va, root_pa=root_pa) + if page_mode not in PAGE_MODES: + raise WalkError(f"page_mode must be one of {PAGE_MODES}, got {page_mode!r}") + if not (0 <= va <= VA_MASK): + return _fault(FAULT_OUT_OF_RANGE, f"va {va:#x} outside 49-bit space", [], + va=va, root_pa=root_pa) + + steps: list[dict] = [] + pa_complete = True + page_used = page_mode + evidence = "validated" + cur = root_pa + + for depth in range(5): + geo = LEVELS[depth] + _hi_bit, lo_bit = geo["va_bits"] + idx = (va >> lo_bit) & ((1 << geo["index_bits"]) - 1) + table_pa = cur + if depth == 3: + off = idx * 16 + raw = mem.read(table_pa + off, 16) + if raw is None: + return _fault(FAULT_TABLE_MISSING, + f"depth-3 dual slot at pa={table_pa + off:#x} not in view", + steps, va=va, root_pa=root_pa) + value128 = int.from_bytes(raw, "little") + low = value128 & 0xFFFF_FFFF_FFFF_FFFF + step = {"depth": depth, "level": geo["hw"], "table_pa": f"{table_pa:#x}", + "index": idx, "entry_pa": f"{table_pa + off:#x}", + "raw": f"{value128:#x}"} + if low & 1: # 2 MB leaf in the low half + try: + pte = decode_entry(low, kind=KIND_VER2, role="pte", level=3) + except GmmuDecodeError as exc: + return _fault(FAULT_UNSUPPORTED, str(exc), steps + [step], + va=va, root_pa=root_pa) + step.update({"role": "pte", "page_size": "2m", + "aperture_name": pte["aperture_name"], + "valid": pte["valid"], "address": pte["address"]}) + steps.append(step) + if not pte["valid"] or pte["address"] is None: + return _fault(FAULT_PTE_INVALID, + f"2M leaf at {table_pa + off:#x} invalid/unsupported", + steps, va=va, root_pa=root_pa) + if not pte["address_complete"]: + pa_complete = False + pa = pte["address"] + (va & 0x1F_FFFF) + return _ok(va, pa, "2m", pte, steps, root_pa, kind, + page_used="2m", evidence=evidence, pa_complete=pa_complete) + try: + dual = decode_dual_pde(value128, pde_kind=KIND_VER2_DUAL, level=3) + except GmmuDecodeError as exc: + return _fault(FAULT_UNSUPPORTED, str(exc), steps + [step], + va=va, root_pa=root_pa) + chosen = None + if page_mode == "64k": + chosen, page_used, evidence = dual["big"], "64k", "modeled" + elif page_mode == "4k": + chosen, page_used = dual["small"], "4k" + else: # auto: prefer BIG for a 64K-aligned VA when valid + if va % 0x10000 == 0 and dual["big"]["address"] is not None: + chosen, page_used, evidence = dual["big"], "64k", "modeled" + else: + chosen, page_used = dual["small"], "4k" + step.update({"role": "dual_pde", "page_mode_used": page_used, + "big": dual["big"], "small": dual["small"]}) + steps.append(step) + if chosen["address"] is None: + return _fault(FAULT_INVALID_APERTURE, + f"dual PDE at {table_pa + off:#x} {page_used} half has " + f"aperture {chosen['aperture_name']}", steps, + va=va, root_pa=root_pa) + if not chosen["address_complete"]: + pa_complete = False + cur = chosen["address"] + continue + + off = idx * geo["entry_size"] + if depth == 4: + # A 64K-granularity leaf table is indexed by VA bits 24:16. + idx = (va >> (16 if page_used == "64k" else 12)) & 0x1FF + off = idx * geo["entry_size"] + raw = mem.read(table_pa + off, geo["entry_size"]) + if raw is None: + return _fault(FAULT_TABLE_MISSING, + f"level {depth} entry at pa={table_pa + off:#x} not in view", + steps, va=va, root_pa=root_pa) + value = int.from_bytes(raw, "little") + step = {"depth": depth, "level": geo["hw"], "table_pa": f"{table_pa:#x}", + "index": idx, "entry_pa": f"{table_pa + off:#x}", "raw": f"{value:#x}"} + if depth < 3: + try: + pde = decode_pde(value, pde_kind=KIND_VER2, level=depth) + except GmmuDecodeError as exc: + return _fault(FAULT_UNSUPPORTED, str(exc), steps + [step], + va=va, root_pa=root_pa) + step.update({"role": "pde", "aperture_name": pde["aperture_name"], + "address": pde["address"]}) + steps.append(step) + if pde["address"] is None: + return _fault(FAULT_INVALID_APERTURE, + f"level {depth} entry at {table_pa + off:#x} has " + f"aperture {pde['aperture_name']}", steps, + va=va, root_pa=root_pa) + if not pde["address_complete"]: + pa_complete = False + cur = pde["address"] + continue + + # depth 4: leaf page table + try: + pte = decode_pte(value, pte_kind=KIND_VER2, level=4) + except GmmuDecodeError as exc: + return _fault(FAULT_UNSUPPORTED, str(exc), steps + [step], + va=va, root_pa=root_pa) + leaf_mask = 0xFFFF if page_used == "64k" else 0xFFF + step.update({"role": "pte", "page_size": page_used, + "aperture_name": pte["aperture_name"], "valid": pte["valid"], + "address": pte["address"], "writable": pte["writable"]}) + steps.append(step) + if not pte["valid"] or pte["address"] is None: + return _fault(FAULT_PTE_INVALID, + f"leaf PTE at {table_pa + off:#x} invalid/unsupported " + f"(raw {value:#x})", steps, va=va, root_pa=root_pa) + if not pte["address_complete"]: + pa_complete = False + pa = pte["address"] + (va & leaf_mask) + return _ok(va, pa, page_used, pte, steps, root_pa, kind, + page_used=page_used, evidence=evidence, pa_complete=pa_complete) + + return _fault(FAULT_UNSUPPORTED, "walk exceeded 5 levels", steps, + va=va, root_pa=root_pa) + + +def _ok(va, pa, page_size, leaf, steps, root_pa, kind, *, page_used, evidence, + pa_complete) -> dict: + report = { + "format": WALK_FORMAT, + "status": "ok", + "va": f"{va:#x}", + "root_pa": f"{root_pa:#x}", + "kind": kind, + "page_mode_used": page_used, + "page_size": page_size, + "evidence": evidence, + "pa": f"{pa:#x}", + "pa_complete": pa_complete, + "aperture_name": leaf.get("aperture_name"), + "writable": leaf.get("writable", True), + "read_only": leaf.get("read_only", False), + "kind_name": leaf.get("kind_name"), + "steps": steps, + "assumptions": [ + "4K/2M paths are validated against a captured channel PDB; the 64K dual path is modelled and reported with evidence='modeled'.", + "A video-memory leaf reports pa_complete=false (high address bits are not claimed by the decoder).", + ], + } + if page_size == "64k": + report["assumptions"].append("64K page-mode selection for auto walks is an assumption (64K-aligned VA prefers the BIG half).") + return report + + +def walk_many(vas, *, root_pa: int, mem: MemoryView, kind: str = KIND_VER2, + page_mode: str = "4k") -> list[dict]: + return [walk(va, root_pa=root_pa, mem=mem, kind=kind, page_mode=page_mode) + for va in vas] + + +# --------------------------------------------------------------------------- +# Captured golden vector: a live channel PDB (numeric only) +# --------------------------------------------------------------------------- + +#: Captured channel-PDB entries from a live run (five VAs of one channel +#: chunk). Used as the walker's regression vector: the walk must reproduce +#: the DMA addresses the capture logged for these VAs. +_CHANNEL_PDB = { + "root_pa": 0x12F16E000, + "root": {0: 0x0000_0000_12F16F0E}, + "pd2": {0: 0x0000_0000_11065E0E}, + "pd1": {0: 0x0000_0000_1106630E}, + "pd0_small": {0: 0x0000_0000_15908D0E}, + "pt": { + 256: 0x0000_0000_2045D90F, + 257: 0x0000_0000_121BD90F, + 258: 0x0000_0000_1521150F, + 259: 0x0000_0000_15A5F50F, + 260: 0x0000_0000_10CF270F, + }, +} + +_CHANNEL_VAS = { + 0x100000: ("gpfifo", 0x2045D9000), + 0x101000: ("push-buffer", 0x121BD9000), + 0x102000: ("copy-src", 0x152115000), + 0x103000: ("copy-dst", 0x15A5F5000), + 0x104000: ("semaphore", 0x10CF27000), +} + + +def channel_pdb_fixture() -> tuple[int, MemoryView]: + """Rebuild the captured channel PDB as a walkable memory view.""" + mem = MemoryView() + root = bytearray(0x1000) + for idx, value in _CHANNEL_PDB["root"].items(): + struct.pack_into("<Q", root, idx * 8, value) + mem.add(0x12F16E000, bytes(root)) + for name, base in (("pd2", 0x12F16F000), ("pd1", 0x11065E000)): + page = bytearray(0x1000) + for idx, value in _CHANNEL_PDB[name].items(): + struct.pack_into("<Q", page, idx * 8, value) + mem.add(base, bytes(page)) + pd0 = bytearray(0x1000) + for idx, small in _CHANNEL_PDB["pd0_small"].items(): + struct.pack_into("<16s", pd0, idx * 16, encode_dual_slot(None, (small >> 8) << 12)) + mem.add(0x110663000, bytes(pd0)) + pt = bytearray(0x1000) + for idx, value in _CHANNEL_PDB["pt"].items(): + struct.pack_into("<Q", pt, idx * 8, value) + mem.add(0x15908D000, bytes(pt)) + return _CHANNEL_PDB["root_pa"], mem + + +def run_demo(out: str | None = None) -> dict: + """Walk the captured channel-PDB vector; PASS iff every VA matches.""" + root_pa, mem = channel_pdb_fixture() + rows = [] + ok = True + for va, (label, expected) in sorted(_CHANNEL_VAS.items()): + report = walk(va, root_pa=root_pa, mem=mem) + got = int(report["pa"], 16) if report["status"] == "ok" else None + match = got == expected + ok = ok and match + rows.append({"label": label, "va": f"{va:#x}", "expected_pa": f"{expected:#x}", + "walk": report, "match": match}) + demo = {"format": "greeneye-gmmu-walk-demo/1", "ok": ok, "rows": rows, + "root_pa": f"{root_pa:#x}"} + if out: + os.makedirs(out, exist_ok=True) + path = os.path.join(out, "gmmu-walk-demo.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(demo, fh, indent=2, sort_keys=True) + fh.write("\n") + demo["artifacts"] = {"demo_json": os.path.abspath(path)} + return demo + + +def _parse_int(text: str) -> int: + try: + return int(text, 0) + except ValueError as exc: + raise WalkError(f"bad integer {text!r}") from exc + + +def _parse_mem(spec: str) -> tuple[int, bytes]: + if "=" not in spec: + raise WalkError(f"--mem needs <pa>=<hex>: {spec!r}") + pa_text, _, hex_text = spec.partition("=") + data = "".join(hex_text.split()) + if len(data) % 2: + raise WalkError(f"odd hex length in {spec!r}") + return _parse_int(pa_text), bytes.fromhex(data) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="gmmu_walk.py", + description="Walk a GMMU VA through captured/constructed page tables (offline).", + ) + sub = parser.add_subparsers(dest="command") + run = sub.add_parser("run", help="walk VAs through a page-table view") + run.add_argument("--pdb-pa", required=True, help="page-directory base PA") + run.add_argument("--va", action="append", required=True, help="virtual address (repeatable)") + run.add_argument("--mem", action="append", default=[], help="<pa>=<hex> table page (repeatable)") + run.add_argument("--kind", default=KIND_VER2, choices=list(SUPPORTED_KINDS)) + run.add_argument("--page-mode", default="4k", choices=list(PAGE_MODES)) + run.add_argument("--out", help="write the JSON report here") + demo = sub.add_parser("demo", help="walk the captured channel-PDB vector") + demo.add_argument("--out", help="output directory") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + if args.command == "run": + mem = MemoryView() + for spec in args.mem: + pa, data = _parse_mem(spec) + mem.add(pa, data) + reports = [walk(_parse_int(va), root_pa=_parse_int(args.pdb_pa), mem=mem, + kind=args.kind, page_mode=args.page_mode) for va in args.va] + doc = {"format": "greeneye-gmmu-walk-set/1", "reports": reports} + for report in reports: + if report["status"] == "ok": + print(f"OK {report['va']} -> {report['pa']} " + f"{report['page_size']} {report['aperture_name']} " + f"writable={report['writable']}") + else: + print(f"FAULT {report['va']} {report['fault']}: {report['fault_message']}") + if args.out: + os.makedirs(args.out, exist_ok=True) + path = os.path.join(args.out, "gmmu-walk.json") + with open(path, "w", encoding="utf-8") as fh: + json.dump(doc, fh, indent=2, sort_keys=True) + fh.write("\n") + print(json.dumps({"report_json": os.path.abspath(path)}, indent=2)) + return 0 if all(r["status"] == "ok" for r in reports) else 3 + if args.command == "demo": + demo = run_demo(args.out) + for row in demo["rows"]: + report = row["walk"] + got = report.get("pa", report.get("fault_message")) + print(f"{'PASS' if row['match'] else 'FAIL'} {row['label']:<12} " + f"{row['va']} -> {got} (expected {row['expected_pa']})") + print(json.dumps({"ok": demo["ok"]}, indent=2)) + return 0 if demo["ok"] else 3 + parser.print_help() + return 2 + except WalkError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/sim/host_traffic.py b/debug/risc-v-gsp-rm-vm/sim/host_traffic.py new file mode 100644 index 0000000..a9dd55d --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/sim/host_traffic.py @@ -0,0 +1,1437 @@ +"""Host-traffic message loop: doorbell + GSP msgq transport + replay (B4/B5). + +Status: **stub wave (in development)**. This module lands the transport half +of the B4/B5 rung (docs/35-GREENEYE-SIMULATOR-DESIGN.md §3.3, docs/45, +handoff-simulator.md §31 item B1): a modelled host doorbell, a command/status +message-queue pair laid out with the real queue arithmetic, queue-element +framing with the real checksum, a small stub dispatcher, and deterministic +replay of an ``rm_capture`` stimulus. The real 570.144 firmware is **not** +run by this module: no real message loop, no RM semantics, no firmware claim. + +Real structures mirrored here (provenance: public OpenRM headers and the +header-derived layouts already used by the project's boot code; every constant +is checked against ``harness/rm_capture.py`` where the two overlap): + +* ``NV_PGSP_QUEUE_HEAD(i) = 0x110c00 + i * 8`` (public Turing ``dev_gsp.h``), + 32-bit RW; the 570.144 host writes it after submitting a command + (``kgspSetCmdQueueHead``), and the command data itself lives in the sysmem + message queues. The simulator reaches the register at ``PRIV+0x110c00``. +* ``msgqTxHeader`` (8 x u32: version/size/msgSize/msgCount/writePtr/flags/ + rxHdrOff/entryOff) and ``msgqRxHeader`` (readPtr), with + ``rxHdrOff = align32(16) = 32`` and ``entryOff = align36(16) = 48`` for the + GSP header alignment of 4 (``msgqTxCreate``). +* ``GSP_MSG_QUEUE_ELEMENT``: 16 B authTag + 16 B aad + checksum/seqNum/ + elemCount/reserved (16 B) + the 32 B ``rpc_message_header_v`` (so the RPC + payload starts at 0x50). Checksum = XOR of 64-bit words, hi32 ^ lo32; the + stored field makes a receiver's recomputation zero (verified). +* The real command queue is created with ``MSGQ_FLAGS_SWAP_RX``; the two rx + headers are cross-linked between the command and status buffers. That + cross-link is modelled, not simplified away. +* The RPC call shape is the one ``harness/rm_capture.py`` decodes + (``rpc_message_header_v`` = 32 bytes, GSP_RM_CONTROL = 76, GSP_RM_ALLOC = + 103); the demo's 512-byte control is real-shaped but comes from no capture. + +What is stubbed: the dispatcher answers only ``RPC_FN_ECHO`` / +``RPC_FN_STATUS_QUERY`` (stub-only function ids) and returns +``NV_ERR_NOT_SUPPORTED`` for every real RM function. That reply is framed and +transported exactly, but it is a stub verdict, never firmware behaviour. + +Determinism: the transport transcript carries no host timestamps and is +byte-identical across runs; only the optional ``greeneye-session/1`` recorder +events carry host ``t`` (documented as non-semantic by ``compare``). Every +run is bounded: a message stages contiguously and splits across at most 16 +ring slots (the wire element maximum; larger messages are refused), ``poll`` +drains only the ring, ``run_cycles`` is capped. No hardware, module, mdev, +guest, network or QEMU is touched. +""" +from __future__ import annotations + +import argparse +import json +import os +import struct +import sys +from dataclasses import dataclass +from typing import Any, Callable + +from .devices.base import Device + +_TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +#: GSP doorbell register block (public Turing ``dev_gsp.h``, block-relative). +QUEUE_HEAD_OFFSET = 0x110C00 +QUEUE_HEAD_STRIDE = 8 +QUEUE_HEAD_COUNT = 8 +#: PRIV aperture base used by the GSP profiles; doorbell window = PRIV + offset. +PRIV_BASE = 0x2000000000000000 +DOORBELL_BASE = PRIV_BASE + QUEUE_HEAD_OFFSET +DOORBELL_SIZE = QUEUE_HEAD_STRIDE * QUEUE_HEAD_COUNT +#: Host writes the queue head to this register as the kick (observed 570.144 +#: submit path writes 0); the model records the value per queue index. +DOORBELL_KICK_QUEUE = 0 +DOORBELL_KICK_VALUE = 0 + +#: ``msgq_priv.h``. +MSGQ_VERSION = 0 +MSGQ_FLAGS_SWAP_RX = 1 +MSGQ_TX_HEADER_BYTES = 32 +MSGQ_RX_HEADER_BYTES = 4 +MSGQ_HEADER_ALIGN = 4 +MSGQ_RX_HDR_OFF = 32 +MSGQ_ENTRY_OFF = 48 +MSGQ_TX_HEADER_FORMAT = "<8I" +MSGQ_RX_HEADER_FORMAT = "<I" + +#: ``message_queue_priv.h``. +GSP_MSG_QUEUE_ELEMENT_SIZE_MIN = 0x1000 +GSP_MSG_QUEUE_ELEMENT_SIZE_MAX = 0x10000 +GSP_MSG_QUEUE_HEADER_SIZE = 0x1000 +GSP_MSG_QUEUE_ELEMENT_HDR_SIZE = 0x30 + +#: ``g_rpc-message-header.h`` / 570.144 wire constants. +RPC_HEADER_BYTES = 32 +RPC_HEADER_FORMAT = "<8I" +RPC_SIGNATURE = 0x43505256 +RPC_HEADER_VERSION = 0x03000000 + +#: Default shared sysmem queue block and sizes (host ``message_queue_cpu.c``). +DEFAULT_CMD_QUEUE_SIZE = 0x40000 +DEFAULT_STAT_QUEUE_SIZE = 0x40000 +SHARED_MEM_BASE = 0x81000000 + +#: Real RM wire function ids used by the capture fixtures. +GSP_RM_CONTROL = 76 +GSP_RM_ALLOC = 103 +GSP_GET_GSP_STATIC_INFO = 65 + +#: Stub-only dispatcher ids (deliberately not RM function numbers). +RPC_FN_ECHO = 0x40 +RPC_FN_STATUS_QUERY = 0x41 + +#: ``nvstatuscodes.h``. +NV_OK = 0 +NV_ERR_NOT_SUPPORTED = 0x56 + +#: Bounded ``run_cycles`` cap (same ceiling as the boot runner's attempts). +MAX_RUN_CYCLES = 2_000_000 + +STATUS_MAGIC = b"GSTS" +STATUS_VERSION = 1 +STATUS_FORMAT = "<4sIIII" + +DEMO_FORMAT = "greeneye-host-traffic-demo/1" +REPLAY_FORMAT = "greeneye-host-traffic-replay/1" +TRANSCRIPT_FORMAT = "greeneye-host-traffic-log/1" +REPLY_FORMAT = "greeneye-host-traffic-reply/1" +STIMULUS_FORMAT = "greeneye-rm-rpc-stimulus/1" +CAPTURE_FORMAT = "greeneye-rm-rpc-capture/1" + +_TX_HEADER = struct.Struct(MSGQ_TX_HEADER_FORMAT) +_RX_HEADER = struct.Struct(MSGQ_RX_HEADER_FORMAT) +_RPC_HEADER = struct.Struct(RPC_HEADER_FORMAT) +_ELEMENT_PREFIX = struct.Struct("<16s16sIIII") + +_LIMITS = [ + "Stub wave: no real firmware message loop is run; the dispatcher handlers " + "are stubs and real RM functions get NV_ERR_NOT_SUPPORTED.", + "Multi-element messages are split across ring slots up to the wire element " + "maximum (0x10000 bytes); larger messages are refused, not truncated.", + "The shared-memory address and queue sizes are fixture defaults, not " + "values read from a real GSP_ARGUMENTS_CACHED transfer.", + "The MSI-X/event path by which the real GSP signals a status-queue write " + "is not modelled; replies are read by polling.", + "L0/L1 framing and ordering only; no timing (L2) and no device-environment " + "(L3) claim.", +] + + +class HostTrafficError(Exception): + """A framing, queue or replay input that cannot be represented.""" + + +def _align_up(value: int, align: int) -> int: + return (value + align - 1) // align * align + + +# --------------------------------------------------------------------------- +# Framing +# --------------------------------------------------------------------------- + + +@dataclass +class RpcMessage: + """One decoded ``rpc_message_header_v`` plus its payload.""" + + header_version: int + signature: int + length: int + function: int + rpc_result: int + rpc_result_private: int + sequence: int + spare: int + params: bytes = b"" + + def to_bytes(self) -> bytes: + return _RPC_HEADER.pack( + self.header_version, + self.signature, + RPC_HEADER_BYTES + len(self.params), + self.function, + self.rpc_result, + self.rpc_result_private, + self.sequence, + self.spare, + ) + self.params + + +@dataclass +class RpcRequest: + """A command message handed to a dispatcher handler.""" + + function: int + sequence: int + params: bytes + element_index: int + raw: bytes + + +@dataclass +class RpcReply: + """A handler result before framing into a status element.""" + + function: int + sequence: int + params: bytes = b"" + result: int = NV_OK + + +def build_rpc_message( + function: int, + params: bytes = b"", + sequence: int = 0, + *, + result: int = NV_OK, + result_private: int = 0, + header_version: int = RPC_HEADER_VERSION, + signature: int = RPC_SIGNATURE, +) -> bytes: + """Frame one ``rpc_message_header_v`` + payload (capture-tool shape).""" + payload = bytes(params) + return _RPC_HEADER.pack( + header_version, + signature, + RPC_HEADER_BYTES + len(payload), + int(function) & 0xFFFFFFFF, + int(result) & 0xFFFFFFFF, + int(result_private) & 0xFFFFFFFF, + int(sequence) & 0xFFFFFFFF, + 0, + ) + payload + + +def parse_rpc_message(data: bytes) -> RpcMessage: + """Decode and validate one RPC message; raises on a bad frame.""" + if len(data) < RPC_HEADER_BYTES: + raise HostTrafficError( + f"rpc message shorter than {RPC_HEADER_BYTES} bytes: {len(data)}" + ) + ( + header_version, + signature, + length, + function, + rpc_result, + rpc_result_private, + sequence, + spare, + ) = _RPC_HEADER.unpack_from(data, 0) + if signature != RPC_SIGNATURE: + raise HostTrafficError( + f"bad rpc signature {signature:#010x} (expected {RPC_SIGNATURE:#010x})" + ) + if length < RPC_HEADER_BYTES or length > len(data): + raise HostTrafficError( + f"rpc length {length:#x} outside [{RPC_HEADER_BYTES:#x}, {len(data):#x}]" + ) + return RpcMessage( + header_version=header_version, + signature=signature, + length=length, + function=function, + rpc_result=rpc_result, + rpc_result_private=rpc_result_private, + sequence=sequence, + spare=spare, + params=bytes(data[RPC_HEADER_BYTES:length]), + ) + + +def compute_message_checksum(data: bytes) -> int: + """The msgq checksum: XOR of 64-bit words, then hi32 ^ lo32. + + Zero-pads to 8 bytes exactly like ``_checkSum32``; a correctly built queue + element recomputes to 0. + """ + padded = bytes(data) + bytes((-len(data)) % 8) + total = 0 + for offset in range(0, len(padded), 8): + total ^= int.from_bytes(padded[offset : offset + 8], "little") + return ((total >> 32) ^ (total & 0xFFFFFFFF)) & 0xFFFFFFFF + + +def build_queue_element( + rpc_bytes: bytes, *, sequence: int = 0, elem_count: int | None = None +) -> bytes: + """Wrap an RPC message in a ``GSP_MSG_QUEUE_ELEMENT`` with checksum.""" + payload = bytes(rpc_bytes) + if elem_count is None: + elem_count = max(1, -(-(GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + len(payload)) + // GSP_MSG_QUEUE_ELEMENT_SIZE_MIN)) + prefix = _ELEMENT_PREFIX.pack(b"\x00" * 16, b"\x00" * 16, 0, + sequence & 0xFFFFFFFF, + elem_count & 0xFFFFFFFF, 0) + body = prefix + payload + checksum = compute_message_checksum(body) + return _ELEMENT_PREFIX.pack(b"\x00" * 16, b"\x00" * 16, checksum, + sequence & 0xFFFFFFFF, + elem_count & 0xFFFFFFFF, 0) + payload + + +def split_element_stream(stream: bytes, msg_size: int) -> list[bytes]: + """Split a contiguous element stream into ``msg_size`` ring slots. + + The real producer stages the whole message in one buffer and copies + ``GSP_MSG_QUEUE_ELEMENT_SIZE_MIN`` bytes per ring slot + (``GspMsgQueueSendCommand``); the tail slot is zero-padded. + """ + blob = bytes(stream) + if not blob: + raise HostTrafficError("empty element stream") + slots = [] + for start in range(0, len(blob), msg_size): + piece = blob[start : start + msg_size] + slots.append(piece + bytes(msg_size - len(piece))) + return slots + + +def parse_queue_element(data: bytes, *, verify: bool = True) -> tuple[dict, RpcMessage]: + """Decode one queue element stream; raises on a bad frame or checksum.""" + if len(data) < GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + RPC_HEADER_BYTES: + raise HostTrafficError(f"queue element too short: {len(data)} bytes") + auth_tag, aad, checksum, seq_num, elem_count, reserved = _ELEMENT_PREFIX.unpack_from(data, 0) + if not 1 <= elem_count <= GSP_MSG_QUEUE_ELEMENT_SIZE_MAX // GSP_MSG_QUEUE_ELEMENT_SIZE_MIN: + raise HostTrafficError(f"implausible elem_count {elem_count}") + rpc = parse_rpc_message(data[GSP_MSG_QUEUE_ELEMENT_HDR_SIZE:]) + total = GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + rpc.length + if total > len(data): + raise HostTrafficError(f"queue element truncated: need {total:#x}, have {len(data):#x}") + if verify and compute_message_checksum(bytes(data[:total])) != 0: + raise HostTrafficError("bad message checksum") + prefix = { + "auth_tag": auth_tag, + "aad": aad, + "checksum": checksum, + "seq_num": seq_num, + "elem_count": elem_count, + "reserved": reserved, + } + return prefix, rpc + + +# --------------------------------------------------------------------------- +# Memory adapters +# --------------------------------------------------------------------------- + + +class RegionMemory: + """A ``base``/``size`` bytearray window with Machine-like read/write.""" + + def __init__(self, base: int, size: int): + if size <= 0: + raise HostTrafficError(f"region size must be positive, got {size}") + self.base = int(base) + self.size = int(size) + self.data = bytearray(self.size) + + def _check(self, addr: int, size: int) -> int: + offset = addr - self.base + if offset < 0 or offset + size > self.size: + raise HostTrafficError( + f"access outside region: addr={addr:#x} size={size}" + ) + return offset + + def read(self, addr: int, size: int) -> bytes: + offset = self._check(addr, size) + return bytes(self.data[offset : offset + size]) + + def write(self, addr: int, data: bytes) -> None: + blob = bytes(data) + offset = self._check(addr, len(blob)) + self.data[offset : offset + len(blob)] = blob + + +class _MachineMemory: + """Adapter from RegionMemory's surface to ``sim.machine.Machine``.""" + + def __init__(self, machine): + self.machine = machine + + def read(self, addr: int, size: int) -> bytes: + return self.machine.read_mem(addr, size) + + def write(self, addr: int, data: bytes) -> None: + self.machine.write_mem(addr, data) + + +def _make_memory(machine, shared_base: int, size: int): + if machine is None: + return RegionMemory(shared_base, size) + return _MachineMemory(machine) + + +# --------------------------------------------------------------------------- +# Message queues +# --------------------------------------------------------------------------- + + +@dataclass +class MsgqTxHeader: + """``msgqTxHeader``: 8 x u32 written by the queue producer.""" + + version: int + size: int + msg_size: int + msg_count: int + write_ptr: int + flags: int + rx_hdr_off: int + entry_off: int + + def to_bytes(self) -> bytes: + return _TX_HEADER.pack( + self.version, + self.size, + self.msg_size, + self.msg_count, + self.write_ptr, + self.flags, + self.rx_hdr_off, + self.entry_off, + ) + + @classmethod + def from_bytes(cls, data: bytes) -> "MsgqTxHeader": + return cls(*_TX_HEADER.unpack_from(data, 0)) + + +class MsgqQueue: + """One msgq buffer: tx header at its base, entries at ``entryOff``. + + ``rx_hdr_addr`` is the physical address of this queue's rx header. The + real command queue keeps it in the status buffer (``MSGQ_FLAGS_SWAP_RX``), + so the caller passes the cross-linked address. + """ + + def __init__( + self, + mem, + base: int, + size: int, + *, + msg_size: int = GSP_MSG_QUEUE_ELEMENT_SIZE_MIN, + header_align: int = MSGQ_HEADER_ALIGN, + flags: int = MSGQ_FLAGS_SWAP_RX, + rx_hdr_addr: int | None = None, + version: int = MSGQ_VERSION, + ): + if size < GSP_MSG_QUEUE_ELEMENT_SIZE_MIN: + raise HostTrafficError(f"queue size {size:#x} below one element") + self.mem = mem + self.tx_base = int(base) + self.size = int(size) + self.msg_size = int(msg_size) + self.header_align = int(header_align) + self.flags = int(flags) + self.version = int(version) + self.rx_hdr_addr = int(rx_hdr_addr) if rx_hdr_addr is not None else base + MSGQ_RX_HDR_OFF + + def format(self) -> MsgqTxHeader: + """Write a fresh tx header and zero the rx header (msgqTxCreate).""" + rx_hdr_off = _align_up(MSGQ_TX_HEADER_BYTES, 1 << self.header_align) + entry_off = _align_up(rx_hdr_off + MSGQ_RX_HEADER_BYTES, 1 << self.header_align) + header = MsgqTxHeader( + version=self.version, + size=self.size, + msg_size=self.msg_size, + msg_count=(self.size - entry_off) // self.msg_size, + write_ptr=0, + flags=self.flags, + rx_hdr_off=rx_hdr_off, + entry_off=entry_off, + ) + self.mem.write(self.tx_base, header.to_bytes()) + self.mem.write(self.rx_hdr_addr, _RX_HEADER.pack(0)) + return header + + def header(self) -> MsgqTxHeader: + return MsgqTxHeader.from_bytes(self.mem.read(self.tx_base, MSGQ_TX_HEADER_BYTES)) + + def _write_header(self, header: MsgqTxHeader) -> None: + self.mem.write(self.tx_base, header.to_bytes()) + + @property + def write_ptr(self) -> int: + return self.header().write_ptr + + @property + def read_ptr(self) -> int: + return _RX_HEADER.unpack_from(self.mem.read(self.rx_hdr_addr, MSGQ_RX_HEADER_BYTES), 0)[0] + + def _entry_addr(self, index: int) -> int: + count = self.header().msg_count + if not 0 <= index < count: + raise HostTrafficError(f"entry index {index} outside 0..{count - 1}") + return self.tx_base + self.header().entry_off + index * self.msg_size + + def write_entry(self, index: int, data: bytes) -> None: + blob = bytes(data) + if len(blob) > self.msg_size: + raise HostTrafficError( + f"entry of {len(blob)} bytes exceeds msgSize {self.msg_size:#x}" + ) + self.mem.write(self._entry_addr(index), blob) + + def read_entry(self, index: int, size: int | None = None) -> bytes: + return self.mem.read(self._entry_addr(index), self.msg_size if size is None else size) + + def publish(self, write_ptr: int) -> int: + header = self.header() + header.write_ptr = int(write_ptr) % header.msg_count + self._write_header(header) + return header.write_ptr + + def set_read(self, read_ptr: int) -> int: + count = self.header().msg_count + value = int(read_ptr) % count + self.mem.write(self.rx_hdr_addr, _RX_HEADER.pack(value)) + return value + + def state(self) -> dict: + header = self.header() + return { + "tx_base": self.tx_base, + "size": self.size, + "msg_size": header.msg_size, + "msg_count": header.msg_count, + "write_ptr": header.write_ptr, + "read_ptr": self.read_ptr, + "flags": header.flags, + "rx_hdr_off": header.rx_hdr_off, + "entry_off": header.entry_off, + "rx_hdr_addr": self.rx_hdr_addr, + } + + +class SharedQueues: + """A command/status msgq pair with the real cross-linked rx headers. + + Layout follows the host's ``GspMsgQueuesInit``: the status buffer starts + ``cmd_size`` bytes after the command buffer, both tx headers carry + ``MSGQ_FLAGS_SWAP_RX``, and each queue's rx header physically lives in the + other buffer (``msgqRxLink``). + """ + + def __init__( + self, + mem, + *, + cmd_base: int = SHARED_MEM_BASE, + stat_base: int | None = None, + cmd_size: int = DEFAULT_CMD_QUEUE_SIZE, + stat_size: int = DEFAULT_STAT_QUEUE_SIZE, + msg_size: int = GSP_MSG_QUEUE_ELEMENT_SIZE_MIN, + swap_rx: bool = True, + ): + self.mem = mem + self.cmd_base = int(cmd_base) + self.stat_base = int(stat_base) if stat_base is not None else self.cmd_base + cmd_size + flags = MSGQ_FLAGS_SWAP_RX if swap_rx else 0 + self.command = MsgqQueue( + mem, + self.cmd_base, + cmd_size, + msg_size=msg_size, + flags=flags, + rx_hdr_addr=self.stat_base + MSGQ_RX_HDR_OFF, + ) + self.status = MsgqQueue( + mem, + self.stat_base, + stat_size, + msg_size=msg_size, + flags=flags, + rx_hdr_addr=self.cmd_base + MSGQ_RX_HDR_OFF, + ) + + def format(self) -> None: + self.command.format() + self.status.format() + + def _place(self, queue: MsgqQueue, element: bytes) -> dict: + slots = split_element_stream(element, queue.msg_size) + count = queue.header().msg_count + if len(slots) > count: + raise HostTrafficError( + f"message needs {len(slots)} elements, queue holds {count}" + ) + index = queue.write_ptr + ptr = index + for slot in slots: + queue.write_entry(ptr, slot) + ptr = (ptr + 1) % count + write_ptr = queue.publish(ptr) + return { + "index": index, + "write_ptr": write_ptr, + "bytes": len(element), + "elements": len(slots), + } + + def _take(self, queue: MsgqQueue) -> dict | None: + write_ptr = queue.write_ptr + read_ptr = queue.read_ptr + if read_ptr == write_ptr: + return None + count = queue.header().msg_count + first = queue.read_entry(read_ptr) + elements = int.from_bytes(first[0x28:0x2C], "little") + if not 1 <= elements <= count: + raise HostTrafficError(f"implausible elem_count {elements}") + slots = [] + ptr = read_ptr + for _ in range(elements): + slots.append(queue.read_entry(ptr)) + ptr = (ptr + 1) % count + stream = b"".join(slots) + total = GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + int.from_bytes( + stream[0x38:0x3C], "little" + ) + element = ( + stream[:total] + if GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + RPC_HEADER_BYTES <= total <= len(stream) + else stream + ) + queue.set_read(ptr) + return {"index": read_ptr, "element": element, "elements": elements} + + def submit_element(self, element: bytes) -> dict: + """Host -> GSP: append one element to the command ring.""" + return self._place(self.command, element) + + def take_command(self) -> dict | None: + """GSP side: consume the next command element, or ``None``.""" + return self._take(self.command) + + def publish_element(self, element: bytes) -> dict: + """GSP -> host: append one element to the status ring.""" + return self._place(self.status, element) + + def take_status(self) -> dict | None: + """Host side: consume the next status element, or ``None``.""" + return self._take(self.status) + + def state(self) -> dict: + return {"command": self.command.state(), "status": self.status.state()} + + +# --------------------------------------------------------------------------- +# Doorbell +# --------------------------------------------------------------------------- + + +class DoorbellModel: + """Recorded queue-head writes; the firmware-visible read-back surface.""" + + def __init__(self, *, count: int = QUEUE_HEAD_COUNT, max_writes: int = 1024): + self.count = int(count) + self.max_writes = int(max_writes) + self.heads = [0] * self.count + self.writes: list[dict] = [] + self.dropped = 0 + self.kicks = 0 + + def ring(self, queue: int, value: int = 0, *, pc: int | None = None) -> None: + queue = int(queue) + if not 0 <= queue < self.count: + raise ValueError(f"queue index {queue} outside 0..{self.count - 1}") + value = int(value) & 0xFFFFFFFF + self.heads[queue] = value + self.kicks += 1 + record = {"queue": queue, "value": value} + if pc is not None: + record["pc"] = int(pc) + if len(self.writes) < self.max_writes: + self.writes.append(record) + else: + self.dropped += 1 + + def head(self, queue: int) -> int: + queue = int(queue) + if not 0 <= queue < self.count: + raise ValueError(f"queue index {queue} outside 0..{self.count - 1}") + return self.heads[queue] + + def state(self) -> dict: + return { + "count": self.count, + "heads": list(self.heads), + "kicks": self.kicks, + "writes": [dict(entry) for entry in self.writes], + "dropped": self.dropped, + } + + def reset(self) -> None: + self.heads = [0] * self.count + self.writes = [] + self.dropped = 0 + self.kicks = 0 + + +class DoorbellBlock(Device): + """``NV_PGSP_QUEUE_HEAD`` window over a :class:`DoorbellModel`. + + Evidence: hardware register header (offset/width/access) for the register; + the model's read-back is storage of the last written value, and the + firmware's read use of this register is **not observed** (documented, not + claimed). Guest stores are recorded with their PC. + """ + + name = "gsp-doorbell" + wants_pc = True + + def __init__(self, doorbell: DoorbellModel | None = None, *, + base: int = DOORBELL_BASE, size: int = DOORBELL_SIZE): + super().__init__(base=base, size=size) + self.doorbell = doorbell if doorbell is not None else DoorbellModel() + + def reset(self) -> None: + if getattr(self, "doorbell", None) is None: + self.doorbell = DoorbellModel() + else: + self.doorbell.reset() + + def _queue(self, offset: int, size: int) -> int | None: + if size != 4 or offset % QUEUE_HEAD_STRIDE: + return None + queue = offset // QUEUE_HEAD_STRIDE + if not 0 <= queue < QUEUE_HEAD_COUNT: + return None + return queue + + def read(self, offset: int, size: int, pc: int = 0) -> int | None: + queue = self._queue(offset, size) + if queue is None: + return None + return self.doorbell.head(queue) + + def write(self, offset: int, size: int, value: int, pc: int = 0) -> bool: + queue = self._queue(offset, size) + if queue is None: + return False + self.doorbell.ring(queue, value, pc=pc) + return True + + def peek(self, offset: int, size: int) -> int | None: + return self.read(offset, size) + + def state(self) -> dict: + return {"doorbell": self.doorbell.state()} + + def load_state(self, data: dict) -> None: + state = data.get("doorbell", data) + self.doorbell.count = int(state.get("count", self.doorbell.count)) + self.doorbell.heads = [int(v) for v in state.get("heads", [0] * self.doorbell.count)] + self.doorbell.kicks = int(state.get("kicks", 0)) + self.doorbell.writes = [dict(entry) for entry in state.get("writes", [])] + self.doorbell.dropped = int(state.get("dropped", 0)) + + +# --------------------------------------------------------------------------- +# Recorder compatibility +# --------------------------------------------------------------------------- + + +class _NullRecorder: + """Stand-in for ``greeneye.record.NullRecorder`` (keeps sim standalone).""" + + dir: str | None = None + + def event(self, event_type: str, **fields) -> None: + pass + + def blob(self, name: str, data: bytes) -> str: + return name + + +# --------------------------------------------------------------------------- +# Host traffic loop +# --------------------------------------------------------------------------- + + +class HostTraffic: + """The modelled host-side transport: submit, dispatch (stub), read reply. + + ``machine`` (optional) supplies simulated RAM for the queue block and lets + :meth:`run_cycles` step the core; without it the queues live in a private + :class:`RegionMemory`. ``recorder`` (optional) is any object with + ``event(type, **fields)`` / ``blob(name, data)``; the transport transcript + stays deterministic regardless. + """ + + def __init__( + self, + *, + machine=None, + recorder=None, + shared_base: int = SHARED_MEM_BASE, + cmd_size: int = DEFAULT_CMD_QUEUE_SIZE, + stat_size: int = DEFAULT_STAT_QUEUE_SIZE, + msg_size: int = GSP_MSG_QUEUE_ELEMENT_SIZE_MIN, + handlers: dict[int, Callable[..., Any]] | None = None, + ): + if machine is None and shared_base == SHARED_MEM_BASE: + mem = RegionMemory(shared_base, cmd_size + stat_size) + else: + mem = _make_memory(machine, shared_base, cmd_size + stat_size) + self.machine = machine + self.recorder = recorder if recorder is not None else _NullRecorder() + self.shared_base = int(shared_base) + self.queues = SharedQueues( + mem, + cmd_base=shared_base, + cmd_size=cmd_size, + stat_size=stat_size, + msg_size=msg_size, + ) + self.queues.format() + self.doorbell = DoorbellModel() + self._tx_seq = 0 + self._rx_seq = 0 + self._next_sequence = 1 + self._submitted = 0 + self._dispatched = 0 + self._replies = 0 + self._events: list[dict] = [] + self._closed = False + self._handlers: dict[int, tuple[str, Callable[..., Any]]] = {} + self.set_handler(RPC_FN_ECHO, self._handle_echo, name="stub:echo") + self.set_handler(RPC_FN_STATUS_QUERY, self._handle_status_query, + name="stub:status_query") + for function, handler in (handlers or {}).items(): + self.set_handler(function, handler) + self._emit( + "host_traffic_config", + shared_base=self.shared_base, + cmd_base=self.queues.command.tx_base, + stat_base=self.queues.status.tx_base, + cmd_size=self.queues.command.size, + stat_size=self.queues.status.size, + msg_size=self.queues.command.msg_size, + handlers=sorted(self.handler_names().items()), + ) + + # -- handler table ------------------------------------------------------ + + def set_handler(self, function: int, handler: Callable[..., Any], *, + name: str | None = None) -> None: + label = name or getattr(handler, "__name__", "handler") + self._handlers[int(function)] = (label, handler) + + def handler_names(self) -> dict[int, str]: + return {function: label for function, (label, _) in sorted(self._handlers.items())} + + def _handle_echo(self, request: RpcRequest) -> RpcReply: + return RpcReply(function=request.function, sequence=request.sequence, + params=request.params) + + def _handle_status_query(self, request: RpcRequest) -> RpcReply: + payload = struct.pack( + STATUS_FORMAT, + STATUS_MAGIC, + STATUS_VERSION, + self._submitted, + self._dispatched, + self._replies, + ) + return RpcReply(function=request.function, sequence=request.sequence, + params=payload) + + # -- recording ---------------------------------------------------------- + + def _emit(self, event_type: str, **fields) -> None: + self.recorder.event(event_type, **fields) + + def _record(self, event: dict) -> None: + self._events.append(event) + + def note_skip(self, index: int, reason: str) -> None: + """Record a replay message that was not submitted (documented).""" + self._record({"type": "skip", "index": int(index), "reason": reason}) + self._emit("host_replay_skip", index=int(index), reason=reason) + + # -- host side ---------------------------------------------------------- + + def submit(self, message: bytes) -> dict: + """Write one framed RPC message to the command ring, then kick.""" + parsed = parse_rpc_message(message) + element = build_queue_element(message, sequence=self._tx_seq) + if len(element) > GSP_MSG_QUEUE_ELEMENT_SIZE_MAX: + raise HostTrafficError( + f"message of {len(element)} bytes exceeds the wire element " + f"maximum of {GSP_MSG_QUEUE_ELEMENT_SIZE_MAX:#x}" + ) + placed = self.queues.submit_element(element) + self._tx_seq += 1 + self.doorbell.ring(DOORBELL_KICK_QUEUE, DOORBELL_KICK_VALUE) + self._submitted += 1 + self._emit("doorbell", queue=DOORBELL_KICK_QUEUE, + value=DOORBELL_KICK_VALUE, source="host") + result = { + "format": "greeneye-host-traffic-submit/1", + "index": placed["index"], + "write_ptr": placed["write_ptr"], + "function": parsed.function, + "sequence": parsed.sequence, + "msgq_sequence": self._tx_seq - 1, + "length": parsed.length, + "params_bytes": len(parsed.params), + "element_bytes": len(element), + "doorbell": {"queue": DOORBELL_KICK_QUEUE, "value": DOORBELL_KICK_VALUE}, + } + self._record({ + "type": "submit", + "index": placed["index"], + "function": parsed.function, + "sequence": parsed.sequence, + "params_bytes": len(parsed.params), + "element_bytes": len(element), + "write_ptr": placed["write_ptr"], + }) + self._emit( + "host_submit", + function=parsed.function, + sequence=parsed.sequence, + length=parsed.length, + params_bytes=len(parsed.params), + element_index=placed["index"], + write_ptr=placed["write_ptr"], + ) + return result + + def submit_rpc(self, function: int, params: bytes = b"", + *, sequence: int | None = None) -> dict: + """Frame and submit one RPC; ``sequence`` defaults to a local counter.""" + if sequence is None: + sequence = self._next_sequence + self._next_sequence = max(self._next_sequence, int(sequence) + 1) + return self.submit(build_rpc_message(function, params, sequence=sequence)) + + def _normalise_reply(self, produced: Any, request: RpcRequest) -> RpcReply: + if produced is None: + return RpcReply(function=request.function, sequence=request.sequence) + if isinstance(produced, RpcReply): + return produced + if isinstance(produced, (bytes, bytearray)): + return RpcReply(function=request.function, sequence=request.sequence, + params=bytes(produced)) + raise HostTrafficError( + f"handler returned {type(produced).__name__}; expected RpcReply, " + "bytes or None" + ) + + def poll(self) -> int: + """GSP side: drain the command ring through the dispatcher (bounded).""" + drained = 0 + while True: + taken = self.queues.take_command() + if taken is None: + break + prefix, request_message = parse_queue_element(taken["element"]) + label, handler = self._handlers.get( + request_message.function, ("stub:unsupported", None) + ) + self._dispatched += 1 + request = RpcRequest( + function=request_message.function, + sequence=request_message.sequence, + params=request_message.params, + element_index=taken["index"], + raw=bytes(taken["element"]), + ) + if handler is None: + reply = RpcReply( + function=request.function, + sequence=request.sequence, + params=b"", + result=NV_ERR_NOT_SUPPORTED, + ) + else: + reply = self._normalise_reply(handler(request), request) + reply_bytes = build_rpc_message( + reply.function, reply.params, sequence=reply.sequence, + result=reply.result, + ) + element = build_queue_element(reply_bytes, sequence=self._rx_seq) + self._rx_seq += 1 + if len(element) > GSP_MSG_QUEUE_ELEMENT_SIZE_MAX: + raise HostTrafficError( + f"reply of {len(element)} bytes exceeds the wire element maximum" + ) + self.queues.publish_element(element) + self._replies += 1 + drained += 1 + self._record({ + "type": "dispatch", + "index": taken["index"], + "function": request.function, + "sequence": request.sequence, + "handler": label, + "result": reply.result, + "reply_bytes": len(reply.params), + }) + self._emit( + "host_dispatch", + function=request.function, + sequence=request.sequence, + handler=label, + result=reply.result, + reply_bytes=len(reply.params), + element_index=taken["index"], + msgq_sequence=prefix["seq_num"], + ) + return drained + + def read_reply(self) -> dict | None: + """Host side: consume the next status element, or ``None``.""" + taken = self.queues.take_status() + if taken is None: + return None + _prefix, message = parse_queue_element(taken["element"]) + record = { + "format": REPLY_FORMAT, + "index": taken["index"], + "function": message.function, + "sequence": message.sequence, + "result": message.rpc_result, + "params": message.params, + "params_hex": message.params.hex(), + "params_bytes": len(message.params), + } + self._record({ + "type": "reply", + "index": taken["index"], + "function": message.function, + "sequence": message.sequence, + "result": message.rpc_result, + "params_bytes": len(message.params), + }) + self._emit( + "host_reply_read", + function=message.function, + sequence=message.sequence, + result=message.rpc_result, + params_bytes=len(message.params), + element_index=taken["index"], + ) + return record + + def run_cycles(self, budget: int) -> dict: + """Run the attached core for a bounded ``budget``, then ``poll``.""" + if isinstance(budget, bool) or not isinstance(budget, int) or not 0 < budget <= MAX_RUN_CYCLES: + raise ValueError(f"budget must be an int in 1..{MAX_RUN_CYCLES}, got {budget!r}") + reason = pc = None + if self.machine is not None: + reason, pc, _a, _b = self.machine.run(budget) + dispatched = self.poll() + return { + "budget": budget, + "reason": reason, + "pc": pc, + "insns": self.machine.insns if self.machine is not None else 0, + "dispatched": dispatched, + "replies": self._replies, + } + + # -- evidence ----------------------------------------------------------- + + def transcript(self) -> dict: + counts: dict[str, int] = {} + for event in self._events: + counts[event["type"]] = counts.get(event["type"], 0) + 1 + return { + "format": TRANSCRIPT_FORMAT, + "events": [dict(event) for event in self._events], + "counts": counts, + } + + def state(self) -> dict: + return { + "format": TRANSCRIPT_FORMAT, + "shared_base": self.shared_base, + "queues": self.queues.state(), + "doorbell": self.doorbell.state(), + "counts": { + "submitted": self._submitted, + "dispatched": self._dispatched, + "replies": self._replies, + }, + "handlers": self.handler_names(), + } + + def doorbell_block(self) -> DoorbellBlock: + return DoorbellBlock(self.doorbell) + + def close(self) -> None: + if self._closed: + return + self._closed = True + payload = json.dumps(self.transcript(), indent=2, sort_keys=True).encode() + self.recorder.blob("host-traffic.json", payload) + self._emit("host_transcript", events=len(self._events)) + + +# --------------------------------------------------------------------------- +# Stimulus replay +# --------------------------------------------------------------------------- + + +def _stimulus_from_capture(source: str | dict) -> dict: + if _TOOL_ROOT not in sys.path: + sys.path.insert(0, _TOOL_ROOT) + try: + from harness.rm_capture import load_capture, stimulus_from_capture + except ImportError as exc: # pragma: no cover - tool checkout broken + raise HostTrafficError(f"harness.rm_capture unavailable: {exc}") from exc + if isinstance(source, dict): + raise HostTrafficError( + "an inline capture document is not accepted; pass the capture JSON " + "path or a stimulus document" + ) + capture = load_capture(source) + return stimulus_from_capture(capture) + + +def load_stimulus(source: str | dict) -> dict: + """Load a replay stimulus from a doc, a stimulus JSON or a capture file.""" + if isinstance(source, dict): + doc = source + else: + if not os.path.isfile(source): + raise HostTrafficError(f"stimulus not found: {source}") + with open(source, "r", encoding="utf-8", errors="replace") as fh: + head = fh.read(4096) + if not head.lstrip().startswith("{"): + return _stimulus_from_capture(source) + try: + with open(source, "r", encoding="utf-8", errors="replace") as fh: + doc = json.load(fh) + except (OSError, json.JSONDecodeError) as exc: + raise HostTrafficError(f"cannot read stimulus {source}: {exc}") from exc + if not isinstance(doc, dict): + raise HostTrafficError("stimulus document must be an object") + fmt = str(doc.get("format") or "") + if fmt == STIMULUS_FORMAT: + return doc + if fmt == CAPTURE_FORMAT: + return _stimulus_from_capture(source) + raise HostTrafficError( + f"unknown stimulus/capture format {fmt!r}; expected {STIMULUS_FORMAT!r} " + f"or {CAPTURE_FORMAT!r}" + ) + + +def _message_to_rpc(message: dict, index: int) -> tuple[int, bytes]: + """Map one stimulus message to ``(wire function, payload)``. + + ``send`` messages carry the wire function and payload directly. ``control`` + and ``alloc`` records carry a named id instead; the capture tool's mapping + (control -> 76, alloc -> 103) is applied, and the captured parameter block + is used as the payload. That identity is an assumption until a capture + with full send payloads exists (documented in docs/51). + """ + kind = message.get("kind") + function = message.get("fn") + if function is None: + if kind == "control": + function = GSP_RM_CONTROL + elif kind == "alloc": + function = GSP_RM_ALLOC + if function is None: + raise HostTrafficError(f"message {index}: no wire function id") + hex_text = message.get("params_hex") or "" + try: + params = bytes.fromhex(hex_text) + except ValueError as exc: + raise HostTrafficError(f"message {index}: bad params_hex: {exc}") from exc + return int(function), params + + +def replay_stimulus(host: HostTraffic, stimulus: str | dict, *, + strict: bool = True) -> dict: + """Submit a stimulus document's messages in order, then read the replies.""" + doc = load_stimulus(stimulus) + messages = doc.get("messages") + if not isinstance(messages, list): + raise HostTrafficError("stimulus document has no 'messages' list") + pairs: list[dict] = [] + skipped: list[dict] = [] + for index, message in enumerate(messages): + if not isinstance(message, dict): + raise HostTrafficError(f"message {index} is not an object") + if message.get("truncated"): + reason = "truncated payload is not replayable as-is" + if strict: + raise HostTrafficError(f"message {index}: {reason}") + host.note_skip(index, reason) + skipped.append({"index": index, "reason": reason}) + continue + function, params = _message_to_rpc(message, index) + submit = host.submit_rpc(function, params) + dispatched = host.poll() + reply = host.read_reply() + if reply is None or dispatched != 1: + raise HostTrafficError( + f"message {index}: transport did not produce exactly one reply" + ) + pairs.append({ + "seq": message.get("seq", index), + "kind": message.get("kind"), + "function": function, + "submit": submit, + "reply": reply, + }) + return { + "format": REPLAY_FORMAT, + "source": doc.get("source"), + "submitted": len(pairs), + "skipped": skipped, + "pairs": pairs, + "transcript": host.transcript(), + "limits": list(_LIMITS), + } + + +# --------------------------------------------------------------------------- +# Demo +# --------------------------------------------------------------------------- + + +def _new_recorder(out_dir: str | None, name: str): + if out_dir is None: + return _NullRecorder(), False + from greeneye.record import Recorder # deferred: keep sim importable alone + + meta = {"kind": name, "host_traffic": True} + return Recorder(out_dir, name=name, meta=meta), True + + +def _json_reply(reply: dict | None) -> dict | None: + if reply is None: + return None + return { + "format": reply["format"], + "index": reply["index"], + "function": reply["function"], + "sequence": reply["sequence"], + "result": reply["result"], + "params_hex": reply["params_hex"], + "params_bytes": reply["params_bytes"], + "read": True, + } + + +def run_demo(out_dir: str | None = None) -> dict: + """Deterministic end-to-end transport demo; no capture, no firmware.""" + from .machine import Machine # deferred (module import stays light) + + machine = Machine(memory_mb=32) + machine.load_segments( + [(machine.ram_base, bytes.fromhex("73001000"))], entry=machine.ram_base + ) + recorder, owns = _new_recorder(out_dir, "host-traffic-demo") + host = HostTraffic(machine=machine, recorder=recorder) + replies: list[dict] = [] + for sequence, (function, params) in enumerate( + ( + (RPC_FN_ECHO, b"\x01\x02\x03\x04"), + (RPC_FN_STATUS_QUERY, b""), + ), + start=1, + ): + host.submit_rpc(function, params, sequence=sequence) + host.run_cycles(64) + reply = host.read_reply() + if reply is not None: + replies.append(_json_reply(reply)) + + control_cmd = 0x2080012B + control_params = ( + (0xC2000006).to_bytes(4, "little") + + (0xABCD0080).to_bytes(4, "little") + + control_cmd.to_bytes(4, "little") + + bytes(500) + ) + host.submit_rpc(GSP_RM_CONTROL, control_params, sequence=3) + host.run_cycles(64) + control_reply = host.read_reply() + if control_reply is not None: + replies.append(_json_reply(control_reply)) + + checks = { + "echo_reply_matches_payload": ( + replies[0]["function"] == RPC_FN_ECHO + and replies[0]["params_hex"] == b"\x01\x02\x03\x04".hex() + and replies[0]["result"] == NV_OK + ), + "status_reply_has_magic": ( + len(replies) > 1 + and bytes.fromhex(replies[1]["params_hex"])[:4] == STATUS_MAGIC + ), + "control_replay_is_512_bytes": host.state()["counts"]["submitted"] == 3, + "control_reply_is_a_framed_stub_verdict": ( + control_reply is not None + and control_reply["function"] == GSP_RM_CONTROL + and control_reply["result"] == NV_ERR_NOT_SUPPORTED + ), + "doorbell_ring_per_submit": host.doorbell.kicks == 3, + "all_replies_read": len(replies) == 3, + } + control = { + "function": GSP_RM_CONTROL, + "cmd": control_cmd, + "bytes": len(control_params), + "reply_function": control_reply["function"] if control_reply else None, + "reply_result": control_reply["result"] if control_reply else None, + "reply_params_bytes": control_reply["params_bytes"] if control_reply else None, + } + payload = { + "format": DEMO_FORMAT, + "ok": all(checks.values()), + "checks": checks, + "machine": { + "ram_base": machine.ram_base, + "insns": machine.insns, + "cycles": machine.cycles, + }, + "shared_base": host.shared_base, + "queue": host.queues.command.state(), + "control": control, + "replies": replies, + "transcript": host.transcript(), + "limits": list(_LIMITS), + } + host.close() + if owns: + os.makedirs(out_dir, exist_ok=True) + payload["artifacts"] = {"session": os.path.abspath(out_dir)} + with open(os.path.join(out_dir, "host-traffic-demo.json"), "w", + encoding="utf-8") as fh: + json.dump(payload, fh, indent=1, sort_keys=True) + fh.write("\n") + recorder.close() + return payload + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="sim.host_traffic", + description=( + "Modelled host traffic loop for the GSP message queues: doorbell, " + "command/status rings, stub dispatch and rm_capture stimulus " + "replay. Offline only; the simulator runs no real firmware here." + ), + ) + sub = parser.add_subparsers(dest="command") + demo_cmd = sub.add_parser("demo", help="deterministic end-to-end demo") + demo_cmd.add_argument("--out", help="session/output directory (optional)") + demo_cmd.add_argument("--json", action="store_true", help="print full JSON") + replay_cmd = sub.add_parser("replay", help="replay an rm_capture stimulus") + replay_cmd.add_argument("--stimulus", required=True, + help="stimulus JSON, capture JSON or raw capture log") + replay_cmd.add_argument("--out", help="session/output directory (optional)") + replay_cmd.add_argument("--allow-truncated", action="store_true", + help="skip truncated messages instead of refusing") + replay_cmd.add_argument("--json", action="store_true", help="print full JSON") + return parser + + +def _cli_summary(payload: dict) -> dict: + summary = { + "format": payload.get("format"), + "ok": payload.get("ok"), + } + if payload.get("checks") is not None: + summary["checks"] = payload["checks"] + if "submitted" in payload: + summary["submitted"] = payload["submitted"] + summary["skipped"] = len(payload.get("skipped", [])) + if payload.get("artifacts"): + summary["artifacts"] = payload["artifacts"] + return summary + + +def _jsonable(value): + if isinstance(value, bytes): + return value.hex() + if isinstance(value, (list, tuple)): + return [_jsonable(item) for item in value] + if isinstance(value, dict): + return {str(key): _jsonable(item) for key, item in value.items()} + return value + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if args.command is None: + parser.print_help(sys.stderr) + return 2 + try: + if args.command == "demo": + payload = run_demo(args.out) + else: + recorder, owns = _new_recorder(args.out, "host-traffic-replay") + host = HostTraffic(recorder=recorder) + payload = replay_stimulus( + host, args.stimulus, strict=not args.allow_truncated + ) + host.close() + if owns: + os.makedirs(args.out, exist_ok=True) + payload["artifacts"] = {"session": os.path.abspath(args.out)} + with open(os.path.join(args.out, "host-traffic-replay.json"), "w", + encoding="utf-8") as fh: + json.dump(_jsonable(payload), fh, indent=1, sort_keys=True) + fh.write("\n") + recorder.close() + except HostTrafficError as exc: + print(f"error: {exc}", file=sys.stderr) + return 2 + if args.json: + print(json.dumps(_jsonable(payload), indent=2, sort_keys=True)) + else: + print(json.dumps(_jsonable(_cli_summary(payload)), indent=2, sort_keys=True)) + if args.command == "demo" and not payload.get("ok"): + return 3 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/debug/risc-v-gsp-rm-vm/sim/libos_port.py b/debug/risc-v-gsp-rm-vm/sim/libos_port.py new file mode 100644 index 0000000..eaa3f95 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/sim/libos_port.py @@ -0,0 +1,386 @@ +"""libos port receive-path model (the B5 message-loop blocker, first half). + +Status: **offline development tool (new, additive)**. This module models the +part of the GSP firmware's libos *port receive* machinery that the B3 +message-loop task needs before it can make progress: the argument contract of +``TASK_PORT_SEND_RECV_WAIT`` (syscall 1), the asynchronous receive post, the +synchronous send/reply, the cross-payload handoff performed when a receive +completes, and the external-interrupt port retirement that a host doorbell wake +performs. + +Evidence kind: **firmware use + RTOS port implementation** (the same evidence +class as :mod:`sim.libos`). Every structure below was observed on the real +570.144 image, not inferred from a stub: + +* the message-loop task posts four asynchronous receives through a wrapper at + firmware vaddr ``0x4018000`` that moves its arguments ``a0-a3`` into + ``a4-a7``, zeroes ``a0-a3`` and ``t1``/``t2``, and executes ``ecall`` with + ``t0 = 1`` (an async receive post): + ``(shuttle 10, port 6, payload 0, size 0)`` immediately followed by + ``INTERRUPT_ARM_EXTERNAL(0xFF)`` -- the external-interrupt receive; + ``(shuttle 11, port 7, payload 0x5ba8e10, size 4)``; + ``(shuttle 9, port 8, payload 0x4015070, size 0x808)``; + ``(shuttle 14, port 13, payload 0x5ba8e20, size 0x24)``; +* it then performs one synchronous send on the default shuttle id (1) to port + 12 with a zero-length payload, ``waitId = 1``, ``timeout = -1`` (infinite); +* it blocks in ``PortWait(LIBOS_SHUTTLE_ANY, timeout = -1)`` (ecall at + ``0x4019df4``); +* on an external interrupt the port that was armed retires with + ``error_code = LIBOS_OK`` and ``payload_size = 0`` (the report path), and the + wait returns with ``a0 = status``, ``a4 = completedSize``, + ``a5 = completedRemoteTaskId``, ``a6 = completedShuttle``. + +The argument contract modelled (mirrors the RTOS ``libosPortSendRecvAndWait``): + +=============== ========================================================= +register meaning at ``ecall`` +=============== ========================================================= +``t0`` syscall number (1) +``a0`` send shuttle id (0 = none) +``a1`` send port id +``a2``/``a3`` send payload address / size +``a4`` receive shuttle id (0 = none) +``a5`` receive port id +``a6``/``a7`` receive payload address / capacity +``t1`` wait id (0 = do not block, ``0xFFFF`` = any) +``t2`` timeout in ns (``0xFFFFFFFFFFFFFFFF`` = infinite) +=============== ========================================================= + +On return: ``a0`` = status, ``a4`` = completed size, ``a5`` = originating task +id, ``a6`` = completed shuttle (local handle). + +Honesty rules (mandatory; docs/51 section 6, docs/06-ACCURACY.md): + +* the model is **data-driven by observation**: port roles come from the posted + receive sequence of the real image; nothing infers a port number that was + not observed; +* every modelled event carries ``"modelled": true`` and a label -- the host + doorbell wake is ``shim:host-doorbell-wake``, the boot handshake send is + ``modelled:peer-consumes-on-send``. These are not firmware behaviour; +* **receives are never completed with invented payloads**: a wake only retires + the external-interrupt receive with the RTOS report contract (rare, zero + length). No command, status or event payload is fabricated; +* wait semantics are strictly "return the first completed shuttle or stop"; + the model never returns from a blocking wait that nothing completed; +* L0/L1 only: no timing (L2) and no device-environment (L3) claim; with a + single modelled task there is no scheduler and no other task is run. + +Development-only, offline: no GPU, kernel module, mdev, guest, network, QEMU, +``/dev`` access or reboot. Nothing here is a product dependency and nothing +is injected into a guest. +""" +from __future__ import annotations + +from sim.libos import ( + INTERRUPT_ARM_EXTERNAL, + NAMES, + TASK_PORT_SEND_RECV_WAIT, +) + +#: Report schema tag for :meth:`LibosPortModel.state`. +REPORT_FORMAT = "greeneye-msgq-port/1" + +#: ``LIBOS_SHUTTLE_ANY`` (RTOS port API). +SHUTTLE_ANY = 0xFFFF +#: The task's default shuttle id (RTOS port API). +SHUTTLE_DEFAULT = 1 +#: ``LIBOS_TIMEOUT_INFINITE``. +TIMEOUT_INFINITE = 0xFFFFFFFFFFFFFFFF +#: LIBOS_OK. +LIBOS_OK = 0 + +#: Observed port roles for the real 570.144 message-loop task. The values are +#: the port ids carried by the task's posted receives (resource handles); the +#: labels describe behaviour only. +PORT_ROLES = { + 6: "extintr", # external-interrupt receive (armed with 0xFF) + 7: "notification", # 4-byte notification buffer + 8: "request", # 0x808-byte request/reply buffer + 12: "handshake", # zero-length boot handshake send target + 13: "event", # 36-byte event buffer +} + +#: Port label used when a port was not part of the observed post sequence. +ROLE_DATA = "data" + +#: The wake contract observed for the external-interrupt receive: the RTOS +#: report retires the waiting shuttle with error LIBOS_OK and a zero payload. +EXTINTR_WAKE_CONTRACT = "shim:host-doorbell-wake" +HANDSHAKE_CONTRACT = "modelled:peer-consumes-on-send" + + +def role_for(port: int) -> str: + """Role label for a port id (observed table; ``"data"`` otherwise).""" + return PORT_ROLES.get(int(port), ROLE_DATA) + + +class LibosPortModel: + """Deterministic model of the libos receive path for one GSP task. + + The model is a ``Machine.syscall_handler``: it owns syscall 1 + (``TASK_PORT_SEND_RECV_WAIT``) and the interrupt-arm call, and delegates + every other syscall to ``base`` (:class:`sim.libos.LibosSyscalls`, which + keeps recording). ``machine.x`` is the only machine state touched. + """ + + name = "libos-port-model" + + def __init__(self, base=None, *, task: str = "gsp-rm-task"): + self.base = base + self.task = task + #: Async receive posts in observation order. + self.posts: list[dict] = [] + #: Pending receives by shuttle id. + self.pending: dict[int, dict] = {} + #: Completed handoffs awaiting a wait (FIFO). + self.completed: list[dict] = [] + #: Synchronous sends observed (send payloads are not copied; the peer + #: side is not modelled). + self.sends: list[dict] = [] + #: Blocking waits observed. + self.waits: list[dict] = [] + #: Labelled model events (every one is explicitly ``modelled``). + self.events: list[dict] = [] + #: External interrupt mask last armed (RTOS ``IRQMSET`` semantics). + #: An installed base model may already have observed an arm before this + #: model was attached (the recorded boot arm), so seed from it. + self.armed_mask = int(getattr(base, "armed_mask", 0)) + #: Recorded host doorbell wakes. + self.host_kicks = 0 + #: Host wakes not yet handed to a receive (the firmware may post and + #: arm after the host has rung the doorbell; the pending interrupt is + #: delivered at the first armed receive). + self.kicks_pending = 0 + #: The request that stopped the run (unmodelled or blocking), if any. + self.stopped: dict | None = None + + # -- observation ------------------------------------------------------- + + def _event(self, name: str, **fields) -> dict: + event = {"event": name, "modelled": True, "task": self.task} + event.update(fields) + self.events.append(event) + return event + + def host_kick(self, *, label=EXTINTR_WAKE_CONTRACT, queue=None, value=None, + bit=None) -> dict: + """Record the host doorbell write; deliver it when a receive is armed. + + The host submit path writes ``NV_PGSP_QUEUE_HEAD(0)``; the mapping from + that write to a falcon interrupt bit is **not modelled** (unobserved). + The wake is delivered to the external-interrupt receive once it has + been posted and interrupts are armed (the firmware may do both after + the host has already rung). Returns the event; ``"delivered"`` is the + retired shuttle id, or ``None`` when the wake remains pending. + """ + self.host_kicks += 1 + self.kicks_pending += 1 + delivered = self._try_kick() + return self._event( + "host_kick", + contract=label, + queue=queue, + value=value, + falcon_irq_bit=bit, + delivered=delivered, + pending=bool(self.kicks_pending), + note=("queue-head write -> falcon interrupt bit is unobserved; " + "the wake is the documented shim"), + ) + + def _try_kick(self): + """Deliver one pending host wake to an armed extintr receive, if any.""" + if not self.kicks_pending or not self.armed_mask: + return None + for shuttle in sorted(self.pending): + post = self.pending[shuttle] + if post["role"] != "extintr": + continue + self.kicks_pending -= 1 + self._complete( + shuttle, + payload_size=0, + origin_task=0, + contract=EXTINTR_WAKE_CONTRACT, + reason="external-interrupt report (armed mask " + f"{self.armed_mask:#x}, pending host doorbell wake)", + ) + return shuttle + return None + + def _complete(self, shuttle: int, *, payload_size: int, origin_task: int, + contract: str, reason: str) -> dict: + post = self.pending.pop(int(shuttle)) + entry = { + "shuttle": int(shuttle), + "port": int(post["port"]), + "role": post["role"], + "payload": int(post["payload"]), + "capacity": int(post["capacity"]), + "payload_size": int(payload_size), + "origin_task": int(origin_task), + "contract": contract, + "reason": reason, + "post_pc": int(post["pc"]), + } + self.completed.append(entry) + self._event("receive_complete", **entry) + return entry + + # -- syscall surface --------------------------------------------------- + + def handle(self, machine, pc: int = 0): + """``Machine`` syscall hook: ``(handled, result)``.""" + num = int(machine.x[5]) + if num == TASK_PORT_SEND_RECV_WAIT: + return self._handle_port(machine, pc) + if num == INTERRUPT_ARM_EXTERNAL: + self.armed_mask |= int(machine.x[10]) + self._event("interrupt_arm_external", mask=int(machine.x[10]), pc=pc) + self._try_kick() + if self.base is not None: + handled, result = self.base.handle(machine, pc) + else: + handled, result = (True, LIBOS_OK) if num == INTERRUPT_ARM_EXTERNAL \ + else (False, None) + if not handled and self.stopped is None: + entry = getattr(self.base, "stopped", None) + number = int(entry["num"]) if entry else num + self.stopped = { + "num": number, + "name": (entry or {}).get("name", NAMES.get(number, f"unknown_{number}")), + "pc": int(pc), + "reason": "unmodelled-syscall", + } + return handled, result + + def _handle_port(self, machine, pc: int): + a0, a1, a2, a3 = (int(machine.x[10]), int(machine.x[11]), + int(machine.x[12]), int(machine.x[13])) + a4, a5, a6, a7 = (int(machine.x[14]), int(machine.x[15]), + int(machine.x[16]), int(machine.x[17])) + t1, t2 = int(machine.x[6]), int(machine.x[7]) + + send = a0 != 0 or a1 != 0 + recv = a4 != 0 or a5 != 0 + + if recv: + self._post(shuttle=a4, port=a5, payload=a6, capacity=a7, pc=pc, + wait_id=t1, timeout=t2) + if send: + self._send(shuttle=a0, port=a1, payload=a2, size=a3, wait_id=t1, + timeout=t2, pc=pc, machine=machine) + return True, LIBOS_OK + if recv: + # Async receive post: the kernel returns LIBOS_OK without touching + # the wait outputs (a4-a6 still hold the receive arguments). + return True, LIBOS_OK + + # Pure blocking wait: return the first completed shuttle, or stop. + if not self.completed: + self._try_kick() + if self.completed: + entry = self.completed.pop(0) + machine.x[14] = entry["payload_size"] & 0xFFFFFFFFFFFFFFFF + machine.x[15] = entry["origin_task"] & 0xFFFFFFFFFFFFFFFF + machine.x[16] = entry["shuttle"] & 0xFFFFFFFFFFFFFFFF + machine.x[17] = 0 + record = { + "pc": int(pc), + "wait_id": t1, + "timeout": t2, + "completed": entry["shuttle"], + "payload_size": entry["payload_size"], + "origin_task": entry["origin_task"], + "role": entry["role"], + "contract": entry["contract"], + } + self.waits.append(record) + self._event("wait_complete", **record) + return True, LIBOS_OK + + record = { + "pc": int(pc), + "wait_id": t1, + "timeout": t2, + "reason": "port-wait-blocks", + "pending": sorted(self.pending), + } + self.waits.append(record) + self._event("wait_block", **record) + if self.stopped is None: + self.stopped = { + "num": TASK_PORT_SEND_RECV_WAIT, + "name": "task_port_send_recv_wait", + "pc": int(pc), + "reason": "port-wait-blocks", + "pending": sorted(self.pending), + } + return False, None + + def _post(self, *, shuttle: int, port: int, payload: int, capacity: int, + pc: int, wait_id: int, timeout: int) -> dict: + post = { + "shuttle": int(shuttle), + "port": int(port), + "role": role_for(port), + "payload": int(payload), + "capacity": int(capacity), + "pc": int(pc), + "wait_id": int(wait_id), + "timeout": int(timeout), + } + self.posts.append(post) + self.pending[int(shuttle)] = post + self._event("receive_post", **post) + self._try_kick() + return post + + def _send(self, *, shuttle: int, port: int, payload: int, size: int, + wait_id: int, timeout: int, pc: int, machine) -> dict: + """Model a synchronous send/reply as completed by its peer. + + Only the boot handshake shape was observed (zero-length payload to port + 12 on the default shuttle). The peer is not modelled: consumption is + an explicit, labelled assumption. The wait outputs receive the RTOS + transcribe contract for a completed send (``a4`` = copied size, + ``a5`` = 0, ``a6`` = send shuttle). + """ + record = { + "shuttle": int(shuttle), + "port": int(port), + "role": role_for(port) if port else "reply", + "payload": int(payload), + "size": int(size), + "pc": int(pc), + "wait_id": int(wait_id), + "timeout": int(timeout), + "contract": HANDSHAKE_CONTRACT, + } + self.sends.append(record) + machine.x[14] = int(size) & 0xFFFFFFFFFFFFFFFF + machine.x[15] = 0 + machine.x[16] = int(shuttle) & 0xFFFFFFFFFFFFFFFF + machine.x[17] = 0 + self._event("send_complete", **record) + return record + + # -- report surface ---------------------------------------------------- + + def state(self) -> dict: + """Deterministic serialisable state for the ``greeneye-msgq-attach/1`` + report (no host timestamps).""" + return { + "format": REPORT_FORMAT, + "task": self.task, + "armed_mask": int(self.armed_mask), + "host_kicks": int(self.host_kicks), + "kicks_pending": int(self.kicks_pending), + "posts": [dict(post) for post in self.posts], + "pending": sorted(self.pending), + "completed": [dict(entry) for entry in self.completed], + "sends": [dict(record) for record in self.sends], + "waits": [dict(record) for record in self.waits], + "events": [dict(event) for event in self.events], + "stopped": dict(self.stopped) if self.stopped else None, + } diff --git a/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py b/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py new file mode 100644 index 0000000..f1530bd --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py @@ -0,0 +1,1122 @@ +"""Message-queue attach: real 570.144 GSP boot to B3 + host ring attach (B3/B4 seam). + +Status: **offline development tool (new, additive)**. This module implements +steps 1--3 of ``docs/51-HOST-TRAFFIC-LOOP.md`` section 7 ("the next step to +B5"): + +1. boot the real 570.144 image to B3 -- the message-loop port wait, the + ``task_port_send_recv_wait`` ecall at ``0x4018038`` -- reusing + :mod:`sim.gsp_boot`'s profile, boot shim and libos service model; +2. attach :class:`sim.host_traffic.DoorbellBlock` and the command/status rings + to the machine **before** the boot run, so the firmware's first queue-head + access is served by the model and recorded with its PC; +3. with ``--doorbell``, submit a real-shaped ``GSP_RM_CONTROL`` from the host + model, wake the port wait exactly once, continue in bounded instructions, + and classify the first ring access the firmware performs (address, width, + value read, and whether it looks like a queue head / Tx header / Rx header / + entry). By default the continuation runs under + :class:`sim.libos_port.LibosPortModel` (the ``--no-port-model`` flag keeps + the first wave's single-wake shim): the model reproduces the observed + async receive posts, the boot handshake send and the host wake's + external-interrupt handoff, and reports the exact unmodelled semantic that + stops the run (``report["missing_semantic"]``) instead of manufacturing a + poll. + +Schema: ``greeneye-msgq-attach/1`` (:func:`assemble_report`). + +Honesty rules (mandatory, docs/51 section 6, docs/06-ACCURACY.md): + +* the real image is executed; nothing here fakes a poll, a reply or a + completion. If the firmware does not reach the wait, faults, or never + touches the rings, the report says so and records the first divergence -- + a run that does not reach the loop is never presented as success; +* the port-wait wake is a documented shim, labelled in + ``report["continuation"]["wake_log"]``; the wake result is not firmware + behaviour and no RM semantics are claimed. Under the port model the wake + uses the RTOS external-interrupt report contract (error 0, zero payload); + without it the first-wave single-wake shim is used; +* the queue layout and arithmetic are the :mod:`sim.host_traffic` models + (real msgq arithmetic), but the queue base addresses are the fixture block; + the real addresses live in ``GSP_ARGUMENTS_CACHED``/queue-init arguments, + which the simulator does not parse yet; +* the dispatcher in :mod:`sim.host_traffic` is a stub and is not driven here: + the tool records what the *firmware* does, not a stub verdict; +* L0/L1 only: no timing (L2) and no device-environment (L3) claim; the cycle + count is the modeled :mod:`sim.cycles` accounting, not a measurement. + +Development-only, offline: no GPU, kernel module, mdev, guest, network, QEMU, +``/dev`` access or reboot. Nothing here is a product dependency and nothing +is injected into a guest. +""" +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import sys +from dataclasses import dataclass + +try: # prefer the harness's established image/profile default (docs/51 seam) + from harness.rehost import DEFAULT_IMAGE, DEFAULT_PROFILE +except Exception: # pragma: no cover - harness import is not always on path + from sim.gsp_boot import DEFAULT_IMAGE, DEFAULT_PROFILE + +from greeneye.device_map import firmware_sha256 +from greeneye.elf import load_elf +from greeneye.sim_backend import launch_native +from sim.decode import K_LOAD, LOAD_SIZE, decode16, decode32 +from sim.devices import load_profile, registry_from_profile +from sim.gsp_boot import ( + MAX_ATTEMPT_INSTRUCTIONS, + MILESTONES, + STOP_REASONS, + _apply_boot_shim, + _rung_from_run, +) +from sim.host_traffic import ( + DOORBELL_BASE, + DOORBELL_SIZE, + GSP_MSG_QUEUE_ELEMENT_HDR_SIZE, + GSP_RM_CONTROL, + MSGQ_RX_HDR_OFF, + MSGQ_TX_HEADER_BYTES, + HostTraffic, + parse_queue_element, +) +from sim.libos import LibosSyscalls, TASK_PORT_SEND_RECV_WAIT, TASK_STATE_QUERY +from sim.libos_port import LibosPortModel +from sim.machine import R_BP, R_BUDGET, R_ECALL, R_FAULT, R_STEP, R_WATCH + +#: Report schema tag. Every session/JSON artifact carries it. +REPORT_FORMAT = "greeneye-msgq-attach/1" + +#: ``run_cycles``-style cap: same ceiling as the boot runner's attempts. +MAX_RUN_CYCLES = MAX_ATTEMPT_INSTRUCTIONS +#: Default total instruction budget for one attach run. +DEFAULT_MAX_CYCLES = MAX_ATTEMPT_INSTRUCTIONS +#: Demo budget: boot-to-B3 needs ~474k instructions; leave a small tail. +DEMO_MAX_CYCLES = 600_000 +#: Bounded evidence: the first ring access plus up to 16 subsequent ones. +MAX_POLLS_RECORDED = 17 +#: Internal stop code: the bounded poll list was filled before the run ended. +R_POLL_LIMIT = 199 + +#: Doorbell label (block-relative register name; public Turing dev_gsp.h). +DOORBELL_WINDOW = "PRIV+0x110c00" + +#: The real-shaped control used for the host kick (same shape as the docs/51 +#: demo control; real-shaped, from no capture). +CONTROL_CMD = 0x2080012B + +#: ``msgqTxHeader`` field names by byte offset (public queue header). +_TX_FIELDS = { + 0: "version", + 4: "size", + 8: "msgSize", + 12: "msgCount", + 16: "writePtr", + 20: "flags", + 24: "rxHdrOff", + 28: "entryOff", +} + +_LIMITS = [ + "The real 570.144 image is executed; no poll, reply or completion is " + "faked. A run that does not reach the message-loop wait reports its first " + "divergence instead of success.", + "The port-wait wake is a documented shim (the libos message-port delivery " + "path is not modelled); wake events carry an explicit label.", + "The port model (sim.libos_port) reproduces the observed receive posts, " + "the boot handshake send and the external-interrupt handoff only; it " + "fabricates no message payload and schedules no second task. The exact " + "unmodelled semantic that stops the run is reported in missing_semantic.", + "Queue base addresses are the sim.host_traffic fixture block; the real " + "addresses live in GSP_ARGUMENTS_CACHED/queue-init arguments, which are " + "not parsed yet.", + "No RM semantics: the sim.host_traffic stub dispatcher is not driven by " + "this tool, so no reply is claimed.", + "L0/L1 only: no timing (L2) and no device-environment (L3) claim; cycle " + "numbers are modeled accounting (sim.cycles), not measurements.", +] + + +class MsgqAttachError(Exception): + """An attach input (image, profile or budget) that cannot be run.""" + + +def _as_int(value, default=0): + if value is None: + return default + if isinstance(value, str): + return int(value, 0) + return int(value) + + +def _sha256_text(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +# --------------------------------------------------------------------------- +# Ring geometry and the pure poll classifier +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class RingGeometry: + """Physical layout of the attached doorbell and command/status rings. + + All fields are plain integers so the classifier is testable without a + machine: ``classify_ring_access`` takes one of these plus an access. + """ + + shared_base: int + cmd_base: int + stat_base: int + cmd_size: int + stat_size: int + msg_size: int + entry_off: int + rx_hdr_off: int + doorbell_base: int + doorbell_size: int + + def to_dict(self) -> dict: + return { + "shared_base": self.shared_base, + "cmd_base": self.cmd_base, + "stat_base": self.stat_base, + "cmd_size": self.cmd_size, + "stat_size": self.stat_size, + "msg_size": self.msg_size, + "entry_off": self.entry_off, + "rx_hdr_off": self.rx_hdr_off, + "doorbell_base": self.doorbell_base, + "doorbell_size": self.doorbell_size, + } + + +def ring_geometry_from_host(host: HostTraffic) -> RingGeometry: + """Read the attached host's queue/doorbell layout into a pure geometry.""" + command = host.queues.command + header = command.header() + return RingGeometry( + shared_base=int(host.shared_base), + cmd_base=int(command.tx_base), + stat_base=int(host.queues.status.tx_base), + cmd_size=int(command.size), + stat_size=int(host.queues.status.size), + msg_size=int(header.msg_size), + entry_off=int(header.entry_off), + rx_hdr_off=int(MSGQ_RX_HDR_OFF), + doorbell_base=int(DOORBELL_BASE), + doorbell_size=int(DOORBELL_SIZE), + ) + + +def classify_ring_access(addr: int, width, value, geometry: RingGeometry) -> dict: + """Classify one firmware access against the ring geometry (pure). + + Returns ``{addr, width, value, in_ring, kind, ...}`` where ``kind`` is one + of ``queue-head`` (the doorbell register), ``tx-header``, ``rx-header``, + ``entry``, ``ring-gap`` or ``other``. The Rx-header note carries the real + ``MSGQ_FLAGS_SWAP_RX`` cross-link: the header physically in the command + buffer at ``+rx_hdr_off`` is the *status* queue's read pointer and vice + versa. + """ + addr = int(addr) + entry = { + "addr": addr, + "width": width, + "value": value, + "in_ring": False, + "kind": "other", + } + if geometry.doorbell_base <= addr < geometry.doorbell_base + geometry.doorbell_size: + offset = addr - geometry.doorbell_base + queue = offset // 8 if offset % 8 == 0 else None + entry.update({ + "in_ring": True, + "kind": "queue-head", + "window": DOORBELL_WINDOW, + "offset": offset, + }) + if queue is not None and 0 <= queue < geometry.doorbell_size // 8: + entry["queue"] = queue + entry["register"] = f"NV_PGSP_QUEUE_HEAD({queue})" + return entry + for buffer_name, base, size in ( + ("command", geometry.cmd_base, geometry.cmd_size), + ("status", geometry.stat_base, geometry.stat_size), + ): + if not base <= addr < base + size: + continue + offset = addr - base + entry.update({"in_ring": True, "buffer": buffer_name}) + if offset < MSGQ_TX_HEADER_BYTES: + entry.update({ + "kind": "tx-header", + "field_offset": offset, + "field": _TX_FIELDS.get(offset, "reserved"), + }) + elif offset == geometry.rx_hdr_off: + owner = "status" if buffer_name == "command" else "command" + entry.update({ + "kind": "rx-header", + "cross_linked_owner": owner, + "semantics": "readPtr", + }) + elif offset >= geometry.entry_off: + relative = offset - geometry.entry_off + slot = relative % geometry.msg_size + entry.update({ + "kind": "entry", + "entry_index": relative // geometry.msg_size, + "slot_offset": slot, + "region": "entry-header" if slot < GSP_MSG_QUEUE_ELEMENT_HDR_SIZE + else "entry-payload", + }) + else: + entry["kind"] = "ring-gap" + return entry + return entry + + +# --------------------------------------------------------------------------- +# Pure report assembly (hermetic-testable) +# --------------------------------------------------------------------------- + + +def _report_sha(report: dict) -> str: + clone = json.loads(json.dumps(report, sort_keys=True, default=str)) + clone.get("determinism", {}).pop("report_sha256", None) + return _sha256_text(json.dumps(clone, sort_keys=True)) + + +def assemble_report(*, image: dict, profile: dict, boot: dict, attach: dict, + doorbell: dict, polls: list, divergence, missing_mmio: dict, + determinism: dict, continuation: dict | None = None, + port_model: dict | None = None, missing_semantic: dict | None = None, + limits=None, notes=None) -> dict: + """Assemble the ``greeneye-msgq-attach/1`` report from plain inputs. + + Pure: no machine, no boot, no I/O. ``polls`` is the ordered list of + classified ring accesses (first = index 0); the report keeps the first and + at most 16 subsequent ones. ``boot["message_loop_wait_reached"]`` and + ``boot.get("skipped")`` select the outcome; a not-reached run keeps its + non-empty divergence reason, never a manufactured success. + + ``port_model`` is the :class:`sim.libos_port.LibosPortModel` state when the + receive path was modelled (``{"enabled": False}`` otherwise), and + ``missing_semantic`` is the exact unmodelled semantic that stopped a + port-model continuation (``None`` when there is none). Both are additive + fields, present on every report. + """ + ordered = [dict(poll) for poll in polls] + first = ordered[0] if ordered else None + subsequent = ordered[1:1 + (MAX_POLLS_RECORDED - 1)] + if boot.get("skipped"): + outcome = "skipped" + elif not boot.get("message_loop_wait_reached"): + outcome = "not-reached" + elif first is not None: + outcome = "wait-reached-poll-classified" + else: + outcome = "wait-reached-no-poll" + ok = outcome in ("wait-reached-poll-classified", "wait-reached-no-poll") + report = { + "format": REPORT_FORMAT, + "ok": ok, + "outcome": outcome, + "image": dict(image), + "profile": dict(profile), + "boot": dict(boot), + "attach": dict(attach), + "doorbell": dict(doorbell), + "continuation": dict(continuation or {}), + "first_poll": first, + "subsequent_polls": subsequent, + "poll_count": len(ordered), + "divergence": divergence, + "missing_mmio": dict(missing_mmio), + "port_model": dict(port_model) if port_model is not None + else {"enabled": False}, + "missing_semantic": dict(missing_semantic) if missing_semantic is not None + else None, + "determinism": dict(determinism), + "limits": list(limits if limits is not None else _LIMITS), + "notes": list(notes or []), + } + report["determinism"]["report_sha256"] = _report_sha(report) + return report + + +def _skipped_report(image: str, profile_path: str, reason: str, + *, doorbell: bool) -> dict: + return assemble_report( + image={"path": os.path.abspath(image), "sha256": None, "size": None}, + profile={"name": os.path.basename(profile_path), + "path": os.path.abspath(profile_path)}, + boot={"ran": False, "skipped": True, "reason": reason, + "message_loop_wait_reached": False, "rung": None, + "instructions_executed": 0}, + attach={}, + doorbell={"requested": bool(doorbell)}, + polls=[], + divergence=None, + missing_mmio={"count": 0, "first": None}, + determinism={}, + ) + + +# --------------------------------------------------------------------------- +# Execution helpers (machine stepping stays in sim.machine) +# --------------------------------------------------------------------------- + + +class _PortWaitWake: + """Host-traffic-gated wake of ``task_port_send_recv_wait`` (documented shim). + + The real libos kernel unblocks the task when a message is delivered through + the port machinery; the simulator does not model that path, so the only + modelled wake is explicit: after the host has rung the doorbell, the first + wait returns LIBOS_OK (0) exactly once and the event is recorded. A + re-entry is recorded and stops the run -- the wait is never silently + repeated as if progress had been made. + """ + + def __init__(self, base_handler, *, max_wakes: int = 1): + self.base_handler = base_handler + self.max_wakes = int(max_wakes) + self.wakes = 0 + self.wake_log: list[dict] = [] + self.reentries: list[dict] = [] + + def handle(self, machine, pc: int = 0): + number = int(machine.x[5]) + if number == TASK_PORT_SEND_RECV_WAIT: + record = { + "num": number, + "name": "task_port_send_recv_wait", + "pc": int(pc), + "a4": int(machine.x[14]), + "a5": int(machine.x[15]), + "a6": int(machine.x[16]), + "a7": int(machine.x[17]), + } + if self.wakes < self.max_wakes: + self.wakes += 1 + record["result"] = "shim:port_wait_wake" + record["returned"] = 0 + self.wake_log.append(record) + return True, 0 + record["result"] = "stop:reentry" + self.reentries.append(record) + return False, None + return self.base_handler.handle(machine, pc) + + def state(self) -> dict: + return { + "wakes": self.wakes, + "wake_log": [dict(entry) for entry in self.wake_log], + "reentries": [dict(entry) for entry in self.reentries], + } + + +def _watchers(geometry: RingGeometry): + """Read-watchpoints over the doorbell and both ring buffers (mask = loads).""" + return ( + (geometry.doorbell_base, geometry.doorbell_base + geometry.doorbell_size, + "doorbell", 2), + (geometry.cmd_base, geometry.cmd_base + geometry.cmd_size, "rings", 2), + (geometry.stat_base, geometry.stat_base + geometry.stat_size, "rings", 2), + ) + + +def _decoded_load(machine, pc: int): + """``(size, rd)`` of the cached load at ``pc``, or ``(None, None)``.""" + decoded = getattr(machine, "_cache", {}).get(pc) + if decoded is None: + try: + fetch = pc + if machine.mpu.active: + translated = machine.mpu.translate(pc) + if translated is not None: + fetch = translated + word = int.from_bytes(machine.read_mem(fetch, 4), "little") + if word & 3 == 3: + decoded = decode32(pc, word) + else: + decoded = decode16(pc, word & 0xFFFF) + except Exception: + return None, None + if decoded[0] != K_LOAD: + return None, None + return int(LOAD_SIZE[decoded[1]]), int(decoded[2]) + + +def _capture_poll(machine, geometry: RingGeometry, pc: int, addr: int) -> dict: + """Record and step over the watched load (a watch fires before execution). + + The load's width/rd come from the decoded instruction; the value is read + from the destination register after the single step, so the recorded value + is the value the firmware actually received. + """ + size, rd = _decoded_load(machine, pc) + step_reason, _spc, fault_cause, fault_tval = machine.step() + value = None + if rd is not None and step_reason != R_FAULT and rd != 0: + value = machine.x[rd] + poll = classify_ring_access(addr, size, value, geometry) + poll["pc"] = int(pc) + poll["executed"] = step_reason == R_STEP + if step_reason == R_FAULT: + poll["step_fault"] = {"cause": int(fault_cause), "tval": int(fault_tval)} + return poll + + +def _run_watching(machine, budget: int, geometry: RingGeometry, + milestones_hit: list, polls: list, + *, max_polls: int = MAX_POLLS_RECORDED): + """Run to ``budget`` (absolute instructions), recording milestone hits and + ring reads; same loop shape as ``sim.gsp_boot._run_with_milestones`` plus + watchpoint step-over. Returns ``(reason, pc, a, b)``. + """ + pending = {addr for name, addr in MILESTONES if name not in milestones_hit} + while True: + remaining = budget - machine.insns + if remaining <= 0: + return R_BUDGET, machine.pc, 0, 0 + watchers = _watchers(geometry) if len(polls) < max_polls else () + reason, pc, a, b = machine.run(remaining, breakpoints=pending, + watchers=watchers) + if reason == R_BP: + for name, addr in MILESTONES: + if addr == pc and name not in milestones_hit: + milestones_hit.append(name) + break + pending = {addr for name, addr in MILESTONES + if name not in milestones_hit} + continue + if reason == R_WATCH: + polls.append(_capture_poll(machine, geometry, pc, a)) + if len(polls) >= max_polls: + return R_POLL_LIMIT, machine.pc, a, b + continue + return reason, pc, a, b + + +def _control_params() -> bytes: + """A real-shaped 512-byte ``GSP_RM_CONTROL`` parameter block (docs/51).""" + return ( + (0xC2000006).to_bytes(4, "little") + + (0xABCD0080).to_bytes(4, "little") + + CONTROL_CMD.to_bytes(4, "little") + + bytes(500) + ) + + +def _verify_command_element(host: HostTraffic, submit: dict) -> dict: + """Parse the just-submitted command element back and verify its checksum.""" + try: + raw = host.queues.command.read_entry(submit["index"], + submit["element_bytes"]) + prefix, message = parse_queue_element(raw) + return { + "checksum_ok": True, + "elem_count": int(prefix["elem_count"]), + "function": int(message.function), + "sequence": int(message.sequence), + "length": int(message.length), + } + except Exception as exc: # pragma: no cover - defensive evidence + return {"checksum_ok": False, "error": str(exc)} + + +def missing_semantic_after_handoff(*, stopped: dict, polls: list) -> dict: + """The exact unmodelled semantics that stopped the port-model continuation. + + Recorded when, after the recorded host wake, the message-loop task + completed the external-interrupt receive, dispatched it, and issued + ``TASK_STATE_QUERY`` (the libos trap/report path) without touching the + command/status rings. Pure: takes the stop record and the classified poll + list; no machine, no I/O. + """ + return { + "stage": "after-extintr-handoff", + "stop": dict(stopped), + "observed": [ + "host doorbell write NV_PGSP_QUEUE_HEAD(0) recorded", + "external-interrupt receive retired with the RTOS report contract " + "(error 0, payload size 0) on the recorded wake", + "completed shuttle dispatched; next action is a task-state query " + "for the reported remote task", + "no command/status ring or doorbell read occurred before the stop", + ], + "missing": [ + "doorbell-write -> falcon interrupt bit mapping: the host submit " + "path writes only NV_PGSP_QUEUE_HEAD; no IRQ register write is " + "observed and the firmware armed mask 0xFF", + "libos interrupt-trap delivery and the interrupted task's saved " + "xcause/xepc consumed by TASK_STATE_QUERY", + "the command-queue consumer (msgq receive) task and the " + "GSP_ARGUMENTS_CACHED/queue-init address handoff; this single-task " + "model schedules no other task", + ], + "next_step": ( + "model TASK_STATE_QUERY's trap state or establish the queue-service " + "task's wake path, then re-run this attach and classify the first " + "ring poll; do not infer a poll from this run"), + "polls": len(polls), + } + + +# --------------------------------------------------------------------------- +# The attach run +# --------------------------------------------------------------------------- + + +def run_attach(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFILE, + max_cycles: int = DEFAULT_MAX_CYCLES, doorbell: bool = False, + port_model: bool = True, json_out: str | None = None, + record_dir: str | None = None) -> dict: + """Boot to the port wait, attach the rings, optionally kick and continue. + + Bounded and deterministic for a fixed image/profile/budget. Returns the + assembled report; on an absent image/profile it returns a clean + ``outcome: "skipped"`` report instead of raising. Never fakes a poll: the + outcome is the observed one, with the first divergence recorded when the + wait is not reached. + + ``port_model`` (default on) installs + :class:`sim.libos_port.LibosPortModel` for the continuation: the async + receive posts, the boot handshake send and the host wake's + external-interrupt handoff are modelled, so the run advances to the + task-state-query the interrupt dispatch performs (or further). With the + model off, the documented single-wake shim of the first wave is used and + the run reports ``port-wait-reentry`` instead. + """ + if isinstance(max_cycles, bool) or not isinstance(max_cycles, int) \ + or not 0 < max_cycles <= MAX_RUN_CYCLES: + raise MsgqAttachError( + f"max_cycles must be an int in 1..{MAX_RUN_CYCLES}, got {max_cycles!r}") + image = os.path.abspath(image) + profile_path = os.path.abspath(profile_path) + if not os.path.isfile(image): + return _skipped_report(image, profile_path, f"image absent: {image}", + doorbell=doorbell) + if not os.path.isfile(profile_path): + return _skipped_report(image, profile_path, + f"profile absent: {profile_path}", + doorbell=doorbell) + + sha = firmware_sha256(image) + size = os.path.getsize(image) + profile = load_profile(profile_path) + load = profile.get("load", {}) + ram_base = _as_int(load.get("ram_base", 0)) + memory_mb = _as_int(load.get("memory_mb", 12288)) + + session = None + notes: list[str] = [] + divergence = None + polls: list[dict] = [] + continuation: dict = {"ran": False} + port_model_state: dict | None = None + missing_semantic: dict | None = None + try: + session = launch_native( + kernel=image, + memory_mb=memory_mb, + ram_base=ram_base, + record=record_dir if record_dir else False, + name="msgq-attach", + device_map=None, + unknown_mmio="lenient", + bias=0, + ) + machine = session.target.machine + machine.devices = registry_from_profile(profile, machine=machine) + machine.translator = getattr(machine.devices, "dma", None) + libos = LibosSyscalls() + machine.syscall_handler = libos.handle + shim_meta = {"enabled": True, "assumptions": []} + shim = dict(profile.get("boot_shim", {})) + shim["_image_path"] = image + _apply_boot_shim(machine, shim, load_elf(image), shim_meta) + + host = HostTraffic(machine=machine, recorder=session.recorder) + machine.devices.add(host.doorbell_block()) + geometry = ring_geometry_from_host(host) + + submit = None + element_check = None + if doorbell: + submit = host.submit_rpc(GSP_RM_CONTROL, _control_params(), + sequence=1) + element_check = _verify_command_element(host, submit) + + milestones_hit: list[str] = [] + reason, pc, a, b = _run_watching(machine, max_cycles, geometry, + milestones_hit, polls) + stopped = dict(libos.stopped) if libos.stopped else None + wait_reached = bool(stopped) and stopped.get("num") == TASK_PORT_SEND_RECV_WAIT + stop_reason = STOP_REASONS.get(reason, str(reason)) + if reason == R_POLL_LIMIT: + stop_reason = "poll-limit" + fault = {"cause": int(a), "tval": int(b), "pc": int(pc)} \ + if reason == R_FAULT else None + rung = _rung_from_run(machine, shim_meta, milestones_hit, libos) + + if not wait_reached: + if reason == R_FAULT: + divergence = { + "kind": "fault", + "reason": (f"fault before the message-loop port wait: " + f"cause={a:#x} tval={b:#x} pc={pc:#x}"), + } + elif reason == R_POLL_LIMIT: + divergence = { + "kind": "poll-limit", + "reason": (f"{MAX_POLLS_RECORDED} ring accesses observed " + f"before the port wait; run stopped by the " + f"bounded poll limit"), + } + elif reason == R_ECALL: + name = (stopped or {}).get("name", "unhandled ecall") + number = (stopped or {}).get("num") + divergence = { + "kind": "ecall", + "reason": (f"unhandled libos syscall before the " + f"message-loop port wait: {name} " + f"(num={number}) at pc={pc:#x}"), + } + else: + divergence = { + "kind": "budget", + "reason": (f"message-loop port wait not reached within " + f"{max_cycles} instructions (rung {rung}, " + f"stop={stop_reason} at pc={pc:#x})"), + } + + if wait_reached and doorbell and port_model: + model = LibosPortModel(base=libos) + machine.syscall_handler = model.handle + last_kick = host.doorbell.writes[-1] if host.doorbell.writes else None + model.host_kick( + queue=int(last_kick["queue"]) if last_kick else None, + value=int(last_kick["value"]) if last_kick else None, + ) + reason2, pc2, a2, b2 = _run_watching(machine, max_cycles, geometry, + milestones_hit, polls) + stop_reason2 = STOP_REASONS.get(reason2, str(reason2)) + if reason2 == R_POLL_LIMIT: + stop_reason2 = "poll-limit" + port_model_state = {"enabled": True, **model.state()} + continuation = { + "ran": True, + "port_model": True, + "stop_reason": stop_reason2, + "stop_pc": int(pc2), + "instructions_executed": int(machine.insns), + "wakes": int(model.host_kicks), + "wake_log": [ + { + "result": entry["contract"], + "returned": 0, + "delivered": entry.get("delivered"), + "pending": entry.get("pending"), + "queue": entry.get("queue"), + "value": entry.get("value"), + } + for entry in model.events + if entry["event"] == "host_kick" + ], + "reentries": [], + "polls": len(polls), + } + if reason2 == R_FAULT: + divergence = { + "kind": "fault", + "reason": (f"fault after the host wake: cause={a2:#x} " + f"tval={b2:#x} pc={pc2:#x}"), + } + elif reason2 == R_ECALL: + port_stopped = model.stopped or ( + dict(libos.stopped) if libos.stopped else None) + number2 = int(port_stopped["num"]) if port_stopped \ + else int(machine.x[5]) + name2 = (port_stopped or {}).get("name", "unhandled ecall") + if number2 == TASK_STATE_QUERY: + missing_semantic = missing_semantic_after_handoff( + stopped=port_stopped or {"num": number2, "pc": int(pc2)}, + polls=polls) + divergence = { + "kind": "task-state-query", + "reason": ( + "after the recorded host wake the external " + "interrupt receive completed (error 0, size 0) and " + "the message-loop task dispatched it; its next " + "syscall is task_state_query " + f"(num={number2}) at pc={pc2:#x}, the libos " + "trap/report path -- no command/status ring or " + "doorbell read occurred, so no poll is reported"), + "stop": dict(port_stopped or {}), + } + elif number2 == TASK_PORT_SEND_RECV_WAIT: + divergence = { + "kind": "port-wait-reentry", + "reason": ( + "the message-loop task re-entered " + "task_port_send_recv_wait after the recorded host " + "wake; nothing completed the wait, so the firmware " + "produced no queue poll"), + "wait_pc": int(pc2), + } + else: + divergence = { + "kind": "ecall", + "reason": (f"unhandled libos syscall after the host " + f"wake: {name2} (num={number2}) at " + f"pc={pc2:#x}"), + } + elif reason2 == R_BUDGET and not polls: + divergence = { + "kind": "budget", + "reason": (f"no ring access observed within the total " + f"budget of {max_cycles} instructions"), + } + elif reason2 == R_POLL_LIMIT: + notes.append( + f"continuation stopped after {MAX_POLLS_RECORDED} recorded " + "ring accesses (bounded by design)") + elif wait_reached and doorbell: + wake = _PortWaitWake(libos, max_wakes=1) + machine.syscall_handler = wake.handle + reason2, pc2, a2, b2 = _run_watching(machine, max_cycles, geometry, + milestones_hit, polls) + stop_reason2 = STOP_REASONS.get(reason2, str(reason2)) + if reason2 == R_POLL_LIMIT: + stop_reason2 = "poll-limit" + continuation = { + "ran": True, + "port_model": False, + "stop_reason": stop_reason2, + "stop_pc": int(pc2), + "instructions_executed": int(machine.insns), + "wakes": int(wake.wakes), + "wake_log": [dict(entry) for entry in wake.wake_log], + "reentries": [dict(entry) for entry in wake.reentries], + "polls": len(polls), + } + if reason2 == R_FAULT: + divergence = { + "kind": "fault", + "reason": (f"fault after the host wake: cause={a2:#x} " + f"tval={b2:#x} pc={pc2:#x}"), + } + elif reason2 == R_ECALL: + number2 = int(machine.x[5]) + if number2 == TASK_PORT_SEND_RECV_WAIT: + divergence = { + "kind": "port-wait-reentry", + "reason": ( + "the message-loop task re-entered " + "task_port_send_recv_wait after the recorded host " + "wake; libos message-port delivery is not modelled, " + "so the firmware produced no queue poll"), + "wait_pc": int(pc2), + } + else: + divergence = { + "kind": "ecall", + "reason": (f"unhandled libos syscall after the host " + f"wake: num={number2} at pc={pc2:#x}"), + } + elif reason2 == R_BUDGET and not polls: + divergence = { + "kind": "budget", + "reason": (f"no ring access observed within the total " + f"budget of {max_cycles} instructions"), + } + elif reason2 == R_POLL_LIMIT: + notes.append( + f"continuation stopped after {MAX_POLLS_RECORDED} recorded " + "ring accesses (bounded by design)") + elif wait_reached and not doorbell: + notes.append( + "continuation not requested (--doorbell off); the port wait was " + "reached but no host kick was sent and no poll was classified") + + ring_reads = [ + access for access in machine.devices.accesses + if access["rw"] == "r" + and geometry.doorbell_base <= access["addr"] + < geometry.doorbell_base + geometry.doorbell_size + ] + ring_writes = [ + access for access in machine.devices.accesses + if access["rw"] == "w" + and geometry.doorbell_base <= access["addr"] + < geometry.doorbell_base + geometry.doorbell_size + ] + doorbell_write = None + if submit is not None and host.doorbell.writes: + last = host.doorbell.writes[-1] + doorbell_write = { + "window": DOORBELL_WINDOW, + "base": int(DOORBELL_BASE), + "offset": int(last["queue"]) * 8, + "queue": int(last["queue"]), + "value": int(last["value"]), + } + doorbell_report = { + "requested": bool(doorbell), + "write": doorbell_write, + "host_kicks": int(host.doorbell.kicks), + "head_read_back": int(host.doorbell.head(0)), + "guest_reads": len(ring_reads), + "guest_writes": len(ring_writes), + } + + attach = dict(geometry.to_dict()) + attach["command_queue"] = host.queues.command.state() + attach["status_queue"] = host.queues.status.state() + attach["command_element"] = element_check + + device_state = json.dumps(machine.devices.state(), sort_keys=True, + default=str) + queue_state = json.dumps(host.queues.state(), sort_keys=True) + doorbell_state = json.dumps(host.doorbell.state(), sort_keys=True) + determinism = { + "instructions": int(machine.insns), + "cycles_modeled": int(machine.cycles), + "cycle_model": machine.cycle_model.name + if getattr(machine, "cycle_model", None) else None, + "max_cycles": int(max_cycles), + "ram_base": int(machine.ram_base), + "ram_size": int(machine.ram_size), + "device_state_sha256": _sha256_text(device_state), + "access_log_len": len(machine.devices.accesses), + "queue_state_sha256": _sha256_text(queue_state), + "doorbell_state_sha256": _sha256_text(doorbell_state), + } + + boot = { + "ran": True, + "skipped": False, + "rung": rung, + "message_loop_wait_reached": wait_reached, + "instructions_executed": int(machine.insns), + "cycles_modeled": int(machine.cycles), + "cycle_model": determinism["cycle_model"], + "instructions_budget": int(max_cycles), + "stop": {"reason": stop_reason, "pc": int(pc), "fault": fault}, + "wait": stopped if wait_reached else None, + "libos_stopped": stopped, + "milestones": list(milestones_hit), + "mmio_unknown": int(machine.mmio_unknown), + "device_unhandled": int(machine.devices.unhandled), + } + missing_mmio = { + "count": int(machine.mmio_unknown), + "first": dict(machine.mmio_log[0]) if machine.mmio_log else None, + } + report = assemble_report( + image={"path": image, "sha256": sha, "size": size}, + profile={"name": profile.get("name"), "path": profile_path}, + boot=boot, + attach=attach, + doorbell=doorbell_report, + polls=polls, + divergence=divergence, + missing_mmio=missing_mmio, + determinism=determinism, + continuation=continuation, + port_model=port_model_state, + missing_semantic=missing_semantic, + notes=notes, + ) + if getattr(session.recorder, "dir", None) is not None: + session.recorder.meta.update({ + "kind": "msgq-attach", + "outcome": report["outcome"], + "rung": rung, + "instructions_executed": int(machine.insns), + "message_loop_wait_reached": wait_reached, + "port_model": bool(report["port_model"].get("enabled")), + "missing_semantic": bool(report["missing_semantic"]), + "first_poll": None if report["first_poll"] is None + else report["first_poll"]["kind"], + }) + session.recorder.blob( + "msgq-attach.json", + json.dumps(report, indent=1, sort_keys=True).encode(), + ) + session.note( + "msgq_attach", + outcome=report["outcome"], + rung=rung, + instructions=machine.insns, + wait_reached=wait_reached, + polls=report["poll_count"], + ) + if json_out: + parent = os.path.dirname(os.path.abspath(json_out)) + os.makedirs(parent, exist_ok=True) + with open(json_out, "w", encoding="utf-8") as handle: + json.dump(report, handle, indent=1, sort_keys=True) + handle.write("\n") + finally: + if session is not None: + session.close() + return report + + +# --------------------------------------------------------------------------- +# Demo and CLI +# --------------------------------------------------------------------------- + + +def run_demo(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFILE, + print_json: bool = False, port_model: bool = True) -> dict: + """Bounded attach demo; cleanly skips (exit 0) when the image is absent.""" + report = run_attach(image=image, profile_path=profile_path, + max_cycles=DEMO_MAX_CYCLES, doorbell=True, + port_model=port_model) + if report["outcome"] == "skipped": + print(f"[msgq-attach] skipped: {report['boot'].get('reason')}") + return report + if print_json: + print(json.dumps(report, indent=1, sort_keys=True)) + else: + _print_report(report) + return report + + +def _print_report(report: dict) -> None: + image = report["image"] + boot = report["boot"] + print(f"format: {report['format']}") + print(f"image: {image['path']}") + print(f"sha256: {image['sha256']} size: {image['size']}") + print(f"outcome: {report['outcome']} (ok={report['ok']})") + print(f"rung: {boot.get('rung')} instructions: {boot.get('instructions_executed')}" + f" cycles (modeled): {boot.get('cycles_modeled')}") + wait = boot.get("wait") + if wait: + print(f"message-loop wait: reached ({wait.get('name')} @ " + f"{wait.get('pc'):#x})") + else: + print("message-loop wait: not reached") + write = report["doorbell"].get("write") + if write: + print(f"doorbell: host write {write['window']} offset={write['offset']:#x} " + f"value={write['value']:#x} (guest reads={report['doorbell']['guest_reads']})") + model = report.get("port_model") or {} + if model.get("enabled"): + posts = model.get("posts", []) + roles = ", ".join(f"port {post['port']}:{post['role']}" + for post in posts) + print(f"port model: enabled posts=[{roles}] " + f"host kicks={model.get('host_kicks', 0)}") + for wait_record in model.get("waits", []): + if wait_record.get("completed") is not None: + print(f" wait @ {wait_record['pc']:#x}: completed shuttle " + f"{wait_record['completed']} ({wait_record['role']}) " + f"size={wait_record['payload_size']} " + f"contract={wait_record['contract']}") + else: + print("port model: disabled") + first = report["first_poll"] + if first: + print(f"first poll: addr={first['addr']:#x} width={first['width']} " + f"value={first['value']} kind={first['kind']}") + else: + print("first poll: none observed") + print(f"polls recorded: {report['poll_count']}" + f" (subsequent listed: {len(report['subsequent_polls'])})") + if report["divergence"]: + print(f"divergence: {report['divergence']['reason']}") + missing = report.get("missing_semantic") + if missing: + print(f"missing semantic ({missing.get('stage')}):") + for item in missing.get("missing", []): + print(f" - {item}") + print("note: development-only, offline; L0/L1 evidence, no timing claim") + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="sim.msgq_attach", + description=( + "Boot the real 570.144 GSP image to the message-loop port wait " + "(B3), attach the host doorbell/msgq rings before the run, " + "optionally kick, continue bounded and classify the first ring " + "poll. Offline only; no hardware, module, guest or network."), + ) + sub = parser.add_subparsers(dest="command") + run_cmd = sub.add_parser( + "run", help="bounded boot + attach + optional kick + poll classify") + run_cmd.add_argument("--image", default=DEFAULT_IMAGE, + help="GSP firmware image (default: the 570.144 image)") + run_cmd.add_argument("--profile", default=DEFAULT_PROFILE, + help="greeneye-sim-profile/1 document") + run_cmd.add_argument("--max-cycles", type=int, default=DEFAULT_MAX_CYCLES, + help=f"total instruction budget (1..{MAX_RUN_CYCLES})") + run_cmd.add_argument("--doorbell", action="store_true", + help="submit a host command and wake the port wait once") + run_cmd.add_argument("--no-port-model", action="store_true", + help="disable the sim.libos_port receive-path model " + "(use the first-wave single-wake shim)") + run_cmd.add_argument("--json", metavar="OUT", default=None, + help="write the full report JSON to OUT") + run_cmd.add_argument("--record", metavar="DIR", default=None, + help="write a greeneye-session/1 directory to DIR") + demo_cmd = sub.add_parser( + "demo", help="bounded canned attach run (skips cleanly if image absent)") + demo_cmd.add_argument("--image", default=DEFAULT_IMAGE, + help="GSP firmware image (default: the 570.144 image)") + demo_cmd.add_argument("--profile", default=DEFAULT_PROFILE, + help="greeneye-sim-profile/1 document") + demo_cmd.add_argument("--no-port-model", action="store_true", + help="disable the sim.libos_port receive-path model") + demo_cmd.add_argument("--json", action="store_true", + help="print the full report JSON") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + if args.command is None: + parser.print_help(sys.stderr) + return 2 + port_model = not getattr(args, "no_port_model", False) + if args.command == "demo": + run_demo(image=args.image, profile_path=args.profile, + print_json=args.json, port_model=port_model) + return 0 + try: + report = run_attach( + image=args.image, + profile_path=args.profile, + max_cycles=args.max_cycles, + doorbell=args.doorbell, + port_model=port_model, + json_out=args.json, + record_dir=args.record, + ) + except MsgqAttachError as exc: + print(f"[msgq-attach] error: {exc}", file=sys.stderr) + return 1 + if report["outcome"] == "skipped": + print(f"[msgq-attach] skipped: {report['boot'].get('reason')}", + file=sys.stderr) + return 2 + _print_report(report) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) |