Switch to use the new scheduler & remove old unused Scheduler & Launcher logic

src/dvsim/cli/run.py

@@ -35,20 +35,15 @@
 from dvsim.instrumentation import InstrumentationFactory, set_instrumentation
 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
 from dvsim.launcher.sge import SgeLauncher
 from dvsim.launcher.slurm import SlurmLauncher
 from dvsim.logging import LOG_LEVELS, configure_logging, log
 from dvsim.runtime.backend import RuntimeBackend
 from dvsim.runtime.registry import BackendType, backend_registry
-from dvsim.scheduler.async_status_printer import StatusPrinter
-from dvsim.scheduler.async_status_printer import get_status_printer as get_async_status_printer
-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
+from dvsim.scheduler.status_printer import StatusPrinter, get_status_printer
+from dvsim.utils import TS_FORMAT, TS_FORMAT_LONG, rm_path, run_cmd_with_timeout
 
 # The different categories that can be passed to the --list argument.
 _LIST_CATEGORIES = ["build_modes", "run_modes", "tests", "regressions"]
@@ -785,10 +780,11 @@ def parse_args(argv: list[str] | None = None):
     dvg.add_argument(
         "--print-interval",
         "-pi",
-        type=int,
+        type=float,
         default=10,
         metavar="N",
-        help="Print status every N seconds.",
+        help="Print status every N seconds (default %(default)d). A zero value means that every"
+        " job status change will cause a print.",
     )
 
     dvg.add_argument(
@@ -817,17 +813,8 @@ def parse_args(argv: list[str] | None = None):
         help=("Use a fake launcher that generates random results"),
     )
 
-    dvg.add_argument(
-        "--experimental-enable-async-scheduler",
-        action="store_true",
-        help="Enable experimental use of the async scheduler (not fully integrated).",
-    )
-
     args = parser.parse_args(argv) if argv else parser.parse_args()
 
-    if args.experimental_enable_async_scheduler:
-        log.warning("DVSim configured to use new experimental async scheduler.")
-
     # Check conflicts
     # interactive and remote, r
     if args.interactive and args.remote:
@@ -926,19 +913,15 @@ def main(argv: list[str] | None = None) -> None:
     RunTest.fixed_seed = args.fixed_seed
 
     # 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
     LsfLauncher.max_parallel = args.max_parallel
     NcLauncher.max_parallel = args.max_parallel
     Launcher.max_odirs = args.max_odirs
     RuntimeBackend.max_output_dirs = args.max_odirs
-    FakeLauncher.max_parallel = args.max_parallel
-    set_launcher_type(is_local=args.local, fake=args.fake)
 
-    # Configure the runtime backend. TODO: deprecate `set_launcher_type` above.
+    # Configure the runtime backend.
     set_backend_type(is_local=args.local, fake=args.fake)
 
     # Configure scheduler instrumentation
@@ -983,12 +966,8 @@ def main(argv: list[str] | None = None) -> None:
         # Now that we have printed the results from the scheduler, we close the
         # status printer, to ensure the status remains relevant in the UI context
         # (for applicable status printers).
-        if args.experimental_enable_async_scheduler:
-            if not args.interactive:
-                status_printer = get_async_status_printer()
-                status_printer.exit()
-        else:
-            status_printer = get_status_printer(args.interactive)
+        if not args.interactive:
+            status_printer = get_status_printer()
             status_printer.exit()
 
     else:

src/dvsim/flow/base.py

@@ -20,14 +20,12 @@
 from dvsim.flow.hjson import set_target_attribute
 from dvsim.job.data import CompletedJobStatus, JobSpec
 from dvsim.job.status import JobStatus
-from dvsim.launcher.factory import get_launcher_cls
 from dvsim.logging import log
 from dvsim.runtime.fake import FakeRuntimeBackend
 from dvsim.runtime.registry import backend_registry
-from dvsim.scheduler.async_core import Scheduler as AsyncScheduler
-from dvsim.scheduler.async_status_printer import create_status_printer
 from dvsim.scheduler.core import Scheduler
 from dvsim.scheduler.log_manager import LogManager
+from dvsim.scheduler.status_printer import create_status_printer
 from dvsim.utils import (
     find_and_substitute_wildcards,
     rm_path,
@@ -465,14 +463,7 @@ def deploy_objects(self) -> Sequence[CompletedJobStatus]:
                 ),
             )
 
-        if self.args.experimental_enable_async_scheduler:
-            return asyncio.run(self.run_scheduler(jobs))
-
-        return Scheduler(
-            items=jobs,
-            launcher_cls=get_launcher_cls(),
-            interactive=self.interactive,
-        ).run()
+        return asyncio.run(self.run_scheduler(jobs))
 
     async def run_scheduler(self, jobs: list[JobSpec]) -> list[CompletedJobStatus]:
         """Run the scheduler with the given set of job specifications."""
@@ -485,7 +476,7 @@ async def run_scheduler(self, jobs: list[JobSpec]) -> list[CompletedJobStatus]:
 
         max_timeout = max((job.timeout_mins for job in jobs if job.timeout_mins), default=0)
 
-        scheduler = AsyncScheduler(
+        scheduler = Scheduler(
             jobs=jobs,
             backends={default_backend.name: default_backend},
             default_backend=default_backend.name,

src/dvsim/job/status.py

@@ -12,8 +12,6 @@
 class JobStatus(Enum):
     """Status of a Job."""
 
-    # SCHEDULED is currently unused in the old sync scheduler, there `SCHEDULED` and `QUEUED`
-    # are combined under `QUEUED`. It is used only 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