Remove Cluster class, replace with FixedNativeWorkerAdapter

Cluster was a multiprocessing.Process subclass that wrapped
FixedNativeWorkerAdapter in a subprocess with an asyncio event loop
solely to handle signals. This intermediate process layer is removed;
workers are now direct children of SchedulerClusterCombo.

- Delete Cluster, ClusterConfig, and the cluster entry point module
- Redirect scaler_cluster and run_cluster.py to the fixed native
  worker adapter entry point (with --num-of-workers alias for compat)
- Add SIGINT/SIGTERM handling to the fixed native adapter entry point
- Update SchedulerClusterCombo, tests, and examples to use
  FixedNativeWorkerAdapter directly
- Update ECS adapter to use scaler_cluster with updated parameters
  (--max-workers replaces --num-of-workers/--worker-names; worker IDs
  are no longer pre-announced since workers self-assign UUIDs)

Author: magniloquency

examples/task_capabilities.py

@@ -6,13 +6,15 @@
 
 import math
 
-from scaler import Client, Cluster
+from scaler import Client
 from scaler.cluster.combo import SchedulerClusterCombo
 from scaler.config.common.logging import LoggingConfig
 from scaler.config.common.worker import WorkerConfig
-from scaler.config.section.cluster import ClusterConfig
+from scaler.config.common.worker_adapter import WorkerAdapterConfig
+from scaler.config.section.fixed_native_worker_adapter import FixedNativeWorkerAdapterConfig
 from scaler.config.section.scheduler import PolicyConfig
-from scaler.config.types.worker import WorkerCapabilities, WorkerNames
+from scaler.config.types.worker import WorkerCapabilities
+from scaler.worker_manager_adapter.baremetal.fixed_native import FixedNativeWorkerAdapter
 
 
 def gpu_task(x: float) -> float:
@@ -36,34 +38,35 @@ def main():
     )
 
     # Adds an additional worker with GPU support
