[Bugfix][EngineCore] Handle malformed ADD frames and input thread failures

tests/v1/engine/test_preprocess_error_handling.py

@@ -1,17 +1,66 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
+import logging
+import queue
+import threading
+from unittest.mock import MagicMock
+
+import msgspec
 import pytest
 import torch.cuda
 
 from vllm import LLM, SamplingParams
 from vllm.platforms import current_platform
-from vllm.v1.engine import EngineCoreRequest
-from vllm.v1.engine.core import EngineCore
+from vllm.v1.engine import EngineCoreRequest, EngineCoreRequestType
+from vllm.v1.engine.core import EngineCore, EngineCoreProc, _decode_add_request
+from vllm.v1.serial_utils import MsgpackDecoder
 
 MODEL_NAME = "hmellor/tiny-random-LlamaForCausalLM"
 
 
+def test_decode_error_does_not_stop_input_processing(
+    caplog: pytest.LogCaptureFixture,
+):
+    """The ADD decoder remains usable after dropping a malformed frame."""
+    malformed_request = msgspec.msgpack.encode(
+        ["bad", [1.5], None, None, None, 0.0, None, None, None]
+    )
+    valid_request = msgspec.msgpack.encode(
+        ["good", [1], None, None, None, 0.0, None, None, None]
+    )
+    decoder = MsgpackDecoder(EngineCoreRequest)
+
+    with caplog.at_level(logging.ERROR, logger="vllm.v1.engine.core"):
+        assert _decode_add_request(decoder, [malformed_request]) is None
+
+    request = _decode_add_request(decoder, [valid_request])
+    assert request is not None
+    assert request.request_id == "good"
+    assert request.prompt_token_ids == [1]
+    assert "Failed to deserialize ADD request" in caplog.text
+
+
+def test_input_socket_thread_failure_is_reported():
+    """An unexpected input-thread failure must reach the engine busy loop."""
+    engine_core = EngineCoreProc.__new__(EngineCoreProc)
+    engine_core.input_queue = queue.Queue()
+    engine_core.process_input_sockets = MagicMock(
+        side_effect=RuntimeError("socket failure")
+    )
+
+    engine_core._run_input_socket_thread([], None, b"engine", threading.Event())
+
+    request_type, request = engine_core.input_queue.get_nowait()
+    assert request_type == EngineCoreRequestType.INPUT_THREAD_FAILED
+    assert request is None
+
+    with pytest.raises(RuntimeError, match="Input socket thread failed"):
+        EngineCoreProc._handle_client_request(
+            engine_core, EngineCoreRequestType.INPUT_THREAD_FAILED, None
+        )
+
+
 def test_preprocess_error_handling(monkeypatch: pytest.MonkeyPatch):
     """Test that preprocessing errors are handled gracefully."""
     if current_platform.is_rocm() or current_platform.is_xpu():

vllm/v1/engine/__init__.py

@@ -297,6 +297,8 @@ class EngineCoreRequestType(enum.Enum):
     EXECUTOR_FAILED = b"\x04"
     # Sentinel to wake up input_queue.get() during shutdown.
     WAKEUP = b"\x05"
+    # Sentinel raised by the input socket thread when it cannot continue.
+    INPUT_THREAD_FAILED = b"\x06"
 
 
 class ReconfigureDistributedRequest(msgspec.Struct):

vllm/v1/engine/core.py

@@ -108,6 +108,16 @@
 _R = TypeVar("_R")  # Return type for collective_rpc
 
 
+def _decode_add_request(
+    decoder: MsgpackDecoder, data_frames: Sequence[bytestr]
+) -> EngineCoreRequest | None:
+    try:
+        return decoder.decode(data_frames)
+    except Exception:
+        logger.exception("Failed to deserialize ADD request; dropping it.")
+        return None
+
+
 class EngineCore:
     """Inner loop of vLLM's Engine."""
 
@@ -1176,7 +1186,7 @@ def __init__(
             # Threads handle Socket <-> Queues and core_busy_loop uses Queue.
             ready_event = threading.Event()
             input_thread = threading.Thread(
-                target=self.process_input_sockets,
+                target=self._run_input_socket_thread,
                 args=(
                     addresses.inputs,
                     addresses.coordinator_input,
@@ -1206,6 +1216,23 @@ def __init__(
                 assert addresses.coordinator_input is not None
                 logger.info("Waiting for READY message from DP Coordinator...")
 
+    def _run_input_socket_thread(
+        self,
+        input_addresses: list[str],
+        coord_input_address: str | None,
+        identity: bytes,
+        ready_event: threading.Event,
+    ) -> None:
+        try:
+            self.process_input_sockets(
+                input_addresses, coord_input_address, identity, ready_event
+            )
+        except BaseException:
+            logger.exception("EngineCore input socket thread failed.")
+            self.input_queue.put_nowait(
+                (EngineCoreRequestType.INPUT_THREAD_FAILED, None)
+            )
+
     @contextmanager
     def _perform_handshakes(
         self,
@@ -1623,6 +1650,8 @@ def _handle_client_request(
             self._invoke_utility_method(method_name, get_result, output, enqueue_output)
         elif request_type == EngineCoreRequestType.EXECUTOR_FAILED:
             raise RuntimeError("Executor failed.")
+        elif request_type == EngineCoreRequestType.INPUT_THREAD_FAILED:
+            raise RuntimeError("Input socket thread failed.")
         else:
             logger.error(
                 "Unrecognized input request type encountered: %s", request_type
@@ -1821,7 +1850,9 @@ def process_input_sockets(
                     # Deserialize the request data.
                     request: Any
                     if request_type == EngineCoreRequestType.ADD:
-                        req: EngineCoreRequest = add_request_decoder.decode(data_frames)
+                        req = _decode_add_request(add_request_decoder, data_frames)
+                        if req is None:
+                            continue
                         try:
                             request = self.preprocess_add_request(req)
                         except MultiModalCacheMissError as e: