Fix hanging issue in DT; fix hang when aborting communicators from Python side; use queue.Queue for engine_exception_q

Signed-off-by: fangyuchu <fangyuchu@qq.com>

Author: fangyuchu

tests/v1/engine/test_client_guard.py

@@ -1,8 +1,8 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
-import asyncio
 import json
+import queue
 import threading
 import time
 from unittest.mock import AsyncMock
@@ -32,21 +32,20 @@ def create_test_thread_safe_dict(initial_data=None):
 
 
 def create_client_guard(
-    engine_exception_q: asyncio.Queue, engine_status_dict: ThreadSafeDict[int, str]
+    engine_exception_q: queue.Queue, engine_status_dict: ThreadSafeDict[int, str]
 ):
     return ClientGuard(
         fault_receiver_addr=FAULT_RECEIVER_ADDR,
         cmd_addr=CMD_ADDR,
         engine_registry=[b"engine_identity"],
         engine_exception_q=engine_exception_q,
-        engine_exception_q_lock=asyncio.Lock(),
         fault_pub_addr=FAULT_PUB_ADDR,
         engine_status_dict=engine_status_dict,
     )
 
 
 def test_client_guard_initialization():
-    engine_exception_q: asyncio.Queue[FaultInfo] = asyncio.Queue()
+    engine_exception_q: queue.Queue[FaultInfo] = queue.Queue()
     engine_status_dict = create_test_thread_safe_dict({1: "Healthy"})
     guard = create_client_guard(engine_exception_q, engine_status_dict)
 
@@ -64,7 +63,7 @@ def test_client_guard_initialization():
 
 @pytest.mark.asyncio
 async def test_handle_fault():
-    engine_exception_q: asyncio.Queue[FaultInfo] = asyncio.Queue()
+    engine_exception_q: queue.Queue[FaultInfo] = queue.Queue()
     engine_status_dict = create_test_thread_safe_dict({1: "Healthy"})
     guard = create_client_guard(engine_exception_q, engine_status_dict)
 
@@ -82,7 +81,7 @@ async def test_handle_fault():
 
 
 def test_fault_receiver():
-    engine_exception_q: asyncio.Queue[FaultInfo] = asyncio.Queue()
+    engine_exception_q: queue.Queue[FaultInfo] = queue.Queue()
     engine_status_dict = create_test_thread_safe_dict({1: "Healthy"})
     guard = create_client_guard(engine_exception_q, engine_status_dict)
 
@@ -97,7 +96,7 @@ def send_test_message():
         socket.close()
         ctx.term()
 
-    sender_thread = threading.Thread(target=send_test_message)
+    sender_thread = threading.Thread(target=send_test_message, daemon=True)
     sender_thread.start()
 
     def check_published_message():
@@ -114,7 +113,7 @@ def check_published_message():
         assert prefix == FAULT_PUB_TOPIC
         assert json.loads(data) == {"1": "Dead"}
 
-    check_thread = threading.Thread(target=check_published_message)
+    check_thread = threading.Thread(target=check_published_message, daemon=True)
     check_thread.start()
 
     time.sleep(0.1)
@@ -130,7 +129,7 @@ def check_published_message():
 
 
 def test_fault_receiver_unhealthy():
-    engine_exception_q: asyncio.Queue[FaultInfo] = asyncio.Queue()
+    engine_exception_q: queue.Queue[FaultInfo] = queue.Queue()
     engine_status_dict = create_test_thread_safe_dict({1: "Healthy"})
     guard = create_client_guard(engine_exception_q, engine_status_dict)
 
@@ -145,7 +144,7 @@ def send_unhealthy_message():
         socket.close()
         ctx.term()
 
-    threading.Thread(target=send_unhealthy_message).start()
+    threading.Thread(target=send_unhealthy_message, daemon=True).start()
     time.sleep(0.1)
 
     assert engine_status_dict[1] == "Unhealthy"
@@ -154,7 +153,7 @@ def send_unhealthy_message():
 
 
 def test_shutdown_guard():
-    engine_exception_q: asyncio.Queue[FaultInfo] = asyncio.Queue()
+    engine_exception_q: queue.Queue[FaultInfo] = queue.Queue()
     engine_status_dict = create_test_thread_safe_dict({1: "Healthy"})
     guard = create_client_guard(engine_exception_q, engine_status_dict)
 
@@ -181,7 +180,7 @@ def test_shutdown_guard():
 
 @pytest.mark.asyncio
 async def test_handle_fault_async():
-    engine_exception_q: asyncio.Queue[FaultInfo] = asyncio.Queue()
+    engine_exception_q: queue.Queue[FaultInfo] = queue.Queue()
     engine_status_dict = create_test_thread_safe_dict({0: "Unhealthy"})
     guard = create_client_guard(engine_exception_q, engine_status_dict)
 
@@ -190,6 +189,7 @@ async def test_handle_fault_async():
     cmd_socket = ctx.socket(zmq.DEALER)
     cmd_socket.setsockopt(zmq.IDENTITY, b"engine_identity")
     cmd_socket.connect(CMD_ADDR)
