Refactor fault tolerance modules by renaming classes to Sentinel and converting engine_registry to a dict

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

Author: fangyuchu

tests/v1/engine/test_client_guard.py

@@ -11,7 +11,7 @@
 import zmq
 
 from vllm.utils.collection_utils import ThreadSafeDict
-from vllm.v1.engine.core_client import ClientGuard
+from vllm.v1.engine.core_client import ClientSentinel
 from vllm.v1.engine.utils import FaultHandler, FaultInfo
 
 FAULT_RECEIVER_ADDR = "tcp://127.0.0.1:8844"
@@ -31,59 +31,59 @@ def create_test_thread_safe_dict(initial_data=None):
     return tsd
 
 
-def create_client_guard(
+def create_client_sentinel(
     engine_exception_q: queue.Queue, engine_status_dict: ThreadSafeDict[int, str]
 ):
-    return ClientGuard(
+    return ClientSentinel(
         fault_receiver_addr=FAULT_RECEIVER_ADDR,
         cmd_addr=CMD_ADDR,
-        engine_registry=[b"engine_identity"],
+        engine_registry={0: b"engine_identity"},
         engine_exception_q=engine_exception_q,
         fault_pub_addr=FAULT_PUB_ADDR,
         engine_status_dict=engine_status_dict,
     )
 
 
-def test_client_guard_initialization():
+def test_client_sentinel_initialization():
     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)
+    sentinel = create_client_sentinel(engine_exception_q, engine_status_dict)
 
-    assert guard.engine_registry == [b"engine_identity"]
-    assert not guard.client_guard_dead
-    assert isinstance(guard.fault_handler, FaultHandler)
-    assert guard.engine_exception_q is engine_exception_q
+    assert sentinel.engine_registry[0] == b"engine_identity"
+    assert not sentinel.client_sentinel_dead
+    assert isinstance(sentinel.fault_handler, FaultHandler)
+    assert sentinel.engine_exception_q is engine_exception_q
 
-    assert guard.fault_receiver_socket.type == zmq.ROUTER
-    assert guard.cmd_socket.type == zmq.ROUTER
-    assert guard.fault_pub_socket.type == zmq.PUB
+    assert sentinel.fault_receiver_socket.type == zmq.ROUTER
+    assert sentinel.cmd_socket.type == zmq.ROUTER
+    assert sentinel.fault_pub_socket.type == zmq.PUB
 
-    guard.shutdown_guard()
+    sentinel.shutdown_sentinel()
 
 
 @pytest.mark.asyncio
 async def test_handle_fault():
     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)
+    sentinel = create_client_sentinel(engine_exception_q, engine_status_dict)
 
     engine_exception_q.put_nowait(
         FaultInfo(engine_id="1", message="test exception", type="test")
     )
 
-    guard.fault_handler.handle_fault = AsyncMock(return_value=True)
+    sentinel.fault_handler.handle_fault = AsyncMock(return_value=True)
 
-    result = await guard.handle_fault("pause", 5)
+    result = await sentinel.handle_fault("pause", 5)
     assert result is True
-    guard.fault_handler.handle_fault.assert_awaited_once_with("pause", 5)
+    sentinel.fault_handler.handle_fault.assert_awaited_once_with("pause", 5)
 
-    guard.shutdown_guard()
+    sentinel.shutdown_sentinel()
 
 
 def test_fault_receiver():
     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)
+    sentinel = create_client_sentinel(engine_exception_q, engine_status_dict)
 
     def send_test_message():
         ctx = zmq.Context()
@@ -125,13 +125,13 @@ def check_published_message():
 
     assert engine_status_dict[1] == "Dead"
 
-    guard.shutdown_guard()
+    sentinel.shutdown_sentinel()
 
 
 def test_fault_receiver_unhealthy():
     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)
+    sentinel = create_client_sentinel(engine_exception_q, engine_status_dict)
 
     def send_unhealthy_message():
         ctx = zmq.Context()
@@ -149,22 +149,22 @@ def send_unhealthy_message():
 
     assert engine_status_dict[1] == "Unhealthy"
 
-    guard.shutdown_guard()
+    sentinel.shutdown_sentinel()
 
 
-def test_shutdown_guard():
+def test_shutdown_sentinel():
     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)
+    sentinel = create_client_sentinel(engine_exception_q, engine_status_dict)
 
-    original_fault_sock = guard.fault_receiver_socket
-    original_cmd_sock = guard.cmd_socket
-    original_pub_sock = guard.fault_pub_socket
-    original_ctx = guard.zmq_ctx
+    original_fault_sock = sentinel.fault_receiver_socket
+    original_cmd_sock = sentinel.cmd_socket
+    original_pub_sock = sentinel.fault_pub_socket
+    original_ctx = sentinel.zmq_ctx
 
-    guard.shutdown_guard()
+    sentinel.shutdown_sentinel()
 
-    assert guard.client_guard_dead is True
+    assert sentinel.client_sentinel_dead is True
 
     with pytest.raises(zmq.ZMQError):
         original_fault_sock.recv()
@@ -182,7 +182,7 @@ def test_shutdown_guard():
 async def test_handle_fault_async():
     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)
+    sentinel = create_client_sentinel(engine_exception_q, engine_status_dict)
 
     time.sleep(0.1)
     ctx = zmq.Context().instance()
@@ -213,11 +213,11 @@ def response_cmd(cmd_socket):
     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)
+    result = await sentinel.handle_fault("retry", 3)
 
     assert result is True
     assert engine_status_dict[0] == "Healthy"
 
     cmd_socket.close()
     ctx.term()
-    guard.shutdown_guard()
+    sentinel.shutdown_sentinel()

tests/v1/engine/test_engine_core_guard.py

@@ -12,21 +12,21 @@
 
 from vllm.utils.network_utils import make_zmq_socket
 from vllm.v1.engine.core import (
-    EngineCoreGuard,
+    EngineCoreSentinel,
     EngineLoopPausedError,
 )
 from vllm.v1.serial_utils import serialize_method_call
 
 CLIENT_CMD_ADDR = "tcp://127.0.0.1:8844"
 WORKER_CMD_ADDR = "tcp://127.0.0.1:8845"
 FAULT_REPORT_ADDR = "tcp://127.0.0.1:8846"
-GUARD_IDENTITY = b"engine_guard_0"
+SENTINEL_IDENTITY = b"engine_sentinel_0"
 
 
-def create_engine_core_guard(
+def create_engine_core_sentinel(
     fault_signal_q: queue.Queue, busy_loop_active: threading.Event
 ):
-    return EngineCoreGuard(
+    return EngineCoreSentinel(
         engine_index=0,
         fault_signal_q=fault_signal_q,
         cmd_q=queue.Queue(),
@@ -35,31 +35,31 @@ def create_engine_core_guard(
         client_cmd_addr=CLIENT_CMD_ADDR,
         worker_cmd_addr=WORKER_CMD_ADDR,
         fault_report_addr=FAULT_REPORT_ADDR,
-        guard_identity=GUARD_IDENTITY,
+        sentinel_identity=SENTINEL_IDENTITY,
         tp_size=1,
         pp_size=1,
         dp_size=1,
     )
 
 
-def test_engine_core_guard_initialization():
+def test_engine_core_sentinel_initialization():
     fault_signal_q: queue.Queue = queue.Queue()
     busy_loop_active = threading.Event()
 
-    guard = create_engine_core_guard(fault_signal_q, busy_loop_active)
+    sentinel = create_engine_core_sentinel(fault_signal_q, busy_loop_active)
 
-    assert guard.engine_index == 0
-    assert guard.tp_size == 1
-    assert guard.pp_size == 1
-    assert not guard.communicator_aborted
-    assert guard.engine_running is True
-    assert guard.daemon is True
+    assert sentinel.engine_index == 0
+    assert sentinel.tp_size == 1
+    assert sentinel.pp_size == 1
+    assert not sentinel.communicator_aborted
+    assert sentinel.engine_running is True
+    assert sentinel.daemon is True
 
-    assert guard.fault_report_socket.type == zmq.DEALER
-    assert guard.client_cmd_socket.type == zmq.DEALER
-    assert guard.worker_cmd_socket.type == zmq.ROUTER
+    assert sentinel.fault_report_socket.type == zmq.DEALER
+    assert sentinel.client_cmd_socket.type == zmq.DEALER
+    assert sentinel.worker_cmd_socket.type == zmq.ROUTER
 
-    guard.shutdown()
+    sentinel.shutdown()
 
 
 @pytest.mark.parametrize("instruction", ["pause", "retry"])
@@ -73,7 +73,7 @@ def test_run_handle_instruction(instruction):
 
     time.sleep(0.1)
 
-    guard = create_engine_core_guard(fault_signal_q, busy_loop_active)
+    sentinel = create_engine_core_sentinel(fault_signal_q, busy_loop_active)
     time.sleep(0.1)
 
     ctx = zmq.Context()
@@ -96,7 +96,7 @@ def mock_worker_receiver(cmd_socket):
         logging.info(identity)
         cmd_socket.send_multipart([b"", json.dumps(response_dict).encode("utf-8")])
 
-    threading.Thread(target=guard.run, daemon=True).start()
+    threading.Thread(target=sentinel.run, daemon=True).start()
     time.sleep(0.1)
 
     param = {"timeout": 3}
@@ -106,7 +106,7 @@ def mock_worker_receiver(cmd_socket):
         param["new_stateless_dp_group_port"] = 23456
     serial_instruction = serialize_method_call(instruction, **param)
     client_socket.send_multipart(
-        [GUARD_IDENTITY, b"", serial_instruction.encode("utf-8")]
+        [SENTINEL_IDENTITY, b"", serial_instruction.encode("utf-8")]
     )
     if instruction == "pause":
         fault_signal_q.put(EngineLoopPausedError(Exception("test error")))
@@ -127,4 +127,4 @@ def mock_worker_receiver(cmd_socket):
 
     client_socket.close()
     worker_cmd_socket.close()
-    guard.shutdown()
+    sentinel.shutdown()

vllm/config/fault_tolerance.py

@@ -39,10 +39,10 @@ class FaultToleranceConfig:
 
     engine_core_cmd_addr: str = ""
     """
-    The ZMQ address between engine_core_guard and worker_guard.
+    The ZMQ address between engine_core_sentinel and worker_sentinel.
     It will be initialized and assigned in EngineCore, then passed
     to the Worker via vllm_config—this is required for the Worker
-    to spin up the WorkerGuard.
+    to spin up the WorkerSentinel.
     """
 
     gloo_comm_timeout: int = 30