runners: improve QEMU runner (RunnerCaps, safer serial/FIFO handling, extra-args)

Author: ritesh006

scripts/west_commands/runners/qemu.py

@@ -1,31 +1,185 @@
-# Copyright (c) 2017 Linaro Limited.
-#
-# SPDX-License-Identifier: Apache-2.0
+# scripts/west_commands/runners/qemu.py
+"""
+Improved QEMU runner for Zephyr (PR-ready starter).
+- Provides RunnerCaps for run/flash/debug where applicable
+- Supports extra CLI args and qemu-binary override
+- Safer handling of temp dir, FIFOs, and PTY fallback
+- Board heuristics that can be extended via runners.yaml
+"""
+from __future__ import annotations
 
-'''Runner stub for QEMU.'''
+import os
+import shlex
+import shutil
+import subprocess
+import tempfile
+import time
+from pathlib import Path
+from typing import List, Optional
 
-from runners.core import RunnerCaps, ZephyrBinaryRunner
+from runners.core import ZephyrBinaryRunner, RunnerConfig, RunnerCaps, log
 
 
-class QemuBinaryRunner(ZephyrBinaryRunner):
-    '''Place-holder for QEMU runner customizations.'''
+class QemuRunner(ZephyrBinaryRunner):
+    name = "qemu"
+    description = "Run Zephyr ELF using QEMU (improved starter runner)"
 
-    @classmethod
-    def name(cls):
-        return 'qemu'
+    # Capabilities: this runner supports 'run' (execute ELF in qemu).
+    # Flash/debug can be added as needed.
+    CAPABILITIES = RunnerCaps(run=True)
 
     @classmethod
-    def capabilities(cls):
-        # This is a stub.
-        return RunnerCaps(commands=set())
+    def add_parser(cls, parser):
+        super().add_parser(parser)
+        parser.add_argument("--qemu-binary", help="Path to qemu-system-<arch> binary")
+        parser.add_argument("--qemu-arg", action="append", help="Extra CLI args to append to QEMU")
+        parser.add_argument("--qemu-serial", default="stdio",
+                            help="Serial backend: stdio (default), file:/path, fifo, pty")
+        parser.add_argument("--qemu-keep-tmp", action="store_true",
+                            help="Keep temporary files/dirs for debugging")
 
     @classmethod
-    def do_add_parser(cls, parser):
-        pass                    # Nothing to do.
+    def create(cls, cfg: RunnerConfig):
+        # Only create runner if an ELF exists and a qemu-capable board is selected.
+        if not cfg or not cfg.elf_file:
+            return None
+        # Optionally, check cfg.board metadata for qemu_ prefix; be permissive for now.
+        return QemuRunner(cfg)
 
