Skip to content

Add a Job Status enum + scheduler refactoring #92

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
AlexJones0 wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Add a Job Status enum + scheduler refactoring #92

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
41ccfdb
refactor: replace assert with RuntimeError
AlexJones0 Jan 29, 2026
232b6f0
refactor: resolve scheduler complexity lint warnings
AlexJones0 Jan 29, 2026
18d5131
fix: resolve JobSpec/str type error
AlexJones0 Jan 29, 2026
6b260bb
fix: Replace 'E' local launcher poll status with a LauncherError
AlexJones0 Feb 5, 2026
403977e
refactor: make Job Status an Enum
AlexJones0 Jan 30, 2026
931f4b7
refactor: Launcher poll always returns a status
AlexJones0 Feb 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions src/dvsim/flow/formal.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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")
Expand Down
9 changes: 5 additions & 4 deletions src/dvsim/flow/one_shot.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
9 changes: 5 additions & 4 deletions src/dvsim/job/data.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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]
Expand Down Expand Up @@ -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."""
17 changes: 9 additions & 8 deletions src/dvsim/job/deploy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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().
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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)
Expand Down
29 changes: 29 additions & 0 deletions src/dvsim/job/status.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
51 changes: 26 additions & 25 deletions src/dvsim/launcher/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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.

"""

Expand Down Expand Up @@ -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)
Expand All @@ -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=[],
Expand Down Expand Up @@ -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],
Expand All @@ -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
Expand All @@ -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
Expand Down
15 changes: 8 additions & 7 deletions src/dvsim/launcher/fake.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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.
Expand All @@ -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 = {
Expand All @@ -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=[]),
)

Expand Down
Loading
Loading