-    base_cluster = cluster._cluster
-    regular_cluster = Cluster(
-        config=ClusterConfig(
-            scheduler_address=base_cluster._address,
-            object_storage_address=None,
+    base_adapter = cluster._worker_adapter
+    gpu_adapter = FixedNativeWorkerAdapter(
+        FixedNativeWorkerAdapterConfig(
+            worker_adapter_config=WorkerAdapterConfig(
+                scheduler_address=base_adapter._address,
+                object_storage_address=None,
+                max_workers=1,
+            ),
             preload=None,
-            worker_names=WorkerNames(["gpu_worker"]),
-            num_of_workers=1,
-            event_loop=base_cluster._event_loop,
+            event_loop=base_adapter._event_loop,
             worker_io_threads=1,
             worker_config=WorkerConfig(
                 per_worker_capabilities=WorkerCapabilities({"gpu": -1}),
-                per_worker_task_queue_size=base_cluster._per_worker_task_queue_size,
-                heartbeat_interval_seconds=base_cluster._heartbeat_interval_seconds,
-                task_timeout_seconds=base_cluster._task_timeout_seconds,
-                death_timeout_seconds=base_cluster._death_timeout_seconds,
-                garbage_collect_interval_seconds=base_cluster._garbage_collect_interval_seconds,
-                trim_memory_threshold_bytes=base_cluster._trim_memory_threshold_bytes,
-                hard_processor_suspend=base_cluster._hard_processor_suspend,
+                per_worker_task_queue_size=base_adapter._task_queue_size,
+                heartbeat_interval_seconds=base_adapter._heartbeat_interval_seconds,
+                task_timeout_seconds=base_adapter._task_timeout_seconds,
+                death_timeout_seconds=base_adapter._death_timeout_seconds,
+                garbage_collect_interval_seconds=base_adapter._garbage_collect_interval_seconds,
+                trim_memory_threshold_bytes=base_adapter._trim_memory_threshold_bytes,
+                hard_processor_suspend=base_adapter._hard_processor_suspend,
             ),
             logging_config=LoggingConfig(
-                paths=base_cluster._logging_paths,
-                level=base_cluster._logging_level,
-                config_file=base_cluster._logging_config_file,
+                paths=base_adapter._logging_paths,
+                level=base_adapter._logging_level,
+                config_file=base_adapter._logging_config_file,
             ),
         )
     )
-    regular_cluster.start()
+    gpu_adapter.start()
 
     with Client(address=cluster.get_address()) as client:
         print("Submitting tasks...")
@@ -82,6 +85,7 @@ def main():
         gpu_future.result()
         cpu_future.result()
 
+    gpu_adapter.shutdown()
     cluster.shutdown()
 
 

src/run_cluster.py

@@ -1,4 +1,4 @@
-from scaler.entry_points.cluster import main
+from scaler.entry_points.worker_manager_baremetal_fixed_native import main
 
 if __name__ == "__main__":
     main()

src/scaler/__init__.py

@@ -1,14 +1,12 @@
 from .about import __version__
 from .client.client import Client, ScalerFuture
 from .client.serializer.mixins import Serializer
-from .cluster.cluster import Cluster
 from .cluster.combo import SchedulerClusterCombo
 from .cluster.scheduler import Scheduler
 
 assert isinstance(__version__, str)
 assert isinstance(Client, type)
 assert isinstance(ScalerFuture, type)
 assert isinstance(Scheduler, type)
-assert isinstance(Cluster, type)
 assert isinstance(SchedulerClusterCombo, type)
 assert isinstance(Serializer, type)

src/scaler/cluster/cluster.py

@@ -1,12 +1,11 @@
 import logging
-import socket
 from typing import Dict, Optional, Tuple
 
-from scaler.cluster.cluster import Cluster
 from scaler.cluster.object_storage_server import ObjectStorageServerProcess
 from scaler.cluster.scheduler import SchedulerProcess
 from scaler.config.common.logging import LoggingConfig
 from scaler.config.common.worker import WorkerConfig
+from scaler.config.common.worker_adapter import WorkerAdapterConfig
 from scaler.config.defaults import (
     DEFAULT_CLIENT_TIMEOUT_SECONDS,
     DEFAULT_GARBAGE_COLLECT_INTERVAL_SECONDS,
@@ -25,12 +24,13 @@
     DEFAULT_WORKER_DEATH_TIMEOUT,
     DEFAULT_WORKER_TIMEOUT_SECONDS,
 )
-from scaler.config.section.cluster import ClusterConfig
+from scaler.config.section.fixed_native_worker_adapter import FixedNativeWorkerAdapterConfig
 from scaler.config.section.scheduler import PolicyConfig
 from scaler.config.types.object_storage_server import ObjectStorageAddressConfig
-from scaler.config.types.worker import WorkerCapabilities, WorkerNames
+from scaler.config.types.worker import WorkerCapabilities
 from scaler.config.types.zmq import ZMQConfig
 from scaler.utility.network_util import get_available_tcp_port
+from scaler.worker_manager_adapter.baremetal.fixed_native import FixedNativeWorkerAdapter
 
 
 class SchedulerClusterCombo:
@@ -89,13 +89,14 @@ def __init__(
         self._object_storage.start()
         self._object_storage.wait_until_ready()  # object storage should be ready before starting the cluster
 
-        self._cluster = Cluster(
-            config=ClusterConfig(
-                scheduler_address=self._address,
-                object_storage_address=self._object_storage_address,
+        self._worker_adapter = FixedNativeWorkerAdapter(
+            FixedNativeWorkerAdapterConfig(
+                worker_adapter_config=WorkerAdapterConfig(
+                    scheduler_address=self._address,
+                    object_storage_address=self._object_storage_address,
+                    max_workers=n_workers,
+                ),
                 preload=None,
-                worker_names=WorkerNames([f"{socket.gethostname().split('.')[0]}" for _ in range(n_workers)]),
-                num_of_workers=n_workers,
                 event_loop=event_loop,
                 worker_io_threads=worker_io_threads,
                 worker_config=WorkerConfig(
@@ -131,7 +132,7 @@ def __init__(
             policy=scaler_policy,
         )
 
-        self._cluster.start()
+        self._worker_adapter.start()
         self._scheduler.start()
         logging.info(f"{self.__get_prefix()} started")
 
@@ -143,9 +144,8 @@ def shutdown(self):
         self._shutdown_called = True
 
         logging.info(f"{self.__get_prefix()} shutdown")
-        self._cluster.terminate()
+        self._worker_adapter.shutdown()
         self._scheduler.terminate()
-        self._cluster.join()
         self._scheduler.join()
 
         # object storage should terminate after the cluster and scheduler.

src/scaler/cluster/combo.py

@@ -1,3 +1,4 @@
+import argparse
 import dataclasses
 from typing import Optional
 
@@ -25,6 +26,11 @@ class FixedNativeWorkerAdapterConfig(ConfigClass):
         metadata=dict(short="-wit", help="set the number of io threads for io backend per worker"),
     )
 
+    @classmethod
+    def configure_parser(cls, parser) -> None:
+        super().configure_parser(parser)
+        parser.add_argument("--num-of-workers", dest="max_workers", type=int, help=argparse.SUPPRESS)
+
     def __post_init__(self) -> None:
         if self.worker_io_threads <= 0:
             raise ValueError("worker_io_threads must be a positive integer.")

src/scaler/config/section/cluster.py

@@ -1,10 +1,3 @@
-from scaler.cluster.cluster import Cluster
-from scaler.config.section.cluster import ClusterConfig
-from scaler.utility.event_loop import register_event_loop
+from scaler.entry_points.worker_manager_baremetal_fixed_native import main
 
-
-def main():
-    cluster_config = ClusterConfig.parse("Scaler Standalone Compute Cluster", "cluster")
-    register_event_loop(cluster_config.event_loop)
-    cluster = Cluster(cluster_config)
-    cluster.run()
+__all__ = ["main"]