fix

Author: hiworldwzj

lightllm/server/api_http.py

@@ -101,7 +101,7 @@ def set_args(self, args):
                 enable_multimodal=args.enable_multimodal,
                 metric_port=args.metric_port,
             )
-            dp_size_in_node = max(1, args.dp // args.nnodes)  # 兼容多机存粹tp的运行模式，这时候 1 // 2 == 0, 需要兼容
+            dp_size_in_node = max(1, args.dp // args.nnodes)  # 兼容多机纯tp的运行模式，这时候 1 // 2 == 0, 需要兼容
             self.shared_token_load = TokenLoad(f"{get_unique_server_name()}_shared_token_load", dp_size_in_node)
 
 

lightllm/server/router/manager.py

@@ -389,11 +389,11 @@ def _can_decode(self, batch: Batch):
         # p d 分离模式下，目前只能使用保守调度，保证请求放入进行decode的时候
         # 显存token肯定是够用的。
         # deepseekv2 dp 模式下,采用保守调度，也肯定够用
-        if self.is_pd_run_mode or self.dp_size > 1 or self.is_safe_schedule:
+        if self.is_pd_run_mode or self.dp_size_in_node > 1 or self.is_safe_schedule:
             return True
 
         # 下面的判定条件，只在 dp 为 1 的情况下启用
-        assert self.dp_size == 1
+        assert self.dp_size_in_node == 1
         return batch.get_batch_decode_need_tokens()[0] + self.get_used_tokens(0) <= self.max_total_token_num
 
     def get_used_tokens(self, dp_index):

lightllm/server/router/req_queue/dp_base_queue.py

@@ -44,11 +44,8 @@ def _merge_batch(self, dp_batches: List[Batch]):
 
     def append(self, req: Req):
         suggested_dp_index = req.sample_params.suggested_dp_index
-        if suggested_dp_index is None or suggested_dp_index >= self.dp_size_in_node or suggested_dp_index < 0:
-            if suggested_dp_index is not None and (
-                suggested_dp_index >= self.dp_size_in_node or suggested_dp_index < 0
-            ):
-                logger.error(f"input req {req.request_id} dp index {suggested_dp_index} has error")
+        if suggested_dp_index >= self.dp_size_in_node or suggested_dp_index < 0:
+            logger.error(f"input req {req.request_id} dp index {suggested_dp_index} has error")
             suggested_dp_index = random.randint(0, self.dp_size_in_node - 1)
             req.sample_params.suggested_dp_index = suggested_dp_index
             self.inner_queues[suggested_dp_index].append(req)
@@ -61,11 +58,8 @@ def extend(self, req_group: List[Req]):
         index = random.randint(0, self.dp_size_in_node - 1)
         for req in req_group:
             suggested_dp_index = req.sample_params.suggested_dp_index
-            if suggested_dp_index is None or suggested_dp_index >= self.dp_size_in_node or suggested_dp_index < 0:
-                if suggested_dp_index is not None and (
-                    suggested_dp_index >= self.dp_size_in_node or suggested_dp_index < 0
-                ):
-                    logger.error(f"input req {req.request_id} dp index {suggested_dp_index} has error")
+            if suggested_dp_index >= self.dp_size_in_node or suggested_dp_index < 0:
+                logger.error(f"input req {req.request_id} dp index {suggested_dp_index} has error")
                 req.sample_params.suggested_dp_index = index
                 self.inner_queues[index].append(req)
             else: