simulator.py

Go to the documentation of this file.

#  Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
#  See https://llvm.org/LICENSE.txt for license information.
#  SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 
import json
import os
import re
import shutil
import signal
import socket
import subprocess
import time
from pathlib import Path
from typing import Dict, List, Optional, Callable, IO, Union
import threading
 
_thisdir = Path(__file__).parent
CosimCollateralDir = _thisdir
_SUPPORTED_SIMULATORS = ("verilator", "questa")
 
 
def is_port_open(port) -> bool:
  """Check if a TCP port is open locally."""
  sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  result = sock.connect_ex(('127.0.0.1', port))
  sock.close()
  return True if result == 0 else False
 
 
def supported_simulators() -> List[str]:
  """Return the simulator backends known to the ESI cosim runtime."""
  return list(_SUPPORTED_SIMULATORS)
 
 
def is_simulator_available(name: str) -> bool:
  """Return True if the requested simulator backend is usable.
 
  This checks the executable and environment conventions used by the Python
  cosim backends, so pytest callers can skip simulator-backed tests before
  generating or compiling hardware collateral.
  """
  name = name.lower()
  if name == "verilator":
    from .verilator import Verilator
    if Verilator._find_verilator_bin() is None:
      return False
    return Verilator._find_verilator_root() is not None
  if name == "questa":
    return shutil.which("vsim") is not None
  raise ValueError(f"Unknown simulator: {name}")
 
 
def available_simulators() -> List[str]:
  """Return the known simulator backends available in this environment."""
  return [
      name for name in _SUPPORTED_SIMULATORS if is_simulator_available(name)
  ]
 
 
class SourceFiles:
 
  def __init__(self, top: str) -> None:
    # User source files.
    self.user: List[Path] = []
    # DPI shared objects.
    self.dpi_so: List[str] = ["EsiCosimDpiServer"]
    # DPI SV files.
    self.dpi_sv: List[Path] = [
        CosimCollateralDir / "Cosim_DpiPkg.sv",
        CosimCollateralDir / "Cosim_Endpoint.sv",
        CosimCollateralDir / "Cosim_CycleCount.sv",
        CosimCollateralDir / "Cosim_Manifest.sv",
    ]
    # Name of the top module.
    self.top = top
 
  def add_file(self, file: Path):
    """Add a single RTL file to the source list."""
    if file.is_file():
      self.user.append(file)
    else:
      raise FileNotFoundError(f"File {file} does not exist")
 
  def add_dir(self, dir: Path):
    """Add all the RTL files in a directory to the source list. Files named in
    the directory's `filelist.f` (as written by ExportVerilog) are added first,
    in the listed order, since some simulators require dependencies such as
    packages to be compiled before their users."""
    listed: List[Path] = []
    filelist = dir / "filelist.f"
    if filelist.is_file():
      for line in filelist.read_text().splitlines():
        line = line.strip()
        if line and (dir / line).is_file():
          listed.append(dir / line)
    for file in listed + sorted(dir.iterdir()):
      if file in self.user:
        continue
      if file.is_file() and (file.suffix == ".sv" or file.suffix == ".v"):
        self.user.append(file)
      elif file.is_dir():
        self.add_dir(file)
 
  def dpi_so_paths(self) -> List[Path]:
    """Return a list of all the DPI shared object files (the loadable
    artifact: ``.so`` on POSIX, ``.dll`` on Windows)."""
    return [self._find_dpi(name, link=False) for name in self.dpi_so]
 
  def dpi_link_paths(self) -> List[Path]:
    """Return a list of files to pass to the linker for the DPI libraries.
    On POSIX this is the same as ``dpi_so_paths()``; on Windows it is the
    import library (``.lib``) sitting next to the DLL."""
    return [self._find_dpi(name, link=True) for name in self.dpi_so]
 
  def _find_dpi(self, name: str, link: bool) -> Path:
    is_windows = os.name == "nt"
 
    def check_path(p: Path) -> Optional[Path]:
      if is_windows:
        suffix = ".lib" if link else ".dll"
        cand = p / f"{name}{suffix}"
      else:
        cand = p / f"lib{name}.so"
      return cand if cand.exists() else None
 
    env = Simulator.get_env()
    if is_windows:
      search_env = env.get("PATH", "")
      env_sep = os.pathsep
    else:
      search_env = env.get("LD_LIBRARY_PATH", "")
      env_sep = ":"
    for path in search_env.split(env_sep):
      if not path:
        continue
      p = check_path(Path(path))
      if p is not None:
        return p
 
    # Check a few directories relative to this file. The build tree puts
    # libraries under ``<package>/../lib`` and ``<package>/../../lib``; wheel
    # installs put them directly in the esiaccel package dir.
    for candidate in (
        _thisdir.parent,
        _thisdir.parent / "lib",
        _thisdir.parent.parent / "lib",
    ):
      p = check_path(candidate)
      if p is not None:
        return p
 
    suffix = (".lib" if link else ".dll") if is_windows else ".so"
    raise FileNotFoundError(f"Could not find {name}{suffix}")
 
  @property
  def rtl_sources(self) -> List[Path]:
    """Return a list of all the RTL source files."""
    return self.dpi_sv + self.user
 
 
