Repository: vllm-project/vllm
Pull request: [BugFix] Fix frontend multiprocessing hang
Now that the OpenAI server has [optional multiprocessing](https://github.com/vllm-project/vllm/pull/6883), there is a hang that can happen if the backend dies during initialization and never replies to the `IS_SERVER_READY` message sent in `async_engine_client.setup()` . 

This PR add an optional timeout to `_send_one_way_rpc_request` so that during the initialization we can check periodically if the server process is still running while we wait for the reply to the `IS_SERVER_READY` message.

FIX #7213 

FYI @njhill @robertgshaw2-neuralmagic 

Problem statement:
[Bug]: With frontend multiprocessing the openai server hangs if there is an initialization error
### Your current environment

The environment is not relevant for this issue.


### 🐛 Describe the bug

With frontend multiprocessing, if there is an error in the initialization if the LLMEngine, the server prints the stack trace and then hangs and has to be killed. The expected behavior is that the server would exit, ideally with a non-zero status.

Reproducing this error is pretty simple. Just add `raise Exception("foo")` at the first line of `AsyncLLEngine.from_engine_args()`. 

Passing `--disable-frontend-multiprocessing` makes this problem go away.

Hints from issue discussion:
This happens with version `1f26efb`.

Commit message: Merge branch 'vllm-project:main' into fix_multiprocess_hang

Patch to review:
diff --git a/vllm/entrypoints/openai/api_server.py b/vllm/entrypoints/openai/api_server.py
index 88f0bd4ee4d..9d69002edd1 100644
--- a/vllm/entrypoints/openai/api_server.py
+++ b/vllm/entrypoints/openai/api_server.py
@@ -116,7 +116,14 @@ async def build_async_engine_client(args) -> AsyncIterator[AsyncEngineClient]:
 
         # Build RPCClient, which conforms to AsyncEngineClient Protocol.
         async_engine_client = AsyncEngineRPCClient(port)
-        await async_engine_client.setup()
+
+        while True:
+            try:
+                await async_engine_client.setup()
+                break
+            except TimeoutError as e:
+                if not rpc_server_process.is_alive():
+                    raise RuntimeError("Server crashed") from e
 
         try:
             yield async_engine_client
diff --git a/vllm/entrypoints/openai/rpc/client.py b/vllm/entrypoints/openai/rpc/client.py
index 45bf88b5bf5..49f55608766 100644
--- a/vllm/entrypoints/openai/rpc/client.py
+++ b/vllm/entrypoints/openai/rpc/client.py
@@ -18,6 +18,9 @@
 from vllm.sampling_params import SamplingParams
 from vllm.transformers_utils.tokenizer_group import init_tokenizer_from_configs
 
+# Time to wait before checking it the server process is alive.
+SERVER_START_TIMEOUT = 1000
+
 
 class AsyncEngineRPCClient:
 
@@ -85,14 +88,19 @@ async def _send_get_data_rpc_request(self, request: RPCUtilityRequest,
 
         return data
 
-    async def _send_one_way_rpc_request(self, request: RPC_REQUEST_TYPE,
-                                        error_message: str):
+    async def _send_one_way_rpc_request(self,
+                                        request: RPC_REQUEST_TYPE,
+                                        error_message: str,
+                                        timeout: int = 0):
         """Send one-way RPC request to trigger an action."""
         with self.socket() as socket:
             # Ping RPC Server with request.
             await socket.send(cloudpickle.dumps(request))
 
             # Await acknowledgement from RPCServer.
+            if timeout > 0 and await socket.poll(timeout) == 0:
+                raise TimeoutError(f"server didn't reply within {timeout} ms")
+
             response = cloudpickle.loads(await socket.recv())
 
         if not isinstance(response, str) or response != VLLM_RPC_SUCCESS_STR:
@@ -117,7 +125,8 @@ async def wait_for_server(self):
 
         await self._send_one_way_rpc_request(
             request=RPCUtilityRequest.IS_SERVER_READY,
-            error_message="Unable to start RPC Server.")
+            error_message="Unable to start RPC Server.",
+            timeout=SERVER_START_TIMEOUT)
 
     async def _get_model_config_rpc(self) -> ModelConfig:
         """Get the ModelConfig object from the RPC Server"""


Code review comments:
diff_hunk: @@ -85,14 +88,19 @@ async def _send_get_data_rpc_request(self, request: RPCUtilityRequest,
 
         return data
 
-    async def _send_one_way_rpc_request(self, request: RPC_REQUEST_TYPE,
-                                        error_message: str):
+    async def _send_one_way_rpc_request(self,
+                                        request: RPC_REQUEST_TYPE,
+                                        error_message: str,
+                                        timeout: int = 0):
         """Send one-way RPC request to trigger an action."""
         with self.socket() as socket:
             # Ping RPC Server with request.
             await socket.send(cloudpickle.dumps(request))
 
             # Await acknowledgement from RPCServer.
+            if timeout > 0 and await socket.poll(timeout) == 0:
path: vllm/entrypoints/openai/rpc/client.py
text: @user1:
```suggestion
            if timeout is not None and await socket.poll(timeout=timeout) == 0:
```

diff_hunk: @@ -116,7 +116,14 @@ async def build_async_engine_client(args) -> AsyncIterator[AsyncEngineClient]:
 
         # Build RPCClient, which conforms to AsyncEngineClient Protocol.
         async_engine_client = AsyncEngineRPCClient(port)
-        await async_engine_client.setup()
+
+        while True:
+            try:
+                await async_engine_client.setup()
+                break
+            except TimeoutError as e:
+                if not rpc_server_process.is_alive():
+                    raise RuntimeError("Server crashed") from e
path: vllm/entrypoints/openai/api_server.py
text: @user1:
Would be best to move this within the `try` block below I think

@author:
Right, so that the clean up code in the finally section is called. Just moving the code into the try block caused another "hang" because it turns out that `context.destroy()` by default waits for 30 seconds if there are undelivered messages, which is the case if the server process crashed. So I added `linger=0` to the socket.close() calls to prevent that. I think this makes sense because in all uses of `self.socket()` the client sends a request and waits for the answer, so in all these cases we know that the server has seen the message.

diff_hunk: @@ -85,14 +88,19 @@ async def _send_get_data_rpc_request(self, request: RPCUtilityRequest,
 
         return data
 
-    async def _send_one_way_rpc_request(self, request: RPC_REQUEST_TYPE,
-                                        error_message: str):
+    async def _send_one_way_rpc_request(self,
+                                        request: RPC_REQUEST_TYPE,
+                                        error_message: str,
+                                        timeout: int = 0):
path: vllm/entrypoints/openai/rpc/client.py
text: @user1:
I think default here should be `None` rather than 0. Could you include `Optional[int]` type

@author:
Oh yeah, C++ programmer habits...