Port the status printers to use the new async scheduler

src/dvsim/cli/run.py

@@ -43,6 +43,7 @@
 from dvsim.launcher.sge import SgeLauncher
 from dvsim.launcher.slurm import SlurmLauncher
 from dvsim.logging import LOG_LEVELS, configure_logging, log
+from dvsim.scheduler.async_status_printer import StatusPrinter
 from dvsim.scheduler.status_printer import get_status_printer
 from dvsim.utils import TS_FORMAT, TS_FORMAT_LONG, Timer, rm_path, run_cmd_with_timeout
 
@@ -884,6 +885,7 @@ def main(argv: list[str] | None = None) -> None:
 
     # Register the common deploy settings.
     Timer.print_interval = args.print_interval
+    StatusPrinter.print_interval = args.print_interval
     LocalLauncher.max_parallel = args.max_parallel
     SlurmLauncher.max_parallel = args.max_parallel
     SgeLauncher.max_parallel = args.max_parallel

src/dvsim/instrumentation/resources.py

@@ -227,7 +227,7 @@ def on_job_status_change(self, job: JobSpec, status: JobStatus) -> None:
         with self._lock:
             running = job_id in self._running_jobs
             started = running or job_id in self._finished_jobs
-            if not started and status != JobStatus.QUEUED:
+            if not started and status not in (JobStatus.SCHEDULED, JobStatus.QUEUED):
                 self._running_jobs[job_id] = JobResourceAggregate(job)
                 running = True
             if running and status.is_terminal:

src/dvsim/instrumentation/timing.py

@@ -99,7 +99,7 @@ def on_job_status_change(self, job: JobSpec, status: JobStatus) -> None:
             job_info = TimingJobFragment(job)
             self._jobs[job_id] = job_info
 
-        if job_info.start_time is None and status != JobStatus.QUEUED:
+        if job_info.start_time is None and status not in (JobStatus.SCHEDULED, JobStatus.QUEUED):
             job_info.start_time = time.perf_counter()
         if status.is_terminal:
             job_info.end_time = time.perf_counter()

src/dvsim/job/status.py

@@ -12,11 +12,14 @@
 class JobStatus(Enum):
     """Status of a Job."""
 
-    QUEUED = auto()
-    RUNNING = auto()
-    PASSED = auto()
-    FAILED = auto()
-    KILLED = auto()
+    # SCHEDULED is currently unused in the old sync scheduler, there `SCHEDULED` and `QUEUED`
+    # are combined under `QUEUED`. It is intended to be used in the new async scheduler.
+    SCHEDULED = auto()  # Waiting for dependencies
+    QUEUED = auto()  # Dependencies satisfied, waiting to be dispatched
+    RUNNING = auto()  # Dispatched to a backend and actively executing
+    PASSED = auto()  # Completed successfully
+    FAILED = auto()  # Completed with failure
+    KILLED = auto()  # Forcibly terminated or never executed
 
     @property
     def shorthand(self) -> str: