"""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 = " 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())