diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/sim')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/sim/libos_port.py | 511 | ||||
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/sim/msgq_attach.py | 354 |
2 files changed, 860 insertions, 5 deletions
diff --git a/debug/risc-v-gsp-rm-vm/sim/libos_port.py b/debug/risc-v-gsp-rm-vm/sim/libos_port.py index eaa3f95..b1f596d 100644 --- a/debug/risc-v-gsp-rm-vm/sim/libos_port.py +++ b/debug/risc-v-gsp-rm-vm/sim/libos_port.py @@ -73,13 +73,33 @@ from __future__ import annotations from sim.libos import ( INTERRUPT_ARM_EXTERNAL, + MEMORY_QUERY, NAMES, + SHUTTLE_RESET, TASK_PORT_SEND_RECV_WAIT, + TASK_STATE_QUERY, ) #: Report schema tag for :meth:`LibosPortModel.state`. REPORT_FORMAT = "greeneye-msgq-port/1" +#: Report schema tag for :meth:`LibosPortNetwork.state`. +NETWORK_REPORT_FORMAT = "greeneye-msgq-port-network/1" + +#: The 8-byte "SPURIOUS" sentinel the firmware compares a queried task's +#: ``xepc`` against (the little-endian bytes of the literal, observed in the +#: image rodata and compared as a u64). A modelled external-interrupt report +#: carries it when no task trapped. +SPURIOUS_XEPC = 0x53554F4952555053 + +#: Modelled external-interrupt cause (interrupt bit | machine external +#: interrupt code) for the doorbell wake's report state. +EXTERNAL_INTERRUPT_XCAUSE = (1 << 63) | 11 + +#: ``LibosTaskState`` size (32 registers + xepc + xcause + xbadaddr + +#: nvriscv), the payload of ``TASK_STATE_QUERY``. +TASK_STATE_BYTES = 288 + #: ``LIBOS_SHUTTLE_ANY`` (RTOS port API). SHUTTLE_ANY = 0xFFFF #: The task's default shuttle id (RTOS port API). @@ -263,6 +283,10 @@ class LibosPortModel: send = a0 != 0 or a1 != 0 recv = a4 != 0 or a5 != 0 + # A receive with a non-zero wait id is a *synchronous* receive (or the + # receive half of a sendrecv): the kernel blocks until a message is + # delivered. Only a wait id of 0 is an asynchronous post. + sync_recv = recv and t1 != 0 if recv: self._post(shuttle=a4, port=a5, payload=a6, capacity=a7, pc=pc, @@ -270,11 +294,54 @@ class LibosPortModel: if send: self._send(shuttle=a0, port=a1, payload=a2, size=a3, wait_id=t1, timeout=t2, pc=pc, machine=machine) - return True, LIBOS_OK - if recv: + if not recv: + return True, LIBOS_OK + if recv and not sync_recv: # Async receive post: the kernel returns LIBOS_OK without touching # the wait outputs (a4-a6 still hold the receive arguments). return True, LIBOS_OK + if sync_recv: + if not self.completed: + self._try_kick() + if self.completed: + entry = self.completed.pop(0) + machine.x[14] = entry["payload_size"] & 0xFFFFFFFFFFFFFFFF + machine.x[15] = entry["origin_task"] & 0xFFFFFFFFFFFFFFFF + machine.x[16] = entry["shuttle"] & 0xFFFFFFFFFFFFFFFF + machine.x[17] = 0 + record = { + "pc": int(pc), + "wait_id": t1, + "timeout": t2, + "completed": entry["shuttle"], + "payload_size": entry["payload_size"], + "origin_task": entry["origin_task"], + "role": entry["role"], + "contract": entry["contract"], + } + self.waits.append(record) + self._event("sync_recv_complete", **record) + return True, LIBOS_OK + record = { + "pc": int(pc), + "wait_id": t1, + "timeout": t2, + "reason": "sync-recv-blocks", + "port": int(a5), + "shuttle": int(a4), + "pending": sorted(self.pending), + } + self.waits.append(record) + self._event("sync_recv_block", **record) + if self.stopped is None: + self.stopped = { + "num": TASK_PORT_SEND_RECV_WAIT, + "name": "task_port_send_recv_wait", + "pc": int(pc), + "reason": "sync-recv-blocks", + "pending": sorted(self.pending), + } + return False, None # Pure blocking wait: return the first completed shuttle, or stop. if not self.completed: @@ -384,3 +451,443 @@ class LibosPortModel: "events": [dict(event) for event in self.events], "stopped": dict(self.stopped) if self.stopped else None, } + + +def task_state_bytes(*, xcause: int, xepc: int, xbadaddr: int = 0, + registers=None, priv_err_info: int = 0) -> bytes: + """Serialise a ``LibosTaskState`` payload (pure; no machine). + + Layout: 32 u64 registers, then ``xepc``, ``xcause``, ``xbadaddr`` and the + NVRISCV error word (``priv_err_info`` u32 + reserved u32). + """ + buf = bytearray(TASK_STATE_BYTES) + for index, value in enumerate(registers or ()): + if 0 <= index < 32: + buf[index * 8:index * 8 + 8] = (int(value) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little") + buf[256:264] = (int(xepc) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little") + buf[264:272] = (int(xcause) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little") + buf[272:280] = (int(xbadaddr) & 0xFFFFFFFFFFFFFFFF).to_bytes(8, "little") + buf[280:284] = (int(priv_err_info) & 0xFFFFFFFF).to_bytes(4, "little") + return bytes(buf) + + +class LibosPortNetwork(LibosPortModel): + """Two-task libos port network (additive, opt-in). + + The real 570.144 GSP image links **two** tasks (an init daemon and the RM + task); the single-hart linear model only ever runs one of them. This + subclass models the libos port machinery between the modelled tasks: + + * receives/posts/completions are kept per task (shuttle ids are + task-local handles and collide across tasks); + * a send is routed to a receive waiting on the same port id in another + task, copying the firmware's own payload bytes (never an invented + payload); the receiver's completion records the sender's task index and + pairs a reply back to the sender's receive half (sendrecv); + * ``TASK_STATE_QUERY`` serves a modelled trap state for the external + interrupt report (the queried state's ``xcause`` is the interrupt cause + and ``xepc`` the "SPURIOUS" sentinel the firmware checks; both are + labelled model values, not firmware observations); + * ``MEMORY_QUERY`` and ``SHUTTLE_RESET`` are acknowledged from the flat + RAM model (labelled); + * ``INTERRUPT_ARM_EXTERNAL`` records the second argument (the port the + report is bound to) in addition to the mask. + + The single-task :class:`LibosPortModel` behaviour and report schema are + unchanged; this class is used only when the runner schedules a second + task. + """ + + name = "libos-port-network" + + def __init__(self, base=None, *, tasks=("init", "rm"), current=None): + task = current or tasks[0] + super().__init__(base=base, task=task) + self.tasks = tuple(tasks) + self.current = task + self.task_completions = {name: [] for name in self.tasks} + self.task_recvs = {name: {} for name in self.tasks} + self.send_queue = {port: [] for port in ()} + self.reply_map = {} + self.cross_task_payloads = [] + self.services = [] + self.queries = [] + self.arm_ports = [] + self.report_trap = None + self.switches = [] + #: Blocked syscall per task (so a resumed ecall is not re-executed as + #: a fresh send/post: the kernel would switch back *after* the wait). + self.blocked = {} + + # -- task identity ----------------------------------------------------- + + def switch(self, task: str, pc: int = 0) -> dict: + """Set the executing task (runner-side cooperative switch).""" + self.current = task + self.task = task + event = self._event("task_switch", pc=int(pc), to=task) + self.switches.append({"pc": int(pc), "to": task}) + return event + + def _task_index(self, name: str) -> int: + return self.tasks.index(name) if name in self.tasks else 0 + + # -- routing helpers --------------------------------------------------- + + def _find_receiver(self, port: int): + for name in self.tasks: + for shuttle in sorted(self.task_recvs[name]): + post = self.task_recvs[name][shuttle] + if post["port"] == int(port): + return post + return None + + def _post(self, *, shuttle: int, port: int, payload: int, capacity: int, + pc: int, wait_id: int, timeout: int) -> dict: + post = { + "shuttle": int(shuttle), + "port": int(port), + "role": role_for(port), + "payload": int(payload), + "capacity": int(capacity), + "pc": int(pc), + "wait_id": int(wait_id), + "timeout": int(timeout), + "task": self.current, + } + self.posts.append(post) + self.task_recvs[self.current][int(shuttle)] = post + self._event("receive_post", **post) + self._try_kick() + return post + + def _complete(self, shuttle: int, *, payload_size: int, origin_task: int, + contract: str, reason: str) -> dict: + owner = self.current + post = None + for name in self.tasks: + if int(shuttle) in self.task_recvs[name]: + owner = name + post = self.task_recvs[name].pop(int(shuttle)) + break + if post is None: # pragma: no cover - caller only completes posts + raise KeyError(f"shuttle {shuttle} is not posted") + entry = { + "shuttle": int(shuttle), + "port": int(post["port"]), + "role": post["role"], + "payload": int(post["payload"]), + "capacity": int(post["capacity"]), + "payload_size": int(payload_size), + "origin_task": int(origin_task), + "contract": contract, + "reason": reason, + "post_pc": int(post["pc"]), + "task": owner, + } + self.completed.append(entry) + self.task_completions[owner].append(entry) + self._event("receive_complete", **entry) + return entry + + def _copy(self, machine, dst: int, src: int, size: int) -> bool: + if size <= 0: + return True + try: + data = machine.read_mem(int(src), int(size)) + machine.write_mem(int(dst), data) + return True + except Exception: + return False + + def _take_completion(self, machine, pc: int): + queue = self.task_completions[self.current] + if not queue: + return None + entry = queue.pop(0) + machine.x[14] = entry["payload_size"] & 0xFFFFFFFFFFFFFFFF + machine.x[15] = entry["origin_task"] & 0xFFFFFFFFFFFFFFFF + machine.x[16] = entry["shuttle"] & 0xFFFFFFFFFFFFFFFF + machine.x[17] = 0 + record = { + "pc": int(pc), + "completed": entry["shuttle"], + "payload_size": entry["payload_size"], + "origin_task": entry["origin_task"], + "role": entry["role"], + "contract": entry["contract"], + "task": self.current, + } + self.waits.append(record) + self._event("wait_complete", **record) + return record + + def _block(self, machine, pc: int, *, reason: str, port: int, + shuttle: int) -> None: + kind = {"port-wait-blocks": "wait", "sync-recv-blocks": "recv", + "sync-send-blocks": "send"}.get(reason, "wait") + self.blocked[self.current] = {"kind": kind, "reason": reason, + "pc": int(pc), "port": int(port), + "shuttle": int(shuttle)} + record = { + "pc": int(pc), + "reason": reason, + "port": int(port), + "shuttle": int(shuttle), + "pending": sorted(s for name in self.tasks + for s in self.task_recvs[name]), + "task": self.current, + } + self.waits.append(record) + self._event("wait_block", **record) + if self.stopped is None: + self.stopped = { + "num": TASK_PORT_SEND_RECV_WAIT, + "name": "task_port_send_recv_wait", + "pc": int(pc), + "reason": reason, + "pending": record["pending"], + "task": self.current, + } + + # -- syscall surface --------------------------------------------------- + + def handle(self, machine, pc: int = 0): + num = int(machine.x[5]) + if num == TASK_PORT_SEND_RECV_WAIT: + return self._handle_network(machine, pc) + if num == INTERRUPT_ARM_EXTERNAL: + mask, port = int(machine.x[10]), int(machine.x[11]) + self.armed_mask |= mask + self.arm_ports.append({"mask": mask, "port": port, "pc": int(pc), + "task": self.current}) + self._event("interrupt_arm_external", mask=mask, port=port, + pc=pc, task=self.current) + self._try_kick() + return True, LIBOS_OK + if num == TASK_STATE_QUERY: + self.queries.append({"pc": int(pc), "task_index": int(machine.x[10]), + "buffer": int(machine.x[11]), + "task": self.current}) + trap = self.report_trap + if trap is None: + # No modelled trap state: the kernel would return an access + # error for an invalid target; return LIBOS_ERROR_INVALID_ACCESS + # (labelled model value) and let the firmware error path run. + return True, 0x80000000 + state = task_state_bytes( + xcause=trap["xcause"], xepc=trap["xepc"], + xbadaddr=trap["xbadaddr"], registers=trap.get("registers"), + ) + try: + machine.write_mem(int(machine.x[11]), state) + except Exception: + return True, 0x80000000 + self._event("task_state_query", **self.queries[-1], + xcause=hex(trap["xcause"]), xepc=hex(trap["xepc"])) + return True, LIBOS_OK + if num == MEMORY_QUERY: + remote, pointer = int(machine.x[10]), int(machine.x[11]) + attrs = 0x1 | 0x2 | 0x8 # read | write | cached (flat model) + resolved = pointer + translator = getattr(machine, "translator", None) + if translator is not None: + sim = translator.translate(pointer) + if sim is not None: + resolved = sim + machine.x[11] = attrs + machine.x[12] = pointer & ~0xFFF + machine.x[13] = 0x1000 + machine.x[14] = 1 # LIBOS_MEMORY_APERTURE_SYSCOH + machine.x[15] = resolved + record = {"pc": int(pc), "remote_task": remote, "pointer": pointer, + "resolved": resolved, "task": self.current} + self.services.append(dict(record, service="memory_query")) + self._event("memory_query", **record) + return True, LIBOS_OK + if num == SHUTTLE_RESET: + record = {"pc": int(pc), "shuttle": int(machine.x[10]), + "task": self.current} + self.services.append(dict(record, service="shuttle_reset")) + self._event("shuttle_reset", **record) + return True, LIBOS_OK + return super().handle(machine, pc) + + def _handle_network(self, machine, pc: int): + resuming = self.blocked.pop(self.current, None) + if resuming is not None and resuming["kind"] in ("recv", "wait"): + record = self._take_completion(machine, pc) + if record is not None: + return True, LIBOS_OK + self._block(machine, pc, reason=resuming["reason"], + port=resuming["port"], shuttle=resuming["shuttle"]) + return False, None + if resuming is not None and resuming["kind"] == "send": + self._block(machine, pc, reason=resuming["reason"], + port=resuming["port"], shuttle=resuming["shuttle"]) + return False, None + a0, a1, a2, a3 = (int(machine.x[10]), int(machine.x[11]), + int(machine.x[12]), int(machine.x[13])) + a4, a5, a6, a7 = (int(machine.x[14]), int(machine.x[15]), + int(machine.x[16]), int(machine.x[17])) + t1, t2 = int(machine.x[6]), int(machine.x[7]) + send = a0 != 0 or a1 != 0 + recv = a4 != 0 or a5 != 0 + sync_recv = recv and t1 != 0 + + if send and a1 == 0: + # Reply: route through the received shuttle's partner. + target = self.reply_map.pop((self.current, a0), None) + if target is not None: + name, shuttle = target + post = self.task_recvs[name].get(shuttle) + if post is not None: + size = min(a3, int(post["capacity"])) + ok = self._copy(machine, post["payload"], a2, size) + entry = self._complete( + shuttle, payload_size=size if ok else 0, + origin_task=self._task_index(self.current), + contract="modelled:cross-task-reply", + reason=f"reply from {self.current} ({size} bytes)") + self.sends.append({"shuttle": a0, "port": 0, "size": a3, + "pc": int(pc), "task": self.current, + "queued": False, "delivered_to": entry["shuttle"]}) + self.cross_task_payloads.append({ + "port": 0, "size": size, "from": self.current, + "to": entry.get("task", name), "kind": "reply"}) + machine.x[14] = size + machine.x[15] = 0 + machine.x[16] = a0 + machine.x[17] = 0 + if t1 == a0 or not recv: + return True, LIBOS_OK + + if send and a1 != 0: + partner = self._find_receiver(a1) + if partner is None: + size = a3 + self.sends.append({"shuttle": a0, "port": a1, "size": size, + "pc": int(pc), "task": self.current, + "queued": True, "payload": a2}) + return self._unrouted_send(machine, pc, a0, a1, a3, t1, + recv, sync_recv) + size = min(a3, int(partner["capacity"])) + ok = self._copy(machine, partner["payload"], a2, size) + entry = self._complete( + int(partner["shuttle"]), payload_size=size if ok else 0, + origin_task=self._task_index(self.current), + contract="modelled:cross-task-send", + reason=f"send from {self.current} port {a1} copied {size}") + self.sends.append({"shuttle": a0, "port": a1, "size": a3, + "pc": int(pc), "task": self.current, + "queued": False, "delivered_to": entry["shuttle"]}) + self.cross_task_payloads.append({ + "port": int(a1), "size": size, "from": self.current, + "to": entry.get("task"), "kind": "send"}) + if a4 or a5: + self.reply_map[(entry.get("task"), entry["shuttle"])] = (self.current, a4) + if t1 == a0: + machine.x[14] = size + machine.x[15] = 0 + machine.x[16] = a0 + machine.x[17] = 0 + return True, LIBOS_OK + if not recv: + return True, LIBOS_OK + + if recv: + self._post(shuttle=a4, port=a5, payload=a6, capacity=a7, pc=pc, + wait_id=t1, timeout=t2) + if sync_recv: + record = self._take_completion(machine, pc) + if record is not None: + return True, LIBOS_OK + self._block(machine, pc, reason="sync-recv-blocks", port=a5, + shuttle=a4) + return False, None + return True, LIBOS_OK + + record = self._take_completion(machine, pc) + if record is not None: + return True, LIBOS_OK + self._block(machine, pc, reason="port-wait-blocks", port=0, shuttle=0) + return False, None + + def _unrouted_send(self, machine, pc, shuttle, port, size, wait_id, + recv, sync_recv): + """A send with no modelled receiver: keep the boot-shape shim. + + The zero-length boot handshake send had no modelled peer and is an + explicit, labelled assumption (``modelled:peer-consumes-on-send``); + every other unrouted send stays queued and blocks its sync sender. + """ + if size != 0: + if wait_id == shuttle: + self._block(machine, pc, reason="sync-send-blocks", port=port, + shuttle=shuttle) + return False, None + return True, LIBOS_OK + machine.x[14] = 0 + machine.x[15] = 0 + machine.x[16] = shuttle + machine.x[17] = 0 + self._event("send_complete", shuttle=shuttle, port=port, size=0, + pc=int(pc), task=self.current, + contract="modelled:peer-consumes-on-send") + if wait_id == shuttle or not recv: + return True, LIBOS_OK + if sync_recv: + record = self._take_completion(machine, pc) + if record is not None: + return True, LIBOS_OK + self._block(machine, pc, reason="sync-recv-blocks", port=0, + shuttle=0) + return False, None + return True, LIBOS_OK + + def _try_kick(self): + """Deliver a pending host wake to the extintr receive (documented shim). + + The bound interrupt port is recorded from the arm call (the real kernel + delivers to the armed port); the modelled report retires the waiting + external-interrupt receive and records the modelled trap state its + ``TASK_STATE_QUERY`` consumer observes. + """ + if not self.kicks_pending: + return None + for name in self.tasks: + for shuttle in sorted(self.task_recvs[name]): + post = self.task_recvs[name][shuttle] + if post["role"] != "extintr": + continue + self.kicks_pending -= 1 + self.report_trap = { + "xcause": EXTERNAL_INTERRUPT_XCAUSE, + "xepc": SPURIOUS_XEPC, + "xbadaddr": 0, + "origin_task": 0, + } + self._complete(shuttle, payload_size=0, origin_task=0, + contract=EXTINTR_WAKE_CONTRACT, + reason="external-interrupt report") + return shuttle + return None + + def state(self) -> dict: + base = super().state() + base.update({ + "format": NETWORK_REPORT_FORMAT, + "tasks": list(self.tasks), + "current": self.current, + "task_completions": {name: [dict(e) for e in entries] + for name, entries in sorted(self.task_completions.items())}, + "recvs": {name: sorted(self.task_recvs[name]) + for name in self.tasks}, + "cross_task_payloads": [dict(p) for p in self.cross_task_payloads], + "services": [dict(s) for s in self.services], + "queries": [dict(q) for q in self.queries], + "arm_ports": [dict(a) for a in self.arm_ports], + "report_trap": dict(self.report_trap) if self.report_trap else None, + "switches": [dict(s) for s in self.switches], + }) + return base diff --git a/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py b/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py index f1530bd..01fc84a 100644 --- a/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py +++ b/debug/risc-v-gsp-rm-vm/sim/msgq_attach.py @@ -85,7 +85,7 @@ from sim.host_traffic import ( parse_queue_element, ) from sim.libos import LibosSyscalls, TASK_PORT_SEND_RECV_WAIT, TASK_STATE_QUERY -from sim.libos_port import LibosPortModel +from sim.libos_port import LibosPortModel, LibosPortNetwork from sim.machine import R_BP, R_BUDGET, R_ECALL, R_FAULT, R_STEP, R_WATCH #: Report schema tag. Every session/JSON artifact carries it. @@ -102,6 +102,16 @@ MAX_POLLS_RECORDED = 17 #: Internal stop code: the bounded poll list was filled before the run ended. R_POLL_LIMIT = 199 +#: Default extra slice cap for the scheduled second task (0 = run it until +#: the total instruction budget is reached). The observed RM task completes +#: its boot handshake with the init task inside a few thousand instructions; +#: its own start-up then runs past the total cap, so the frontier is reported +#: at the cap, never as success. +DEFAULT_SECOND_TASK_BUDGET = 0 + +#: Wait reasons that mean "this task is blocked and another task may run". +_BLOCK_REASONS = ("port-wait-blocks", "sync-recv-blocks", "sync-send-blocks") + #: Doorbell label (block-relative register name; public Turing dev_gsp.h). DOORBELL_WINDOW = "PRIV+0x110c00" @@ -297,6 +307,7 @@ def assemble_report(*, image: dict, profile: dict, boot: dict, attach: dict, doorbell: dict, polls: list, divergence, missing_mmio: dict, determinism: dict, continuation: dict | None = None, port_model: dict | None = None, missing_semantic: dict | None = None, + tasks: dict | None = None, schedule: list | None = None, limits=None, notes=None) -> dict: """Assemble the ``greeneye-msgq-attach/1`` report from plain inputs. @@ -343,6 +354,9 @@ def assemble_report(*, image: dict, profile: dict, boot: dict, attach: dict, else {"enabled": False}, "missing_semantic": dict(missing_semantic) if missing_semantic is not None else None, + "tasks": dict(tasks) if tasks is not None + else {"format": "greeneye-image-tasks/1", "tasks": []}, + "schedule": [dict(step) for step in (schedule or [])], "determinism": dict(determinism), "limits": list(limits if limits is not None else _LIMITS), "notes": list(notes or []), @@ -533,6 +547,153 @@ def _verify_command_element(host: HostTraffic, submit: dict) -> dict: return {"checksum_ok": False, "error": str(exc)} +def image_tasks(image: str) -> dict: + """Discover the linked libos tasks from the image's section header table. + + Pure stdlib ELF64 parsing (no symbols are present in the image): the + ``section_task_<name>_elf_text_instance`` sections give each task's text + range and the ``section_task_<name>_stack_instance`` sections its stack; + the task entry is the first pointer inside the text range found in the + ``.manifest`` section (the firmware's own task record). Data-driven: an + image without section names, manifest or match returns an empty table and + the caller stays single-task. + + Returns ``{"format": "greeneye-image-tasks/1", "tasks": [...]}``. + """ + empty = {"format": "greeneye-image-tasks/1", "tasks": []} + try: + with open(image, "rb") as handle: + blob = handle.read() + except OSError: + return empty + if len(blob) < 64 or blob[:4] != b"\x7fELF" or blob[4] != 2: + return empty + shoff = int.from_bytes(blob[0x28:0x30], "little") + shentsize = int.from_bytes(blob[0x3A:0x3C], "little") + shnum = int.from_bytes(blob[0x3C:0x3E], "little") + shstrndx = int.from_bytes(blob[0x3E:0x40], "little") + if not (0 < shentsize and 0 < shnum and shstrndx < shnum + and shoff + shentsize * shnum <= len(blob)): + return empty + shstr = None + raw_sections = [] + for index in range(shnum): + base = shoff + index * shentsize + if base + 64 > len(blob): + return empty + fields = blob[base:base + 64] + name_off = int.from_bytes(fields[0:4], "little") + addr = int.from_bytes(fields[16:24], "little") + offset = int.from_bytes(fields[24:32], "little") + size = int.from_bytes(fields[32:40], "little") + raw_sections.append((name_off, addr, offset, size)) + _, shstr_addr, shstr_off, shstr_size = raw_sections[shstrndx] + shstr = blob[shstr_off:shstr_off + shstr_size] + + def name_at(name_off): + end = shstr.find(b"\x00", name_off) + return shstr[name_off:end].decode("latin1") if end >= 0 else "" + + sections = {} + for name_off, addr, offset, size in raw_sections: + name = name_at(name_off) + if name: + sections[name] = {"addr": addr, "offset": offset, "size": size, "name": name} + tasks = {} + for name, section in sections.items(): + if name.startswith(".section_task_") and name.endswith("_elf_text_instance"): + task = name[len(".section_task_"):-len("_elf_text_instance")] + tasks.setdefault(task, {})["text"] = {"base": section["addr"], + "size": section["size"]} + elif name.startswith(".section_task_") and name.endswith("_stack_instance"): + task = name[len(".section_task_"):-len("_stack_instance")] + tasks.setdefault(task, {})["stack"] = { + "base": section["addr"], "size": section["size"], + "top": section["addr"] + section["size"]} + manifest = sections.get(".manifest") + if manifest: + data = blob[manifest["offset"]:manifest["offset"] + manifest["size"]] + for index in range(0, len(data) - 7, 8): + value = int.from_bytes(data[index:index + 8], "little") + for task, record in tasks.items(): + text = record.get("text") + if text and "entry" not in record \ + and text["base"] <= value < text["base"] + text["size"]: + record["entry"] = value + ordered = [] + for task in sorted(tasks): + record = tasks[task] + if "text" not in record or "entry" not in record: + continue + ordered.append({ + "name": task, + "entry": int(record["entry"]), + "text_base": int(record["text"]["base"]), + "text_size": int(record["text"]["size"]), + "stack_base": int(record.get("stack", {}).get("base", 0)), + "stack_size": int(record.get("stack", {}).get("size", 0)), + "stack_top": int(record.get("stack", {}).get("top", 0)), + }) + return {"format": "greeneye-image-tasks/1", "tasks": ordered} + + +def _task_for_pc(tasks_doc: dict, pc: int) -> dict | None: + for task in tasks_doc.get("tasks", []): + if task["text_base"] <= pc < task["text_base"] + task["text_size"]: + return task + return None + + +def missing_semantic_after_second_task(*, tasks_doc: dict, schedule: list, + frontier: dict, polls: list) -> dict: + """The exact unmodelled semantics at the two-task frontier (pure). + + Recorded when the scheduled second task (the RM task discovered from the + image) is blocked or out of its bounded slice without touching the rings. + """ + first = schedule[0] if schedule else {} + second = schedule[-1] if schedule else {} + return { + "stage": "second-task-frontier", + "tasks": [task["name"] for task in tasks_doc.get("tasks", [])], + "observed": [ + "the host doorbell was delivered as the RTOS external-interrupt " + "report and the init task consumed it as a spurious/EINT report " + "(modelled trap state, TASK_STATE_QUERY served)", + "the init task re-posted its receive and blocked in its port wait", + "the second task (the RM task, entry and stack discovered from " + "the image section/manifest records) was scheduled and completed " + "its boot port handshake with the init task", + "no command/status ring or doorbell read occurred within the " + "bounded instruction budget; the RM task's own start-up continues " + "past the cap before its first further port syscall", + ], + "missing": [ + "the queue consumer's start-up is not reachable inside the 2M " + "instruction cap from its entry; a data-driven post-init resume " + "point (or a qualified budget) is the smallest next semantic", + "the GSP_ARGUMENTS_CACHED / message-queue init handoff: the " + "simulator never delivers the queue base addresses the host " + "actually gives the firmware, so the rings stay at the fixture " + "block and no firmware queue poll can be observed", + "the doorbell -> interrupt binding for the consumer task: the " + "bound interrupt port recorded from INTERRUPT_ARM_EXTERNAL is " + "not the port the RM task receives the queue kick on", + ], + "next_step": ( + "extend the schedule with a data-driven post-init entry for the " + "RM task (or raise the cap with an admission record), deliver the " + "GSP_ARGUMENTS_CACHED/queue-init addresses from the host model, " + "then re-run this attach and classify the first ring poll; do not " + "infer a poll from this run"), + "schedule": [dict(step) for step in schedule], + "frontier": dict(frontier) if frontier else None, + "polls": len(polls), + "first_task": first.get("task"), + "second_task": second.get("task"), + } + + def missing_semantic_after_handoff(*, stopped: dict, polls: list) -> dict: """The exact unmodelled semantics that stopped the port-model continuation. @@ -578,7 +739,9 @@ def missing_semantic_after_handoff(*, stopped: dict, polls: list) -> dict: def run_attach(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFILE, max_cycles: int = DEFAULT_MAX_CYCLES, doorbell: bool = False, - port_model: bool = True, json_out: str | None = None, + port_model: bool = True, second_task: bool = True, + second_task_budget: int = DEFAULT_SECOND_TASK_BUDGET, + json_out: str | None = None, record_dir: str | None = None) -> dict: """Boot to the port wait, attach the rings, optionally kick and continue. @@ -595,11 +758,25 @@ def run_attach(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFIL task-state-query the interrupt dispatch performs (or further). With the model off, the documented single-wake shim of the first wave is used and the run reports ``port-wait-reentry`` instead. + + ``second_task`` (default on) additionally discovers the image's linked + tasks (section/manifest records) and schedules the second one (the RM + task) when the init task blocks: the boot port handshake is modelled with + the firmware's own payloads, bounded by the total instruction budget (an + optional tighter ``second_task_budget`` slice cap, 0 = until the total + budget). The second task's own start-up runs past the cap, so the report + names that frontier instead of manufacturing a poll; the single-task + behaviour is kept with ``second_task=False``. """ if isinstance(max_cycles, bool) or not isinstance(max_cycles, int) \ or not 0 < max_cycles <= MAX_RUN_CYCLES: raise MsgqAttachError( f"max_cycles must be an int in 1..{MAX_RUN_CYCLES}, got {max_cycles!r}") + if isinstance(second_task_budget, bool) or not isinstance(second_task_budget, int) \ + or not 0 <= second_task_budget <= MAX_RUN_CYCLES: + raise MsgqAttachError( + f"second_task_budget must be an int in 0..{MAX_RUN_CYCLES} " + f"(0 = until the total budget), got {second_task_budget!r}") image = os.path.abspath(image) profile_path = os.path.abspath(profile_path) if not os.path.isfile(image): @@ -616,11 +793,13 @@ def run_attach(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFIL load = profile.get("load", {}) ram_base = _as_int(load.get("ram_base", 0)) memory_mb = _as_int(load.get("memory_mb", 12288)) + tasks_doc = image_tasks(image) session = None notes: list[str] = [] divergence = None polls: list[dict] = [] + schedule: list[dict] = [] continuation: dict = {"ran": False} port_model_state: dict | None = None missing_semantic: dict | None = None @@ -699,7 +878,163 @@ def run_attach(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFIL f"stop={stop_reason} at pc={pc:#x})"), } - if wait_reached and doorbell and port_model: + if wait_reached and doorbell and port_model and second_task: + # Two-task schedule: the image's second linked task (the RM task) + # is data-discovered and seeded when the init task blocks. + primary = _task_for_pc(tasks_doc, int(pc)) or (tasks_doc["tasks"][0] + if tasks_doc["tasks"] else None) + secondary = next((task for task in tasks_doc["tasks"] if task is not primary), + None) if primary else None + if primary is not None and secondary is not None: + model = LibosPortNetwork( + base=libos, tasks=(primary["name"], secondary["name"]), + current=primary["name"]) + machine.syscall_handler = model.handle + last_kick = host.doorbell.writes[-1] if host.doorbell.writes else None + model.host_kick( + queue=int(last_kick["queue"]) if last_kick else None, + value=int(last_kick["value"]) if last_kick else None, + ) + schedule: list[dict] = [] + saved = {primary["name"]: None, secondary["name"]: None} + started = {primary["name"]: True, secondary["name"]: False} + entries = {primary["name"]: primary, secondary["name"]: secondary} + reason2, pc2, a2, b2 = _run_watching(machine, max_cycles, geometry, + milestones_hit, polls) + for _ in range(8): + stop = model.stopped if reason2 == R_ECALL else None + if polls or stop is None \ + or stop.get("reason") not in _BLOCK_REASONS \ + or reason2 == R_POLL_LIMIT: + break + current = model.current + saved[current] = {"registers": list(machine.x), + "pc": int(machine.pc)} + schedule.append({"task": current, "pc": int(pc2), + "reason": stop.get("reason"), + "insns": int(machine.insns)}) + other = secondary["name"] if current == primary["name"] \ + else primary["name"] + if not started[other]: + if machine.insns >= max_cycles: + break + started[other] = True + record = entries[other] + machine.x = [0] * 32 + machine.x[2] = int(record.get("stack_top", 0)) + machine.pc = int(record["entry"]) + model.switch(other, pc=int(machine.pc)) + model.stopped = None + schedule.append({"task": other, "pc": int(machine.pc), + "reason": "scheduled", + "insns": int(machine.insns)}) + slice_end = max_cycles if not second_task_budget \ + else min(max_cycles, + machine.insns + int(second_task_budget)) + reason2, pc2, a2, b2 = _run_watching( + machine, slice_end, geometry, milestones_hit, polls) + continue + if model.task_completions.get(other): + machine.x = list(saved[other]["registers"]) + machine.pc = int(saved[other]["pc"]) + model.switch(other, pc=int(machine.pc)) + model.stopped = None + schedule.append({"task": other, "pc": int(machine.pc), + "reason": "resumed", + "insns": int(machine.insns)}) + reason2, pc2, a2, b2 = _run_watching( + machine, max_cycles, geometry, milestones_hit, polls) + continue + break + stop_reason2 = STOP_REASONS.get(reason2, str(reason2)) + if reason2 == R_POLL_LIMIT: + stop_reason2 = "poll-limit" + port_model_state = {"enabled": True, **model.state()} + continuation = { + "ran": True, + "port_model": True, + "second_task": True, + "stop_reason": stop_reason2, + "stop_pc": int(pc2), + "instructions_executed": int(machine.insns), + "wakes": int(model.host_kicks), + "wake_log": [ + { + "result": entry["contract"], + "returned": 0, + "delivered": entry.get("delivered"), + "pending": entry.get("pending"), + "queue": entry.get("queue"), + "value": entry.get("value"), + } + for entry in model.events + if entry["event"] == "host_kick" + ], + "reentries": [], + "polls": len(polls), + "tasks": list(model.tasks), + "switches": len(model.switches), + } + if reason2 == R_FAULT: + divergence = { + "kind": "fault", + "reason": (f"fault after the host wake: cause={a2:#x} " + f"tval={b2:#x} pc={pc2:#x}"), + } + elif reason2 == R_ECALL: + port_stopped = model.stopped or ( + dict(libos.stopped) if libos.stopped else None) + number2 = int(port_stopped["num"]) if port_stopped \ + else int(machine.x[5]) + name2 = (port_stopped or {}).get("name", "unhandled ecall") + if not polls: + attack = (port_stopped or {}) + missing_semantic = missing_semantic_after_second_task( + tasks_doc=tasks_doc, schedule=schedule, + frontier=attack, polls=polls) + divergence = { + "kind": "second-task-frontier", + "reason": ( + "after the recorded host wake the init task " + "consumed the external-interrupt report and " + "blocked; the discovered second task (the RM " + "task) was scheduled, completed its boot port " + "handshake and then blocked without touching " + "the rings -- " + f"{name2} (num={number2}) at pc={pc2:#x}"), + "stop": dict(attack), + } + elif reason2 == R_BUDGET and not polls: + frontier = { + "task": model.current, + "reason": "second-task-budget" if second_task_budget + else "total-budget", + "pc": int(pc2), + } + missing_semantic = missing_semantic_after_second_task( + tasks_doc=tasks_doc, schedule=schedule, + frontier=frontier, polls=polls) + if second_task_budget: + divergence = { + "kind": "second-task-frontier", + "reason": (f"the discovered second task ran its " + f"bounded slice ({second_task_budget} " + f"instructions) without a ring read " + f"(total budget {max_cycles})"), + } + else: + divergence = { + "kind": "second-task-frontier", + "reason": (f"the discovered second task ran to the " + f"total budget of {max_cycles} " + f"instructions without a ring read " + f"(its start-up continues past the cap)"), + } + elif reason2 == R_POLL_LIMIT: + notes.append( + f"continuation stopped after {MAX_POLLS_RECORDED} " + "recorded ring accesses (bounded by design)") + elif wait_reached and doorbell and port_model: model = LibosPortModel(base=libos) machine.syscall_handler = model.handle last_kick = host.doorbell.writes[-1] if host.doorbell.writes else None @@ -935,6 +1270,8 @@ def run_attach(*, image: str = DEFAULT_IMAGE, profile_path: str = DEFAULT_PROFIL continuation=continuation, port_model=port_model_state, missing_semantic=missing_semantic, + tasks=tasks_doc, + schedule=schedule, notes=notes, ) if getattr(session.recorder, "dir", None) is not None: @@ -1069,6 +1406,14 @@ def build_parser() -> argparse.ArgumentParser: run_cmd.add_argument("--no-port-model", action="store_true", help="disable the sim.libos_port receive-path model " "(use the first-wave single-wake shim)") + run_cmd.add_argument("--no-second-task", action="store_true", + help="keep the single-task port model (do not schedule " + "the image's second linked task)") + run_cmd.add_argument("--second-task-budget", type=int, + default=DEFAULT_SECOND_TASK_BUDGET, + help="extra instruction slice cap for the scheduled " + "second task (0 = until the total budget, " + f"default {DEFAULT_SECOND_TASK_BUDGET})") run_cmd.add_argument("--json", metavar="OUT", default=None, help="write the full report JSON to OUT") run_cmd.add_argument("--record", metavar="DIR", default=None, @@ -1093,6 +1438,7 @@ def main(argv: list[str] | None = None) -> int: parser.print_help(sys.stderr) return 2 port_model = not getattr(args, "no_port_model", False) + second_task = not getattr(args, "no_second_task", False) if args.command == "demo": run_demo(image=args.image, profile_path=args.profile, print_json=args.json, port_model=port_model) @@ -1104,6 +1450,8 @@ def main(argv: list[str] | None = None) -> int: max_cycles=args.max_cycles, doorbell=args.doorbell, port_model=port_model, + second_task=second_task, + second_task_budget=args.second_task_budget, json_out=args.json, record_dir=args.record, ) |