fix

Author: wangzaijun

lightllm/server/router/manager.py

@@ -51,7 +51,6 @@ def __init__(self, args: StartArgs, router_port, detokenization_port, metric_por
         self.schedule_time_interval = args.schedule_time_interval  # 默认30ms 的调度周期
         # 兼容多机纯tp的运行模式，这时候 1 // 2 == 0, 需要兼容
         self.dp_size_in_node = max(1, args.dp // self.nnodes)
-        self.dp_world_size = self.world_size // self.dp_size
         self.is_multinode_tp = args.nnodes > 1 and args.dp == 1
         self.is_multinode_tp_master = self.is_multinode_tp and args.node_rank == 0
         self.is_multinode_tp_slave = self.is_multinode_tp and args.node_rank != 0
@@ -114,14 +113,12 @@ async def wait_to_model_ready(self):
         self.mem_queues: List[torch.multiprocessing.Queue] = [
             torch.multiprocessing.Queue() for _ in range(self.node_world_size)
         ]
-        self.result_queues: List[mp.Queue] = [mp.Queue() for _ in range(self.node_world_size)]
         self.rpc_event = multiprocessing.Event()
         self.rpc_finished_event = multiprocessing.Event()
 
         assert (self.world_size % self.nnodes) == 0
         node_world_size = self.world_size // self.nnodes
         for rank_id in range(self.node_rank * node_world_size, (self.node_rank + 1) * node_world_size):
-
             rpc_model = await start_model_process(
                 args=self.args,
                 rank=rank_id,
@@ -130,8 +127,7 @@ async def wait_to_model_ready(self):
                 rpc_event=self.rpc_event,
                 rpc_finished_event=self.rpc_finished_event,
                 info_queue=self.info_queue,
-                result_queue=self.result_queues[rank_id % node_world_size],
-                mem_queue=self.mem_queues[rank_id % node_world_size],
+                mem_queue=self.mem_queues[(rank_id % node_world_size)],
                 router_lock=self.router_lock,
             )
             self.model_rpc_servers.append(rpc_model)
@@ -184,7 +180,7 @@ async def wait_to_model_ready(self):
                 get_unique_server_name(),
                 self.max_total_token_num,
                 node_world_size=self.node_world_size,
-                dp_world_size=self.dp_world_size,
+                dp_world_size=self.world_size // self.dp_size,
             )
         self.req_queue = build_req_queue(self.args, self, self.dp_size_in_node)
         logger.info(f"use req queue {self.req_queue.__class__.__name__}")
@@ -197,30 +193,6 @@ async def wait_to_model_ready(self):
 
             start_prefill_kv_move_manager_process(self.args, self.info_queue, self.mem_queues)
 
-        if self.args.run_mode == "nixl_prefill":
-            from lightllm.server.router.model_infer.mode_backend.pd_nixl.pd_remote_prefill import (
-                start_pd_remote_prefill_server_process,
-            )
-
-            dist_info = DistInfo(
-                self.world_size,
-                self.nnodes,
-                self.dp_size,
-                self.dp_world_size,
-                self.dp_size_in_node,
-                self.node_world_size,
-            )
-
-            start_pd_remote_prefill_server_process(
-                self.args.pd_node_id,
-                dist_info=dist_info,
-                http_server_port=self.args.pd_nixl_remote_prefill_http_port,
-                server_port=self.args.pd_nixl_remote_prefill_port,
-                from_backend_queue=self.info_queue,
-                to_backend_queues=self.result_queues,
-                agent_meta_queues=self.mem_queues,
-            )
-
         if self.args.run_mode == "decode":
             # 启动 decode kv move 管理进程
             from lightllm.server.router.model_infer.mode_backend.continues_batch.pd_mode.decode_node_impl import (
@@ -229,28 +201,6 @@ async def wait_to_model_ready(self):
 
             start_decode_kv_move_manager_process(self.args, self.info_queue, self.mem_queues)
 
-        if self.args.run_mode == "nixl_decode":
-            from lightllm.server.router.model_infer.mode_backend.pd_nixl.pd_remote_prefill import (
-                start_pd_remote_prefill_client_process,
-            )
-
-            dist_info = DistInfo(
-                self.world_size,
-                self.nnodes,
-                self.dp_size,
-                self.dp_world_size,
-                self.dp_size_in_node,
-                self.node_world_size,
-            )
-
-            start_pd_remote_prefill_client_process(
-                self.args.pd_node_id,
-                dist_info,
-                from_backend_queue=self.info_queue,
-                to_backend_queues=self.result_queues,
-                agent_meta_queues=self.mem_queues,
-            )
-
         return
 
     def _get_schedule_time_interval(self):
@@ -459,7 +409,8 @@ def _add_req(self, group_req_indexes: GroupReqIndexes):
             req._router_stop_str_matched = False
 
             if isinstance(req, PDNIXLChunkedPrefillReq):
-                req.set_dp_world_size(self.dp_world_size)
+                dp_world_size = self.world_size // self.dp_size
+                req.set_dp_world_size(dp_world_size)
 
             req_group.append(req)