"""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, }