-    @classmethod
-    def do_create(cls, cfg, args):
-        return QemuBinaryRunner(cfg)
+    def __init__(self, cfg: RunnerConfig):
+        super().__init__(cfg)
+        self.cfg = cfg
+        self.elf = cfg.elf_file
+        self.build_dir = Path(cfg.build_dir) if cfg.build_dir else Path.cwd()
+        self._tmpdir: Optional[Path] = None
+        self._created: List[Path] = []
+
+    def do_run(self, command: List[str], timeout: int = 0, **kwargs) -> int:
+        # Locate qemu binary
+        qemu_bin = kwargs.get("qemu_binary") or self._guess_qemu_binary()
+        if not qemu_bin or not shutil.which(qemu_bin):
+            raise RuntimeError("QEMU binary not found. Install qemu-system-* or pass --qemu-binary")
+
+        # Create temporary directory for FIFOs/PTY files
+        self._tmpdir = Path(tempfile.mkdtemp(prefix="zephyr-qemu-"))
+        log.debug("QEMU tmpdir: %s", str(self._tmpdir))
+
+        # Prepare serial/backend args
+        serial_spec = kwargs.get("qemu_serial") or "stdio"
+        serial_args = self._prepare_serial(serial_spec)
+
+        # Build base qemu command
+        qemu_cmd = [qemu_bin]
+        qemu_cmd += self._platform_flags()
+        qemu_cmd += ["-kernel", str(self.elf)]
+        qemu_cmd += serial_args
+
+        # Add extra args passed via --qemu-arg (list)
+        extra_args = kwargs.get("qemu_arg") or []
+        if isinstance(extra_args, str):
+            extra_args = shlex.split(extra_args)
+        for a in extra_args:
+            if isinstance(a, str):
+                qemu_cmd += shlex.split(a)
+
+        # Print diagnostic
+        log.inf("Running QEMU: %s", " ".join(shlex.quote(x) for x in qemu_cmd))
+
+        proc = subprocess.Popen(qemu_cmd, cwd=str(self.build_dir))
+        try:
+            # Wait (timeout==0 implies wait forever)
+            ret = proc.wait(timeout=None if timeout == 0 else timeout)
+            return ret
+        finally:
+            if not kwargs.get("qemu_keep_tmp"):
+                self._cleanup()
+
+    # Helper methods -------------------------------------------------
+
+    def _guess_qemu_binary(self) -> Optional[str]:
+        # Heuristic order: prefer 64-bit x86 qemu for qemu_x86 boards
+        candidates = ("qemu-system-x86_64", "qemu-system-x86", "qemu-system-riscv64",
+                      "qemu-system-arm", "qemu-system-aarch64")
+        for name in candidates:
+            if shutil.which(name):
+                return name
+        return None
+
+    def _platform_flags(self) -> List[str]:
+        """
+        Return a small set of default machine/memory flags depending on board name.
+        Extend this to parse runners.yaml or board metadata as needed.
+        """
+        board = (self.cfg.board or "").lower() if hasattr(self.cfg, "board") else ""
+        board_dir = str(self.cfg.board_dir) if hasattr(self.cfg, "board_dir") else ""
+
+        if "qemu_x86" in board or "qemu_x86" in board_dir:
+            # qemu_x86 default used in Zephyr's samples; these match the qemu.cmake defaults
+            return ["-M", "q35", "-m", "32", "-cpu", "qemu32,+nx,+pae", "-no-reboot", "-nographic",
+                    "-machine", "acpi=off", "-net", "none", "-icount", "shift=5,align=off,sleep=off", "-rtc", "clock=vm"]
+        # cortex-m defaults (example)
+        if "qemu_cortex_m" in board or "lm3s" in board_dir:
+            return ["-M", "lm3s6965evb", "-nographic"]
+        # fallback minimal
+        return ["-nographic"]
+
+    def _prepare_serial(self, spec: str) -> List[str]:
+        """
+        Build qemu serial args based on 'spec' (stdio, file:/path, fifo, pty).
+        Prefer chardev+serial+mon setup consistent with qemu.cmake defaults.
+        """
+        if spec == "stdio":
+            return ["-chardev", "stdio,id=con,mux=on", "-serial", "chardev:con", "-mon", "chardev=con,mode=readline"]
+        if spec.startswith("file:"):
+            path = spec.split(":", 1)[1]
+            path = os.path.expanduser(path)
+            Path(path).parent.mkdir(parents=True, exist_ok=True)
+            return ["-serial", f"file:{path}"]
+        if spec == "fifo" or spec.startswith("fifo:"):
+            return self._create_fifos()
+        if spec == "pty":
+            # QEMU's pty support on Linux usually creates a PTY automatically with -serial pty
+            return ["-serial", "pty"]
+        # default fallback
+        return ["-chardev", "stdio,id=con,mux=on", "-serial", "chardev:con", "-mon", "chardev=con,mode=readline"]
+
+    def _create_fifos(self) -> List[str]:
+        """
+        Create two FIFOs under the tmpdir. Returns a qemu -serial option referencing the output FIFO
+        (for simple log capture). This function intentionally uses file: because pipe: syntax differs.
+        """
+        if not self._tmpdir:
+            self._tmpdir = Path(tempfile.mkdtemp(prefix="zephyr-qemu-"))
+        in_fifo = self._tmpdir / "qemu-serial-in"
+        out_fifo = self._tmpdir / "qemu-serial-out"
+        try:
+            # Remove if stale, then create
+            for p in (in_fifo, out_fifo):
+                try:
+                    if p.exists():
+                        p.unlink()
+                except Exception:
+                    pass
+                os.mkfifo(str(p))
+                self._created.append(p)
+            # Return a qemu arg that writes serial output to a file FIFO (readable by user/tools)
+            return ["-serial", f"file:{str(out_fifo)}"]
+        except Exception as e:
+            log.err("Failed to create FIFOs (%s), falling back to stdio", e)
+            return ["-chardev", "stdio,id=con,mux=on", "-serial", "chardev:con", "-mon", "chardev=con,mode=readline"]
 
-    def do_run(self, command, **kwargs):
-        pass
+    def _cleanup(self):
+        # remove created files and tempdir
+        for p in self._created:
+            try:
+                if p.exists():
+                    p.unlink()
+            except Exception:
+                pass
+        if self._tmpdir:
+            try:
+                shutil.rmtree(self._tmpdir)
+            except Exception:
+                pass
+            self._tmpdir = None
+            self._created = []