diff --git a/fastdeploy/entrypoints/openai/api_server.py b/fastdeploy/entrypoints/openai/api_server.py index 2c108dda0af..d0ac8f147b7 100644 --- a/fastdeploy/entrypoints/openai/api_server.py +++ b/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())}") diff --git a/fastdeploy/output/token_processor.py b/fastdeploy/output/token_processor.py index 795e8d072a5..ae1decc6021 100644 --- a/fastdeploy/output/token_processor.py +++ b/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}", diff --git a/fastdeploy/worker/gpu_model_runner.py b/fastdeploy/worker/gpu_model_runner.py index 4dbf153208f..276ea1ae33d 100644 --- a/fastdeploy/worker/gpu_model_runner.py +++ b/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() diff --git a/fastdeploy/worker/metax_model_runner.py b/fastdeploy/worker/metax_model_runner.py index b4a238139c8..1053ad2fe0c 100644 --- a/fastdeploy/worker/metax_model_runner.py +++ b/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() diff --git a/fastdeploy/worker/xpu_model_runner.py b/fastdeploy/worker/xpu_model_runner.py index ded76bdcc1d..7c8f58be4d4 100644 --- a/fastdeploy/worker/xpu_model_runner.py +++ b/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() diff --git a/tests/entrypoints/openai/test_api_server.py b/tests/entrypoints/openai/test_api_server.py index 4ae1d58adcd..316e6aae3a2 100644 --- a/tests/entrypoints/openai/test_api_server.py +++ b/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()