class SimProcess:
 
  def __init__(self,
               proc: subprocess.Popen,
               port: int,
               threads: Optional[List[threading.Thread]] = None,
               gui: bool = False):
    self.proc = proc
    self.port = port
    self.threads: List[threading.Thread] = threads or []
    self.gui = gui
 
  def force_stop(self):
    """Make sure to stop the simulation no matter what."""
    if self.proc:
      if os.name == "nt":
        # The child was started with CREATE_NEW_PROCESS_GROUP, so CTRL_BREAK
        # is delivered to that group only.
        try:
          self.proc.send_signal(signal.CTRL_BREAK_EVENT)
        except (OSError, ValueError):
          pass
      else:
        os.killpg(os.getpgid(self.proc.pid), signal.SIGINT)
      # Allow the simulation time to flush its outputs.
      try:
        self.proc.wait(timeout=1.0)
      except subprocess.TimeoutExpired:
        # If the simulation doesn't exit of its own free will, kill it.
        self.proc.kill()
 
    # Join reader threads (they should exit once pipes are closed).
    for t in self.threads:
      t.join()
 
 
class Simulator:
 
  CompileCommand = List[str]
  CompileFunction = Callable[[], Optional[int]]
  CompileStep = Union[CompileCommand, CompileFunction]
 
  # Some RTL simulators don't use stderr for error messages. Everything goes to
  # stdout. Boo! They should feel bad about this. Also, they can specify that
  # broken behavior by overriding this.
  UsesStderr = True
 
  def __init__(self,
               sources: SourceFiles,
               run_dir: Path,
               debug: bool,
               save_waveform: bool = False,
               run_stdout_callback: Optional[Callable[[str], None]] = None,
               run_stderr_callback: Optional[Callable[[str], None]] = None,
               compile_stdout_callback: Optional[Callable[[str], None]] = None,
               compile_stderr_callback: Optional[Callable[[str], None]] = None,
               make_default_logs: bool = True,
               macro_definitions: Optional[Dict[str, Optional[str]]] = None):
    """Simulator base class.
 
    Optional sinks can be provided for capturing output. If not provided,
    the simulator will write to log files in `run_dir`.
 
    Args:
      sources: SourceFiles describing RTL/DPI inputs.
      run_dir: Directory where build/run artifacts are placed.
      debug: Enable cosim debug mode.
      save_waveform: When True and debug=True, dump simulator waveforms to a
        waveform file. The exact format depends on the backend (e.g. FST for
        Verilator, VCD for Questa). Requires debug to be enabled.
      run_stdout_callback: Line-based callback for runtime stdout.
      run_stderr_callback: Line-based callback for runtime stderr.
      compile_stdout_callback: Line-based callback for compile stdout.
      compile_stderr_callback: Line-based callback for compile stderr.
      make_default_logs: If True and corresponding callback is not supplied,
        create log file and emit via internally-created callback.
      macro_definitions: Optional dictionary of macro definitions to be defined
        during compilation.
    """
    self.sources = sources
    self.run_dir = run_dir
    self.debug = debug
    self.save_waveform = save_waveform
    self.macro_definitions = macro_definitions
 
    # Unified list of any log file handles we opened.
    self._default_files: List[IO[str]] = []
 
    def _ensure_default(cb: Optional[Callable[[str], None]], filename: str):
      """Return (callback, file_handle_or_None) with optional file creation.
 
      Behavior:
        * If a callback is provided, return it unchanged with no file.
        * If no callback and make_default_logs is False, return (None, None).
        * If no callback and make_default_logs is True, create a log file and
          return a writer callback plus the opened file handle.
      """
      if cb is not None:
        return cb, None
      if not make_default_logs:
        return None, None
      p = self.run_dir / filename
      p.parent.mkdir(parents=True, exist_ok=True)
      logf = p.open("w+")
      self._default_files.append(logf)
 
      def _writer(line: str, _lf=logf):
        _lf.write(line + "\n")
        _lf.flush()
 
      return _writer, logf
 
    # Initialize all four (compile/run stdout/stderr) uniformly.
    self._compile_stdout_cb, self._compile_stdout_log = _ensure_default(
        compile_stdout_callback, 'compile_stdout.log')
    self._compile_stderr_cb, self._compile_stderr_log = _ensure_default(
        compile_stderr_callback, 'compile_stderr.log')
    self._run_stdout_cb, self._run_stdout_log = _ensure_default(
        run_stdout_callback, 'sim_stdout.log')
    self._run_stderr_cb, self._run_stderr_log = _ensure_default(
        run_stderr_callback, 'sim_stderr.log')
 
  @staticmethod
  def get_env() -> Dict[str, str]:
    """Get the environment variables to locate shared objects."""
 
    env = os.environ.copy()
    # Directories that may contain the ESI runtime / cosim DPI shared
    # libraries (build tree and wheel install layouts).
    lib_dirs = [
        str(_thisdir.parent),
        str(_thisdir.parent / "lib"),
        str(_thisdir.parent.parent / "lib"),
    ]
    sep = os.pathsep
    if os.name == "nt":
      # Windows resolves DLL loads via PATH (and the loader's own search
      # order). Make sure both the package dir (wheel layout) and any
      # build-tree ``lib`` dirs are visible.
      env["PATH"] = sep.join(lib_dirs) + sep + env.get("PATH", "")
    else:
      env["LIBRARY_PATH"] = env.get("LIBRARY_PATH",
                                    "") + ":" + ":".join(lib_dirs)
      env["LD_LIBRARY_PATH"] = env.get("LD_LIBRARY_PATH",
                                       "") + ":" + ":".join(lib_dirs)
    return env
 
  def compile_commands(self) -> List[CompileStep]:
    """Return the compile steps for the simulator.
 
    Each step may be either a shell command (`List[str]`) or a Python callback
    (`Callable[[], Optional[int]]`). Python callbacks should return `0` or
    `None` on success and a non-zero integer on failure.
    """
    assert False, "Must be implemented by subclass"
 
  def _run_compile_command(self, cmd: CompileCommand) -> int:
    ret = self._start_process_with_callbacks(cmd,
                                             env=Simulator.get_env(),
                                             cwd=None,
                                             stdout_cb=self._compile_stdout_cb,
                                             stderr_cb=self._compile_stderr_cb,
                                             wait=True)
    if isinstance(ret, int) and ret != 0:
      print("====== Compilation failure")
 
      # Always print both stdout and stderr so that linker errors (which go
      # to stdout for cmake/ninja) are not silently hidden.
      if self._compile_stdout_log is not None:
        self._compile_stdout_log.seek(0)
        stdout_content = self._compile_stdout_log.read()
        if stdout_content:
          print(stdout_content)
      if self._compile_stderr_log is not None:
        self._compile_stderr_log.seek(0)
        stderr_content = self._compile_stderr_log.read()
        if stderr_content:
          print(stderr_content)
 
    return ret if isinstance(ret, int) else 1
 
  def _run_compile_step(self, step: CompileStep) -> int:
    if callable(step):
      ret = step()
      if ret is None:
        return 0
      if not isinstance(ret, int):
        raise TypeError("compile step callback must return int or None")
      return ret
    return self._run_compile_command(step)
 
  def compile(self) -> int:
    cmds = self.compile_commands()
    self.run_dir.mkdir(parents=True, exist_ok=True)
    for step in cmds:
      ret = self._run_compile_step(step)
      if ret != 0:
        return ret
    return 0
 
  def run_command(self, gui: bool) -> List[str]:
    """Return the command to run the simulation."""
    assert False, "Must be implemented by subclass"
 
  @property
  def waveform_extension(self) -> str:
    """File extension for waveform dumps.
 
    Subclasses should override if their format differs.  The Verilator C++
    driver writes FST (via ``VerilatedFstC``); the generic SV driver uses
    ``$dumpfile/$dumpvars`` which produces VCD.
    """
    return ".vcd"
 
  def run_proc(self, gui: bool = False) -> SimProcess:
    """Run the simulation process. Returns the Popen object and the port which
      the simulation is listening on.
 
    If user-provided stdout/stderr sinks were supplied in the constructor,
    they are used. Otherwise, log files are created in `run_dir`.
    """
    self.run_dir.mkdir(parents=True, exist_ok=True)
 
    env_gui = os.environ.get("COSIM_GUI", "0")
    if env_gui != "0":
      gui = True
 
    # Erase the config file if it exists. We don't want to read
    # an old config.
    portFileName = self.run_dir / "cosim.cfg"
    if os.path.exists(portFileName):
      os.remove(portFileName)
 
    # Run the simulation.
    simEnv = Simulator.get_env()
    if self.debug:
      debug_file = (self.run_dir / "cosim_debug.log").resolve()
      simEnv["COSIM_DEBUG_FILE"] = str(debug_file)
      if "DEBUG_PERIOD" not in simEnv:
        # Slow the simulation down to one tick per millisecond.
        simEnv["DEBUG_PERIOD"] = "1"
      if self.save_waveform:
        waveform_file = (self.run_dir /
                         f"cosim_waveform{self.waveform_extension}").resolve()
        simEnv["SAVE_WAVE"] = str(waveform_file)
    rcmd = self.run_command(gui)
    # Start process with asynchronous output capture.
    proc, threads = self._start_process_with_callbacks(
        rcmd,
        env=simEnv,
        cwd=self.run_dir,
        stdout_cb=self._run_stdout_cb,
        stderr_cb=self._run_stderr_cb,
        wait=False)
 
    # Get the port which the simulation RPC selected.
    checkCount = 0
    while (not os.path.exists(portFileName)) and \
            proc.poll() is None:
      time.sleep(0.1)
      checkCount += 1
      if checkCount > 500 and not gui:
        raise Exception(f"Cosim never wrote cfg file: {portFileName}")
    port = -1
    while port < 0:
      portFile = open(portFileName, "r")
      for line in portFile.readlines():
        m = re.match("port: (\\d+)", line)
        if m is not None:
          port = int(m.group(1))
      portFile.close()
 
    # The cosim server writes ``cosim.cfg`` after its TCP listen socket is
    # bound and the accept thread has been spawned. So we don't need to wait for
    # the port to be opened.
    return SimProcess(proc=proc, port=port, threads=threads, gui=gui)
 
  def _start_process_with_callbacks(
      self, cmd: List[str], env: Optional[Dict[str, str]], cwd: Optional[Path],
      stdout_cb: Optional[Callable[[str],
                                   None]], stderr_cb: Optional[Callable[[str],
                                                                        None]],
      wait: bool) -> int | tuple[subprocess.Popen, List[threading.Thread]]:
    """Start a subprocess and stream its stdout/stderr to callbacks.
 
    If wait is True, blocks until process completes and returns its exit code.
    If wait is False, returns the Popen object (threads keep streaming).
    """
    if os.name == "posix":
      proc = subprocess.Popen(cmd,
                              stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE,
                              env=env,
                              cwd=cwd,
                              text=True,
                              preexec_fn=os.setsid)
    else:  # windows
      proc = subprocess.Popen(cmd,
                              stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE,
                              env=env,
                              cwd=cwd,
                              text=True,
                              creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
 
    def _reader(pipe, cb):
      if pipe is None:
        return
      for raw in pipe:
        if raw.endswith('\n'):
          raw = raw[:-1]
        if cb:
          try:
            cb(raw)
          except Exception as e:
            print(f"Exception in simulator output callback: {e}")
 
    threads: List[threading.Thread] = [
        threading.Thread(target=_reader,
                         args=(proc.stdout, stdout_cb),
                         daemon=True),
        threading.Thread(target=_reader,
                         args=(proc.stderr, stderr_cb),
                         daemon=True),
    ]
    for t in threads:
      t.start()
    if wait:
      for t in threads:
        t.join()
      return proc.wait()
    return proc, threads
 
  def run(self,
          inner_command: str,
          gui: bool = False,
          server_only: bool = False) -> int:
    """Start the simulation then run the command specified. Kill the simulation
    when the command exits."""
 
    # 'simProc' is accessed in the finally block. Declare it here to avoid
    # syntax errors in that block.
    simProc = None
    try:
      simProc = self.run_proc(gui=gui)
      if server_only:
        # wait for user input to kill the server
        input(
            f"Running in server-only mode on port {simProc.port} - Press anything to kill the server..."
        )
        return 0
      else:
        # Run the inner command, passing the connection info via environment vars.
        testEnv = os.environ.copy()
        testEnv["ESI_COSIM_PORT"] = str(simProc.port)
        testEnv["ESI_COSIM_HOST"] = "localhost"
        ret = subprocess.run(inner_command, cwd=os.getcwd(),
                             env=testEnv).returncode
        if simProc.gui:
          print("GUI mode - waiting for simulator to exit...")
          simProc.proc.wait()
        return ret
    finally:
      if simProc and simProc.proc.poll() is None:
        simProc.force_stop()
 
 
def load_macro_definitions(path: Path) -> Dict[str, Optional[str]]:
  """Read RTL macro definitions from a JSON file.
 
  The file must hold an object mapping macro name to value, where null defines
  the macro without assigning one::
 
      {"SPB_256BIT_DATA": "1", "SYNTHESIS": null}
 
  This lets a source generator that only discovers its macros while running --
  a build system reading its own project description, for instance -- hand them
  to the simulator without the caller having to know them up front.
 
  Args:
    path: The JSON file to read.
 
  Returns:
    The macro mapping, suitable for `get_simulator`'s `macro_definitions`.
 
  Raises:
    FileNotFoundError: If *path* does not exist.
    ValueError: If *path* is not a JSON object of name/value pairs.
  """
  if not path.is_file():
    raise FileNotFoundError(f"Macro definitions file not found: {path}")
 
  try:
    definitions = json.loads(path.read_text())
  except ValueError as e:
    raise ValueError(f"{path} is not valid JSON: {e}") from e
 
  if not isinstance(definitions, dict):
    raise ValueError(f"{path} must contain a JSON object mapping macro name "
                     f"to value, got {type(definitions).__name__}")
 
  macros: Dict[str, Optional[str]] = {}
  for name, value in definitions.items():
    if value is not None and not isinstance(value, (str, int, float, bool)):
      raise ValueError(f"{path}: macro '{name}' has a "
                       f"{type(value).__name__} value; expected a scalar or "
                       f"null")
    macros[str(name)] = None if value is None else str(value)
  return macros
 
 
def get_simulator(
    name: str,
    sources: SourceFiles,
    rundir: Path,
    debug: bool,
    save_waveform: bool = False,
    macro_definitions: Optional[Dict[str, Optional[str]]] = None) -> Simulator:
  """Create a simulator backend.
 
  Args:
    name: Simulator backend name.
    sources: SourceFiles describing RTL/DPI inputs.
    rundir: Directory where build/run artifacts are placed.
    debug: Enable cosim debug mode.
    save_waveform: Save simulator waveforms when debug mode is enabled.
    macro_definitions: Optional mapping of macro names to values. A value of
      None defines the macro without assigning a value.
  """
  name = name.lower()
  if name == "verilator":
    from .verilator import Verilator
    return Verilator(sources,
                     rundir,
                     debug,
                     save_waveform=save_waveform,
                     macro_definitions=macro_definitions)
  elif name == "questa":
    from .questa import Questa
    return Questa(sources,
                  rundir,
                  debug,
                  save_waveform=save_waveform,
                  macro_definitions=macro_definitions)
  else:
    raise ValueError(f"Unknown simulator: {name}")