fix req queue init error.

Author: hiworldwzj

lightllm/server/router/batch.py

@@ -8,11 +8,11 @@
 
 
 class Batch:
-    def __init__(self, batch_id, reqs: List[Req], dp_size: int):
+    def __init__(self, batch_id, reqs: List[Req], dp_size_in_node: int):
         self.batch_id = batch_id
         self.reqs = reqs
         self.id_to_reqs = {req.request_id: req for req in reqs}
-        self.dp_size = dp_size
+        self.dp_size_in_node = dp_size_in_node
         return
 
     def input_tokens(self):
@@ -22,7 +22,7 @@ def input_tokens(self):
         return batch_input_tokens
 
     def get_batch_decode_need_tokens(self):
-        new_batch_decode_need_tokens = [0 for _ in range(self.dp_size)]  # for chunked prefill
+        new_batch_decode_need_tokens = [0 for _ in range(self.dp_size_in_node)]  # for chunked prefill
 
         for req in self.reqs:
             req_dp_index = req.sample_params.suggested_dp_index

lightllm/server/router/req_queue/chunked_prefill/impl.py

@@ -5,8 +5,8 @@
 
 
 class ChunkedPrefillQueue(BaseQueue):
-    def __init__(self, args, router, dp_index, dp_size) -> None:
-        super().__init__(args, router, dp_index, dp_size)
+    def __init__(self, args, router, dp_index, dp_size_in_node) -> None:
+        super().__init__(args, router, dp_index, dp_size_in_node)
 
     def _init_cache_list(self, current_batch: Batch, is_busy):
         if current_batch is not None:
@@ -99,7 +99,7 @@ def generate_new_batch(self, current_batch: Batch, limit_router_queue_length: in
                 break
 
         if len(can_run_list) != 0:
-            new_batch = Batch(uuid.uuid4().int, can_run_list, dp_size=self.dp_size)
+            new_batch = Batch(uuid.uuid4().int, can_run_list, dp_size_in_node=self.dp_size_in_node)
             for req in abort_req_list:
                 self.router.shm_req_manager.put_back_req_obj(req)
             self.waiting_req_list = self.waiting_req_list[len(can_run_list) + aborted_count :]

lightllm/server/router/req_queue/continues_batch/beam_impl.py

@@ -5,8 +5,8 @@
 
 
 class BeamContinuesBatchQueue(BaseQueue):
-    def __init__(self, args, router, dp_index, dp_size) -> None:
-        super().__init__(args, router, dp_index, dp_size)
+    def __init__(self, args, router, dp_index, dp_size_in_node) -> None:
+        super().__init__(args, router, dp_index, dp_size_in_node)
         return
 
     def _init_cache_list(self, current_batch: Batch, is_busy):
@@ -129,7 +129,7 @@ def generate_new_batch(self, current_batch: Batch, limit_router_queue_length: in
                         req.is_paused = False
 
         if len(can_run_list) != 0:
-            new_batch = Batch(uuid.uuid4().int, can_run_list, dp_size=self.dp_size)
+            new_batch = Batch(uuid.uuid4().int, can_run_list, dp_size_in_node=self.dp_size_in_node)
             for req in abort_req_list:
                 self.router.shm_req_manager.put_back_req_obj(req)
 

lightllm/server/router/req_queue/continues_batch/impl.py

@@ -6,8 +6,8 @@
 
 
 class ContinuesBatchQueue(BaseQueue):
-    def __init__(self, args, router, dp_index, dp_size) -> None:
-        super().__init__(args, router, dp_index, dp_size)
+    def __init__(self, args, router, dp_index, dp_size_in_node) -> None:
+        super().__init__(args, router, dp_index, dp_size_in_node)
 
     def _init_cache_list(self, current_batch: Batch, is_busy):
         if current_batch is not None:
@@ -100,7 +100,7 @@ def generate_new_batch(self, current_batch: Batch, limit_router_queue_length: in
             else:
                 break
         if len(can_run_list) != 0:
-            new_batch = Batch(uuid.uuid4().int, can_run_list, dp_size=self.dp_size)
+            new_batch = Batch(uuid.uuid4().int, can_run_list, dp_size_in_node=self.dp_size_in_node)
             for req in abort_req_list:
                 self.router.shm_req_manager.put_back_req_obj(req)
             self.waiting_req_list = self.waiting_req_list[len(can_run_list) + aborted_count :]

lightllm/server/router/req_queue/continues_batch/pd_decode_impl.py

@@ -9,8 +9,8 @@
 
 
 class ContinuesBatchQueueForPDDecode(BaseQueue):
-    def __init__(self, args, router, dp_index, dp_size) -> None:
-        super().__init__(args, router, dp_index, dp_size)
+    def __init__(self, args, router, dp_index, dp_size_in_node) -> None:
+        super().__init__(args, router, dp_index, dp_size_in_node)
 
     def _init_cache_list(self, current_batch: Batch, is_busy):
         if current_batch is not None:
@@ -47,7 +47,7 @@ def generate_new_batch(self, current_batch: Batch, limit_router_queue_length: in
                 break
 
         if len(can_run_list) != 0:
-            new_batch = Batch(uuid.uuid4().int, can_run_list, dp_size=self.dp_size)
+            new_batch = Batch(uuid.uuid4().int, can_run_list, dp_size_in_node=self.dp_size_in_node)
             for req in abort_req_list:
                 self.router.shm_req_manager.put_back_req_obj(req)
             self.waiting_req_list = self.waiting_req_list[len(can_run_list) + aborted_count :]