feat: [launcher] add fake launcher to produce random results

This is useful when generating fake run data for developing
report templates for example. It is also useful for investigating
the scheduler behaviour without having to spawn sub processes.

With further development this could be useful for testing the scheduler.

Signed-off-by: James McCorrie <[email protected]>

Author: machshev

src/dvsim/cli.py

@@ -34,6 +34,7 @@
 from dvsim.job.deploy import RunTest
 from dvsim.launcher.base import Launcher
 from dvsim.launcher.factory import set_launcher_type
+from dvsim.launcher.fake import FakeLauncher
 from dvsim.launcher.local import LocalLauncher
 from dvsim.launcher.lsf import LsfLauncher
 from dvsim.launcher.nc import NcLauncher
@@ -785,6 +786,12 @@ def parse_args():
         help=("Print dvsim tool messages but don't actually run any command"),
     )
 
+    dvg.add_argument(
+        "--fake",
+        action="store_true",
+        help=("Use a fake launcher that generates random results"),
+    )
+
     args = parser.parse_args()
 
     if args.version:
@@ -843,7 +850,7 @@ def main() -> None:
         args.cfg = os.path.join(proj_root, cfg_path)
 
     # Add timestamp to args that all downstream objects can use.
-    curr_ts = datetime.datetime.utcnow()
+    curr_ts = datetime.datetime.now(datetime.UTC)
     args.timestamp_long = curr_ts.strftime(TS_FORMAT_LONG)
     args.timestamp = curr_ts.strftime(TS_FORMAT)
 
@@ -863,11 +870,11 @@ def main() -> None:
     LsfLauncher.max_parallel = args.max_parallel
     NcLauncher.max_parallel = args.max_parallel
     Launcher.max_odirs = args.max_odirs
-    set_launcher_type(args.local)
+    FakeLauncher.max_parallel = args.max_parallel
+    set_launcher_type(is_local=args.local, fake=args.fake)
 
     # Build infrastructure from hjson file and create the list of items to
     # be deployed.
-    global cfg
     cfg = make_cfg(args.cfg, args, proj_root)
 
     # List items available for run if --list switch is passed, and exit.

src/dvsim/launcher/factory.py

@@ -5,6 +5,7 @@
 import os
 
 from dvsim.launcher.base import Launcher
+from dvsim.launcher.fake import FakeLauncher
 from dvsim.launcher.local import LocalLauncher
 from dvsim.launcher.lsf import LsfLauncher
 from dvsim.launcher.nc import NcLauncher
@@ -23,7 +24,7 @@
 _LAUNCHER_CLS: type[Launcher] | None = None
 
 
-def set_launcher_type(is_local: bool = False) -> None:
+def set_launcher_type(is_local: bool = False, fake: bool = False) -> None:
     """Set the launcher type that will be used to launch the jobs.
 
     The env variable `DVSIM_LAUNCHER` is used to identify what launcher system
@@ -35,6 +36,8 @@ def set_launcher_type(is_local: bool = False) -> None:
     launcher = os.environ.get("DVSIM_LAUNCHER", "local")
     if is_local:
         launcher = "local"
+    if fake:
+        launcher = "fake"
     Launcher.variant = launcher
 
     global _LAUNCHER_CLS
@@ -53,6 +56,9 @@ def set_launcher_type(is_local: bool = False) -> None:
     elif launcher == "slurm":
         _LAUNCHER_CLS = SlurmLauncher
 
+    elif launcher == "fake":
+        _LAUNCHER_CLS = FakeLauncher
+
     # These custom launchers are site specific. They may not be committed to
     # the open source repo.
     elif launcher == "edacloud" and EDACLOUD_LAUNCHER_EXISTS:

src/dvsim/launcher/fake.py

@@ -0,0 +1,72 @@
+# Copyright lowRISC contributors (OpenTitan project).
+# Licensed under the Apache License, Version 2.0, see LICENSE for details.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Fake Launcher that returns random results."""
+
+from random import choice, random
+from typing import TYPE_CHECKING
+
+from dvsim.launcher.base import ErrorMessage, Launcher
+
+if TYPE_CHECKING:
+    from dvsim.job.deploy import CovReport, Deploy, RunTest
+
+
+def _run_test_handler(deploy: "RunTest") -> str:
+    """Handle a RunTest deploy job."""
+    return choice(("P", "F"))
+
+
+def _cov_report_handler(deploy: "CovReport") -> str:
+    """Handle a CompileSim deploy job."""
+    keys = [
+        "score",
+        "line",
+        "cond",
+        "toggle",
+        "fsm",
+        "branch",
+        "assert",
+        "group",
+    ]
+
+    deploy.cov_results_dict = {k: f"{random() * 100:.2f} %" for k in keys}
+
+    return "P"
+
+
+_DEPLOY_HANDLER = {
+    "RunTest": _run_test_handler,
+    "CovReport": _cov_report_handler,
+}
+
+
+class FakeLauncher(Launcher):
+    """Launch jobs and return fake results."""
+
+    # Poll job's completion status every this many seconds
+    poll_freq = 0
+
+    def __init__(self, deploy: "Deploy") -> None:
+        """Initialize common class members."""
+        super().__init__(deploy)
+
+    def _do_launch(self) -> None:
+        """Do the launch."""
+
+    def poll(self) -> str | None:
+        """Check status of the running process."""
+        deploy_cls = self.deploy.__class__.__name__
+        if deploy_cls in _DEPLOY_HANDLER:
+            return _DEPLOY_HANDLER[deploy_cls](deploy=self.deploy)
+
+        # Default result is Pass
+        return "P"
+
+    def kill(self) -> None:
+        """Kill the running process."""
+        self._post_finish(
+            "K",
+            ErrorMessage(line_number=None, message="Job killed!", context=[]),
+        )