+    time.sleep(0.1)
 
     uuid = None
 
@@ -210,13 +210,14 @@ def response_cmd(cmd_socket):
         execute_result = {"engine_index": 0, "success": True, "method_uuid": uuid}
         cmd_socket.send_multipart([b"", json.dumps(execute_result).encode("utf-8")])
 
-    threading.Thread(target=receive_cmd, args=(cmd_socket,)).start()
-    threading.Thread(target=response_cmd, args=(cmd_socket,)).start()
+    threading.Thread(target=receive_cmd, args=(cmd_socket,), daemon=True).start()
+    threading.Thread(target=response_cmd, args=(cmd_socket,), daemon=True).start()
 
     result = await guard.handle_fault("retry", 3)
 
     assert result is True
     assert engine_status_dict[0] == "Healthy"
 
     cmd_socket.close()
+    ctx.term()
     guard.shutdown_guard()

tests/v1/engine/test_engine_core_guard.py

@@ -113,7 +113,9 @@ def mock_worker_receiver(cmd_socket):
     elif instruction == "retry":
         busy_loop_active.set()
 
-    threading.Thread(target=mock_worker_receiver, args=(worker_cmd_socket,)).start()
+    threading.Thread(
+        target=mock_worker_receiver, args=(worker_cmd_socket,), daemon=True
+    ).start()
 
     time.sleep(0.1)
     identity, _, msg = client_socket.recv_multipart()

vllm/v1/engine/core.py

@@ -199,7 +199,9 @@ def _stop_worker_execution(self, soft_pause: bool, timeout: int = 2) -> bool:
             pause_method = "pause_by_abort_communicators"
             self.communicator_aborted = True
 
-        success = self._execute_worker_method(pause_method, timeout=timeout)
+        success = self._execute_worker_method(
+            pause_method, timeout=timeout, worker_timeout=timeout
+        )
         return success
 
     def _execute_worker_method(self, method_name, timeout: int = 5, **kwargs) -> bool:
@@ -238,18 +240,19 @@ def _execute_cmd(self, cmd_str):
         try:
             success = run_method(self, method, args=(), kwargs=method_params)
             self.logger("Command (%s) succeeded: %s", method, success, level="info")
-
+            reason = None
         except Exception as e:
             self.logger(
-                "Error executing method %s: %s %s",
+                "Error executing method %s: %s, %s",
                 method,
                 type(e).__name__,
                 e,
                 level="error",
             )
             success = False
+            reason = f"{type(e).__name__}: {e}"
 
-        self._send_execution_result(success, method_uuid)
+        self._send_execution_result(success, method_uuid, reason)
 
     def pause(self, timeout: int = 1, soft_pause: bool = True) -> bool:
         """
@@ -297,6 +300,9 @@ def retry(self, new_stateless_dp_group_port: int, timeout: int = 1):
         This instruction tells the EngineCore to continue its busy loop
         after being suspended due to an exception.
         """
+        if self.engine_running:
+            return True
+
         start_time = time.monotonic()
 
         success = self._execute_worker_method("restore_worker", timeout=timeout)
@@ -320,14 +326,19 @@ def retry(self, new_stateless_dp_group_port: int, timeout: int = 1):
         remaining_timeout = max(0, timeout - elapsed)
         success = self.busy_loop_active.wait(timeout=remaining_timeout)
         self.engine_running = success
+        assert self.cmd_q.empty(), "cmd_q must be empty after execution"
         return success
 
-    def _send_execution_result(self, success: bool, method_uuid: str):
+    def _send_execution_result(
+        self, success: bool, method_uuid: str, reason: str | None
+    ):
         msg = {
             "engine_index": self.engine_index,
             "success": success,
             "method_uuid": method_uuid,
         }
+        if not success and reason is not None:
+            msg["reason"] = reason
         msg_bytes = json.dumps(msg).encode("utf-8")
         self.client_cmd_socket.send_multipart([b"", msg_bytes])
 
@@ -949,7 +960,7 @@ def __init__(
                 # Track whether the busy loop is currently active.
                 self.busy_loop_active = threading.Event()
                 self.fault_signal_q: queue.Queue[Exception] = queue.Queue()
-                self.cmd_q: queue.Queue[str | None] = queue.Queue()
+                self.cmd_q: queue.Queue[str | None] = queue.Queue(maxsize=1)
                 self.engine_recovery_timeout = ft_config.engine_recovery_timeout
                 engine_core_guard_ids = addresses.engine_core_guard_identities
                 assert engine_core_guard_ids is not None

vllm/v1/engine/core_client.py

@@ -351,8 +351,7 @@ def __init__(
         fault_receiver_addr: str,
         cmd_addr: str,
         engine_registry: list[bytes],
-        engine_exception_q: asyncio.Queue[FaultInfo],
-        engine_exception_q_lock: asyncio.Lock,
+        engine_exception_q: queue.Queue[FaultInfo],
         fault_pub_addr: str,
         engine_status_dict: ThreadSafeDict[int, str],
     ):
@@ -372,17 +371,14 @@ def __init__(
             ctx=self.zmq_ctx, path=fault_pub_addr, socket_type=zmq.PUB, bind=True
         )
 
-        self.engine_exception_q: asyncio.Queue[FaultInfo] = engine_exception_q
-
-        self.engine_exception_q_lock = engine_exception_q_lock
+        self.engine_exception_q: queue.Queue[FaultInfo] = engine_exception_q
 
         self.engine_status_dict: ThreadSafeDict[int, str] = engine_status_dict
 
         self.fault_handler = FaultHandler(
             self.cmd_socket,
             self.engine_registry,
             self.engine_exception_q,
-            self.engine_exception_q_lock,
             self.engine_status_dict,
         )
 
@@ -449,7 +445,7 @@ def fault_receiver(self):
                 )
 
                 # Pause healthy engines on fault.
-                # Pause will be invoked again during fault-tolerance handling,
+                # Pause can be invoked again during fault-tolerance handling,
                 # so it's unnecessary to track whether all engines are currently
                 # paused.
                 self.fault_handler.submit_fault("pause", 5, soft_pause=False)
@@ -684,7 +680,7 @@ def __init__(
                 self.start_engine_core_monitor()
 
             if vllm_config.fault_tolerance_config.enable_fault_tolerance:
-                self.engine_exception_q: asyncio.Queue[FaultInfo] = asyncio.Queue()
+                self.engine_exception_q: queue.Queue[FaultInfo] = queue.Queue()
                 assert addresses.fault_report_addr is not None, (
                     "addresses.fault_report_addr should not be None at fault tolerance"
                     " scenario"
@@ -694,7 +690,6 @@ def __init__(
                     " scenario"
                 )
                 self.engine_registry = addresses.engine_core_guard_identities
-                self.engine_exception_q_lock = asyncio.Lock()
                 assert self.engine_registry is not None
                 assert addresses.fault_pub_socket_addr is not None, (
                     "addresses.fault_pub_socket_addr should not be None at"
@@ -708,7 +703,6 @@ def __init__(
                     addresses.client_cmd_addr,
                     self.engine_registry,
                     self.engine_exception_q,
-                    self.engine_exception_q_lock,
                     addresses.fault_pub_socket_addr,
                     self.engine_status_dict,
                 )
@@ -758,7 +752,7 @@ def monitor_actors():
             if not all_actors:
                 return
             while True:
-                for actor in all_actors:
+                for actor in all_actors[:]:
                     actor_id = actor._actor_id.hex()
                     if actor in engine_manager.local_engine_actors:
                         actor_index = engine_manager.local_engine_actors.index(actor)

vllm/v1/engine/utils.py

@@ -5,6 +5,7 @@
 import json
 import multiprocessing
 import os
+import queue
 import time
 import uuid
 import weakref
@@ -1224,20 +1225,6 @@ def generate_identity_group(peer1, peer2, use, n):
     return identitys
 
 
-async def get_queue_snapshot(queue: asyncio.Queue, queue_lock: asyncio.Lock) -> list:
-    """Thread-safe snapshot of the exception queue."""
-    async with queue_lock:
-        items = []
-        # get item at first
-        while not queue.empty():
-            item = queue.get_nowait()
-            items.append(item)
-        # put item into queue again
-        for item in items:
-            queue.put_nowait(item)
-    return items
-
-
 def broadcast_instruction(
     cmd_socket,
     target_identities: set[bytes] | list[bytes],
@@ -1346,13 +1333,11 @@ def __init__(
         self,
         cmd_socket: zmq.Socket,
         client_cmd_registry: list[bytes],
-        engine_exception_q: asyncio.Queue[FaultInfo],
-        engine_exception_q_lock: asyncio.Lock,
+        engine_exception_q: queue.Queue[FaultInfo],
         engine_status_dict: ThreadSafeDict[int, str],
     ) -> None:
         self.cmd_socket = cmd_socket
         self.engine_exception_q = engine_exception_q
-        self.engine_exception_q_lock = engine_exception_q_lock
         self.engine_status_dict: ThreadSafeDict[int, str] = engine_status_dict
         self.engine_identity_to_index: dict[bytes, int] = {
             identity: i for i, identity in enumerate(client_cmd_registry)
@@ -1394,9 +1379,8 @@ async def _dispatcher(self):
     def retry(self, **kwargs):
         if "Dead" in self.engine_status_dict.values():
             self.logger(
-                "engine_core dead unexpectedly, retry is impossible,"
-                "shutdown will be performed",
-                level="info",
+                "Engine core is dead; retry won't work.",
+                level="warning",
             )
             return False, set(), kwargs
 
@@ -1472,7 +1456,12 @@ async def _handle_fault_internal(
         if instruction == "retry" and all_success:
             for engine_index, _ in self.engine_status_dict.items():
                 self.engine_status_dict[engine_index] = "Healthy"
-            # todo: should we also clear the engine_exception_q here?
+            while not self.engine_exception_q.empty():
+                try:
+                    self.engine_exception_q.get_nowait()
+                except queue.Empty:
+                    break
+
         return all_success
 
     async def handle_fault(self, instruction: str, timeout: int, **kwargs) -> bool: