[APIServer][BugFix] Fix gunicorn fork deadlock and multi-dp pipe issue in get_save_output_v1

fastdeploy/entrypoints/openai/api_server.py

@@ -319,6 +319,8 @@ async def lifespan(app: FastAPI):
     try:
         if envs.FD_ENABLE_ASYNC_LLM:
             await llm_engine.shutdown()
+        else:
+            llm_engine._exit_sub_services()
         await engine_client.connection_manager.close()
         engine_client.zmq_client.close()
         from prometheus_client import multiprocess
@@ -793,7 +795,16 @@ def launch_api_server() -> None:
     }
 
     try:
-        StandaloneApplication(app, options).run()
+        if args.workers > 1:
+            StandaloneApplication(app, options).run()
+        else:
+            uvicorn.run(
+                app,
+                host=args.host,
+                port=args.port,
+                log_config=UVICORN_CONFIG,
+            )
+
     except Exception as e:
         api_server_logger.error(f"launch sync http server error, {e}, {str(traceback.format_exc())}")
 

fastdeploy/output/token_processor.py

@@ -422,21 +422,17 @@ def process_sampling_results_use_zmq(self):
         """
         if self.speculative_decoding:
             raise NotImplementedError("GET_SAVE_OUTPUT_V1 does not support speculative decoding")
-        rank_id = self.cfg.parallel_config.local_data_parallel_id
         while True:
             try:
-                if (
-                    self.cfg.parallel_config.enable_expert_parallel and self.cfg.parallel_config.data_parallel_size > 1
-                ) or (rank_id == 0):
-                    receive_datas = self.zmq_server.recv_pyobj()
-                    assert isinstance(receive_datas, list)
-                    if envs.FD_DEBUG:
-                        llm_logger.debug(f"token_processor receive_data {receive_datas}")
+                receive_datas = self.zmq_server.recv_pyobj()
+                assert isinstance(receive_datas, list)
+                if envs.FD_DEBUG:
+                    llm_logger.debug(f"token_processor receive_data {receive_datas}")
 
-                    self._reschedule_preempt_task_use_zmq(receive_datas)
+                self._reschedule_preempt_task_use_zmq(receive_datas)
 
-                    batch_result = self._process_batch_output_use_zmq(receive_datas)
-                    self.postprocess(batch_result)
+                batch_result = self._process_batch_output_use_zmq(receive_datas)
+                self.postprocess(batch_result)
             except Exception as e:
                 log_request_error(
                     message="Receive message:{receive_datas}, error:{error}, {traceback}",

fastdeploy/worker/gpu_model_runner.py

@@ -259,8 +259,9 @@ def __init__(
         self.async_output_queue = None
         if envs.FD_USE_GET_SAVE_OUTPUT_V1:
             port = self.fd_config.parallel_config.local_engine_worker_queue_port
-            logger.info(f"zmq client get_save_output_rank{local_rank}_{port}")
-            self.zmq_client = ZmqIpcClient(name=f"get_save_output_rank{local_rank}_{port}", mode=zmq.PUSH)
+            rank = self.fd_config.parallel_config.local_data_parallel_id
+            logger.info(f"zmq client get_save_output_rank{rank}_{port}")
+            self.zmq_client = ZmqIpcClient(name=f"get_save_output_rank{rank}_{port}", mode=zmq.PUSH)
             self.zmq_client.connect()
             self.zmq_client.socket.SNDTIMEO = 3000
             self.async_output_queue: queue.Queue = queue.Queue()

fastdeploy/worker/metax_model_runner.py

@@ -210,8 +210,9 @@ def __init__(
         self.async_output_queue = None
         if envs.FD_USE_GET_SAVE_OUTPUT_V1:
             port = self.fd_config.parallel_config.local_engine_worker_queue_port
-            logger.info(f"zmq client get_save_output_rank{local_rank}_{port}")
-            self.zmq_client = ZmqIpcClient(name=f"get_save_output_rank{local_rank}_{port}", mode=zmq.PUSH)
+            rank = self.fd_config.parallel_config.local_data_parallel_id
+            logger.info(f"zmq client get_save_output_rank{rank}_{port}")
+            self.zmq_client = ZmqIpcClient(name=f"get_save_output_rank{rank}_{port}", mode=zmq.PUSH)
             self.zmq_client.connect()
             self.zmq_client.socket.SNDTIMEO = 3000
             self.async_output_queue: queue.Queue = queue.Queue()

fastdeploy/worker/xpu_model_runner.py

@@ -197,8 +197,9 @@ def __init__(
         self.async_output_queue = None
         if envs.FD_USE_GET_SAVE_OUTPUT_V1:
             port = self.fd_config.parallel_config.local_engine_worker_queue_port
-            logger.info(f"zmq client get_save_output_rank{local_rank}_{port}")
-            self.zmq_client = ZmqIpcClient(name=f"get_save_output_rank{local_rank}_{port}", mode=zmq.PUSH)
+            rank = self.fd_config.parallel_config.local_data_parallel_id
+            logger.info(f"zmq client get_save_output_rank{rank}_{port}")
+            self.zmq_client = ZmqIpcClient(name=f"get_save_output_rank{rank}_{port}", mode=zmq.PUSH)
             self.zmq_client.connect()
             self.zmq_client.socket.SNDTIMEO = 3000
             self.async_output_queue: queue.Queue = queue.Queue()

tests/entrypoints/openai/test_api_server.py

@@ -667,12 +667,23 @@ def test_launchers_and_controller():
         with pytest.raises(Exception):
             api_server.launch_api_server()
 
+    # workers > 1 branch: StandaloneApplication
+    api_server.args.workers = 2
     with (
         patch("fastdeploy.entrypoints.openai.api_server.is_port_available", return_value=True),
         patch("fastdeploy.entrypoints.openai.api_server.StandaloneApplication.run", side_effect=RuntimeError("fail")),
     ):
         api_server.launch_api_server()
 
+    # workers == 1 branch: uvicorn.run
+    api_server.args.workers = 1
+    with (
+        patch("fastdeploy.entrypoints.openai.api_server.is_port_available", return_value=True),
+        patch("fastdeploy.entrypoints.openai.api_server.uvicorn.run", side_effect=RuntimeError("fail")) as uv_run,
+    ):
+        api_server.launch_api_server()
+        uv_run.assert_called_once()
+
     with patch("fastdeploy.entrypoints.openai.api_server.uvicorn.run") as uv_run:
         api_server.run_metrics_server()
         api_server.run_controller_server()