Rename scheduler managers to controllers

Author: sharpener6

scaler/io/ymq/ymq.pyi

@@ -22,7 +22,9 @@ class Message:
     address: Bytes
     payload: Bytes
 
-    def __init__(self, address: Bytes | bytes | SupportsBytes | None, payload: Bytes | bytes | SupportsBytes) -> None: ...
+    def __init__(
+        self, address: Bytes | bytes | SupportsBytes | None, payload: Bytes | bytes | SupportsBytes
+    ) -> None: ...
     def __repr__(self) -> str: ...
     def __str__(self) -> str: ...
 
@@ -38,32 +40,35 @@ class IOContext:
 
     def __init__(self, num_threads: int = 1) -> None: ...
     def __repr__(self) -> str: ...
-
     def createIOSocket(self, /, identity: str, socket_type: IOSocketType) -> Awaitable[IOSocket]:
         """Create an io socket with an identity and socket type"""
 
-
 class IOSocket:
     identity: str
     socket_type: IOSocketType
 
     def __repr__(self) -> str: ...
-
     async def send(self, message: Message) -> None:
         """Send a message to one of the socket's peers"""
+
     async def recv(self) -> Message:
         """Receive a message from one of the socket's peers"""
+
     async def bind(self, address: str) -> None:
         """Bind the socket to an address and listen for incoming connections"""
+
     async def connect(self, address: str) -> None:
         """Connect to a remote socket"""
 
     def send_sync(self, message: Message) -> None:
         """Send a message to one of the socket's peers synchronously"""
+
     def recv_sync(self) -> Message:
         """Receive a message from one of the socket's peers synchronously"""
+
     def bind_sync(self, address: str) -> None:
         """Bind the socket to an address and listen for incoming connections synchronously"""
+
     def connect_sync(self, address: str) -> None:
         """Connect to a remote socket synchronously"""
 

scaler/scheduler/managers/__init__.py scaler/scheduler/controllers/__init__.pyscaler/scheduler/managers/__init__.py renamed to scaler/scheduler/controllers/__init__.py

@@ -5,12 +5,12 @@
 from scaler.io.async_connector import AsyncConnector
 from scaler.protocol.python.message import StateBalanceAdvice
 from scaler.scheduler.allocate_policy.mixins import TaskAllocatePolicy
-from scaler.scheduler.managers.mixins import TaskManager
+from scaler.scheduler.controllers.mixins import TaskController
 from scaler.utility.identifiers import WorkerID, TaskID
 from scaler.utility.mixins import Looper
 
 
-class VanillaBalanceManager(Looper):
+class VanillaBalanceController(Looper):
     def __init__(self, load_balance_trigger_times: int, task_allocate_policy: TaskAllocatePolicy):
         self._load_balance_trigger_times = load_balance_trigger_times
 
@@ -22,13 +22,13 @@ def __init__(self, load_balance_trigger_times: int, task_allocate_policy: TaskAl
         self._binder: Optional[AsyncBinder] = None
         self._binder_monitor: Optional[AsyncConnector] = None
 
-        self._task_manager: Optional[TaskManager] = None
+        self._task_controller: Optional[TaskController] = None
 
-    def register(self, binder: AsyncBinder, binder_monitor: AsyncConnector, task_manager: TaskManager):
+    def register(self, binder: AsyncBinder, binder_monitor: AsyncConnector, task_controller: TaskController):
         self._binder = binder
         self._binder_monitor = binder_monitor
 
-        self._task_manager = task_manager
+        self._task_controller = task_controller
 
     async def routine(self):
         current_advice = self._task_allocate_policy.balance()

…er/scheduler/managers/balance_manager.py …eduler/controllers/balance_controller.pyscaler/scheduler/managers/balance_manager.py renamed to scaler/scheduler/controllers/balance_controller.py scaler/scheduler/managers/balance_manager.py renamed to scaler/scheduler/controllers/balance_controller.py

@@ -13,14 +13,14 @@
     TaskCancel,
 )
 from scaler.protocol.python.status import ClientManagerStatus
-from scaler.scheduler.managers.mixins import ClientManager, ObjectManager, TaskManager, WorkerManager
+from scaler.scheduler.controllers.mixins import ClientController, ObjectController, TaskController, WorkerController
 from scaler.utility.exceptions import ClientShutdownException
 from scaler.utility.identifiers import ClientID, TaskID
 from scaler.utility.mixins import Looper, Reporter
 from scaler.utility.one_to_many_dict import OneToManyDict
 
 
