path: root/debug/risc-v-gsp-rm-vm/sim/gmmu_walk.py

#!/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())