update log & fix dp pause

Author: shihaobai

lightllm/server/router/manager.py

@@ -244,17 +244,19 @@ async def loop_for_fwd(
                         estimated_peak_token_count = self.shared_token_load.get_estimated_peak_token_count(d_i)
                         logger.debug(
                             f"dp_i {d_i} current batch size: {len(self.running_batch.reqs)} \n"
-                            f"dp_i {d_i} paused req num: {self.req_queue.get_paused_req_num()} \n"
+                            f"dp_i {d_i} paused req num: {self.req_queue.get_paused_req_num(d_i)} \n"
                             f"dp_i {d_i} frozen token num: {frozen_token_num} \n"
                             f"dp_i {d_i} estimated_peak_token_count: {estimated_peak_token_count} \n"
                             f"dp_i {d_i} token used ratio: {token_ratio1} not contain prompt cache tree unrefed token\n"
                             f"dp_i {d_i} token used ratio: {token_ratio2} contain prompt cache tree unrefed token"
                         )
+                        self.metric_client.gauge_set(
+                            "lightllm_batch_pause_size", self.req_queue.get_paused_req_num(d_i)
+                        )
                 # pd decode mode need to update token_load more frequently
                 self.req_queue.update_token_load(self.running_batch, force_update=self.is_pd_decode_mode)
                 self.stats_tool.print_stats()
                 self.metric_client.gauge_set("lightllm_batch_current_size", len(self.running_batch.reqs))
-                self.metric_client.gauge_set("lightllm_batch_pause_size", self.req_queue.get_paused_req_num())
                 self.metric_client.gauge_set("lightllm_queue_size", self.req_queue.get_wait_req_num())
                 self.metric_client.gauge_set(
                     "lightllm_batch_current_max_tokens",
@@ -358,15 +360,13 @@ async def _step(self):
 
         # Check if need pause some requests for decode.
         for dp_index in range(self.dp_size_in_node):
-            if self._can_decode(self.running_batch, dp_index=dp_index):
-                continue
-            else:
+            while not self._can_decode(self.running_batch, dp_index=dp_index):
                 # pause strategy
                 paused_reqs = select_paused_reqs(
                     self.running_batch, self.pause_strategy, self.req_queue, self.max_total_token_num, dp_index=dp_index
                 )
                 await self._pause_reqs(paused_reqs)
-                logger.debug(f"DP index {dp_index} pasues req num: {self.req_queue.get_paused_req_num()}")
+                logger.debug(f"DP index {dp_index} pasues req num: {self.req_queue.get_paused_req_num(dp_index)}")
                 self.has_wait_tokens = 0
 
         # Decode

lightllm/server/router/req_queue/base_queue.py

@@ -38,7 +38,7 @@ def extend(self, req_group: List[Req]):
         self.waiting_req_list.extend(req_group)
         return
 
-    def get_paused_req_num(self):
+    def get_paused_req_num(self, fake_dp_index: int = 0):
         return len(self.pause_req_dict)
 
     def get_wait_req_num(self):

lightllm/server/router/req_queue/dp_base_queue.py

@@ -26,8 +26,8 @@ def get_dp_queue(self, dp_index: int):
         assert dp_index < self.dp_size_in_node, "dp index out of range"
         return self.inner_queues[dp_index]
 
-    def get_paused_req_num(self):
-        return sum(queue.get_paused_req_num() for queue in self.inner_queues)
+    def get_paused_req_num(self, dp_index: int = 0):
+        return self.inner_queues[dp_index].get_paused_req_num()
 
     def get_wait_req_num(self):
         return sum(queue.get_wait_req_num() for queue in self.inner_queues)