diff options
Diffstat (limited to 'debug/risc-v-gsp-rm-vm/greeneye/target.py')
| -rw-r--r-- | debug/risc-v-gsp-rm-vm/greeneye/target.py | 223 |
1 files changed, 223 insertions, 0 deletions
diff --git a/debug/risc-v-gsp-rm-vm/greeneye/target.py b/debug/risc-v-gsp-rm-vm/greeneye/target.py new file mode 100644 index 0000000..efcb1d1 --- /dev/null +++ b/debug/risc-v-gsp-rm-vm/greeneye/target.py @@ -0,0 +1,223 @@ +"""QEMU target lifecycle for green-eye. + +Owns the ``qemu-system-riscv64`` process, the gdbstub endpoint, the serial +capture and the QEMU trace log. Both firmware paths are supported: + +* ``kernel=`` fast path - QEMU loads an ELF linked for RAM and sets the PC; +* ``segments=`` probe path - each PT_LOAD segment is placed with a loader + device (raw), so a firmware linked for GPU physical addresses can be + relocated into RAM and observed. +""" +from __future__ import annotations + +import os +import shutil +import signal +import socket +import subprocess +import tempfile +import time + +from .errors import TargetError + +DEFAULT_QEMU = "qemu-system-riscv64" + + +def find_qemu() -> str: + env = os.environ.get("GREENEYE_QEMU") + if env: + return env + found = shutil.which(DEFAULT_QEMU) + if not found: + raise TargetError( + f"{DEFAULT_QEMU} not found. Install it (Ubuntu: qemu-system-misc) " + "or set GREENEYE_QEMU to the binary." + ) + return found + + +def free_port() -> int: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", 0)) + return sock.getsockname()[1] + + +class QemuTarget: + def __init__( + self, + *, + machine: str = "virt", + memory_mb: int = 256, + gdb_port: int = 0, + bios: str = "none", + extra_args: list[str] | None = None, + qemu: str | None = None, + kernel: str | None = None, + segments: list[tuple[int, bytes]] | None = None, + dlog: str | None = None, + uart_log: str | None = None, + qemu_log: str | None = None, + workdir: str | None = None, + keep_alive: bool = False, + ): + self.qemu = qemu or find_qemu() + self.machine = machine + self.memory_mb = memory_mb + self.host = "127.0.0.1" + self.port = gdb_port or free_port() + self.bios = bios + self.extra_args = list(extra_args or []) + self.kernel = os.path.abspath(kernel) if kernel else None + self.segments = list(segments or []) + self.dlog = dlog + self.workdir = workdir or tempfile.mkdtemp(prefix="greeneye-qemu-") + os.makedirs(self.workdir, exist_ok=True) + self.uart_log = uart_log or os.path.join(self.workdir, "uart.log") + self.qemu_log = qemu_log or os.path.join(self.workdir, "qemu.log") + self.keep_alive = keep_alive + self.process: subprocess.Popen | None = None + self._segment_files: list[str] = [] + self._cmd: list[str] = [] + + # -- lifecycle --------------------------------------------------------- + + def _build_command(self, paused: bool) -> list[str]: + cmd = [ + self.qemu, + "-machine", + self.machine, + "-m", + f"{self.memory_mb}M", + "-display", + "none", + "-monitor", + "none", + "-serial", + f"file:{self.uart_log}", + "-gdb", + f"tcp:{self.host}:{self.port}", + ] + if self.bios: + cmd += ["-bios", self.bios] + if paused: + cmd += ["-S"] + if self.kernel: + cmd += ["-kernel", self.kernel] + for index, (addr, blob) in enumerate(self.segments): + seg_path = os.path.join(self.workdir, f"segment-{index:02d}-{addr:#x}.bin") + with open(seg_path, "wb") as fh: + fh.write(blob) + self._segment_files.append(seg_path) + cmd += ["-device", f"loader,file={seg_path},addr={addr:#x},force-raw=on"] + if self.dlog: + cmd += ["-d", self.dlog, "-D", self.qemu_log] + cmd += self.extra_args + return cmd + + def start(self, paused: bool = True, timeout: float = 15.0) -> "QemuTarget": + self._cmd = self._build_command(paused) + try: + self.process = subprocess.Popen( + self._cmd, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + cwd=self.workdir, + start_new_session=True, + ) + except OSError as exc: + raise TargetError(f"cannot start QEMU: {exc}") from exc + try: + self.wait_for_port(timeout=timeout) + except TargetError: + self.kill() + raise + return self + + def wait_for_port(self, timeout: float = 15.0) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if self.process is not None and self.process.poll() is not None: + stderr = b"" + if self.process.stderr is not None: + try: + stderr = self.process.stderr.read() or b"" + except OSError: + pass + raise TargetError( + f"QEMU exited early rc={self.process.returncode}: " + f"{stderr.decode('utf-8', 'replace')[-2000:]}" + ) + try: + with socket.create_connection((self.host, self.port), timeout=0.25): + return + except OSError: + time.sleep(0.05) + raise TargetError(f"gdbstub port {self.port} not ready after {timeout}s") + + def alive(self) -> bool: + return self.process is not None and self.process.poll() is None + + def kill(self, timeout: float = 5.0) -> None: + if self.process is None: + return + if self.process.poll() is None: + try: + os.killpg(os.getpgid(self.process.pid), signal.SIGTERM) + except (ProcessLookupError, PermissionError): + try: + self.process.terminate() + except ProcessLookupError: + pass + try: + self.process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + try: + os.killpg(os.getpgid(self.process.pid), signal.SIGKILL) + except (ProcessLookupError, PermissionError): + try: + self.process.kill() + except ProcessLookupError: + pass + self.process.wait(timeout=timeout) + if self.process.stderr is not None: + try: + self.process.stderr.close() + except OSError: + pass + self.process = None + + # -- observation ------------------------------------------------------- + + def read_uart(self) -> str: + try: + with open(self.uart_log, "rb") as fh: + return fh.read().decode("utf-8", "replace") + except FileNotFoundError: + return "" + + def wait_for_uart(self, marker: str, timeout: float = 5.0, interval: float = 0.05) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if marker in self.read_uart(): + return True + time.sleep(interval) + return False + + def read_qemu_log(self) -> str: + try: + with open(self.qemu_log, "rb") as fh: + return fh.read().decode("utf-8", "replace") + except FileNotFoundError: + return "" + + @property + def command_line(self) -> list[str]: + return list(self._cmd) + + def shell_command(self) -> str: + import shlex + + return " ".join(shlex.quote(part) for part in self._cmd) + + def __repr__(self) -> str: # pragma: no cover - cosmetic + return f"QemuTarget(port={self.port}, alive={self.alive()})" |