"""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. Alongside the instruction count (virtual time, ``insns``) the machine keeps a ``cycles`` counter advanced by :mod:`sim.cycles`, a table-driven **modeled** instruction-class cost. It changes no execution behaviour: with the default model enabled, ``insns``, stops and register/memory state are exactly as before; ``cycle_model=False`` leaves ``cycles`` at 0. See ``docs/32-GSP-CYCLE-MODEL.md``. """ from __future__ import annotations import mmap import struct from .cycles import DEFAULT_MODEL from .mpu import MSTATUS, WATCHED_CSRS, RiscvMpu from .decode import ( A_ADD, A_AND, A_DIV, A_DIVU, A_MUL, A_MULH, A_MULHSU, A_MULHU, A_OR, A_REM, 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_SRLI, I_SRLIW, I_XORI, K_AMO, K_AUIPC, K_BRANCH, K_CSR, K_EBREAK, K_ECALL, K_FENCE, 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") #: CSRs whose writes update the MMU model (M-mode protection unit + mstatus). _MPU_CSRS = frozenset(WATCHED_CSRS) | {MSTATUS} 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. ``cycle_model`` selects the modeled cycle accounting (:mod:`sim.cycles`): ``None`` (default) uses :data:`sim.cycles.DEFAULT_MODEL`, a :class:`~sim.cycles.CycleModel` instance uses that table, and ``False`` disables accounting (``cycles`` stays 0). Accounting never changes execution. """ 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, cycle_model=None, devices=None, reset_regs=None, translator=None, ): if unknown_mmio not in UNKNOWN_MMIO_MODES: raise ValueError(f"unknown_mmio must be one of {UNKNOWN_MMIO_MODES}, got {unknown_mmio!r}") if cycle_model is False: cycle_model = None elif cycle_model is None: cycle_model = DEFAULT_MODEL self.cycle_model = cycle_model self.cycles = 0 if cycle_model is None: self._cycle_base = None self._cycle_op = None self._cycle_mem = 0 self._cycle_io = 0 else: self._cycle_base = cycle_model.base_costs self._cycle_op = cycle_model.op_costs self._cycle_mem = cycle_model.memory_latency self._cycle_io = cycle_model.mmio_latency 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.devices = devices self.translator = translator # aperture -> RAM seam (sim.devices.dma) self.mpu = RiscvMpu() # M-mode protection unit (sim.mpu) #: Optional guest-ECALL service (``handle(machine) -> (handled, ret)``); #: when absent (or unhandled) ``run`` stops with ``R_ECALL`` as before. self.syscall_handler = None 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: """Set initial register values (boot shim: a0-a5 bootargs, etc.). Accepts integer indices (0-31) or register names (``"a0"``..``"a7"``, ``"ra"``, ``"sp"``, ``"tp"``, ``"s0"``.., ``"t0"``..). x0 is ignored. """ names = {"zero": 0, "ra": 1, "sp": 2, "gp": 3, "tp": 4, "t0": 5, "t1": 6, "t2": 7, "s0": 8, "fp": 8, "s1": 9, "a0": 10, "a1": 11, "a2": 12, "a3": 13, "a4": 14, "a5": 15, "a6": 16, "a7": 17, "s2": 18, "s3": 19, "s4": 20, "s5": 21, "s6": 22, "s7": 23, "s8": 24, "s9": 25, "s10": 26, "s11": 27, "t3": 28, "t4": 29, "t5": 30, "t6": 31} for key, value in regs.items(): index = key if isinstance(key, int) else names.get(str(key)) if index is None: raise ValueError(f"unknown register name {key!r}") self.set_reg(int(index), int(value)) def insn_count(self) -> int: """Virtual time (instruction count) for modeled clock devices.""" return self.insns # -- image loading ----------------------------------------------------- def load_segments(self, plan, entry: int) -> 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 elif self.translator is not None and self.translator.translate(addr) is not None: sim = self.translator.translate(addr) off = sim - self.ram_base if off < 0 or off + size > self.ram_size: raise Fault(5, addr) out += self.ram[off : off + size] addr += size size = 0 elif self.devices is not None: peeked = self.devices.peek(addr, size) if peeked is None: raise Fault(5, addr) out += peeked addr += size size = 0 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 elif self.translator is not None and self.translator.translate(here) is not None: sim = self.translator.translate(here) off = sim - self.ram_base if off < 0 or off + remaining > self.ram_size: raise Fault(7, here) self.ram[off : off + remaining] = data[offset:] offset = len(data) elif self.devices is not None and self.devices.write( here, remaining, int.from_bytes(data[offset:], "little") ): offset = len(data) 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 _aperture_load(self, addr: int, size: int) -> int | None: """Serve a load from a translated memory aperture (FBGPA/SYSGPA).""" if self.translator is None: return None sim = self.translator.translate(addr) if sim is None: return None off = sim - self.ram_base if off < 0 or off + size > self.ram_size: return None if size == 8: return self._u64.unpack_from(self.ram, off)[0] if size == 4: return self._u32.unpack_from(self.ram, off)[0] if size == 2: return self._u16.unpack_from(self.ram, off)[0] if size == 1: return self.ram[off] return None def _aperture_store(self, addr: int, size: int, value: int) -> bool: """Serve a store into a translated memory aperture (FBGPA/SYSGPA).""" if self.translator is None: return False sim = self.translator.translate(addr) if sim is None: return False off = sim - self.ram_base if off < 0 or off + size > self.ram_size: return False if size == 8: self._u64p(self.ram, off, value & 0xFFFFFFFFFFFFFFFF) elif size == 4: self._u32p(self.ram, off, value & 0xFFFFFFFF) elif size == 2: self._u16p(self.ram, off, value & 0xFFFF) else: self.ram[off] = value & 0xFF self.invalidate() return True 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) value = self._aperture_load(addr, size) if value is not None: return value if self.devices is not None: handled, value = self.devices.read(addr, size, pc=pc) if handled: return value 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 if self._aperture_store(addr, size, value): return if self.devices is not None and self.devices.write(addr, size, value, pc=pc): 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 # M-mode protection unit (sim.mpu): translation is consulted only when # an enabled entry exists; no match is treated as identity. mpu = self.mpu mpu_on = mpu.active # Modeled cycle accounting (sim.cycles): pure table lookups, no timing. cycle_base = self._cycle_base cycle_op = self._cycle_op cycle_mem = self._cycle_mem cycle_io = self._cycle_io counting = cycle_base is not None cycle_sum = 0 while n < budget: if has_bps and pc in breakpoints: reason = R_BP a = pc break fetch_pc = pc if mpu_on: translated = mpu.translate(pc) if translated is not None: fetch_pc = translated off = fetch_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 counting: kc = cycle_base[kind] cycle_sum += cycle_op[sub] if kc < 0 else kc 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 mem_ea = ea if mpu_on: translated = mpu.translate(ea) if translated is not None: mem_ea = translated o = mem_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] if counting: cycle_sum += cycle_mem pc = nxt else: try: v = self._mmio_load(mem_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) if counting: cycle_sum += cycle_io 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 mem_ea = ea if mpu_on: translated = mpu.translate(ea) if translated is not None: mem_ea = translated o = mem_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 # Self-modifying code: drop any cached decode overlapping # the written range (firmware patches code at runtime). # Both the virtual address and its translated backing are # dropped so a fetch through either mapping re-decodes. for target in ((ea, mem_ea) if mem_ea != ea else (ea,)): p = target & ~1 end = target + size while p < end: cache.pop(p, None) p += 2 if counting: cycle_sum += cycle_mem pc = nxt else: try: self._mmio_store(mem_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) if counting: cycle_sum += cycle_io 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 if csr in _MPU_CSRS: mpu.on_csr_write(csr, self.csrs.get(csr, 0)) mpu_on = mpu.active 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 if counting: cycle_sum += cycle_mem if 0 <= ea - base <= limit - width else cycle_io 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 elif self.syscall_handler is not None: handled, result = self.syscall_handler(self, pc) if not handled: reason = R_ECALL a = pc b = x[5] # syscall number (t0), for diagnostics break if result is not None: x[10] = result & MASK64 pc = nxt else: reason = R_ECALL a = pc b = x[5] 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 self.cycles += cycle_sum if single_step and reason == R_BUDGET: reason = R_STEP return (reason, pc, a, b) def device_state(self) -> dict | None: """Serialisable state of the attached device registry (or None).""" if self.devices is None or not hasattr(self.devices, "state"): return None return self.devices.state() # -- guest memory helpers (AMO) ---------------------------------------- def _guest_load(self, addr: int, width: int, pc: int = 0) -> int: mem = addr if self.mpu.active: translated = self.mpu.translate(addr) if translated is not None: mem = translated off = mem - 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(mem, width, pc) def _guest_store(self, addr: int, width: int, value: int, pc: int = 0) -> None: mem = addr if self.mpu.active: translated = self.mpu.translate(addr) if translated is not None: mem = translated off = mem - 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) p = addr & ~1 end = addr + width while p < end: self._cache.pop(p, None) p += 2 return self._mmio_store(mem, width, value, pc) #: Task-facing alias: the native simulator machine is :class:`Machine`. SimMachine = Machine