Exclude stopping status from process reaper (#151)

Author: AlanCoding

dispatcherd/protocols.py

@@ -110,8 +110,15 @@ class PoolWorker(Protocol):
 
     async def start_task(self, message: dict) -> None: ...
 
+    @property
     def is_ready(self) -> bool: ...
 
+    @property
+    def counts_for_capacity(self) -> bool: ...
+
+    @property
+    def expected_alive(self) -> bool: ...
+
     def get_status_data(self) -> dict[str, Any]:
         """Used for worker status control-and-reply command"""
         ...

dispatcherd/service/pool.py

@@ -34,10 +34,6 @@ def __init__(self, worker_id: int, process: ProcessProxy) -> None:
         self.status: Literal['initialized', 'spawned', 'starting', 'ready', 'stopping', 'exited', 'error', 'retired'] = 'initialized'
         self.exit_msg_event = asyncio.Event()
 
-    def is_ready(self) -> bool:
-        """Worker is ready to receive task requests"""
-        return bool(self.status == 'ready')
-
     async def start(self) -> None:
         if self.status != 'initialized':
             logger.error(f'Worker {self.worker_id} status is not initialized, can not start, status={self.status}')
@@ -52,10 +48,26 @@ async def start(self) -> None:
         logger.debug(f'Worker {self.worker_id} pid={self.process.pid} subprocess has spawned')
         self.status = 'starting'  # Not ready until it sends callback message
 
+    @property
+    def is_ready(self) -> bool:
+        """Worker is ready to receive task requests"""
+        return bool(self.status == 'ready')
+
     @property
     def counts_for_capacity(self) -> bool:
+        """Worker is ready to accept work or may become ready very soon, relevant for scale-up decisions"""
         return bool(self.status in ('initialized', 'spawned', 'starting', 'ready'))
 
+    @property
+    def expected_alive(self) -> bool:
+        """Worker is expected to have an active process"""
+        return bool(self.status in ('starting', 'ready'))
+
+    @property
+    def inactive(self) -> bool:
+        """No further shutdown or callback messages are expected from this worker"""
+        return bool(self.status in ('exited', 'error', 'initialized'))
+
     async def start_task(self, message: dict) -> None:
         self.current_task = message  # NOTE: this marks this worker as busy
         self.process.message_queue.put(message)
@@ -136,11 +148,6 @@ def mark_finished_task(self) -> None:
         self.started_at = None
         self.finished_count += 1
 
-    @property
-    def inactive(self) -> bool:
-        "Return True if no further shutdown or callback messages are expected from this worker"
-        return self.status in ['exited', 'error', 'initialized']
-
     def next_wakeup(self) -> Optional[float]:
         """Used by next-run-runner for setting wakeups for task timeouts"""
         if self.is_active_cancel:
@@ -348,7 +355,7 @@ async def manage_old_workers(self) -> None:
         remove_ids = []
         for worker in current_workers:
             # Check if the worker has died unexpectedly.
-            if worker.status not in ['retired', 'error', 'exited', 'initialized', 'spawned'] and not worker.process.is_alive():
+            if worker.expected_alive and not worker.process.is_alive():
                 logger.error(f'Worker {worker.worker_id} pid={worker.process.pid} has died unexpectedly, status was {worker.status}')
 
                 if worker.current_task:

dispatcherd/service/queuer.py

@@ -20,7 +20,7 @@ def count(self) -> int:
 
     def get_free_worker(self) -> Optional[PoolWorker]:
         for candidate_worker in self.workers:
-            if (not candidate_worker.current_task) and candidate_worker.is_ready():
+            if (not candidate_worker.current_task) and candidate_worker.is_ready:
                 return candidate_worker
         return None