1 files changed, 390 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/sim_backend.py b/debug/risc-v-gsp-rm-vm/greeneye/sim_backend.py
new file mode 100644
index 0000000..d17076f
--- /dev/null
+++ b/debug/risc-v-gsp-rm-vm/greeneye/sim_backend.py
@@ -0,0 +1,390 @@
+"""Native green-eye execution backend (M1): deterministic RV64IMAC in Python.
+
+``ge.launch(kernel=..., backend="native")`` builds a :class:`SimTarget` around
+:class:`sim.machine.Machine` and exposes a :class:`SimClient` whose method
+surface matches :class:`greeneye.rsp.RspClient`, so ``greeneye.session.Session``
+runs unchanged against it.  ``trace=`` enables the M1.5 exec/access trace
+(:mod:`sim.trace`), written to the session as ``blobs/trace.jsonl``.
+``device_map=`` and ``unknown_mmio="lenient"`` (H2.1/H2.2 start) serve
+unmodelled MMIO accesses from a harvested first-touch map instead of faulting;
+lenient accesses are logged to ``target.mmio_log`` and counted in
+``target.mmio_unknown``.
+
+Virtual time is instruction count: ``continue_(timeout)`` executes at most
+``timeout * SIM_IPS`` instructions (SIM_IPS = 1 MIPS), deterministically.  The
+wall-clock budgets ``Session.run`` derives from its deadline therefore bound
+the run; the stop/watch/hook sequence within a fixed instruction budget is
+reproducible on every host.
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+from .elf import load_elf
+from .errors import GreenEyeError, RspProtocolError, RspTimeout
+from .rsp import Register, StopReply
+from .session import Session
+from .symbols import SymbolDB
+
+_TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if _TOOL_ROOT not in sys.path:  # `sim/` lives beside the greeneye package
+    sys.path.insert(0, _TOOL_ROOT)
+
+from sim.machine import (  # noqa: E402
+    R_BUDGET,
+    R_FAULT,
+    R_WATCH,
+    Machine,
+)
+from sim.trace import make_trace  # noqa: E402
+#: Virtual instructions per second used to convert wall timeouts to budgets.
+SIM_IPS = 1_000_000
+#: Safety valve for ``continue_(timeout=None)``: never run unbounded.
+MAX_UNBOUNDED_BUDGET = 20_000_000
+
+#: Canonical register names, as exposed by QEMU's target.xml for riscv:rv64.
+ABI_NAMES = (
+    ["zero", "ra", "sp", "gp", "tp", "t0", "t1", "t2", "s0", "s1"]
+    + [f"a{i}" for i in range(8)]
+    + [f"s{i}" for i in range(2, 12)]
+    + [f"t{i}" for i in range(3, 7)]
+)
+#: xN -> ABI-name aliases (x0 == zero, x1 == ra, ...).
+_X_TO_ABI = {f"x{i}": ABI_NAMES[i] for i in range(32)}
+_X_TO_ABI["fp"] = "s0"
+_X_TO_ABI["return"] = "ra"
+_ABI_TO_INDEX = {name: i for i, name in enumerate(ABI_NAMES)}
+
+
+class SimClient:
+    """Drop-in replacement for :class:`RspClient` over a native Machine."""
+
+    def __init__(self, machine: Machine, *, sim_ips: int = SIM_IPS, trace=None):
+        self.machine = machine
+        self.sim_ips = max(1, int(sim_ips))
+        self.timeout = 10.0
+        self.packet_size = 0x1000
+        self.trace = trace  # sim.trace.Trace | None
+        self._recorder = None
+        self.registers = [
+            Register(number=i, name=name, bitsize=64, offset=i * 8)
+            for i, name in enumerate(ABI_NAMES)
+        ] + [Register(number=32, name="pc", bitsize=64, offset=32 * 8)]
+        self._by_name = {r.name: r for r in self.registers}
+        self._by_name.update(_X_TO_ABI)
+        self._by_number = {r.number: r for r in self.registers}
+        self.features = "PacketSize=1000;swbreak+;hwbreak+;greeneye-native+"
+        self.connected = True
+        self._bps: set[int] = set()
+        self._hw_bps: set[int] = set()
+        self._watches: dict[tuple[int, int, str], None] = {}
+
+    # -- lifecycle ---------------------------------------------------------
+
+    def connect(self, retries: int = 1, retry_delay: float = 0.1) -> None:
+        self.connected = True
+
+    def sync(self) -> None:
+        pass  # no server to hand-shake with
+
+    def close(self) -> None:
+        self.connected = False
+        if self.trace is not None:
+            self.trace.flush(self._recorder)
+
+    def __enter__(self) -> "SimClient":
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self.close()
+
+    # -- registers ---------------------------------------------------------
+
+    def read_registers(self) -> dict[str, int]:
+        x = self.machine.x
+        out = {name: x[i] for i, name in enumerate(ABI_NAMES)}
+        out["fp"] = out["s0"]  # QEMU's target.xml names x8 "fp"
+        out["pc"] = self.machine.pc
+        return out
+
+    def _index(self, name_or_number: str | int) -> int:
+        if isinstance(name_or_number, int):
+            if 0 <= name_or_number < 32:
+                return name_or_number
+            if name_or_number == 32:
+                return 32
+            raise RspProtocolError(f"unknown register number {name_or_number}")
+        key = _X_TO_ABI.get(name_or_number, name_or_number)
+        if key == "pc":
+            return 32
+        index = _ABI_TO_INDEX.get(key)
+        if index is None:
+            raise RspProtocolError(f"unknown register {name_or_number!r}")
+        return index
+
+    def read_reg(self, name_or_number: str | int) -> int:
+        index = self._index(name_or_number)
+        if index == 32:
+            return self.machine.pc
+        return self.machine.reg(index)
+
+    def write_reg(self, name_or_number: str | int, value: int) -> None:
+        index = self._index(name_or_number)
+        if index == 32:
+            self.machine.pc = value & ((1 << 64) - 1)
+        else:
+            self.machine.set_reg(index, value)
+
+    @property
+    def pc(self) -> int:
+        return self.machine.pc
+
+    # -- memory ------------------------------------------------------------
+
+    def read_mem(self, addr: int, size: int, chunk: int = 0) -> bytes:
+        try:
+            return self.machine.read_mem(addr, size)
+        except Exception as exc:
+            raise RspProtocolError(str(exc)) from exc
+
+    def write_mem(self, addr: int, data: bytes, chunk: int = 0) -> None:
+        try:
+            self.machine.write_mem(addr, data)
+        except Exception as exc:
+            raise RspProtocolError(str(exc)) from exc
+
+    def read_cstring(self, addr: int, max_len: int = 256) -> str:
+        raw = self.read_mem(addr, max_len)
+        return raw.split(b"\x00", 1)[0].decode("utf-8", "replace")
+
+    # -- breakpoints / watchpoints ----------------------------------------
+
+    def set_sw_break(self, addr: int, kind: int = 4) -> None:
+        self._bps.add(addr)
+
+    def clear_sw_break(self, addr: int, kind: int = 4) -> None:
+        self._bps.discard(addr)
+
+    def set_hw_break(self, addr: int, kind: int = 4) -> None:
+        self._hw_bps.add(addr)
+
+    def clear_hw_break(self, addr: int, kind: int = 4) -> None:
+        self._hw_bps.discard(addr)
+
+    _WATCH_KINDS = {"w": 2, "r": 3, "a": 4}
+    _WATCH_REVERSE = {"w": "watch", "r": "rwatch", "a": "awatch"}
+
+    def set_watch(self, addr: int, size: int = 8, kind: str = "w") -> None:
+        if kind not in self._WATCH_KINDS:
+            raise RspProtocolError(f"unknown watchpoint kind {kind!r}")
+        self._watches[(addr, size, kind)] = None
+
+    def clear_watch(self, addr: int, size: int = 8, kind: str = "w") -> None:
+        self._watches.pop((addr, size, kind), None)
+
+    def _watcher_list(self):
+        out = []
+        for (addr, size, kind) in self._watches:
+            mask = (2 if kind in ("r", "a") else 0) | (1 if kind in ("w", "a") else 0)
+            out.append((addr, addr + size, kind, mask))
+        return tuple(out)
+
+    # -- run control -------------------------------------------------------
+
+    def _budget(self, timeout: float | None) -> int | None:
+        if timeout is None:
+            return None
+        return max(1, int(timeout * self.sim_ips))
+
+    def _to_stop(self, result) -> StopReply:
+        reason, pc, a, b = result
+        if reason == R_WATCH:
+            kind = self._WATCH_REVERSE.get(b, "watch")
+            return StopReply(
+                raw=f"T05watch:{a:x};",
+                signal=5,
+                watch_addr=a,
+                watch_kind=kind,
+            )
+        if reason == R_FAULT:
+            signal = 4 if a == 2 else 11
+            return StopReply(raw=f"T0{signal:x}", signal=signal)
+        # bp, ebreak, ecall and step all report SIGTRAP, like the gdbstub.
+        return StopReply(raw="T05", signal=5)
+
+    def continue_(self, timeout: float | None = None) -> StopReply:
+        budget = self._budget(timeout)
+        if budget is None:
+            budget = MAX_UNBOUNDED_BUDGET
+        result = self.machine.run(
+            budget,
+            breakpoints=frozenset(self._bps | self._hw_bps),
+            watchers=self._watcher_list(),
+            trace=self.trace,
+        )
+        if result[0] == R_BUDGET:
+            raise RspTimeout(f"native budget exhausted after {budget} instructions")
+        return self._to_stop(result)
+
+    def step(self, timeout: float | None = None) -> StopReply:
+        result = self.machine.step(watchers=self._watcher_list(), trace=self.trace)
+        return self._to_stop(result)
+
+    def interrupt(self, timeout: float | None = None) -> StopReply:
+        return StopReply(raw="T02", signal=2)
+
+    # -- monitor -----------------------------------------------------------
+
+    def monitor(self, cmd: str) -> str:
+        return ""
+
+
+class SimTarget:
+    """QemuTarget-shaped handle for a native Machine (no process, no socket)."""
+
+    def __init__(
+        self,
+        machine: Machine,
+        *,
+        memory_mb: int = 256,
+        ram_base: int = 0x80000000,
+        sim_ips: int = SIM_IPS,
+        keep_alive: bool = False,
+        trace=None,
+    ):
+        self.machine = machine
+        self.memory_mb = memory_mb
+        self.ram_base = ram_base
+        self.keep_alive = keep_alive
+        self.client = SimClient(machine, sim_ips=sim_ips, trace=trace)
+        self.kernel: str | None = None
+        self.workdir: str | None = None
+        self._alive = True
+
+    # QemuTarget lifecycle surface ----------------------------------------
+
+    def start(self, paused: bool = True, timeout: float = 0.0) -> "SimTarget":
+        return self
+
+    def wait_for_port(self, timeout: float = 0.0) -> None:
+        return None
+
+    def alive(self) -> bool:
+        return self._alive
+
+    def kill(self, timeout: float = 0.0) -> None:
+        self._alive = False
+        self.client.close()
+
+    def read_uart(self) -> str:
+        return self.machine.uart.text()
+
+    @property
+    def mmio_log(self) -> list[dict]:
+        """Lenient-mode log (empty unless ``unknown_mmio="lenient"``)."""
+        return self.machine.mmio_log
+
+    @property
+    def mmio_unknown(self) -> int:
+        """Count of unmodelled MMIO accesses served leniently."""
+        return self.machine.mmio_unknown
+
+    def wait_for_uart(self, marker: str, timeout: float = 5.0, interval: float = 0.05) -> bool:
+        # Execution is synchronous: by the time the caller looks, the marker
+        # either exists or the budget was too small.  No polling needed.
+        return marker in self.read_uart()
+
+    def read_qemu_log(self) -> str:
+        return ""
+
+    @property
+    def command_line(self) -> list[str]:
+        return [f"native-sim:{self.kernel or '<none>'}"]
+
+    def shell_command(self) -> str:
+        return " ".join(self.command_line)
+
+    def __repr__(self) -> str:  # pragma: no cover - cosmetic
+        return f"SimTarget(pc={self.machine.pc:#x}, alive={self._alive})"
+
+
+def launch_native(
+    *,
+    kernel: str | None = None,
+    memory_mb: int = 256,
+    record=None,
+    symbols: SymbolDB | None = None,
+    name: str = "session",
+    load_symbols: bool = True,
+    keep_alive: bool = False,
+    sim_ips: int = SIM_IPS,
+    trace=None,
+    device_map=None,
+    unknown_mmio: str = "fault",
+    learn_mmio: bool = False,
+    bias: int = 0,
+) -> Session:
+    """Boot the native simulator and return a ready Session.
+
+    ``trace`` enables the R4 execution/access trace: ``True`` for defaults, a
+    dict of ``{"exec": bool, "start": int, "end": int, "max_events": int}``,
+    or a pre-built :class:`sim.trace.Trace`.  The trace is written to the
+    session as ``blobs/trace.jsonl`` when the Session closes; tracing implies
+    recording, so a session directory is created when none was given.
+
+    ``device_map`` is a :class:`greeneye.device_map.DeviceMap` (or a path to a
+    saved map) describing modelled MMIO; ``unknown_mmio="lenient"`` serves
+    unmodelled accesses from it (or zero) instead of faulting, logging each in
+    ``machine.mmio_log`` and counting ``machine.mmio_unknown``.
+    ``learn_mmio=True`` writes the observed values back into the map.
+    ``bias`` relocates PT_LOAD segments and the entry for images linked for
+    GPU physical addresses (e.g. ``0x80000000`` for the real GSP fwimage).
+    """
+    try:
+        tracer = make_trace(trace)
+    except ValueError as exc:
+        raise GreenEyeError(f"invalid trace spec: {exc}") from exc
+    if tracer is not None and not record:
+        record = True  # the trace artifact lives in the session
+    if isinstance(device_map, (str, os.PathLike)):
+        from .device_map import DeviceMap
+
+        device_map = DeviceMap.load(os.fspath(device_map))
+    machine = Machine(
+        memory_mb=memory_mb,
+        device_map=device_map,
+        unknown_mmio=unknown_mmio,
+        learn_mmio=learn_mmio,
+        bias=bias,
+    )
+    target = SimTarget(
+        machine,
+        memory_mb=memory_mb,
+        sim_ips=sim_ips,
+        keep_alive=keep_alive,
+        trace=tracer,
+    )
+    target.kernel = kernel
+    symdb = symbols or SymbolDB()
+    meta = {"backend": "native", "sim_ips": sim_ips, "firmware": kernel}
+    if unknown_mmio != "fault" or device_map is not None:
+        meta["unknown_mmio"] = unknown_mmio
+    session = Session(
+        client=target.client,
+        target=target,
+        symbols=symdb,
+        record=record,
+        name=name,
+        meta=meta,
+    )
+    if tracer is not None:
+        target.client._recorder = session.recorder
+        session.recorder.meta["trace"] = {**tracer.config()}
+    if kernel:
+        image = load_elf(kernel)
+        if not image.is_riscv:
+            raise GreenEyeError(f"{kernel}: not a RISC-V ELF image")
+        machine.load_segments(image.flat_load_plan(bias=bias), image.relocated_entry(bias=bias))
+        session.load_image(image, bias=bias, load_symbols=load_symbols)
+    return session