diff --git a/src/dvsim/logging.py b/src/dvsim/logging.py index 03979e4b..674b382b 100644 --- a/src/dvsim/logging.py +++ b/src/dvsim/logging.py @@ -64,7 +64,7 @@ def __init__(self, name: str) -> None: def verbose(self, msg: object, *args: object) -> None: """Log a verbose msg.""" - self.log(self.VERBOSE, msg, *args) + self.log(self.VERBOSE, msg, *args, stacklevel=2) def set_logfile( self, diff --git a/src/dvsim/scheduler/resources.py b/src/dvsim/scheduler/resources.py index 8914c204..ce00306b 100644 --- a/src/dvsim/scheduler/resources.py +++ b/src/dvsim/scheduler/resources.py @@ -10,6 +10,7 @@ from typing import Protocol from dvsim.job.data import JobSpec, ResourceMapping +from dvsim.job.status import JobStatus from dvsim.logging import log __all__ = ( @@ -19,6 +20,10 @@ ) +# Used to format logs +MIN_FIELD_WIDTH = 5 + + class ResourceProvider(Protocol): """An abstraction of something that provides resource availability info to the scheduler.""" @@ -73,6 +78,13 @@ def __init__( self._provider = provider self._missing_policy = missing_policy self._usage = defaultdict(int) + # Index tracking number of jobs with each status, per resource + self._status_counts: defaultdict[str, dict[JobStatus, int]] = defaultdict( + lambda: defaultdict(int) + ) + # The number of characters used to represent the largest field in the displayed table + self._int_field_width: int + self._str_field_width: int async def can_allocate(self, request: ResourceMapping) -> bool: """Check if a given resource request can be allocated, given current usage and limits.""" @@ -109,6 +121,29 @@ def release(self, request: ResourceMapping) -> None: if amount is not None: self._usage[resource] -= amount + def update_status_counts(self, spec: JobSpec, old: JobStatus, new: JobStatus) -> None: + """Update the index that tracks job status counts per resource.""" + for resource in spec.resources or {}: + if status_counts := self._status_counts.get(resource): + status_counts[old] = status_counts[old] - 1 + status_counts[new] = status_counts[new] + 1 + else: + log.warning("Unrecognised resource: %s", resource) + + def log_status_counts(self) -> None: + """Log job status counts per resource. + + This should be registered as a callback on job status change in Scheduler only when verbose + logging is enabled. + """ + for resource, status_counts in self._status_counts.items(): + status_count_str = ", ".join( + f"{status.value[0]}: {count:>{self._int_field_width}}" + for status, count in status_counts.items() + ) + log_msg = f"[ Resource: %-{self._str_field_width}s ] [ %s ]" + log.verbose(log_msg, resource, status_count_str) + def _log_usage(self, capacity: ResourceMapping, used: ResourceMapping) -> None: """Debug log individual job resource usage aggregates.""" if log.isEnabledFor(log.DEBUG): @@ -201,3 +236,22 @@ async def validate_jobs(self, jobs: Iterable[JobSpec]) -> None: self._log_usage(capacity, aggregate) self._emit_validation_errors(missing_resource_errors, limit_exceeded_errors) + + def init_status_counts(self, jobs: Iterable[JobSpec]) -> None: + """Initialise an index tracking the number of jobs with each status, per resource.""" + for job in jobs: + for resource in job.resources or {}: + if not self._status_counts.get(resource): + self._status_counts[resource] = dict.fromkeys(JobStatus, 0) + self._status_counts[resource][JobStatus.SCHEDULED] += 1 + + self._int_field_width = max( + ( + len(str(job_status_counts[JobStatus.SCHEDULED])) + for job_status_counts in self._status_counts.values() + ), + default=MIN_FIELD_WIDTH, + ) + self._str_field_width = max( + (len(resource) for resource in self._status_counts), default=MIN_FIELD_WIDTH + ) diff --git a/src/dvsim/scheduler/runner.py b/src/dvsim/scheduler/runner.py index 0244fff3..c2877170 100644 --- a/src/dvsim/scheduler/runner.py +++ b/src/dvsim/scheduler/runner.py @@ -8,6 +8,7 @@ import dvsim.instrumentation.runtime as instrumentation from dvsim.job.data import CompletedJobStatus, JobSpec +from dvsim.logging import log from dvsim.runtime.backend import RuntimeBackend from dvsim.runtime.fake import FakePolicy, FakeRuntimeBackend from dvsim.runtime.registry import backend_registry @@ -95,6 +96,9 @@ async def run_scheduler( # Convert to list so that first use doesn't consume the Iterable jobs = list(jobs) + if log.isEnabledFor(log.VERBOSE) and resource_manager: + resource_manager.init_status_counts(jobs) + scheduler = Scheduler( jobs=jobs, backends={backend.name: backend}, @@ -112,6 +116,14 @@ async def run_scheduler( ), ) + # Setup additional verbose logging + if log.isEnabledFor(log.VERBOSE) and resource_manager: + scheduler.add_job_status_change_callback( + lambda spec, old, new, resources=resource_manager: ( + resources.update_status_counts(spec, old, new) or resources.log_status_counts() + ) + ) + if not interactive: status_printer = create_status_printer(jobs) diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 8619c81a..9099a9b3 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -4,8 +4,10 @@ """Test the DVSim scheduler.""" +import logging import multiprocessing import os +import random import sys import threading import time @@ -22,10 +24,13 @@ from dvsim.job.data import CompletedJobStatus, JobSpec, WorkspaceConfig from dvsim.job.status import JobStatus from dvsim.launcher.base import ErrorMessage, Launcher, LauncherBusyError, LauncherError +from dvsim.logging import _VERBOSE_LOG_LEVEL from dvsim.report.data import IPMeta, ToolMeta from dvsim.runtime.legacy import LegacyLauncherAdapter from dvsim.scheduler.core import Scheduler from dvsim.scheduler.resources import ResourceManager, StaticResourceProvider +from dvsim.scheduler.runner import run_scheduler +from dvsim.tool.utils import _SUPPORTED_SIM_TOOLS __all__ = () @@ -34,6 +39,10 @@ DEFAULT_TIMEOUT = 2 SIGNAL_TEST_TIMEOUT = 5 +# Used for randomly selected simulation tool names +SEED = 42 +random.seed(SEED) + @dataclass class MockJob: @@ -255,9 +264,14 @@ def ip_meta_factory(**overrides: str | None) -> IPMeta: return IPMeta(**meta) -def tool_meta_factory(name: str = "test_tool", version: str = "test_version") -> ToolMeta: +def tool_meta_factory(name: str = "", version: str = "test_version") -> ToolMeta: """Create a ToolMeta from a set of default values, for use in testing.""" - return ToolMeta(name=name, version=version) + if name: + return ToolMeta(name=name, version=version) + return ToolMeta( + name=random.choice([*_SUPPORTED_SIM_TOOLS]), # noqa: S311 + version=version, + ) def build_workspace( @@ -1037,3 +1051,65 @@ def test_signal_kill(tmp_path: Path, *, sig: int, repeat: bool, long_poll: bool) proc.join() pytest.fail("Scheduler hung and was terminated") assert_that(proc.exitcode, equal_to(0)) + + +class TestLogging: + """Tests for the logging functionality of the scheduler.""" + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.timeout(DEFAULT_TIMEOUT) + async def test_resource_logging(fxt: Fxt, caplog: pytest.LogCaptureFixture) -> None: + """Test logging of job status counts per tool. + + This test modifies TestSchedulingPriority::test_blocked_weight_starvation + """ + start_job = job_spec_factory(fxt.tmp_path, name="start") + short_blocker = job_spec_factory(fxt.tmp_path, name="short", dependencies=["start"]) + long_blocker = job_spec_factory(fxt.tmp_path, name="long", dependencies=["start"]) + high = job_spec_factory(fxt.tmp_path, name="high", dependencies=["long"], weight=1000000) + jobs = [start_job, short_blocker, long_blocker, high] + jobs += make_many_jobs( + fxt.tmp_path, + n=5, + weight=1, + dependencies=["short"], + vary_targets=True, + ) + # Status counts are tracked per resource, so have each job use its tool as a resource + jobs = [job.model_copy(update={"resources": {job.tool.name: 1}}) for job in jobs] + fxt.mock_ctx.set_config( + short_blocker, + MockJob(status_thresholds=[(0, JobStatus.RUNNING), (1, JobStatus.PASSED)]), + ) + fxt.mock_ctx.set_config( + long_blocker, + MockJob(status_thresholds=[(0, JobStatus.RUNNING), (5, JobStatus.PASSED)]), + ) + + # Capture logging and set debug level + dvsim_logger = logging.getLogger("dvsim") + dvsim_logger.addHandler(caplog.handler) + caplog.set_level(_VERBOSE_LOG_LEVEL, logger="dvsim") + try: + result = await run_scheduler( + jobs=jobs, + max_parallel=len(jobs), + interactive=True, + backend=fxt.mock_legacy_backend, + resource_manager=ResourceManager(StaticResourceProvider({"A": 5, "B": 10})), + ) + finally: + dvsim_logger.removeHandler(caplog.handler) + + _assert_result_status(result, len(jobs)) + + expected_tools = {job.tool.name for job in jobs} + logged_tools = set() + for record in caplog.records: + if record.module == "resources" and record.levelno == _VERBOSE_LOG_LEVEL: + args = record.args + if isinstance(args, tuple) and args: + logged_tools.add(str(args[0]).strip()) + + assert_that(logged_tools, equal_to(expected_tools))