bug fix for max len prefill check error (#650)

Author: hiworldwzj

lightllm/common/basemodel/basemodel.py

@@ -166,7 +166,14 @@ def _check_mem_size(self):
         return
 
     def _init_req_manager(self):
-        self.req_manager = ReqManager(self.max_req_num, self.max_seq_length, self.mem_manager)
+        create_max_seq_len = 0
+
+        if self.batch_max_tokens is not None:
+            create_max_seq_len = max(create_max_seq_len, self.batch_max_tokens)
+        if self.max_seq_length is not None:
+            create_max_seq_len = max(create_max_seq_len, self.max_seq_length)
+
+        self.req_manager = ReqManager(self.max_req_num, create_max_seq_len, self.mem_manager)
         return
 
     def _init_infer_layer(self):

lightllm/server/httpserver/manager.py

@@ -67,6 +67,7 @@ def __init__(
         assert self.pd_mode in [NodeRole.P, NodeRole.D, NodeRole.NORMAL]
         self.id_gen = ReqIDGenerator()
         self.first_time_costs = MovingAverage()
+        self.per_token_costs = MovingAverage()
         # 有的模型的vocab size 读取tokenizer和config.json中不一致
         self.vocab_size = max(get_vocab_size(args.model_dir), self.tokenizer.vocab_size)
 
@@ -340,6 +341,7 @@ async def _wait_to_token_package(
                             pass
                         total_cost_time_ms = (time.time() - start_time) * 1000
                         mean_per_token_cost_time_ms = (total_cost_time_ms - first_token_cost_ms) / out_token_counter
+                        self.per_token_costs.add(mean_per_token_cost_time_ms)
                         x_request_id = request.headers.get("X-Request-Id", "")
                         x_session_id = request.headers.get("X-Session-Id", "")
                         prompt_cache_len = metadata.pop("prompt_cache_len", 0)
@@ -441,6 +443,7 @@ async def timer_to_pd_master(self):
                         await asyncio.sleep(3)
                         if log_count % 5 == 0:
                             logger.info(f"mean first cost: {self.first_time_costs.average()} ms")
+                            logger.info(f"mean per token cost: {self.per_token_costs.average()} ms")
 
             except Exception as e:
                 logger.error("connetion to pd_master has error")

lightllm/server/httpserver_for_pd_master/manager.py

@@ -41,7 +41,9 @@ def __init__(
         self.id_to_event: Dict[int, asyncio.Event] = {}
         self.session = None
         self.first_time_costs = MovingAverage()
-        self.create_session_costs = MovingAverage()
+        self.prefill_create_session_costs = MovingAverage()
+        self.decode_create_session_costs = MovingAverage()
+        self.per_token_costs = MovingAverage()
         return
 
     async def register_pd(self, pd_info_json):
@@ -181,7 +183,7 @@ async def fetch_stream(
             req = await self._to_req_info(prompt, sampling_params, multimodal_params)
             create_start_time = time.time()
             async with self.session.post(p_node.to_llm_url(), json=req) as response:
-                self.create_session_costs.add((time.time() - create_start_time) * 1000)
+                self.prefill_create_session_costs.add((time.time() - create_start_time) * 1000)
                 if response.status == 200:
                     async for line in response.content:
                         line = line.decode("utf-8").strip()
@@ -217,7 +219,9 @@ async def fetch_stream(
             sampling_params.suggested_dp_index = event.upkv_status.dp_index
 
             req = await self._to_req_info(prompt_ids, sampling_params, multimodal_params)
+            create_start_time = time.time()
             async with self.session.post(d_node.to_llm_url(), json=req) as response:
+                self.decode_create_session_costs.add((time.time() - create_start_time) * 1000)
                 if response.status == 200:
                     async for line in response.content:
                         line = line.decode("utf-8").strip()
@@ -269,6 +273,7 @@ async def _wait_to_token_package(
 
         total_cost_time_ms = (time.time() - start_time) * 1000
         mean_per_token_cost_time_ms = (total_cost_time_ms - first_token_cost_ms) / out_token_counter
+        self.per_token_costs.add(mean_per_token_cost_time_ms)
         x_request_id = request.headers.get("X-Request-Id", "")
         x_session_id = request.headers.get("X-Session-Id", "")
         prompt_cache_len = metadata.pop("prompt_cache_len", 0)
@@ -312,5 +317,7 @@ async def handle_loop(self):
             # 可以做一个定时任务
             await asyncio.sleep(20)
             logger.info(f"mean first cost: {self.first_time_costs.average()} ms")
-            logger.info(f"create_session_costs: {self.create_session_costs.average()} ms")
+            logger.info(f"prefill mean create_session_costs: {self.prefill_create_session_costs.average()} ms")
+            logger.info(f"decode mean create_session_costs: {self.decode_create_session_costs.average()} ms")
+            logger.info(f"mean per token cost: {self.per_token_costs.average()} ms")
         return

lightllm/server/router/manager.py

@@ -426,6 +426,9 @@ def _update_init_status_to_batch(self, batch: Batch, req_to_req_status):
 
     def _update_out_status_to_batch(self, batch: Batch, req_to_out_status):
         new_batch_decode_need_tokens = [0 for _ in range(self.dp_size)]  # 只有在 splitfuse 模式下有意义
+
+        start_time = 0
+        # extral_info 字段如果推理后端输入时间标记, 则用来评估序列化所占用的时间, 主要用于调试时使用
         for req_id, (
             req_status,
             cur_kv_len,
@@ -434,6 +437,8 @@ def _update_out_status_to_batch(self, batch: Batch, req_to_out_status):
             finish_status_value,
             extral_info,
         ) in req_to_out_status.items():
+            if extral_info is not None:
+                start_time = max(start_time, extral_info)
             req: Req = batch.id_to_reqs[req_id]
             req.req_status = req_status
             req.cur_kv_len = cur_kv_len
@@ -446,6 +451,9 @@ def _update_out_status_to_batch(self, batch: Batch, req_to_out_status):
             new_batch_decode_need_tokens[req_dp_index] += req.get_decode_need_tokens()
 
         batch.batch_decode_need_tokens = new_batch_decode_need_tokens
+        rpyc_cost_time = (time.time() - start_time) * 1000
+        if 8 <= rpyc_cost_time <= 1000:
+            logger.warning(f"rpyc use too much time {rpyc_cost_time} ms, batch_size {len(req_to_out_status)}")
         return
 
     def _can_decode(self, batch: Batch):