mv connection_manager init (#3901)

Co-authored-by: Yuanle Liu <yuanlehome@163.com>

Author: ltd0924

fastdeploy/entrypoints/openai/api_server.py

@@ -185,6 +185,7 @@ async def lifespan(app: FastAPI):
         workers=args.workers,
         tool_parser=args.tool_call_parser,
     )
+    await engine_client.connection_manager.initialize()
     app.state.dynamic_load_weight = args.dynamic_load_weight
     model_handler = OpenAIServingModels(
         model_paths,

fastdeploy/entrypoints/openai/serving_chat.py

@@ -74,12 +74,6 @@ def __init__(
             self.master_ip = "0.0.0.0"
         api_server_logger.info(f"master ip: {self.master_ip}")
 
-    async def _ensure_connection_manager(self):
-        """ensure connection manager initialized"""
-        if not self.engine_client.connection_initialized:
-            await self.engine_client.connection_manager.initialize()
-            self.engine_client.connection_initialized = True
-
     def _check_master(self):
         return self.engine_client.is_master
 
@@ -208,7 +202,6 @@ async def chat_completion_stream_generator(
         api_server_logger.info(f"create chat completion request: {request_id}")
 
         try:
-            await self._ensure_connection_manager()
             dealer, response_queue = await self.engine_client.connection_manager.get_connection(request_id)
             dealer.write([b"", request_id.encode("utf-8")])
             choices = []
@@ -421,7 +414,6 @@ async def chat_completion_full_generator(
 
         include_stop_str_in_output = request.include_stop_str_in_output
         try:
-            await self._ensure_connection_manager()
             dealer, response_queue = await self.engine_client.connection_manager.get_connection(request_id)
             dealer.write([b"", request_id.encode("utf-8")])
             final_res = None

fastdeploy/entrypoints/openai/serving_completion.py

@@ -51,12 +51,6 @@ def __init__(self, engine_client, models, pid, ips, max_waiting_time):
         else:
             self.master_ip = "0.0.0.0"
 
-    async def _ensure_connection_manager(self):
-        """ensure connection manager initialized"""
-        if not self.engine_client.connection_initialized:
-            await self.engine_client.connection_manager.initialize()
-            self.engine_client.connection_initialized = True
-
     def _check_master(self):
         return self.engine_client.is_master
 
@@ -208,7 +202,6 @@ async def completion_full_generator(
         try:
             request_ids = [f"{request_id}-{i}" for i in range(num_choices)]
             # create dealer
-            await self._ensure_connection_manager()
             dealer, response_queue = await self.engine_client.connection_manager.get_connection(
                 request_id, num_choices
             )
@@ -314,7 +307,6 @@ async def completion_stream_generator(
         Process the stream completion request.
         """
         try:
-            await self._ensure_connection_manager()
             dealer, response_queue = await self.engine_client.connection_manager.get_connection(
                 request_id, num_choices
             )