reject requests when engine is in fault status

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

Author: fangyuchu

vllm/engine/protocol.py

@@ -171,7 +171,7 @@ async def handle_fault(
         """send fault tolerance instruction to the engine"""
         raise NotImplementedError
 
-    async def exception_reporter(self):
+    async def get_fault_info(self):
         """report exception from engine_core"""
         raise NotImplementedError
 

vllm/entrypoints/api_server.py

@@ -95,8 +95,8 @@ async def process_fault_tolerance_instruction(request: Request) -> Response:
 async def get_fault_info() -> Response:
     """Health check."""
     assert engine is not None
-    engine_exception_dict = await engine.exception_reporter()
-    return Response(json.dumps(engine_exception_dict), status_code=200)
+    engine_status_dict = await engine.get_fault_info()
+    return Response(json.dumps(engine_status_dict), status_code=200)
 
 
 @with_cancellation

vllm/entrypoints/openai/api_server.py

@@ -263,6 +263,20 @@ async def validate_json_request(raw_request: Request):
         )
 
 
+async def check_engine_fault(raw_request: Request):
+    client = engine_client(raw_request)
+    assert hasattr(client, "engine_core")
+    core_client = client.engine_core
+    if (
+        hasattr(core_client, "client_sentinel")
+        and core_client.client_sentinel.is_faulted.is_set()
+    ):
+        raise HTTPException(
+            status_code=503,
+            detail="Service is in faulted state, cannot process requests.",
+        )
+
+
 router = APIRouter()
 
 
@@ -395,7 +409,7 @@ async def get_server_load_metrics(request: Request):
 
 @router.post(
     "/tokenize",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
         HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
@@ -430,7 +444,7 @@ async def tokenize(request: TokenizeRequest, raw_request: Request):
 
 @router.post(
     "/detokenize",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
         HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
@@ -505,7 +519,7 @@ async def _convert_stream_to_sse_events(
 
 @router.post(
     "/v1/responses",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
         HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
@@ -598,7 +612,7 @@ async def cancel_responses(response_id: str, raw_request: Request):
 
 @router.post(
     "/v1/messages",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
         HTTPStatus.BAD_REQUEST.value: {"model": AnthropicErrorResponse},
@@ -654,7 +668,7 @@ def translate_error_response(response: ErrorResponse) -> JSONResponse:
 
 @router.post(
     "/v1/chat/completions",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
         HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
@@ -695,7 +709,7 @@ async def create_chat_completion(request: ChatCompletionRequest, raw_request: Re
 
 @router.post(
     "/v1/completions",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.OK.value: {"content": {"text/event-stream": {}}},
         HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
@@ -741,7 +755,7 @@ async def create_completion(request: CompletionRequest, raw_request: Request):
 
 @router.post(
     "/v1/embeddings",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
         HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
@@ -784,7 +798,7 @@ async def create_embedding(
 
 @router.post(
     "/pooling",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
         HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
@@ -820,7 +834,10 @@ async def create_pooling(request: PoolingRequest, raw_request: Request):
     assert_never(generator)
 
 
-@router.post("/classify", dependencies=[Depends(validate_json_request)])
+@router.post(
+    "/classify",
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
+)
 @with_cancellation
 @load_aware_call
 async def create_classify(request: ClassificationRequest, raw_request: Request):
@@ -849,7 +866,7 @@ async def create_classify(request: ClassificationRequest, raw_request: Request):
 
 @router.post(
     "/score",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
         HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
@@ -882,7 +899,7 @@ async def create_score(request: ScoreRequest, raw_request: Request):
 
 @router.post(
     "/v1/score",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
         HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
@@ -979,7 +996,7 @@ async def create_translations(
 
 @router.post(
     "/rerank",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
         HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
@@ -1011,7 +1028,7 @@ async def do_rerank(request: RerankRequest, raw_request: Request):
 
 @router.post(
     "/v1/rerank",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
         HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
@@ -1030,7 +1047,7 @@ async def do_rerank_v1(request: RerankRequest, raw_request: Request):
 
 @router.post(
     "/v2/rerank",
-    dependencies=[Depends(validate_json_request)],
+    dependencies=[Depends(validate_json_request), Depends(check_engine_fault)],
     responses={
         HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
         HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
@@ -1303,8 +1320,8 @@ async def get_fault_info(
     raw_request: Request,
 ):
     client = engine_client(raw_request)
-    engine_exception_dict = await client.exception_reporter()
-    return JSONResponse(content=engine_exception_dict)
+    engine_status_dict = await client.get_fault_info()
+    return JSONResponse(content=engine_status_dict)
 
 
 # NOTE: Construct the TypeAdapters only once

vllm/v1/engine/async_llm.py

@@ -785,7 +785,7 @@ async def handle_fault(
         """send fault tolerance instruction to the engine"""
         return await self.engine_core.handle_fault(instruction, timeout, **kwargs)
 
-    async def exception_reporter(self):
+    async def get_fault_info(self):
         """report exception in engine core"""
         return await self.engine_core.fault_reporter()
 

vllm/v1/engine/core_client.py

@@ -6,6 +6,7 @@
 import multiprocessing
 import queue
 import sys
+import threading
 import time
 import uuid
 import weakref
@@ -355,6 +356,7 @@ def __init__(
         fault_pub_addr: str,
         engine_status_dict: ThreadSafeDict[int, str],
     ):
+        self.is_faulted = threading.Event()
         self.engine_registry = engine_registry
         self.zmq_ctx = zmq.Context()
         self.fault_receiver_socket = make_zmq_socket(
@@ -411,12 +413,15 @@ async def handle_fault(self, instruction: str, timeout: int, **kwargs) -> bool:
         to handle system anomalies, ensuring stable operation or graceful degradation
         of the relevant components.
         """
-        return await run_method(
+        result = await run_method(
             self.fault_handler,
             "handle_fault",
             args=(instruction, timeout),
             kwargs=kwargs,
         )
+        if result:
+            self.is_faulted.clear()
+        return result
 
     def fault_receiver(self):
         """
@@ -443,7 +448,7 @@ def fault_receiver(self):
                 self.fault_pub_socket.send_string(
                     f"vllm_fault|{json.dumps(self.engine_status_dict.to_dict())}"
                 )
-
+                self.is_faulted.set()
                 # Pause healthy engines on fault.
                 # Pause can be invoked again during fault-tolerance handling,
                 # so it's unnecessary to track whether all engines are currently