1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
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,
}
|
