fix

Author: sufubao

lightllm/server/core/objs/req.py

@@ -25,19 +25,20 @@ class FinishStatus(ctypes.Structure):
     NO_FINISH = 0
     FINISHED_STOP = 1
     FINISHED_LENGTH = 2
+    FINISHED_ERROR = 3
 
     def __init__(self, init_state=NO_FINISH):
         self.status = init_state
 
     def set_status(self, new_status):
-        assert 0 <= new_status <= 2
+        assert 0 <= new_status <= 3
         self.status = new_status
 
     def get_status(self):
         return self.status
 
     def is_finished(self):
-        return self.FINISHED_STOP <= self.status <= self.FINISHED_LENGTH
+        return self.FINISHED_STOP <= self.status <= self.FINISHED_ERROR
 
     def is_stopped(self):
         return self.status == self.FINISHED_STOP
@@ -50,6 +51,8 @@ def get_finish_reason(self):
             return "stop"
         elif self.status == self.FINISHED_LENGTH:
             return "length"
+        elif self.status == self.FINISHED_ERROR:
+            return "error"
         return None
 
 

lightllm/server/detokenization/manager.py

@@ -47,24 +47,30 @@ def _init_get_token_id_to_token_str(self):
         return
 
     def _add_new_group_req_index(self, recv_obj: GroupReqIndexes):
+        from lightllm.server.core.objs import FinishStatus
+
         for req_index in recv_obj.shm_req_indexes:
             req = self.shm_req_manager.get_req_obj_by_index(req_index)
-            req.link_prompt_ids_shm_array()
-            req.link_logprobs_shm_array()
-
-            logger.debug(
-                f"detokenization recv req id {req.request_id} " f"cost time {time.time() - recv_obj.time_mark} s"
-            )
-
-            # p d 分离模式，decode节点的解码需要做一些特殊的修复。
-            decode_req = DecodeReq(req, self.is_pd_decode_mode)
-            if self.is_pd_decode_mode:
-                decode_req = decode_mode_fix(decode_req, self.tokenizer, self.eos_id)
-            # token_healing mode 的特殊初始化
-            if self.args.token_healing_mode:
-                decode_req.init_token_healing_prefix_str(self.token_id_to_token, self.tokenizer)
-
-            self.req_id_to_out[req.request_id] = decode_req
+            try:
+                req.link_prompt_ids_shm_array()
+                req.link_logprobs_shm_array()
+
+                logger.debug(
+                    f"detokenization recv req id {req.request_id} " f"cost time {time.time() - recv_obj.time_mark} s"
+                )
+
+                # p d 分离模式，decode节点的解码需要做一些特殊的修复。
+                decode_req = DecodeReq(req, self.is_pd_decode_mode)
+                if self.is_pd_decode_mode:
+                    decode_req = decode_mode_fix(decode_req, self.tokenizer, self.eos_id)
+                # token_healing mode 的特殊初始化
+                if self.args.token_healing_mode:
+                    decode_req.init_token_healing_prefix_str(self.token_id_to_token, self.tokenizer)
+
+                self.req_id_to_out[req.request_id] = decode_req
+            except Exception as e:
+                req.finish_status.set_status(FinishStatus.FINISHED_ERROR)
+                raise e
         return
 
     def handle_loop(self):
@@ -79,9 +85,8 @@ def handle_loop(self):
                         try:
                             self._add_new_group_req_index(recv_obj=recv_obj)
                         except Exception:
-                            # TODO: publish an ERROR finish_status back to httpserver so the
-                            # client gets a 500 instead of hanging until disconnect.
                             logger.exception("add new group req index has exception")
+                            self.pub_to_httpserver.send_pyobj(None, protocol=pickle.HIGHEST_PROTOCOL)
 
                     # 当队列中存在较多的请求时，将一次接受的数量上调
                     recv_max_count = min(int(recv_max_count * 1.3), 256)

lightllm/server/httpserver/manager.py

@@ -298,7 +298,11 @@ async def generate(
         # 用于等待 pd_master 下发的交换信息
         nixl_pd_event: asyncio.Event = None,
     ) -> AsyncGenerator[Tuple[int, str, dict, FinishStatus], None]:
+        group_request_id = None
         if isinstance(prompt, str):
+            # Guard against extremely long string prompts that might stall the tokenizer
+            # or cause excessive memory usage before tokenization.
+            # 8 characters per token is a conservative heuristic (avg is ~4).
             max_prompt_chars = self.max_req_total_len * 8
             if len(prompt) > max_prompt_chars:
                 raise ValueError(

lightllm/server/router/manager.py

@@ -347,16 +347,16 @@ def _filter_reqs_from_running_batch(self):
                 # Settle any output-token tail produced after the last window boundary,
                 # so windowed TPS does not lose the req's last tokens.
                 self.status_reporter.discard_req(req)
-                # Aborted/disconnected requests can leave a partial output_len that
-                # would bias the EMA toward shorter generations; skip them.
-                if req.is_aborted:
-                    continue
                 self.status_reporter.on_request_completed(
                     input_len=req.input_len,
                     output_len=req.shm_cur_output_len,
                     cache_len=req.prompt_cache_len,
                     mtp_accepted=req.mtp_accepted_token_num,
                 )
+                # Aborted/disconnected requests can leave a partial output_len that
+                # would bias the EMA toward shorter generations; skip them.
+                if req.is_aborted:
+                    continue
                 self.router_statics.update(req.candetoken_out_len)
             self.running_batch.filter_out_finished_req(self.shm_req_manager)
             if self.running_batch.is_clear():