"""Deterministic RV64IMAC machine: RAM, NS16550A UART, debugger-visible state. The execution loop is a monolithic dispatch over the cached decode tuples from ``sim.decode``. Everything guest-visible is a pure function of the loaded image and the instruction budget handed to ``run``: no host clock, no threads, no entropy. RAM is an anonymous ``mmap`` (lazily paged, so a 256 MiB region is cheap); all other addresses resolve to the UART or fault. """ from __future__ import annotations import mmap import struct from .decode import ( A_ADD, A_AND, A_DIV, A_DIVU, A_MUL, A_MULH, A_MULHSU, A_MULHU, A_OR, A_REM, A_REMU, A_SLL, A_SLT, A_SLTU, A_SRA, A_SRL, A_SUB, A_XOR, I_ADDI, I_ADDIW, I_ANDI, I_ORI, I_SLLI, I_SLLIW, I_SLTI, I_SLTIU, I_SRAI, I_SRAIW, I_SRLI, I_SRLIW, I_XORI, K_AMO, K_AUIPC, K_BRANCH, K_CSR, K_EBREAK, K_ECALL, K_FENCE, K_ILLEGAL, K_JAL, K_JALR, K_LOAD, K_LUI, K_MRET, K_OP, K_OP32, K_OPI, K_OPI32, K_STORE, LOAD_SIZE, MASK64, STORE_SIZE, decode16, decode32, ) # Stop reasons returned by Machine.run(). R_BUDGET = 0 R_BP = 1 R_WATCH = 2 R_EBREAK = 3 R_FAULT = 4 R_STEP = 5 R_ECALL = 6 #: Accepted ``unknown_mmio`` policies for accesses outside RAM/UART/the map. UNKNOWN_MMIO_MODES = ("fault", "lenient") UART_BASE = 0x10000000 UART_SIZE = 0x100 # RV64I M A C advertised in misa (MXL=2, I=bit8, M=bit12, A=bit0, C=bit2). MISA_RV64IMAC = 0x8000000000001105 class Fault(Exception): """Guest-visible trap: unmapped access or illegal instruction.""" def __init__(self, cause: int, tval: int = 0): super().__init__(f"cause={cause:#x} tval={tval:#x}") self.cause = cause self.tval = tval class Uart16550: """Behavioural NS16550A: TX is captured, LSR always reports THR empty.""" def __init__(self): self.tx = bytearray() self.ier = 0 self.lcr = 0 self.mcr = 0 self.scr = 0 self.dll = 0 self.dlm = 0 def read(self, off: int, size: int = 1) -> int: if off == 0: return self.dll # RBR (no RX modelled) / DLL when DLAB=1 if off == 1: return self.ier if off == 2: return 0x01 # IIR: no pending interrupt if off == 3: return self.lcr if off == 4: return self.mcr if off == 5: return 0x60 # LSR: THRE | TEMT if off == 6: return 0xB0 # MSR: CTS|DSR|DCD if off == 7: return self.scr return 0 def write(self, off: int, value: int, size: int = 1) -> None: value &= 0xFF if off == 0: if self.lcr & 0x80: self.dll = value else: self.tx.append(value) elif off == 1: self.ier = value elif off == 3: self.lcr = value elif off == 4: self.mcr = value elif off == 7: self.scr = value # FCR (2) and reserved offsets are write-only/no-ops. def text(self) -> str: return self.tx.decode("utf-8", "replace") def _signed64(value: int) -> int: return value - 0x10000000000000000 if value & 0x8000000000000000 else value def _signed32(value: int) -> int: value &= 0xFFFFFFFF return value - 0x100000000 if value & 0x80000000 else value def _sext32(value: int) -> int: value &= 0xFFFFFFFF return (value ^ 0x80000000) - 0x80000000 def _watch_match(watchers, ea: int, size: int, rw_bit: int): end = ea + size for wa, we, wk, mask in watchers: if (mask & rw_bit) and ea < we and wa < end: return wk return None class Machine: """One deterministic RV64IMAC hart plus RAM and UART. ``device_map`` is a :class:`greeneye.device_map.DeviceMap` (or any object with the same ``lookup``/``record`` surface) describing modelled MMIO. ``unknown_mmio`` selects what happens to an access that is neither RAM, UART nor in the map: ``"fault"`` (default, M1 parity) stops as before; ``"lenient"`` returns the mapped behaviour or zero, records the access in :attr:`mmio_log`, counts it in :attr:`mmio_unknown`, and, with ``learn_mmio=True``, writes the observation back into the map. Nothing is logged or counted when lenient mode is off, and debugger ``read_mem``/ ``write_mem`` stay strict. """ def __init__( self, memory_mb: int = 256, ram_base: int = 0x80000000, device_map=None, unknown_mmio: str = "fault", learn_mmio: bool = False, bias: int = 0, ): if unknown_mmio not in UNKNOWN_MMIO_MODES: raise ValueError(f"unknown_mmio must be one of {UNKNOWN_MMIO_MODES}, got {unknown_mmio!r}") self.ram_base = ram_base self.ram_size = memory_mb * 1024 * 1024 self.ram = mmap.mmap(-1, self.ram_size) self.uart = Uart16550() self.device_map = device_map self.unknown_mmio = unknown_mmio self.learn_mmio = bool(learn_mmio) self.bias = bias # PC relocation for learned-map firmware addresses self.mmio_log: list[dict] = [] self.mmio_unknown = 0 self.x = [0] * 32 self.pc = ram_base self.insns = 0 self.csrs: dict[int, int] = {0x301: MISA_RV64IMAC} self._cache: dict[int, tuple] = {} self._u16 = struct.Struct(" None: """Place ``(addr, bytes)`` PT_LOAD results and set the PC.""" for addr, data in plan: offset = addr - self.ram_base if offset < 0 or offset + len(data) > self.ram_size: raise Fault(1, addr) self.ram[offset : offset + len(data)] = data self.pc = entry & MASK64 self.invalidate() def invalidate(self) -> None: """Drop the decode cache after debugger-visible code changes.""" self._cache.clear() # -- register access (debugger surface) -------------------------------- def reg(self, index: int) -> int: if index == 0: return 0 return self.x[index] def set_reg(self, index: int, value: int) -> None: if index != 0: self.x[index] = value & MASK64 # -- raw memory access (debugger surface) ------------------------------ def read_mem(self, addr: int, size: int) -> bytes: out = bytearray() while size > 0: if self.ram_base <= addr < self.ram_base + self.ram_size: take = min(size, self.ram_base + self.ram_size - addr) off = addr - self.ram_base out += self.ram[off : off + take] addr += take size -= take elif UART_BASE <= addr < UART_BASE + UART_SIZE: out.append(self.uart.read(addr - UART_BASE)) addr += 1 size -= 1 else: raise Fault(5, addr) return bytes(out) def write_mem(self, addr: int, data: bytes) -> None: self.invalidate() offset = 0 while offset < len(data): here = addr + offset remaining = len(data) - offset if self.ram_base <= here < self.ram_base + self.ram_size: take = min(remaining, self.ram_base + self.ram_size - here) ro = here - self.ram_base self.ram[ro : ro + take] = data[offset : offset + take] offset += take elif UART_BASE <= here < UART_BASE + UART_SIZE: self.uart.write(here - UART_BASE, data[offset]) offset += 1 else: raise Fault(7, here) # -- MMIO (guest side) ------------------------------------------------- def _mapped_mmio(self, addr: int, size: int, write: bool) -> tuple[bool, int]: """Consult the device map; ``(handled, value)`` for a modelled access.""" dm = self.device_map if dm is None: return False, 0 spec = dm.lookup(addr, size, write) if spec == "unimplemented": return False, 0 if spec.startswith("value:"): return True, int(spec[6:], 16) return True, 0 # "zero": mapped but no read value observed def _log_unknown(self, pc: int, addr: int, size: int, rw: str, value: int = 0) -> None: """Lenient fallback: log, count, and optionally learn one access.""" self.mmio_unknown += 1 self.mmio_log.append({"pc": pc, "addr": addr, "size": size, "rw": rw}) if self.learn_mmio: if self.device_map is None: from greeneye.device_map import DeviceMap # lazy: sim stays usable alone self.device_map = DeviceMap(meta={"source": "learn"}) self.device_map.record(addr, size, rw, value, pc=pc, bias=self.bias, source="trace") def _mmio_load(self, addr: int, size: int, pc: int = 0) -> int: if UART_BASE <= addr < UART_BASE + UART_SIZE: return self.uart.read(addr - UART_BASE) handled, value = self._mapped_mmio(addr, size, False) if handled: return value if self.unknown_mmio == "lenient": self._log_unknown(pc, addr, size, "r") return 0 raise Fault(5, addr) def _mmio_store(self, addr: int, size: int, value: int, pc: int = 0) -> None: if UART_BASE <= addr < UART_BASE + UART_SIZE: self.uart.write(addr - UART_BASE, value) return handled, _ = self._mapped_mmio(addr, size, True) if handled: if self.learn_mmio and self.device_map is not None: self.device_map.record(addr, size, "w", value, pc=pc, bias=self.bias, source="trace") return if self.unknown_mmio == "lenient": self._log_unknown(pc, addr, size, "w", value) return raise Fault(7, addr) # -- execution --------------------------------------------------------- def step(self, watchers=(), trace=None): return self.run(1, breakpoints=(), watchers=watchers, single_step=True, trace=trace) def run(self, budget: int, breakpoints=(), watchers=(), single_step: bool = False, trace=None): """Execute up to ``budget`` instructions. Returns ``(reason, pc, a, b)``; ``a``/``b`` carry the watched address and kind (R_WATCH) or the trap cause and tval (R_FAULT). When ``trace`` is a :class:`sim.trace.Trace`, every executed instruction and every completed memory access in its address range is recorded. """ x = self.x ram = self.ram base = self.ram_base limit = self.ram_size cache = self._cache i8 = self._i8.unpack_from i16 = self._i16.unpack_from i32 = self._i32.unpack_from u16 = self._u16.unpack_from u32 = self._u32.unpack_from u64 = self._u64.unpack_from u16p = self._u16p u32p = self._u32p u64p = self._u64p has_bps = bool(breakpoints) pc = self.pc n = 0 reason = R_BUDGET a = b = 0 traced = trace is not None active = False while n < budget: if has_bps and pc in breakpoints: reason = R_BP a = pc break off = pc - base if off < 0 or off + 4 > limit: reason = R_FAULT a = 1 b = pc break hw = ram[off] | (ram[off + 1] << 8) if hw & 3 == 3: d = cache.get(pc) if d is None: d = decode32(pc, hw | (ram[off + 2] << 16) | (ram[off + 3] << 24)) cache[pc] = d else: d = cache.get(pc) if d is None: d = decode16(pc, hw) cache[pc] = d kind, sub, rd, r1, r2, imm, nxt = d n += 1 if traced: insn_pc = pc active = trace.begin(insn_pc, self.insns + n) # -- loads / stores (most frequent in firmware hot loops) ------ if kind == K_LOAD: ea = (x[r1] + imm) & MASK64 size = LOAD_SIZE[sub] if watchers: hit = _watch_match(watchers, ea, size, 2) if hit is not None: reason = R_WATCH a = ea b = hit break o = ea - base if 0 <= o <= limit - size: if active: trace.mem(insn_pc, ea, size, False, self.insns + n) if sub == 3: if rd: x[rd] = u64(ram, o)[0] elif sub == 2: if rd: x[rd] = i32(ram, o)[0] elif sub == 4: if rd: x[rd] = ram[o] elif sub == 0: if rd: x[rd] = i8(ram, o)[0] elif sub == 5: if rd: x[rd] = u16(ram, o)[0] elif sub == 6: if rd: x[rd] = u32(ram, o)[0] else: if rd: x[rd] = i16(ram, o)[0] pc = nxt else: try: v = self._mmio_load(ea, size, pc) except Fault as f: reason = R_FAULT a = f.cause b = f.tval break if rd: x[rd] = v if active: trace.mem(insn_pc, ea, size, False, self.insns + n) pc = nxt continue if kind == K_STORE: ea = (x[r1] + imm) & MASK64 size = STORE_SIZE[sub] if watchers: hit = _watch_match(watchers, ea, size, 1) if hit is not None: reason = R_WATCH a = ea b = hit break o = ea - base value = x[r2] if 0 <= o <= limit - size: if active: trace.mem(insn_pc, ea, size, True, self.insns + n) if sub == 3: u64p(ram, o, value) elif sub == 2: u32p(ram, o, value & 0xFFFFFFFF) elif sub == 1: u16p(ram, o, value & 0xFFFF) else: ram[o] = value & 0xFF pc = nxt else: try: self._mmio_store(ea, size, value, pc) except Fault as f: reason = R_FAULT a = f.cause b = f.tval break if active: trace.mem(insn_pc, ea, size, True, self.insns + n) pc = nxt continue # -- control flow ---------------------------------------------- if kind == K_BRANCH: va = x[r1] vb = x[r2] if sub == 0: take = va == vb elif sub == 1: take = va != vb elif sub == 4: take = _signed64(va) < _signed64(vb) elif sub == 5: take = _signed64(va) >= _signed64(vb) elif sub == 6: take = va < vb else: take = va >= vb pc = imm if take else nxt elif kind == K_JAL: if rd: x[rd] = nxt pc = imm elif kind == K_JALR: target = (x[r1] + imm) & MASK64 target &= ~1 if rd: x[rd] = nxt pc = target elif kind == K_AUIPC: if rd: x[rd] = (pc + imm) & MASK64 pc = nxt elif kind == K_LUI: if rd: x[rd] = imm pc = nxt # -- integer ALU ----------------------------------------------- elif kind == K_OPI: va = x[r1] if r1 else 0 if sub == I_ADDI: rv = (va + imm) & MASK64 elif sub == I_SLLI: rv = (va << (imm & 63)) & MASK64 elif sub == I_SLTI: rv = 1 if _signed64(va) < imm else 0 elif sub == I_SLTIU: rv = 1 if va < (imm & MASK64) else 0 elif sub == I_XORI: rv = va ^ (imm & MASK64) elif sub == I_ORI: rv = va | (imm & MASK64) elif sub == I_ANDI: rv = va & (imm & MASK64) elif sub == I_SRLI: rv = va >> (imm & 63) else: # I_SRAI rv = (_signed64(va) >> (imm & 63)) & MASK64 if rd: x[rd] = rv pc = nxt elif kind == K_OP: va = x[r1] vb = x[r2] if sub == A_ADD: rv = (va + vb) & MASK64 elif sub == A_SLL: rv = (va << (vb & 63)) & MASK64 elif sub == A_SLT: rv = 1 if _signed64(va) < _signed64(vb) else 0 elif sub == A_SLTU: rv = 1 if va < vb else 0 elif sub == A_XOR: rv = va ^ vb elif sub == A_SRL: rv = va >> (vb & 63) elif sub == A_OR: rv = va | vb elif sub == A_AND: rv = va & vb elif sub == A_SUB: rv = (va - vb) & MASK64 elif sub == A_SRA: rv = (_signed64(va) >> (vb & 63)) & MASK64 elif sub == A_MUL: rv = (va * vb) & MASK64 elif sub == A_MULH: rv = ((_signed64(va) * _signed64(vb)) >> 64) & MASK64 elif sub == A_MULHSU: rv = ((_signed64(va) * vb) >> 64) & MASK64 elif sub == A_MULHU: rv = ((va * vb) >> 64) & MASK64 elif sub == A_DIV: sa, sb = _signed64(va), _signed64(vb) if sb == 0: rv = MASK64 elif sa == -(1 << 63) and sb == -1: rv = 1 << 63 else: rv = (abs(sa) // abs(sb)) & MASK64 if (sa < 0) != (sb < 0): rv = (-rv) & MASK64 elif sub == A_DIVU: rv = MASK64 if vb == 0 else va // vb elif sub == A_REM: sa, sb = _signed64(va), _signed64(vb) if sb == 0 or (sa == -(1 << 63) and sb == -1): rv = 0 else: rv = (abs(sa) % abs(sb)) & MASK64 if sa < 0: rv = (-rv) & MASK64 else: # A_REMU rv = va if vb == 0 else va % vb if rd: x[rd] = rv pc = nxt elif kind == K_OPI32: va = x[r1] if r1 else 0 if sub == I_ADDIW: rv = (va + imm) & 0xFFFFFFFF elif sub == I_SLLIW: rv = (va << (imm & 31)) & 0xFFFFFFFF elif sub == I_SRLIW: rv = (va & 0xFFFFFFFF) >> (imm & 31) else: # I_SRAIW rv = (_signed32(va) >> (imm & 31)) & 0xFFFFFFFF if rd: x[rd] = _sext32(rv) & MASK64 pc = nxt elif kind == K_OP32: va = x[r1] vb = x[r2] if sub == A_ADD: rv = (va + vb) & 0xFFFFFFFF elif sub == A_SUB: rv = (va - vb) & 0xFFFFFFFF elif sub == A_SLL: rv = (va << (vb & 31)) & 0xFFFFFFFF elif sub == A_SRL: rv = (va & 0xFFFFFFFF) >> (vb & 31) elif sub == A_SRA: rv = (_signed32(va) >> (vb & 31)) & 0xFFFFFFFF elif sub == A_MUL: rv = (va * vb) & 0xFFFFFFFF elif sub == A_DIV: sa, sb = _signed32(va), _signed32(vb) if sb == 0: rv = 0xFFFFFFFF elif sa == -(1 << 31) and sb == -1: rv = 1 << 31 else: rv = (abs(sa) // abs(sb)) & 0xFFFFFFFF if (sa < 0) != (sb < 0): rv = (-rv) & 0xFFFFFFFF elif sub == A_DIVU: b32 = vb & 0xFFFFFFFF rv = 0xFFFFFFFF if b32 == 0 else (va & 0xFFFFFFFF) // b32 elif sub == A_REM: sa, sb = _signed32(va), _signed32(vb) if sb == 0 or (sa == -(1 << 31) and sb == -1): rv = 0 else: rv = (abs(sa) % abs(sb)) & 0xFFFFFFFF if sa < 0: rv = (-rv) & 0xFFFFFFFF else: # A_REMU b32 = vb & 0xFFFFFFFF rv = (va & 0xFFFFFFFF) if b32 == 0 else (va & 0xFFFFFFFF) % b32 if rd: x[rd] = _sext32(rv) & MASK64 pc = nxt # -- system / atomics / traps ---------------------------------- elif kind == K_FENCE: pc = nxt elif kind == K_CSR: csr = imm old = self.csrs.get(csr, 0) if sub in (5, 6, 7): src = r1 # zimm if sub == 5: if rd: x[rd] = old self.csrs[csr] = src else: if rd: x[rd] = old if src: self.csrs[csr] = (old | src) if sub == 6 else (old & ~src) else: src = x[r1] if sub == 1: if rd: x[rd] = old self.csrs[csr] = src & MASK64 elif sub == 2: if rd: x[rd] = old if r1: self.csrs[csr] = (old | src) & MASK64 elif sub == 3: if rd: x[rd] = old if r1: self.csrs[csr] = (old & ~src) & MASK64 else: reason = R_FAULT a = 2 b = csr break pc = nxt elif kind == K_AMO: width = 8 if sub & 0x40 else 4 op5 = sub & 0x3F ea = x[r1] & MASK64 if watchers: hit = _watch_match(watchers, ea, width, 1) if hit is not None: reason = R_WATCH a = ea b = hit break try: old = self._guest_load(ea, width, pc) if active: trace.mem(insn_pc, ea, width, False, self.insns + n) if op5 == 0x02: # lr if rd: x[rd] = old elif op5 == 0x03: # sc self._guest_store(ea, width, x[r2], pc) if active: trace.mem(insn_pc, ea, width, True, self.insns + n) if rd: x[rd] = 0 else: rhs = x[r2] mask = 0xFFFFFFFF if width == 4 else MASK64 ov = old & mask rv = rhs & mask if op5 == 0x00: new = (ov + rv) & mask elif op5 == 0x01: new = rv elif op5 == 0x04: new = ov ^ rv elif op5 == 0x08: new = ov | rv elif op5 == 0x0C: new = ov & rv elif op5 == 0x10: new = ov if _signed64(ov) < _signed64(rv) else rv elif op5 == 0x14: new = ov if _signed64(ov) > _signed64(rv) else rv elif op5 == 0x18: new = ov if ov < rv else rv else: # 0x1C new = ov if ov > rv else rv self._guest_store(ea, width, new, pc) if active: trace.mem(insn_pc, ea, width, True, self.insns + n) if rd: x[rd] = (_sext32(old) & MASK64) if width == 4 else old pc = nxt except Fault as f: reason = R_FAULT a = f.cause b = f.tval break elif kind == K_EBREAK: if single_step: pc = nxt else: reason = R_EBREAK a = pc break elif kind == K_ECALL: if single_step: pc = nxt else: reason = R_ECALL a = pc break elif kind == K_MRET: target = self.csrs.get(0x341, 0) & MASK64 if target: pc = target else: reason = R_FAULT a = 2 b = 0 break else: # K_ILLEGAL reason = R_FAULT a = 2 b = 0 break self.pc = pc self.insns += n if single_step and reason == R_BUDGET: reason = R_STEP return (reason, pc, a, b) # -- guest memory helpers (AMO) ---------------------------------------- def _guest_load(self, addr: int, width: int, pc: int = 0) -> int: off = addr - self.ram_base if 0 <= off <= self.ram_size - width: if width == 8: return self._u64.unpack_from(self.ram, off)[0] return self._u32.unpack_from(self.ram, off)[0] return self._mmio_load(addr, width, pc) def _guest_store(self, addr: int, width: int, value: int, pc: int = 0) -> None: off = addr - self.ram_base if 0 <= off <= self.ram_size - width: if width == 8: self._u64p(self.ram, off, value) else: self._u32p(self.ram, off, value & 0xFFFFFFFF) return self._mmio_store(addr, width, value, pc) #: Task-facing alias: the native simulator machine is :class:`Machine`. SimMachine = Machine