fix ci

Author: ltd0924

fastdeploy/engine/common_engine.py

@@ -198,7 +198,11 @@ def start_worker_queue_service(self):
                 local_data_parallel_size=self.cfg.parallel_config.data_parallel_size,
             )
 
-            if self.cfg.cache_config.enable_prefix_caching or self.cfg.splitwise_role != "mixed":
+            if (
+                self.cfg.cache_config.enable_prefix_caching
+                or self.cfg.splitwise_role != "mixed"
+                and self.cfg.parallel_config.local_data_parallel_id == 0
+            ):
                 self.cache_task_queue = EngineCacheQueue(
                     address=(
                         self.cfg.master_ip,
@@ -726,7 +730,9 @@ def start_cache_service(self, device_ids, ipc_signal_suffix):
             tensor_parallel_size=self.cfg.tensor_parallel_size,
             device_ids=device_ids,
             pod_ip=self.cfg.master_ip,
-            engine_worker_queue_port=self.cfg.engine_worker_queue_port,
+            engine_worker_queue_port=int(
+                self.cfg.engine_worker_queue_port[self.cfg.parallel_config.local_data_parallel_id]
+            ),
             pid_suffix=ipc_signal_suffix,
         )
 

fastdeploy/engine/engine.py

@@ -109,8 +109,7 @@ def start(self, api_server_pid=None):
         start_time = time.time()
 
         self.api_server_pid = api_server_pid
-        self.engine_pid = os.getpid()
-        self.ipc_signal_suffix = self.engine_pid if self.api_server_pid is None else self.api_server_pid
+        self.ipc_signal_suffix = self.cfg.engine_worker_queue_port[0]
         self._init_worker_signals()
 
         self.data_processor = self.input_processor.create_processor()
@@ -445,7 +444,7 @@ def _start_worker_service(self):
             f" --enc_dec_block_num {self.cfg.cache_config.enc_dec_block_num}"
             f" --eos_tokens_lens {self.data_processor.eos_token_id_len}"
             f" --pad_token_id {self.data_processor.pad_token_id}"
-            f" --engine_pid {self.engine_pid}"
+            f" --engine_pid {self.cfg.engine_worker_queue_port[0]}"
             f" --max_num_batched_tokens {self.cfg.max_num_batched_tokens}"
             f" --splitwise_role {self.cfg.splitwise_role}"
             f" --kv_cache_ratio {self.cfg.cache_config.kv_cache_ratio}"
@@ -600,20 +599,19 @@ def launch_components(self):
             self.engine.scheduler.start(role, host_ip, disaggregate)
 
         if not envs.FD_ENABLE_MULTI_API_SERVER:
-            time.sleep(1)
             if self.cfg.parallel_config.enable_expert_parallel and self.cfg.parallel_config.data_parallel_size > 1:
                 self.dp_processed = []
                 for i in range(
                     1,
                     self.cfg.parallel_config.data_parallel_size // self.cfg.nnode,
                 ):
-                    time.sleep(1)
+                    time.sleep(3)
                     self.dp_processed.append(
                         multiprocessing.Process(
                             target=start_data_parallel_service,
                             args=(
                                 self.cfg,
-                                i + self.cfg.node_rank * self.cfg.worker_num_per_node,
+                                i,
                             ),
                         )
                     )

fastdeploy/engine/expert_service.py

@@ -62,6 +62,7 @@ def __init__(self, cfg, local_data_parallel_id):
                 )
             else:
                 self.cfg.cache_config.pd_comm_port = [self.cfg.cache_config.pd_comm_port[local_data_parallel_id]]
+        self.cfg.parallel_config.local_data_parallel_id = local_data_parallel_id
 
         self.engine = EngineSevice(self.cfg)
         if self.cfg.scheduler_config.name == "splitwise":
@@ -83,7 +84,7 @@ def start(self, ipc_signal_suffix, local_data_parallel_id):
             self.api_server_pid = ipc_signal_suffix
             self.engine.start_zmq_service(ipc_signal_suffix)
         else:
-            ipc_signal_suffix = os.getpid()
+            ipc_signal_suffix = self.cfg.engine_worker_queue_port[0]
 
         llm_logger.info(f"start expert service {local_data_parallel_id}")
         if self.cfg.splitwise_role != "mixed":

fastdeploy/inter_communicator/ipc_signal.py

@@ -78,7 +78,9 @@ def __init__(
             name = name + f".{suffix}"
 
         if create:
-            assert not shared_memory_exists(name), f"ShareMemory: {name} already exists"
+            if shared_memory_exists(name):
+                print(f"ShareMemory: {name} already exists, delete it")
+                SharedMemory(name=name, create=False).unlink()
             self.shm = SharedMemory(create=True, size=array.nbytes, name=name)
             self.value: np.ndarray = np.ndarray(array.shape, dtype=array.dtype, buffer=self.shm.buf)
             self.value[:] = array  # Initialize with input array data

fastdeploy/worker/worker_process.py

@@ -152,19 +152,7 @@ def __init__(self, fd_config: FDConfig, ranks: int = 1, local_rank: int = 0) ->
         # TODO(gongshaotian): Use worker factory to get worker
         self.worker = get_worker(fd_config=fd_config, local_rank=self.local_rank, rank=self.ranks)
 
-        # Initialize task queue
-        task_address = (
-            self.parallel_config.pod_ip,
-            self.parallel_config.engine_worker_queue_port,
-        )
         self.max_chips_per_node = 16 if current_platform.is_iluvatar() else 8
-        self.task_queue = TaskQueue(
-            address=task_address,
-            is_server=False,
-            num_client=self.parallel_config.tensor_parallel_size,
-            client_id=self.parallel_config.tensor_parallel_rank,
-            local_data_parallel_id=self.parallel_config.expert_parallel_rank,
-        )
 
     def init_health_status(self) -> None:
         """
@@ -440,6 +428,20 @@ def init_device(self) -> None:
         """Initialize device and Construct model runner"""
         self.worker.init_device()
 
+    def start_queue_service(self):
+        # Initialize task queue
+        task_address = (
+            self.parallel_config.pod_ip,
+            self.parallel_config.engine_worker_queue_port,
+        )
+        self.task_queue = TaskQueue(
+            address=task_address,
+            is_server=False,
+            num_client=self.parallel_config.tensor_parallel_size,
+            client_id=self.parallel_config.tensor_parallel_rank,
+            local_data_parallel_id=self.parallel_config.expert_parallel_rank,
+        )
+
     def load_model(self) -> None:
         """Load weights and create model"""
 
@@ -773,6 +775,8 @@ def run_worker_proc() -> None:
     # Initialize health status
     worker_proc.init_health_status()
 
+    worker_proc.start_queue_service()
+
     # Start event loop
     if fd_config.parallel_config.use_ep:
         # TODO(wufeisheng): Delete this branch