Add a Job Status enum + scheduler refactoring

src/dvsim/flow/formal.py

@@ -10,6 +10,7 @@
 
 from dvsim.flow.one_shot import OneShotCfg
 from dvsim.job.data import CompletedJobStatus
+from dvsim.job.status import JobStatus
 from dvsim.logging import log
 from dvsim.utils import subst_wildcards
 
@@ -230,7 +231,7 @@ def _gen_results(self, results: Sequence[CompletedJobStatus]) -> None:
         assert len(self.deploy) == 1
         mode = self.deploy[0]
 
-        if complete_job.status == "P":
+        if complete_job.status == JobStatus.PASSED:
             result_data = Path(
                 subst_wildcards(self.build_dir, {"build_mode": mode.name}),
                 "results.hjson",
@@ -262,7 +263,7 @@ def _gen_results(self, results: Sequence[CompletedJobStatus]) -> None:
         else:
             summary += ["N/A", "N/A", "N/A"]
 
-        if complete_job.status != "P":
+        if complete_job.status != JobStatus.PASSED:
             results_str += "\n## List of Failures\n" + "".join(complete_job.fail_msg.message)
 
         messages = self.result.get("messages")

src/dvsim/flow/one_shot.py

@@ -12,6 +12,7 @@
 from dvsim.flow.base import FlowCfg
 from dvsim.job.data import CompletedJobStatus
 from dvsim.job.deploy import CompileOneShot
+from dvsim.job.status import JobStatus
 from dvsim.logging import log
 from dvsim.modes import BuildMode, Mode
 from dvsim.utils import rm_path
@@ -92,10 +93,10 @@ def _expand(self) -> None:
 
             # Set directories with links for ease of debug / triage.
             self.links = {
-                "D": self.scratch_path + "/" + "dispatched",
-                "P": self.scratch_path + "/" + "passed",
-                "F": self.scratch_path + "/" + "failed",
-                "K": self.scratch_path + "/" + "killed",
+                JobStatus.DISPATCHED: self.scratch_path + "/" + "dispatched",
+                JobStatus.PASSED: self.scratch_path + "/" + "passed",
+                JobStatus.FAILED: self.scratch_path + "/" + "failed",
+                JobStatus.KILLED: self.scratch_path + "/" + "killed",
             }
 
             # Use the default build mode for tests that do not specify it

src/dvsim/job/data.py

@@ -14,6 +14,7 @@
 
 from pydantic import BaseModel, ConfigDict
 
+from dvsim.job.status import JobStatus
 from dvsim.launcher.base import ErrorMessage, Launcher
 from dvsim.report.data import IPMeta, ToolMeta
 
@@ -97,13 +98,13 @@ class JobSpec(BaseModel):
     """Output directory for the job results files."""
     log_path: Path
     """Path for the job log file."""
-    links: Mapping[str, Path]
+    links: Mapping[JobStatus, Path]
     """Path for links directories."""
 
     # TODO: remove the need for these callables here
     pre_launch: Callable[[Launcher], None]
     """Callback function for pre-launch actions."""
-    post_finish: Callable[[str], None]
+    post_finish: Callable[[JobStatus], None]
     """Callback function for tidy up actions once the job is finished."""
 
     pass_patterns: Sequence[str]
@@ -153,7 +154,7 @@ class CompletedJobStatus(BaseModel):
     simulated_time: float
     """Simulation time."""
 
-    status: str
-    """Job status string [P,F,K,...]"""
+    status: JobStatus
+    """Status of the job."""
     fail_msg: ErrorMessage
     """Error message."""

src/dvsim/job/deploy.py

@@ -14,6 +14,7 @@
 from tabulate import tabulate
 
 from dvsim.job.data import JobSpec, WorkspaceConfig
+from dvsim.job.status import JobStatus
 from dvsim.job.time import JobTime
 from dvsim.launcher.base import Launcher
 from dvsim.logging import log
@@ -347,10 +348,10 @@ def callback(launcher: Launcher) -> None:
 
         return callback
 
-    def post_finish(self) -> Callable[[str], None]:
+    def post_finish(self) -> Callable[[JobStatus], None]:
         """Get post finish callback."""
 
-        def callback(status: str) -> None:
+        def callback(status: JobStatus) -> None:
             """Perform additional post-finish activities (callback).
 
             This is invoked by launcher::_post_finish().
@@ -641,12 +642,12 @@ def callback(launcher: Launcher) -> None:
 
         return callback
 
-    def post_finish(self) -> Callable[[str], None]:
+    def post_finish(self) -> Callable[[JobStatus], None]:
         """Get post finish callback."""
 
-        def callback(status: str) -> None:
+        def callback(status: JobStatus) -> None:
             """Perform tidy up tasks."""
-            if status != "P":
+            if status != JobStatus.PASSED:
                 # Delete the coverage data if available.
                 rm_path(self.cov_db_test_dir)
 
@@ -812,16 +813,16 @@ def _set_attrs(self) -> None:
         self.cov_results = ""
         self.cov_results_dict = {}
 
-    def post_finish(self) -> Callable[[str], None]:
+    def post_finish(self) -> Callable[[JobStatus], None]:
         """Get post finish callback."""
 
-        def callback(status: str) -> None:
+        def callback(status: JobStatus) -> None:
             """Extract the coverage results summary for the dashboard.
 
             If the extraction fails, an appropriate exception is raised, which must
             be caught by the caller to mark the job as a failure.
             """
-            if self.dry_run or status != "P":
+            if self.dry_run or status != JobStatus.PASSED:
                 return
 
             plugin = get_sim_tool_plugin(tool=self.sim_cfg.tool)

src/dvsim/job/status.py

@@ -0,0 +1,29 @@
+# Copyright lowRISC contributors (OpenTitan project).
+# Licensed under the Apache License, Version 2.0, see LICENSE for details.
+# SPDX-License-Identifier: Apache-2.0
+
+"""An enum definition for the various job statuses."""
+
+from enum import Enum
+
+__all__ = ("JobStatus",)
+
+
+class JobStatus(Enum):
+    """Status of a Job."""
+
+    QUEUED = 0
+    DISPATCHED = 1
+    PASSED = 2
+    FAILED = 3
+    KILLED = 4
+
+    @property
+    def shorthand(self) -> str:
+        """Shorthand for the job status, e.g. 'D' for 'Dispatched'."""
+        return self.name[0]
+
+    @property
+    def ended(self) -> bool:
+        """Whether this status corresponds to some ended job."""
+        return self in (JobStatus.PASSED, JobStatus.FAILED, JobStatus.KILLED)

src/dvsim/launcher/base.py

@@ -15,6 +15,7 @@
 
 from pydantic import BaseModel, ConfigDict
 
+from dvsim.job.status import JobStatus
 from dvsim.job.time import JobTime
 from dvsim.logging import log
 from dvsim.tool.utils import get_sim_tool_plugin
@@ -118,7 +119,7 @@ def __init__(self, job_spec: "JobSpec") -> None:
         # _check_status() method, but eventually updated by the _post_finish()
         # method, in case any of the cleanup tasks fails. This value is finally
         # returned to the Scheduler by the poll() method.
-        self.status = None
+        self.status = JobStatus.QUEUED
 
         # Return status of the process running the job.
         self.exit_code = None
@@ -210,18 +211,18 @@ def _make_odir(self) -> None:
 
         Path(self.job_spec.odir).mkdir(exist_ok=True, parents=True)
 
-    def _link_odir(self, status: str) -> None:
+    def _link_odir(self, status: JobStatus) -> None:
         """Soft-links the job's directory based on job's status.
 
-        The dispatched, passed and failed directories in the scratch area
+        The DISPATCHED, PASSED and FAILED directories in the scratch area
         provide a quick way to get to the job that was executed.
         """
         dest = Path(self.job_spec.links[status], self.job_spec.qual_name)
         mk_symlink(path=self.job_spec.odir, link=dest)
 
         # Delete the symlink from dispatched directory if it exists.
-        if status != "D":
-            old = Path(self.job_spec.links["D"], self.job_spec.qual_name)
+        if status != JobStatus.DISPATCHED:
+            old = Path(self.job_spec.links[JobStatus.DISPATCHED], self.job_spec.qual_name)
             rm_path(old)
 
     def _dump_env_vars(self, exports: Mapping[str, str]) -> None:
@@ -258,28 +259,28 @@ def launch(self) -> None:
         self._do_launch()
 
     @abstractmethod
-    def poll(self) -> str | None:
+    def poll(self) -> JobStatus:
         """Poll the launched job for completion.
 
         Invokes _check_status() and _post_finish() when the job completes.
 
         Returns:
-            status of the job or None
+            status of the job
 
         """
 
     @abstractmethod
     def kill(self) -> None:
         """Terminate the job."""
 
-    def _check_status(self) -> tuple[str, ErrorMessage | None]:
-        """Determine the outcome of the job (P/F if it ran to completion).
+    def _check_status(self) -> tuple[JobStatus, ErrorMessage | None]:
+        """Determine the outcome of the job (PASSED/FAILED if it ran to completion).
 
         Returns:
             (status, err_msg) extracted from the log, where the status is
-            "P" if the it passed, "F" otherwise. This is invoked by poll() just
-            after the job finishes. err_msg is an instance of the named tuple
-            ErrorMessage.
+            PASSED if the job passed, FAILED otherwise. This is invoked by
+            poll() just after the job finishes. err_msg is an instance of the
+            named tuple ErrorMessage.
 
         """
 
@@ -307,7 +308,7 @@ def _find_patterns(patterns: Sequence[str], line: str) -> Sequence[str] | None:
             return None
 
         if self.job_spec.dry_run:
-            return "P", None
+            return JobStatus.PASSED, None
 
         # Only one fail pattern needs to be seen.
         chk_failed = bool(self.job_spec.fail_patterns)
@@ -324,7 +325,7 @@ def _find_patterns(patterns: Sequence[str], line: str) -> Sequence[str] | None:
             ) as f:
                 lines = f.readlines()
         except OSError as e:
-            return "F", ErrorMessage(
+            return JobStatus.FAILED, ErrorMessage(
                 line_number=None,
                 message=f"Error opening file {self.job_spec.log_path}:\n{e}",
                 context=[],
@@ -368,7 +369,7 @@ def _find_patterns(patterns: Sequence[str], line: str) -> Sequence[str] | None:
                     # If failed, then nothing else to do. Just return.
                     # Provide some extra lines for context.
                     end = cnt + 5
-                    return "F", ErrorMessage(
+                    return JobStatus.FAILED, ErrorMessage(
                         line_number=cnt + 1,
                         message=line.strip(),
                         context=lines[cnt:end],
@@ -384,32 +385,32 @@ def _find_patterns(patterns: Sequence[str], line: str) -> Sequence[str] | None:
         # exit code for whatever reason, then show the last 10 lines of the log
         # as the failure message, which might help with the debug.
         if self.exit_code != 0:
-            return "F", ErrorMessage(
+            return JobStatus.FAILED, ErrorMessage(
                 line_number=None,
                 message="Job returned non-zero exit code",
                 context=lines[-10:],
             )
         if chk_passed:
-            return "F", ErrorMessage(
+            return JobStatus.FAILED, ErrorMessage(
                 line_number=None,
                 message=f"Some pass patterns missing: {pass_patterns}",
                 context=lines[-10:],
             )
-        return "P", None
+        return JobStatus.PASSED, None
 
-    def _post_finish(self, status: str, err_msg: ErrorMessage) -> None:
+    def _post_finish(self, status: JobStatus, err_msg: ErrorMessage) -> None:
         """Do post-completion activities, such as preparing the results.
 
         Must be invoked by poll(), after the job outcome is determined.
 
         Args:
-            status: status of the job, either 'P', 'F' or 'K'.
+            status: status of the completed job (must be either PASSED, FAILED or KILLED).
             err_msg: an instance of the named tuple ErrorMessage.
 
         """
-        assert status in ["P", "F", "K"]
+        assert status.ended
         self._link_odir(status)
-        log.debug("Item %s has completed execution: %s", self, status)
+        log.debug("Item %s has completed execution: %s", self, status.shorthand)
 
         try:
             # Run the target-specific cleanup tasks regardless of the job's
@@ -419,16 +420,16 @@ def _post_finish(self, status: str, err_msg: ErrorMessage) -> None:
         except Exception as e:
             # If the job had already failed, then don't do anything. If it's
             # cleanup task failed, then mark the job as failed.
-            if status == "P":
-                status = "F"
+            if status == JobStatus.PASSED:
+                status = JobStatus.FAILED
                 err_msg = ErrorMessage(
                     line_number=None,
                     message=f"{e}",
                     context=[f"{e}"],
                 )
 
         self.status = status
-        if self.status != "P":
+        if self.status != JobStatus.PASSED:
             assert err_msg
             assert isinstance(err_msg, ErrorMessage)
             self.fail_msg = err_msg

src/dvsim/launcher/fake.py

@@ -7,6 +7,7 @@
 from random import choice
 from typing import TYPE_CHECKING
 
+from dvsim.job.status import JobStatus
 from dvsim.launcher.base import ErrorMessage, Launcher
 
 if TYPE_CHECKING:
@@ -16,12 +17,12 @@
 __all__ = ("FakeLauncher",)
 
 
-def _run_test_handler(job_spec: "JobSpec") -> str:
+def _run_test_handler(job_spec: "JobSpec") -> JobStatus:
     """Handle a RunTest deploy job."""
-    return choice(("P", "F"))
+    return choice((JobStatus.PASSED, JobStatus.FAILED))
 
 
-def _cov_report_handler(job_spec: "JobSpec") -> str:
+def _cov_report_handler(job_spec: "JobSpec") -> JobStatus:
     """Handle a CompileSim deploy job."""
     # TODO: this hack doesn't work any more and needs implementing by writing
     # a file that can be parsed as if it's been generated by the tool.
@@ -38,7 +39,7 @@ def _cov_report_handler(job_spec: "JobSpec") -> str:
     # ]
     # job_spec.cov_results_dict = {k: f"{random() * 100:.2f} %" for k in keys}
 
-    return "P"
+    return JobStatus.PASSED
 
 
 _DEPLOY_HANDLER = {
@@ -56,19 +57,19 @@ class FakeLauncher(Launcher):
     def _do_launch(self) -> None:
         """Do the launch."""
 
-    def poll(self) -> str | None:
+    def poll(self) -> JobStatus:
         """Check status of the running process."""
         deploy_cls = self.job_spec.job_type
         if deploy_cls in _DEPLOY_HANDLER:
             return _DEPLOY_HANDLER[deploy_cls](job_spec=self.job_spec)
 
         # Default result is Pass
-        return "P"
+        return JobStatus.PASSED
 
     def kill(self) -> None:
         """Kill the running process."""
         self._post_finish(
-            "K",
+            JobStatus.KILLED,
             ErrorMessage(line_number=None, message="Job killed!", context=[]),
         )