diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/sim/libos.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/sim/libos.py | 217 |
1 files changed, 217 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/sim/libos.py b/debug/risc-v-gsp-rm-vm/sim/libos.py new file mode 100644 index 0000000..e8280fd --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/sim/libos.py @@ -0,0 +1,217 @@ +"""Minimal, honest libos syscall surface for the native GSP boot. + +Evidence kind: **firmware use + RTOS syscall table.** The GSP application is +a libos task that requests kernel services with ``ecall`` (the syscall number +travels in ``t0``); the native ``Machine`` stops on ``ecall`` by default, so +booting the real image past kernel init needs a deterministic service model. + +Rules: + +* every requested syscall is recorded (number, name, arguments, pc, + instruction count) so the session shows exactly what the firmware asked for; +* only *no-ops and modelled* services are acknowledged: task yield, cache + invalidation, external-interrupt arming, profiler enable/query, and the + timer read (returned from the modeled instruction count, labelled + modeled); +* **waits are never faked**: ``PROCESSOR_SUSPEND``, + ``TASK_PORT_SEND_RECV_WAIT``, ``TASK_PANIC`` and every unimplemented + syscall return *not handled*, which stops the run with ``R_ECALL`` so the + caller can classify the stop (message-loop wait / panic / unsupported + service) instead of inventing progress; +* task switching, DMA, memory queries, pagetable registration, partition + switch and interrupt delivery are not modelled. + +No timing, scheduling or security claim is made; this is development +tooling for the boot ladder (docs/35 ยง3.3). +""" +from __future__ import annotations + +# Syscall numbers (RTOS ``libos_syscalls.h``). +TASK_PORT_SEND_RECV_WAIT = 1 +TASK_YIELD = 2 +PROCESSOR_SUSPEND = 3 +DCACHE_INVALIDATE = 4 +MEMORY_QUERY = 5 +TASK_STATE_QUERY = 6 +SHUTTLE_RESET = 7 +TASK_DMA_REGISTER_TCM = 8 +TASK_DMA_COPY = 9 +TASK_DMA_FLUSH = 10 +TASK_MEMORY_READ = 11 +INIT_REGISTER_PAGETABLES = 12 +PROFILER_ENABLE = 13 +TASK_EXIT = 14 +TASK_PANIC = 15 +INTERRUPT_ARM_EXTERNAL = 16 +TIMER_GET_NS = 17 +DCACHE_INVALIDATE_ALL = 18 +PROFILER_GET_COUNTER_MASK = 19 +PARTITION_SWITCH = 20 +LIMIT_POW2 = 32 + +NAMES = { + TASK_PORT_SEND_RECV_WAIT: "task_port_send_recv_wait", + TASK_YIELD: "task_yield", + PROCESSOR_SUSPEND: "processor_suspend", + DCACHE_INVALIDATE: "dcache_invalidate", + MEMORY_QUERY: "memory_query", + TASK_STATE_QUERY: "task_state_query", + SHUTTLE_RESET: "shuttle_reset", + TASK_DMA_REGISTER_TCM: "task_dma_register_tcm", + TASK_DMA_COPY: "task_dma_copy", + TASK_DMA_FLUSH: "task_dma_flush", + TASK_MEMORY_READ: "task_memory_read", + INIT_REGISTER_PAGETABLES: "init_register_pagetables", + PROFILER_ENABLE: "profiler_enable", + TASK_EXIT: "task_exit", + TASK_PANIC: "task_panic", + INTERRUPT_ARM_EXTERNAL: "interrupt_arm_external", + TIMER_GET_NS: "timer_get_ns", + DCACHE_INVALIDATE_ALL: "dcache_invalidate_all", + PROFILER_GET_COUNTER_MASK: "profiler_get_counter_mask", + PARTITION_SWITCH: "partition_switch", + LIMIT_POW2: "limit_pow2", +} + +#: Services acknowledged as no-ops (result 0); everything else stops the run. +_NOOP_SYSCALLS = frozenset({ + TASK_YIELD, DCACHE_INVALIDATE, DCACHE_INVALIDATE_ALL, PROFILER_ENABLE, +}) + + +class LibosSyscalls: + """Deterministic libos service model: record, acknowledge, or stop.""" + + name = "libos-syscalls" + + def __init__(self) -> None: + self.calls: list[dict] = [] + self.counts: dict[int, int] = {} + self.armed_mask = 0 + #: Last registered TCM bounce buffer (``task_dma_register_tcm``). + self.tcm: dict | None = None + #: Last inverted-pagetable registration (``init_register_pagetables``). + self.pagetables: dict | None = None + self.dma_copies = 0 + self.dma_bytes = 0 + #: (number, a0, a1, pc, insns) of the last request. + self.last: tuple | None = None + #: The request that stopped the run (not handled), if any. + self.stopped: dict | None = None + + def handle(self, machine, pc: int = 0): + """``Machine`` ecall hook: ``(handled, result)``. + + Unhandled requests leave ``machine.run`` stopping with ``R_ECALL``; + the number is still recorded here and in ``stopped``. + """ + num = machine.x[5] + a0, a1 = machine.x[10], machine.x[11] + entry = { + "num": num, + "name": NAMES.get(num, f"unknown_{num}"), + "a0": a0, + "a1": a1, + "pc": pc, + "insns": machine.insns, + } + self.calls.append(entry) + self.counts[num] = self.counts.get(num, 0) + 1 + self.last = (num, a0, a1, pc, machine.insns) + if num == INTERRUPT_ARM_EXTERNAL: + self.armed_mask |= a0 + return True, 0 + if num == TIMER_GET_NS: + # Modeled time: instruction count * 1 ns (docs/36: modeled, not + # calibrated; no timing claim). + return True, machine.insns + if num == TASK_DMA_REGISTER_TCM: + # The kernel validates the VA against its pagetable and stores it + # as the DMA bounce buffer; without a paging model the region is + # recorded and the observed result (LIBOS_OK) returned. + self.tcm = {"va": a0, "size": a1} + return True, 0 + if num == TASK_DMA_COPY: + return self._dma_copy(machine, a0, a1, machine.x[12], pc) + if num == INIT_REGISTER_PAGETABLES: + # The kernel builds its inverted page table from this registration + # and returns LIBOS_OK. The simulator does not model the + # inverted table (DMA resolution goes through the flat RAM + MPU + # + aperture seam); the registration is recorded as evidence. + self.pagetables = {"base": a0, "entries": a1, "probe_count": machine.x[12]} + return True, 0 + if num == TASK_DMA_FLUSH: + return True, 0 + if num in _NOOP_SYSCALLS: + return True, 0 + if num == PROFILER_GET_COUNTER_MASK: + return True, 0 + self.stopped = entry + return False, None + + # -- DMA service model ------------------------------------------------- + + def _resolve(self, machine, va: int): + """VA -> simulator RAM offset through the MPU and aperture seam.""" + if machine.mpu.active: + translated = machine.mpu.translate(va) + if translated is not None: + va = translated + off = va - machine.ram_base + if 0 <= off < machine.ram_size: + return off + translator = machine.translator + if translator is not None: + sim = translator.translate(va) + if sim is not None: + off = sim - machine.ram_base + if 0 <= off < machine.ram_size: + return off + return None + + def _dma_copy(self, machine, dst_va: int, src_va: int, size: int, pc: int = 0): + """Modeled DMA copy: equivalent to a RAM memcpy in the flat model. + + The real kernel drives the falcon DMA engine through TCM; the + simulator keeps all regions in one RAM, so the copy effect is the + same. Both sides must resolve or the run stops (never a faked copy). + """ + if size == 0: + return True, 0 + dst = self._resolve(machine, dst_va) + src = self._resolve(machine, src_va) + if dst is None or src is None: + self.stopped = {"num": TASK_DMA_COPY, "name": "task_dma_copy", + "a0": dst_va, "a1": src_va, "size": size, + "reason": "unresolved", "pc": pc} + return False, None + ram = machine.ram + if (0 <= dst <= machine.ram_size - size + and 0 <= src <= machine.ram_size - size): + ram[dst:dst + size] = bytes(ram[src:src + size]) + else: + # Straddling or out-of-range: byte-wise through the bounds. + for i in range(size): + if dst + i >= machine.ram_size or src + i >= machine.ram_size: + self.stopped = {"num": TASK_DMA_COPY, "name": "task_dma_copy", + "a0": dst_va, "a1": src_va, "size": size, + "reason": "out-of-range", "pc": pc} + return False, None + ram[dst + i] = ram[src + i] + machine.invalidate() + self.dma_copies += 1 + self.dma_bytes += size + return True, 0 + + def state(self) -> dict: + return { + "calls": len(self.calls), + "counts": {str(k): v for k, v in sorted(self.counts.items())}, + "armed_mask": self.armed_mask, + "tcm": self.tcm, + "pagetables": self.pagetables, + "dma_copies": self.dma_copies, + "dma_bytes": self.dma_bytes, + "last": list(self.last) if self.last else None, + "stopped": self.stopped, + } |