remove all redundant_enable

Signed-off-by: wanghanqingLYT <wanghanqing3@huawei.com>

Author: wanghanqingLYT

vllm_ascend/eplb/core/eplb_device_transfer_loader.py

@@ -96,7 +96,7 @@ def asyn_expert_weight_transfer(self, reqs):
 
         self.state = ExpertWeightUpdateState.TRANSFERRING
 
-    def update_expert_map_and_weight(self, reqs, redundant_enable):
+    def update_expert_map_and_weight(self, reqs):
         # Only after send/recv tasks have been luanched, expert_map and weight can be updated
         if self.state != ExpertWeightUpdateState.TRANSFERRING:
             return
@@ -113,9 +113,8 @@ def update_expert_map_and_weight(self, reqs, redundant_enable):
                                                self.updated_expert_map)
 
         # update log2phy_map
-        if redundant_enable:
-            self.eplb_adaptor.do_update_log2phy_map(self.layer_id,
-                                                    self.updated_log2phy_map)
+        self.eplb_adaptor.do_update_log2phy_map(self.layer_id,
+                                                self.updated_log2phy_map)
 
         # update expert weight
         buffer_tensor_id = 0

vllm_ascend/eplb/core/eplb_worker.py

@@ -31,18 +31,13 @@
 
 class EplbWorker:
 
-    def __init__(self,
-                 shared_dict,
-                 policy_type,
-                 enable_d2d: bool = True,
-                 redundant_enable=0):
+    def __init__(self, shared_dict, policy_type, enable_d2d: bool = True):
         self.policy_type = policy_type
         self.policy = PolicyFactory.generate_policy(policy_type,
                                                     DynamicConfig())
         self.shared_dict = shared_dict
         self.old_expert_maps = None
         self.enable_d2d = enable_d2d
-        self.redundant_enable = redundant_enable
         self.rank_id = dist.get_rank()
 
     def do_update(self):
@@ -381,11 +376,8 @@ def pack_update_info(self, update_info_generator):
 
             maps.append(new_expert_map[self.rank_id].numpy().tolist())
 
-            if self.redundant_enable:
-                log2phy_map = generate_log2phy_map(new_expert_map)
-                log2phy_all.append(log2phy_map[self.rank_id].numpy().tolist())
-            else:
-                log2phy_all.append([])
+            log2phy_map = generate_log2phy_map(new_expert_map)
+            log2phy_all.append(log2phy_map[self.rank_id].numpy().tolist())
 
             layer_ids.append(layer_id)
 
@@ -398,7 +390,6 @@ def __init__(self,
                  shared_dict,
                  planner_q,
                  block_update_q,
-                 redundant_enable,
                  policy_type: int = 0,
                  enable_d2d: bool = True):
         """
@@ -412,11 +403,10 @@ def __init__(self,
         self.enable_d2d = enable_d2d
         self.planner_q = planner_q
         self.block_update_q = block_update_q
-        self.redundant_enable = redundant_enable
 
         # Create EplbWorker instance
         self.worker = EplbWorker(self.shared_dict, self.policy_type,
-                                 self.enable_d2d, self.redundant_enable)
+                                 self.enable_d2d)
 
     def worker_process(self, planner_q, block_update_q):
         """

vllm_ascend/eplb/eplb_updator.py

@@ -45,7 +45,6 @@ def init_eplb(self, expert_map_path):
         self.rank_id = dist.get_rank()
         self.num_expert_load_gather = 10
         self.periodic_load_gather = True
-        self.redundant_enable = (expert_map_path is not None)
         self.num_iterations_eplb_update: torch.int64 = self.ascend_config.num_iterations_eplb_update
         self.expert_map_path = expert_map_path
 
@@ -83,7 +82,6 @@ def init_eplb(self, expert_map_path):
         self.eplb = EplbProcess(shared_dict=self.shared_dict,
                                 planner_q=self.planner_block_queue,
                                 block_update_q=self.block_update_queue,
-                                redundant_enable=self.redundant_enable,
                                 policy_type=1,
                                 enable_d2d=True)
 
@@ -132,10 +130,8 @@ def forward_before(self):
         if self.update_expert_weight_flag():
             (expert_send_info, expert_recv_info, updated_expert_map,
              log2phy_map, layer_id) = self.update_info_all.pop(0)
-            if self.redundant_enable:
-                log2phy_map_this_rank = torch.from_numpy(
-                    numpy.array(log2phy_map))
-                self.eplb_loader.set_log2phy_map(log2phy_map_this_rank)
+            log2phy_map_this_rank = torch.from_numpy(numpy.array(log2phy_map))
+            self.eplb_loader.set_log2phy_map(log2phy_map_this_rank)
             updated_expert_map_this_rank = torch.from_numpy(
                 numpy.array(updated_expert_map))
             # logger.info(f"check update info, layer = {layer_id}, send = {expert_send_info_this_rank}, recv = {expert_recv_info_this_rank}")
@@ -159,8 +155,7 @@ def forward_end(self):
             self.wakeup_eplb_worker()
 
         if self.update_expert_weight_flag():
-            self.eplb_loader.update_expert_map_and_weight(
-                self.reqs, self.redundant_enable)
+            self.eplb_loader.update_expert_map_and_weight(self.reqs)
 
         self.update_iteration()