-class VanillaClientManager(ClientManager, Looper, Reporter):
+class VanillaClientController(ClientController, Looper, Reporter):
     def __init__(self, client_timeout_seconds: int, protected: bool, storage_address: ObjectStorageAddress):
         self._client_timeout_seconds = client_timeout_seconds
         self._protected = protected
@@ -30,25 +30,25 @@ def __init__(self, client_timeout_seconds: int, protected: bool, storage_address
 
         self._binder: Optional[AsyncBinder] = None
         self._binder_monitor: Optional[AsyncConnector] = None
-        self._object_manager: Optional[ObjectManager] = None
-        self._task_manager: Optional[TaskManager] = None
-        self._worker_manager: Optional[WorkerManager] = None
+        self._object_controller: Optional[ObjectController] = None
+        self._task_controller: Optional[TaskController] = None
+        self._worker_controller: Optional[WorkerController] = None
 
         self._client_last_seen: Dict[ClientID, Tuple[float, ClientHeartbeat]] = dict()
 
     def register(
         self,
         binder: AsyncBinder,
         binder_monitor: AsyncConnector,
-        object_manager: ObjectManager,
-        task_manager: TaskManager,
-        worker_manager: WorkerManager,
+        object_controller: ObjectController,
+        task_controller: TaskController,
+        worker_controller: WorkerController,
     ):
         self._binder = binder
         self._binder_monitor = binder_monitor
-        self._object_manager = object_manager
-        self._task_manager = task_manager
-        self._worker_manager = worker_manager
+        self._object_controller = object_controller
+        self._task_controller = task_controller
+        self._worker_controller = worker_controller
 
     def get_client_task_ids(self, client_id: ClientID) -> Set[TaskID]:
         return self._client_to_task_ids.get_values(client_id)
@@ -89,7 +89,7 @@ async def on_client_disconnect(self, client_id: ClientID, request: ClientDisconn
         if self._protected:
             return
 
-        await self._worker_manager.on_client_shutdown(client_id)
+        await self._worker_controller.on_client_shutdown(client_id)
 
         raise ClientShutdownException(f"received client shutdown from {client_id!r}, quitting")
 
@@ -118,12 +118,12 @@ async def __on_client_disconnect(self, client_id: ClientID):
             self._client_last_seen.pop(client_id)
 
         await self.__cancel_tasks(client_id)
-        self._object_manager.clean_client(client_id)
+        self._object_controller.clean_client(client_id)
 
     async def __cancel_tasks(self, client_id: ClientID):
         if client_id not in self._client_to_task_ids.keys():
             return
 
         tasks = self._client_to_task_ids.get_values(client_id).copy()
         for task in tasks:
-            await self._task_manager.on_task_cancel(client_id, TaskCancel.new_msg(task))
+            await self._task_controller.on_task_cancel(client_id, TaskCancel.new_msg(task))

…ler/scheduler/managers/client_manager.py …heduler/controllers/client_controller.pyscaler/scheduler/managers/client_manager.py renamed to scaler/scheduler/controllers/client_controller.py scaler/scheduler/managers/client_manager.py renamed to scaler/scheduler/controllers/client_controller.py

@@ -9,7 +9,7 @@
 from scaler.io.async_object_storage_connector import AsyncObjectStorageConnector
 from scaler.protocol.python.common import ObjectMetadata, TaskStatus
 from scaler.protocol.python.message import GraphTask, GraphTaskCancel, StateGraphTask, Task, TaskCancel, TaskResult
-from scaler.scheduler.managers.mixins import ClientManager, GraphTaskManager, ObjectManager, TaskManager
+from scaler.scheduler.controllers.mixins import ClientController, GraphTaskController, ObjectController, TaskController
 from scaler.utility.graph.topological_sorter import TopologicalSorter
 from scaler.utility.identifiers import ClientID, ObjectID, TaskID
 from scaler.utility.many_to_many_dict import ManyToManyDict
@@ -47,7 +47,7 @@ class _Graph:
     running_task_ids: Set[TaskID] = dataclasses.field(default_factory=set)
 
 
-class VanillaGraphTaskManager(GraphTaskManager, Looper, Reporter):
+class VanillaGraphTaskController(GraphTaskController, Looper, Reporter):
     """
     A = func()
     B = func2(A)
@@ -73,9 +73,9 @@ def __init__(self):
         self._binder_monitor: Optional[AsyncConnector] = None
         self._connector_storage: Optional[AsyncObjectStorageConnector] = None
 
-        self._client_manager: Optional[ClientManager] = None
-        self._task_manager: Optional[TaskManager] = None
-        self._object_manager: Optional[ObjectManager] = None
+        self._client_controller: Optional[ClientController] = None
+        self._task_controller: Optional[TaskController] = None
+        self._object_controller: Optional[ObjectController] = None
 
         self._unassigned: Queue = Queue()
 
@@ -87,16 +87,16 @@ def register(
         binder: AsyncBinder,
         binder_monitor: AsyncConnector,
         connector_storage: AsyncObjectStorageConnector,
-        client_manager: ClientManager,
-        task_manager: TaskManager,
-        object_manager: ObjectManager,
+        client_controller: ClientController,
+        task_controller: TaskController,
+        object_controller: ObjectController,
     ):
         self._binder = binder
         self._binder_monitor = binder_monitor
         self._connector_storage = connector_storage
-        self._client_manager = client_manager
-        self._task_manager = task_manager
-        self._object_manager = object_manager
+        self._client_controller = client_controller
+        self._task_controller = task_controller
+        self._object_controller = object_controller
 
     async def on_graph_task(self, client_id: ClientID, graph_task: GraphTask):
         await self._unassigned.put((client_id, graph_task))
@@ -141,7 +141,7 @@ def get_status(self) -> Dict:
     async def __add_new_graph(self, client_id: ClientID, graph_task: GraphTask):
         graph = {}
 
-        self._client_manager.on_task_begin(client_id, graph_task.task_id)
+        self._client_controller.on_task_begin(client_id, graph_task.task_id)
 
         tasks = dict()
         depended_task_id_to_task_id: ManyToManyDict[TaskID, TaskID] = ManyToManyDict()
@@ -199,7 +199,7 @@ async def __check_one_graph(self, graph_task_id: TaskID):
                 function_args=[self.__get_argument_object(graph_task_id, arg) for arg in task_info.task.function_args],
             )
 
-            await self._task_manager.on_task_new(graph_info.client, task)
+            await self._task_controller.on_task_new(graph_info.client, task)
 
     async def __mark_node_done(self, result: TaskResult):
         graph_task_id = self._task_id_to_graph_task_id.pop(result.task_id)
@@ -240,7 +240,7 @@ async def __cancel_one_graph(self, graph_task_id: TaskID, result: TaskResult):
             result_metadata = result.metadata
             result_object_ids = [ObjectID(object_id_bytes) for object_id_bytes in result.results]
             result_objects = [
-                (object_id, self._object_manager.get_object_name(object_id)) for object_id in result_object_ids
+                (object_id, self._object_controller.get_object_name(object_id)) for object_id in result_object_ids
             ]
             await self.__clean_all_running_nodes(graph_task_id, result_status, result_metadata, result_objects)
             await self.__clean_all_inactive_nodes(graph_task_id, result_status, result_metadata, result_objects)
@@ -261,7 +261,7 @@ async def __clean_all_running_nodes(
         # cancel all running tasks
         for task_id in running_task_ids:
             new_result_object_ids = await self.__duplicate_objects(graph_info.client, result_objects)
-            await self._task_manager.on_task_cancel(graph_info.client, TaskCancel.new_msg(task_id))
+            await self._task_controller.on_task_cancel(graph_info.client, TaskCancel.new_msg(task_id))
             await self.__mark_node_done(
                 TaskResult.new_msg(
                     task_id, result_status, result_metadata, [bytes(object_id) for object_id in new_result_object_ids]
@@ -292,7 +292,7 @@ async def __clean_all_inactive_nodes(
                 )
 
     async def __finish_one_graph(self, graph_task_id: TaskID, result_status: TaskStatus):
-        self._client_manager.on_task_finish(graph_task_id)
+        self._client_controller.on_task_finish(graph_task_id)
         info = self._graph_task_id_to_graph.pop(graph_task_id)
         await self._binder.send(info.client, TaskResult.new_msg(graph_task_id, result_status, metadata=b""))
 
@@ -329,7 +329,7 @@ async def __clean_intermediate_result(self, graph_task_id: TaskID, task_id: Task
                 continue
 
             # delete intermediate results as they are not needed anymore
-            self._object_manager.on_del_objects(graph_info.client, set(graph_info.tasks[argument].result_object_ids))
+            self._object_controller.on_del_objects(graph_info.client, set(graph_info.tasks[argument].result_object_ids))
 
     async def __duplicate_objects(
         self, owner: ClientID, result_objects: List[Tuple[ObjectID, bytes]]
@@ -350,4 +350,6 @@ async def __duplicate_object(
     ):
         await self._connector_storage.duplicate_object_id(object_id, new_object_id)
 
-        self._object_manager.on_add_object(owner, new_object_id, ObjectMetadata.ObjectContentType.Object, object_name)
+        self._object_controller.on_add_object(
+            owner, new_object_id, ObjectMetadata.ObjectContentType.Object, object_name
+        )

…aler/scheduler/managers/graph_manager.py …cheduler/controllers/graph_controller.pyscaler/scheduler/managers/graph_manager.py renamed to scaler/scheduler/controllers/graph_controller.py scaler/scheduler/managers/graph_manager.py renamed to scaler/scheduler/controllers/graph_controller.py

@@ -0,0 +1,55 @@
+from typing import Optional
+
+import psutil
+
+from scaler.io.async_binder import AsyncBinder
+from scaler.io.async_connector import AsyncConnector
+from scaler.protocol.python.message import StateScheduler
+from scaler.protocol.python.status import Resource
+from scaler.scheduler.controllers.mixins import (
+    ClientController,
+    InformationController,
+    ObjectController,
+    TaskController,
+    WorkerController,
+)
+from scaler.utility.mixins import Looper
+
+
+class VanillaInformationController(InformationController, Looper):
+    def __init__(self, binder: AsyncConnector):
+        self._monitor_binder: AsyncConnector = binder
+        self._process = psutil.Process()
+
+        self._binder: Optional[AsyncBinder] = None
+        self._client_controller: Optional[ClientController] = None
+        self._object_controller: Optional[ObjectController] = None
+        self._task_controller: Optional[TaskController] = None
+        self._worker_controller: Optional[WorkerController] = None
+
+    def register_managers(
+        self,
+        binder: AsyncBinder,
+        client_controller: ClientController,
+        object_controller: ObjectController,
+        task_controller: TaskController,
+        worker_controller: WorkerController,
+    ):
+        self._binder = binder
+        self._client_controller = client_controller
+        self._object_controller = object_controller
+        self._task_controller = task_controller
+        self._worker_controller = worker_controller
+
+    async def routine(self):
+        await self._monitor_binder.send(
+            StateScheduler.new_msg(
+                binder=self._binder.get_status(),
+                scheduler=Resource.new_msg(int(self._process.cpu_percent() * 10), self._process.memory_info().rss),
+                rss_free=psutil.virtual_memory().available,
+                client_manager=self._client_controller.get_status(),
+                object_manager=self._object_controller.get_status(),
+                task_manager=self._task_controller.get_status(),
+                worker_manager=self._worker_controller.get_status(),
+            )
+        )

scaler/scheduler/controllers/information_controller.py

@@ -18,7 +18,7 @@
 from scaler.utility.mixins import Reporter
 
 
-class ObjectManager(Reporter):
+class ObjectController(Reporter):
     @abc.abstractmethod
     async def on_object_instruction(self, source: bytes, request: ObjectInstruction):
         raise NotImplementedError()
@@ -50,7 +50,7 @@ def get_object_name(self, object_id: ObjectID) -> bytes:
         raise NotImplementedError()
 
 
-class ClientManager(Reporter):
+class ClientController(Reporter):
     @abc.abstractmethod
     def get_client_task_ids(self, client_id: ClientID) -> Set[TaskID]:
         raise NotImplementedError()
@@ -80,7 +80,7 @@ async def on_client_disconnect(self, client_id: ClientID, request: ClientDisconn
         raise NotImplementedError()
 
 
-class GraphTaskManager(Reporter):
+class GraphTaskController(Reporter):
     @abc.abstractmethod
     async def on_graph_task(self, client_id: ClientID, graph_task: GraphTask):
         raise NotImplementedError()
@@ -98,7 +98,7 @@ def is_graph_sub_task(self, task_id: TaskID) -> bool:
         raise NotImplementedError()
 
 
-class TaskManager(Reporter):
+class TaskController(Reporter):
     @abc.abstractmethod
     async def on_task_new(self, client_id: ClientID, task: Task):
         raise NotImplementedError()
@@ -116,7 +116,7 @@ async def on_task_reroute(self, task_id: TaskID):
         raise NotImplementedError()
 
 
-class WorkerManager(Reporter):
+class WorkerController(Reporter):
     @abc.abstractmethod
     async def assign_task_to_worker(self, task: Task) -> bool:
         raise NotImplementedError()
@@ -154,5 +154,5 @@ def get_worker_ids(self) -> Set[WorkerID]:
         raise NotImplementedError()
 
 
-class InformationManager(metaclass=abc.ABCMeta):
+class InformationController(metaclass=abc.ABCMeta):
     pass