Refactor pause engine-MindIE-LLM-AtomGit

已合并

Refactor pause engine #586

yuzechen创建于 3月17日从Jechin/MindIE-LLM： feature/stop_engine_with_force_stop 合入到 Ascend/MindIE-LLM： dev

Refactor pause engine #586

已合并

yuzechen创建于 3月17日

从Jechin/MindIE-LLM feature/stop_engine_with_force_stop合入到Ascend/MindIE-LLM dev

共 10 个文件变更+863-150

Mmindie_llm/text_generator/adapter/generator_aclgraph.py+3-23

	@@ -224,29 +224,9 @@ class GeneratorAclGraph(GeneratorBackend):
224	def update_cache_after_switch_pd_role(self):	224	def update_cache_after_switch_pd_role(self):
225	self.cache_pool.allocate_npu_cache()	225	self.cache_pool.allocate_npu_cache()
226		226
227	- def execute_recover_command(self, command: str) -> dict:	227	+ def _execute_cmd_reinit_npu(self):
228	- '''	228	+ torch_npu.npu.restart_device(self.npu_device_id)
229	- Execute recover related command.	229	+ torch_npu.distributed.reinit_process_group(rebuild_link=False)
230	- Args:
231	- command (str): recover command, including "CMD_PAUSE_ENGINE".
232	- Returns:
233	- Tuple[int, str]: (return code, error message). return code: 1 for success, 0 for failure.
234	- '''
235	- error_msg = ""
236	- # Recover command execution result, 0 for success, 1 for failure.
237	- command_result = 1
238	- try:
239	- if (command == "CMD_PAUSE_ENGINE"):
240	- command_result = torch_npu.npu.stop_device(self.npu_device_id)
241	- elif (command == "CMD_REINIT_NPU"):
242	- torch_npu.npu.restart_device(self.npu_device_id)
243	- command_result = 0
244	- except Exception as e:
245	- error_msg = f"Execute recover command {command} failed, exception msg: {e}"
246	- logger.error(error_msg, ErrorCode.TEXT_GENERATOR_INTERNAL_ERROR)
247	- error_msg = str(e)
248	- ret_dict = {"command_result": command_result, "error_msg": error_msg, "npu_device_id": self.npu_device_id}
249	- return ret_dict
250		230
251	def _warm_up(self, model_inputs: ModelInput, **kwargs) -> None:	231	def _warm_up(self, model_inputs: ModelInput, **kwargs) -> None:
252	# NOTE: To ensure compatibility with atb graph, the current warmup procedure is:	232	# NOTE: To ensure compatibility with atb graph, the current warmup procedure is:

Mmindie_llm/text_generator/adapter/generator_backend.py+128-0

	@@ -8,9 +8,13 @@
8	# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.	8	# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
9	# See the Mulan PSL v2 for more details.	9	# See the Mulan PSL v2 for more details.
10		10
		11	+import threading
		12	+import time
11	from typing import Any, Dict, Iterable, List, Optional, Tuple, Union	13	from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
12		14
13	import numpy as np	15	import numpy as np
		16	+import torch
		17	+import torch_npu
14		18
15	from ..samplers.sampler import Sampler	19	from ..samplers.sampler import Sampler
16	from ..utils.config import SamplerConfig	20	from ..utils.config import SamplerConfig
	@@ -19,8 +23,11 @@ from ..utils.sampling_output import SamplingOutput
19	from ..utils.sampling_metadata import SamplingMetadata, SamplingData, SamplingParam	23	from ..utils.sampling_metadata import SamplingMetadata, SamplingData, SamplingParam
20	from ...modeling.model_wrapper import get_model_wrapper	24	from ...modeling.model_wrapper import get_model_wrapper
21	from ...utils.decorators.time_decorator import timer	25	from ...utils.decorators.time_decorator import timer
		26	+from ...utils.log.error_code import ErrorCode
		27	+from ...utils.log.logging import logger
22	from ...utils.tensor import op	28	from ...utils.tensor import op
23	from ...utils.validation import parse_config, ParseType, MODEL_CONFIG_KEY_TYPE	29	from ...utils.validation import parse_config, ParseType, MODEL_CONFIG_KEY_TYPE
		30	+from .recovery_utils import check_and_recover_uce_in_cache
24		31
25	MAX_WORLD_SIZE = 1048576	32	MAX_WORLD_SIZE = 1048576
26	MAX_KEY_LENGTH = 256	33	MAX_KEY_LENGTH = 256
	@@ -128,6 +135,11 @@ class GeneratorBackend:
128	self.enable_dap = False	135	self.enable_dap = False
129	self.obfuscation_func = None	136	self.obfuscation_func = None
130	self.device = None	137	self.device = None
		138	+ self.cache_pool = None
		139	+
		140	+ # Thread-safe mechanism for detecting FORCE STOP exception
		141	+ self.force_stop_exception_occurred = threading.Event()
		142	+ self.is_fault_device = False
131		143
132	self.max_position_embeddings = self.model_wrapper.max_position_embeddings	144	self.max_position_embeddings = self.model_wrapper.max_position_embeddings
133		145
	@@ -157,6 +169,37 @@ class GeneratorBackend:
157	def set_device(self):	169	def set_device(self):
158	pass	170	pass
159		171
		172	+ def notify_force_stop_exception(self):
		173	+ '''
		174	+ Notify that a FORCE STOP exception has occurred in the inference thread.
		175	+ This method should be called from the inference thread when catching FORCE STOP exceptions.
		176	+ '''
		177	+ self.force_stop_exception_occurred.set()
		178	+ logger.info(f"FORCE STOP exception detected and notified for device {self.npu_device_id}")
		179	+
		180	+ def execute_recover_command(self, command: str) -> dict:
		181	+ '''
		182	+ Execute recover related command.
		183	+ Args:
		184	+ command (str): recover command, including "CMD_PAUSE_ENGINE".
		185	+ Returns:
		186	+ dict: {"command_result": int, "error_msg": str, "npu_device_id": int}.
		187	+ command_result: 0 for success, 1 for failure.
		188	+ '''
		189	+ error_msg = ""
		190	+ command_result = 1
		191	+ try:
		192	+ if command == "CMD_PAUSE_ENGINE":
		193	+ command_result, error_msg = self._execute_cmd_pause_engine()
		194	+ elif command == "CMD_REINIT_NPU":
		195	+ self._execute_cmd_reinit_npu()
		196	+ command_result = 0
		197	+ except Exception as e:
		198	+ error_msg = f"Execute recover command {command} failed, exception msg: {e}"
		199	+ logger.error(error_msg, ErrorCode.TEXT_GENERATOR_INTERNAL_ERROR)
		200	+ error_msg = str(e)
		201	+ return {"command_result": command_result, "error_msg": error_msg, "npu_device_id": self.npu_device_id}
		202	+
160	def build_inputs(self, conversations: List[List[Dict[str, str]]], **kwargs) -> List[List[int]]:	203	def build_inputs(self, conversations: List[List[Dict[str, str]]], **kwargs) -> List[List[int]]:
161	return [self.model_wrapper.make_context(conversation, **kwargs) for conversation in conversations]	204	return [self.model_wrapper.make_context(conversation, **kwargs) for conversation in conversations]
162		205
	@@ -216,3 +259,88 @@ class GeneratorBackend:
216	else:	259	else:
217	output = self.sampler(logits, sampling_metadata)	260	output = self.sampler(logits, sampling_metadata)
218	return output	261	return output
		262	+
		263	+ def _execute_cmd_pause_engine(self):
		264	+ self.force_stop_exception_occurred.clear()
		265	+ wait_exception_time = 10.0
		266	+ time.sleep(wait_exception_time)
		267	+ if torch_npu.npu.stop_device(self.npu_device_id) != 0:
		268	+ error_msg = "Stop device failed"
		269	+ command_result = 1
		270	+ else:
		271	+ uce_command_result, uce_error_msg = self._handle_uce_error()
		272	+ if uce_command_result == 1:
		273	+ command_result = uce_command_result
		274	+ error_msg = uce_error_msg
		275	+ elif uce_command_result == 2:
		276	+ command_result = 0
		277	+ error_msg = ""
		278	+ elif not self._wait_for_force_stop_exception():
		279	+ command_result = 1
		280	+ error_msg = "Timeout waiting for FORCE STOP exception"
		281	+ else:
		282	+ command_result = 0
		283	+ error_msg = ""
		284	+ return command_result, error_msg
		285	+
		286	+ def _execute_cmd_reinit_npu(self):
		287	+ '''Reinitialize NPU. Subclasses must override with backend-specific logic.'''
		288	+ raise NotImplementedError("Subclasses must implement _execute_cmd_reinit_npu")
		289	+
		290	+ def _wait_for_force_stop_exception(self):
		291	+ if not self.is_fault_device:
		292	+ timeout = 60.0
		293	+ exception_detected = self.force_stop_exception_occurred.wait(timeout=timeout)
		294	+ if exception_detected:
		295	+ logger.info(
		296	+ f"FORCE STOP exception detected for device {self.npu_device_id}, "
		297	+ "stop_device execution successful"
		298	+ )
		299	+ return True
		300	+ else:
		301	+ logger.warning(
		302	+ f"Timeout waiting for FORCE STOP exception for device {self.npu_device_id} "
		303	+ f"after {timeout} seconds"
		304	+ )
		305	+ return False
		306	+ else:
		307	+ return True
		308	+
		309	+ def _handle_uce_error(self):
		310	+ '''Check and recover UCE error in kvcache. Returns (command_result, error_msg).'''
		311	+ command_result = 0
		312	+ error_msg = ""
		313	+ res = torch.npu.check_uce_in_memory(self.npu_device_id)
		314	+ if res == 2 or res == 3:
		315	+ logger.info(f"Encountered HBM UCE error, check_uce_in_memory result: {res}")
		316	+ command_result = 2
		317	+ if not self._check_and_recover_uce_in_kvcache():
		318	+ logger.warning(f"HBM UCE address not in any kvcache, should trigger reschedule")
		319	+ command_result = 1
		320	+ error_msg = "HBM uce address not overlap kvcache address, should trigger reschedule"
		321	+ elif res == 1:
		322	+ logger.warning(f"Encountered HBM UCE error, but unknown UCE address, should trigger reschedule")
		323	+ command_result = 1
		324	+ error_msg = "HBM uce address unknown, should trigger reschedule"
		325	+ return command_result, error_msg
		326	+
		327	+ def _check_and_recover_uce_in_kvcache(self):
		328	+ '''Check and recover UCE error in kvcache. Returns True if recovered, False otherwise.'''
		329	+ uce_addr_list = torch_npu.npu._get_uce_addr()
		330	+ logger.info(f"UCE address list: {uce_addr_list}")
		331	+ if len(uce_addr_list) == 0:
		332	+ return False
		333	+ for addr_entry in uce_addr_list:
		334	+ uce_addr = addr_entry["ptr"]
		335	+ addr_size = addr_entry["size"]
		336	+ uce_addr_start = uce_addr
		337	+ uce_addr_end = uce_addr_start + addr_size
		338	+
		339	+ for n in range(self.cache_pool.kvcache_settings.num_layers):
		340	+ k_cache = self.cache_pool.npu_cache[n][0]
		341	+ if check_and_recover_uce_in_cache(uce_addr_start, uce_addr_end, k_cache, n, "kcache"):
		342	+ return True
		343	+ v_cache = self.cache_pool.npu_cache[n][1]
		344	+ if check_and_recover_uce_in_cache(uce_addr_start, uce_addr_end, v_cache, n, "vcache"):
		345	+ return True
		346	+ return False

Mmindie_llm/text_generator/adapter/generator_torch.py+5-80

 from ...utils.file_utils import standardize_path, check_file_safety
 from ...utils.env import ENV
++
 ASCEND_310B = 240
         raise ValueError(message)
+-def is_uce_error_addr_overlap_tensor_addr(uce_addr_start, uce_addr_end, tensor_addr_start, tensor_addr_end):
+-    return (uce_addr_start >= tensor_addr_start) and (uce_addr_end <= tensor_addr_end)
+-
+-
+-def get_tensor_address_range(input_tensor):
+-    addr_start = input_tensor.data_ptr()
+-    addr_end = addr_start + input_tensor.numel() * input_tensor.element_size()
+-    return addr_start, addr_end
+-
+-
+-def check_and_recover_uce_in_cache(uce_addr_start, uce_addr_end, cache_tensor, layer_idx, cache_type):
+-    cache_addr_start, cache_addr_end = get_tensor_address_range(cache_tensor)
+-    if is_uce_error_addr_overlap_tensor_addr(uce_addr_start, uce_addr_end, cache_addr_start, cache_addr_end):
+-        torch_npu.npu._recovery.update_npu_tensor_to_safe(cache_tensor)
+-        logger.info(f"HBM UCE address in {cache_type} of layer {layer_idx}, update {cache_type} to safe")
+-        return True
+-    else:
+-        logger.info(
+-            f"HBM UCE address not in {cache_type} address ({cache_addr_start}, {cache_addr_end}) of layer {layer_idx}"
+-        )
+-        return False
+-
 STABLE_SORT_TYPE = "stable"
     def update_cache_after_switch_pd_role(self):
         self.cache_pool.allocate_npu_cache()
++    def _execute_cmd_reinit_npu(self):
++        torch_npu.npu.restart_device(self.npu_device_id)
++        self.model_wrapper.model_runner.reset_execution_status()
++        self.model_wrapper.resume_hccl_comm()
+-            command (str): recover command, including "CMD_PAUSE_ENGINE".
+-        Returns:
+-            Tuple[int, str]: (return code, error message). return code: 1 for success, 0 for failure.
+-        '''
+-        error_msg = ""
+-        # Recover command execution result, 0 for success, 1 for failure.
+-        command_result = 1
+-        try:
+-            if (command == "CMD_PAUSE_ENGINE"):
+-                command_result = torch_npu.npu.stop_device(self.npu_device_id)
+-                command_result, error_msg = self._handle_uce_error()
+-            elif (command == "CMD_REINIT_NPU"):
+-                torch_npu.npu.restart_device(self.npu_device_id)
+-                self.model_wrapper.model_runner.reset_execution_status()
+-                command_result = 0
+-        except Exception as e:
+-            error_msg = f"Execute recover command {command} failed, exception msg: {e}"
+-            logger.error(error_msg, ErrorCode.TEXT_GENERATOR_INTERNAL_ERROR)
+-            error_msg = str(e)
+-        ret_dict = {"command_result": command_result, "error_msg": error_msg, "npu_device_id": self.npu_device_id}
+-        return ret_dict
     def _sort_model_inputs_by_adapter_ids(self, model_inputs):
         adapter_ids = model_inputs.adapter_ids
                         f"and use it properly. Exception msg: {e}"
             logger.error(error_msg, ErrorCode.TEXT_GENERATOR_INTERNAL_ERROR)
             raise RuntimeError(error_msg) from e
+-
+-    def _handle_uce_error(self):
+-        command_result = 0
+-        error_msg = ""
+-        res = torch.npu.check_uce_in_memory(self.npu_device_id)
+-        if res == 2 or res == 3:
+-            logger.info(f"Encountered HBM UCE error, check_uce_in_memory result: {res}")
+-            if not self._check_and_recover_uce_in_kvcache():
+-                logger.warning(f"HBM UCE address not in any kvcache, should trigger reschedule")
+-                command_result = 1
+-                error_msg = "HBM uce address not overlap kvcache address, should trigger reschedule"
+-        return command_result, error_msg
+-
+-    def _check_and_recover_uce_in_kvcache(self):
+-        uce_addr_list = torch_npu.npu._get_uce_addr()
+-        logger.info(f"UCE address list: {uce_addr_list}")
+-        if len(uce_addr_list) == 0:
+-            return False
+-        for addr_entry in uce_addr_list:
+-            uce_addr = addr_entry["ptr"]
+-            addr_size = addr_entry["size"]
+-            uce_addr_start = uce_addr
+-            uce_addr_end = uce_addr_start + addr_size
+-
+-            for n in range(self.cache_pool.kvcache_settings.num_layers):
+-                k_cache = self.cache_pool.npu_cache[n][0]
+-                if check_and_recover_uce_in_cache(uce_addr_start, uce_addr_end, k_cache, n, "kcache"):
+-                    return True
+-
+-                v_cache = self.cache_pool.npu_cache[n][1]
+-                if check_and_recover_uce_in_cache(uce_addr_start, uce_addr_end, v_cache, n, "vcache"):
+-                    return True
+-        return False

Amindie_llm/text_generator/adapter/recovery_utils.py+36-0

@@ -0,0 +1,36 @@
	1	+# Copyright (c) Huawei Technologies Co., Ltd. 2024-2025. All rights reserved.
	2	+# MindIE is licensed under Mulan PSL v2.
	3	+# You can use this software according to the terms and conditions of the Mulan PSL v2.
	4	+# You may obtain a copy of Mulan PSL v2 at:
	5	+# http://license.coscl.org.cn/MulanPSL2
	6	+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
	7	+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
	8	+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
	9	+# See the Mulan PSL v2 for more details.
	10	+
	11	+import torch_npu
	12	+
	13	+from ...utils.log.logging import logger
	14	+
	15	+
	16	+def is_uce_error_addr_overlap_tensor_addr(uce_addr_start, uce_addr_end, tensor_addr_start, tensor_addr_end):
	17	+ return (uce_addr_start >= tensor_addr_start) and (uce_addr_end <= tensor_addr_end)
	18	+
	19	+
	20	+def get_tensor_address_range(input_tensor):
	21	+ addr_start = input_tensor.data_ptr()
	22	+ addr_end = addr_start + input_tensor.numel() * input_tensor.element_size()
	23	+ return addr_start, addr_end
	24	+
	25	+
	26	+def check_and_recover_uce_in_cache(uce_addr_start, uce_addr_end, cache_tensor, layer_idx, cache_type):
	27	+ cache_addr_start, cache_addr_end = get_tensor_address_range(cache_tensor)
	28	+ if is_uce_error_addr_overlap_tensor_addr(uce_addr_start, uce_addr_end, cache_addr_start, cache_addr_end):
	29	+ torch_npu.npu._recovery.update_npu_tensor_to_safe(cache_tensor)
	30	+ logger.info(f"HBM UCE address in {cache_type} of layer {layer_idx}, update {cache_type} to safe")
	31	+ return True
	32	+ else:
	33	+ logger.info(
	34	+ f"HBM UCE address not in {cache_type} address ({cache_addr_start}, {cache_addr_end}) of layer {layer_idx}"
	35	+ )
	36	+ return False

Mmindie_llm/text_generator/generator.py+9-15

	@@ -48,7 +48,7 @@ from mindie_llm.text_generator.utils.request import Request
48	from mindie_llm.utils.decorators.time_decorator import timer	48	from mindie_llm.utils.decorators.time_decorator import timer
49	from mindie_llm.utils.env import ENV	49	from mindie_llm.utils.env import ENV
50	from mindie_llm.utils.log import ErrorCode, logger, print_log	50	from mindie_llm.utils.log import ErrorCode, logger, print_log
51	-from mindie_llm.utils.log.error_code import ErrorCodeException, convert_exception_to_error_code	51	+from mindie_llm.utils.log.error_code import ErrorCodeException, convert_exception_to_error_code, is_force_stop_exception
52	from mindie_llm.utils.status import MindieLlmStatusCode	52	from mindie_llm.utils.status import MindieLlmStatusCode
53	from mindie_llm.utils.tensor import npu	53	from mindie_llm.utils.tensor import npu
54	from mindie_llm.text_generator.utils.separate_deployment_engine import (	54	from mindie_llm.text_generator.utils.separate_deployment_engine import (
	@@ -554,6 +554,9 @@ class Generator(PDInterface):
554	else:	554	else:
555	raise e	555	raise e
556	except Exception as e:	556	except Exception as e:
		557	+ if isinstance(e, ErrorCodeException):
		558	+ self.generator_backend.is_fault_device = True
		559	+ raise e
		姜凯强3月18日 [问题分类] 软件结构 [具体问题] 变量置为true是否和下面重叠 [修改建议] 在下面同一置true yuzechen 3月19日评论： 3月19日评论：
557	error_code = convert_exception_to_error_code(str(e))	560	error_code = convert_exception_to_error_code(str(e))
558		561
559	# Handle PyTorch OOM(Only supports Torch 2.6+ native exception)	562	# Handle PyTorch OOM(Only supports Torch 2.6+ native exception)
	@@ -572,9 +575,13 @@ class Generator(PDInterface):
572	f'{error_code.name} fault happened in generate_token, error code: {error_code.value}.'	575	f'{error_code.name} fault happened in generate_token, error code: {error_code.value}.'
573	)	576	)
574	logger.error(message)	577	logger.error(message)
		578	+ self.generator_backend.is_fault_device = True
575	raise ErrorCodeException(error_code) from e	579	raise ErrorCodeException(error_code) from e
576	print_log(self.rank, logger.error, f'Unknown exception: {e}')	580	print_log(self.rank, logger.error, f'Unknown exception: {e}')
577	if self.is_inference_pause:	581	if self.is_inference_pause:
		582	+ if is_force_stop_exception(e):
		583	+ logger.info(f"FORCE STOP exception detected in generator.generate_token: {e}")
		584	+ self.generator_backend.notify_force_stop_exception()
578	return GenerationOutput.make_empty()	585	return GenerationOutput.make_empty()
579	raise e	586	raise e
580		587
	@@ -756,13 +763,6 @@ class Generator(PDInterface):
756	"npu_device_id": self.npu_device_id	763	"npu_device_id": self.npu_device_id
757	}	764	}
758		765
759	- # Only 'atb' backend supports recovery commands
760	- if self.backend_type != 'atb':
761	- error_msg = f"Recovery commands are only supported by 'atb' backend, got: {self.backend_type!r}"
762	- logger.error(error_msg)
763	- ret_dict[error_msg_key] = error_msg
764	- return ret_dict
765	-
766	logger.info(f"Executing recover command {command} on NPU device {self.npu_device_id}.")	766	logger.info(f"Executing recover command {command} on NPU device {self.npu_device_id}.")
767	# Dispatch by command	767	# Dispatch by command
768	if command == "CMD_REINIT_NPU":	768	if command == "CMD_REINIT_NPU":
	@@ -770,10 +770,6 @@ class Generator(PDInterface):
770	self.infer_context.reset_all_context()	770	self.infer_context.reset_all_context()
771	self.plugin_manager.error_code_collected_in_async = None	771	self.plugin_manager.error_code_collected_in_async = None
772	ret_dict = self.generator_backend.execute_recover_command(command)	772	ret_dict = self.generator_backend.execute_recover_command(command)
773	- if ret_dict[command_res_key] == 0:
774	- acl.rt.set_device(self.npu_device_id)
775	- self.model_wrapper.resume_hccl_comm()
776	- ret_dict[command_res_key] = 0 # success
777	except Exception as e:	773	except Exception as e:
778	error_msg = f"Failed to execute recovery command {command!r}: {e}"	774	error_msg = f"Failed to execute recovery command {command!r}: {e}"
779	logger.exception(error_msg)	775	logger.exception(error_msg)
	@@ -799,14 +795,12 @@ class Generator(PDInterface):
799	# Delegate pause to generator backend	795	# Delegate pause to generator backend
800	self.is_inference_pause = True	796	self.is_inference_pause = True
801	self.plugin_manager.is_inference_pause = True	797	self.plugin_manager.is_inference_pause = True
802	- time.sleep(20)
803	-
804	ret_dict = self.generator_backend.execute_recover_command(command)	798	ret_dict = self.generator_backend.execute_recover_command(command)
805		799
806	elif command == "CMD_PAUSE_ENGINE_ROCE":	800	elif command == "CMD_PAUSE_ENGINE_ROCE":
807	# Delegate pause to generator backend	801	# Delegate pause to generator backend
808	self.is_inference_pause = True	802	self.is_inference_pause = True
809	- self.plugin.is_inference_pause = True	803	+ self.plugin_manager.is_inference_pause = True
810	ret_dict[command_res_key] = 0	804	ret_dict[command_res_key] = 0
811		805
812	elif command == "CMD_CLEAR_TRANSER":	806	elif command == "CMD_CLEAR_TRANSER":

Mmindie_llm/text_generator/plugins/plugin_manager.py+10-1

	@@ -37,7 +37,7 @@ from mindie_llm.utils.decorators.time_decorator import timer
37	from mindie_llm.utils.env import ENV	37	from mindie_llm.utils.env import ENV
38	from mindie_llm.utils.log import logger, HandlerType	38	from mindie_llm.utils.log import logger, HandlerType
39	from mindie_llm.utils.prof.profiler import span_start, span_end, span_req, span_attr, count_block	39	from mindie_llm.utils.prof.profiler import span_start, span_end, span_req, span_attr, count_block
40	-from mindie_llm.utils.log.error_code import ErrorCodeException, convert_exception_to_error_code	40	+from mindie_llm.utils.log.error_code import ErrorCodeException, convert_exception_to_error_code, is_force_stop_exception
41		41
42	if TYPE_CHECKING:	42	if TYPE_CHECKING:
43	from mindie_llm.text_generator.utils import (	43	from mindie_llm.text_generator.utils import (
	@@ -244,6 +244,10 @@ class PluginManager:
244	except Exception as e:	244	except Exception as e:
245	if self.is_inference_pause:	245	if self.is_inference_pause:
246	logger.info(f"Mocking response due to inference pause for trace_ids={trace_ids}.")	246	logger.info(f"Mocking response due to inference pause for trace_ids={trace_ids}.")
		247	+ # Check for FORCE STOP exception and notify generator_backend if it's GeneratorTorch
		248	+ if is_force_stop_exception(e):
		249	+ logger.info(f"FORCE STOP exception detected in plugin_manager.generate_token: {e}")
		250	+ self.generator_backend.notify_force_stop_exception()
		姜凯强3月18日 [问题分类] 编程规范 [具体问题] 这段以及下面重复的，是不是放到generator.py里面写一次就行 yuzechen 3月19日评论： 3月19日评论：
247	return GenerationOutput.make_empty()	251	return GenerationOutput.make_empty()
248	logger.exception(	252	logger.exception(
249	f"Error encountered in generate_token (trace_ids={trace_ids}). "	253	f"Error encountered in generate_token (trace_ids={trace_ids}). "
	@@ -654,6 +658,11 @@ class PluginManager:
654	except Exception as e:	658	except Exception as e:
655	trace_ids = getattr(model_input_wrapper, 'trace_ids', 'unknown')	659	trace_ids = getattr(model_input_wrapper, 'trace_ids', 'unknown')
656		660
		661	+ # Check for FORCE STOP exception and notify generator_backend if it's GeneratorTorch
		662	+ if is_force_stop_exception(e):
		663	+ logger.info(f"FORCE STOP exception detected in plugin_manager.forward_loop: {e}")
		664	+ self.generator_backend.notify_force_stop_exception()
		665	+
		姜凯强3月18日同上 yuzechen 3月19日评论： 3月19日评论：
657	error_code = convert_exception_to_error_code(str(e))	666	error_code = convert_exception_to_error_code(str(e))
658		667
659	# Handle PyTorch OOM(Only supports Torch 2.6+ native exception)	668	# Handle PyTorch OOM(Only supports Torch 2.6+ native exception)

Mmindie_llm/utils/log/error_code.py+10-0

	@@ -78,3 +78,13 @@ def convert_exception_to_error_code(exception_str: str):
78	if exception_key in exception_str:	78	if exception_key in exception_str:
79	return error_code	79	return error_code
80	return None	80	return None
		81	+
		82	+
		83	+def is_force_stop_exception(exception: Exception) -> bool:
		84	+ '''Check if the exception is a FORCE STOP exception.'''
		85	+ if not isinstance(exception, RuntimeError):
		86	+ return False
		87	+ exception_str = str(exception).upper()
		88	+ if "FORCE STOP" in exception_str:
		89	+ return True
		90	+ return False

Atests/pythontest/npu/text_generator/adapter/test_generator_backend.py+535-0

@@ -0,0 +1,535 @@
	1	+# Copyright (c) Huawei Technologies Co., Ltd. 2025. All rights reserved.
	2	+# MindIE is licensed under Mulan PSL v2.
	3	+# You can use this software according to the terms and conditions of the Mulan PSL v2.
	4	+# You may obtain a copy of Mulan PSL v2 at:
	5	+# http://license.coscl.org.cn/MulanPSL2
	6	+# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
	7	+# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
	8	+# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
	9	+# See the Mulan PSL v2 for more details.
	10	+
	11	+import unittest
	12	+from unittest.mock import MagicMock, patch
	13	+
	14	+import numpy as np
	15	+import torch
	16	+
	17	+from mindie_llm.text_generator.adapter.generator_backend import GeneratorBackend
	18	+from mindie_llm.text_generator.utils.model_input import ModelInput
	19	+from mindie_llm.text_generator.utils.sampling_metadata import SamplingMetadata, SamplingData, SamplingParam
	20	+GENERATOR_BACKEND_AVAILABLE = True
	21	+_import_error = None
	22	+
	23	+MOCKED_GET_MODEL_WRAPPER = "mindie_llm.text_generator.adapter.generator_backend.get_model_wrapper"
	24	+MOCKED_SAMPLER = "mindie_llm.text_generator.adapter.generator_backend.Sampler"
	25	+
	26	+
	27	+def get_default_model_config():
	28	+ """Return minimal model config for GeneratorBackend init."""
	29	+ return {
	30	+ 'backend_type': 'atb',
	31	+ 'npu_device_id': 0,
	32	+ 'local_rank': 0,
	33	+ 'rank': 0,
	34	+ 'world_size': 1,
	35	+ 'trust_remote_code': False,
	36	+ }
	37	+
	38	+
	39	+def create_mock_model_wrapper():
	40	+ """Create a mock model wrapper for tests."""
	41	+ mock_wrapper = MagicMock()
	42	+ mock_wrapper.config = MagicMock()
	43	+ mock_wrapper.config_dict = {'max_position_embeddings': 32768}
	44	+ mock_wrapper.model_info = None
	45	+ mock_wrapper.max_position_embeddings = 32768
	46	+ mock_wrapper.forward = MagicMock(return_value=torch.tensor([[0.1, 0.2, 0.7]]))
	47	+ mock_wrapper.make_context = MagicMock(return_value=[1, 2, 3])
	48	+ return mock_wrapper
	49	+
	50	+
	51	+@unittest.skipUnless(GENERATOR_BACKEND_AVAILABLE, f"torch_npu not installed: {_import_error or 'ok'}")
	52	+class TestGeneratorBackend(unittest.TestCase):
	53	+ """Unit tests for GeneratorBackend."""
	54	+
	55	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	56	+ def test_init_success(self, mock_get_wrapper):
	57	+ """Test successful initialization with valid config."""
	58	+ mock_wrapper = create_mock_model_wrapper()
	59	+ mock_get_wrapper.return_value = mock_wrapper
	60	+
	61	+ config = get_default_model_config()
	62	+ backend = GeneratorBackend(config)
	63	+
	64	+ self.assertEqual(backend.rank, 0)
	65	+ self.assertEqual(backend.world_size, 1)
	66	+ self.assertEqual(backend.npu_device_id, 0)
	67	+ self.assertEqual(backend.max_position_embeddings, 32768)
	68	+ mock_get_wrapper.assert_called_once()
	69	+
	70	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	71	+ def test_init_world_size_invalid_small(self, mock_get_wrapper):
	72	+ """Test init raises ValueError when world_size < 1."""
	73	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	74	+ config = get_default_model_config()
	75	+ config['world_size'] = 0
	76	+
	77	+ with self.assertRaises(ValueError) as cm:
	78	+ GeneratorBackend(config)
	79	+ self.assertIn("World size should be in the range of 1 to 1048576", str(cm.exception))
	80	+
	81	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	82	+ def test_init_world_size_invalid_large(self, mock_get_wrapper):
	83	+ """Test init raises ValueError when world_size > MAX_WORLD_SIZE."""
	84	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	85	+ config = get_default_model_config()
	86	+ config['world_size'] = 1048577
	87	+
	88	+ with self.assertRaises(ValueError) as cm:
	89	+ GeneratorBackend(config)
	90	+ self.assertIn("World size should be in the range of 1 to 1048576", str(cm.exception))
	91	+
	92	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	93	+ def test_init_rank_invalid_negative(self, mock_get_wrapper):
	94	+ """Test init raises ValueError when rank < 0."""
	95	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	96	+ config = get_default_model_config()
	97	+ config['rank'] = -1
	98	+
	99	+ with self.assertRaises(ValueError) as cm:
	100	+ GeneratorBackend(config)
	101	+ self.assertIn("Rank should be in the range of 0 to world_size - 1", str(cm.exception))
	102	+
	103	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	104	+ def test_init_rank_invalid_exceeds_world_size(self, mock_get_wrapper):
	105	+ """Test init raises ValueError when rank >= world_size."""
	106	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	107	+ config = get_default_model_config()
	108	+ config['rank'] = 1
	109	+ config['world_size'] = 1
	110	+
	111	+ with self.assertRaises(ValueError) as cm:
	112	+ GeneratorBackend(config)
	113	+ self.assertIn("Rank should be in the range of 0 to world_size - 1", str(cm.exception))
	114	+
	115	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	116	+ def test_init_local_rank_invalid(self, mock_get_wrapper):
	117	+ """Test init raises ValueError when local_rank is invalid."""
	118	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	119	+ config = get_default_model_config()
	120	+ config['local_rank'] = 2
	121	+ config['world_size'] = 1
	122	+
	123	+ with self.assertRaises(ValueError) as cm:
	124	+ GeneratorBackend(config)
	125	+ self.assertIn("Local rank should be in the range of 0 to world_size - 1", str(cm.exception))
	126	+
	127	+ def test_repeat_sample_param_none(self):
	128	+ """Test repeat_sample_param returns None when param_tensor is None."""
	129	+ result = GeneratorBackend.repeat_sample_param(None, [1, 2, 3])
	130	+ self.assertIsNone(result)
	131	+
	132	+ def test_repeat_sample_param_valid(self):
	133	+ """Test repeat_sample_param with valid tensors."""
	134	+ param_tensor = [torch.tensor([[1.0, 2.0]]), torch.tensor([[3.0, 4.0]])]
	135	+ tokens_num_per_batch = [2, 1]
	136	+ result = GeneratorBackend.repeat_sample_param(param_tensor, tokens_num_per_batch)
	137	+ self.assertIsNotNone(result)
	138	+ self.assertEqual(result.shape[0], 3) # 2 + 1
	139	+
	140	+ @patch(MOCKED_SAMPLER)
	141	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	142	+ def test_configure_sampler(self, mock_get_wrapper, mock_sampler_cls):
	143	+ """Test configure_sampler calls sampler.configure."""
	144	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	145	+ mock_sampler = MagicMock()
	146	+ mock_sampler_cls.return_value = mock_sampler
	147	+ backend = GeneratorBackend(get_default_model_config())
	148	+ sampling_metadata = MagicMock()
	149	+
	150	+ backend.configure_sampler(sampling_metadata)
	151	+ mock_sampler.configure.assert_called_once_with(sampling_metadata)
	152	+
	153	+ @patch(MOCKED_SAMPLER)
	154	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	155	+ def test_init_sampler(self, mock_get_wrapper, mock_sampler_cls):
	156	+ """Test init_sampler calls sampler.initialize."""
	157	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	158	+ mock_sampler = MagicMock()
	159	+ mock_sampler_cls.return_value = mock_sampler
	160	+ backend = GeneratorBackend(get_default_model_config())
	161	+ backend.device = 'cpu'
	162	+
	163	+ backend.init_sampler(2) # eos_token_id
	164	+ mock_sampler.initialize.assert_called_once_with('cpu', 2)
	165	+
	166	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	167	+ def test_set_device(self, mock_get_wrapper):
	168	+ """Test set_device does nothing (pass)."""
	169	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	170	+ backend = GeneratorBackend(get_default_model_config())
	171	+ backend.set_device() # Should not raise
	172	+
	173	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	174	+ def test_notify_force_stop_exception(self, mock_get_wrapper):
	175	+ """Test notify_force_stop_exception sets event."""
	176	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	177	+ backend = GeneratorBackend(get_default_model_config())
	178	+
	179	+ self.assertFalse(backend.force_stop_exception_occurred.is_set())
	180	+ backend.notify_force_stop_exception()
	181	+ self.assertTrue(backend.force_stop_exception_occurred.is_set())
	182	+
	183	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	184	+ def test_execute_recover_command_unknown(self, mock_get_wrapper):
	185	+ """Test execute_recover_command with unknown command."""
	186	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	187	+ backend = GeneratorBackend(get_default_model_config())
	188	+
	189	+ result = backend.execute_recover_command("UNKNOWN_CMD")
	190	+ self.assertEqual(result["command_result"], 1)
	191	+ self.assertEqual(result["npu_device_id"], 0)
	192	+
	193	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	194	+ def test_execute_recover_command_cmd_reinit_npu(self, mock_get_wrapper):
	195	+ """Test execute_recover_command with CMD_REINIT_NPU catches NotImplementedError."""
	196	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	197	+ backend = GeneratorBackend(get_default_model_config())
	198	+
	199	+ result = backend.execute_recover_command("CMD_REINIT_NPU")
	200	+ self.assertEqual(result["command_result"], 1)
	201	+ self.assertIn("Subclasses must implement", result["error_msg"])
	202	+
	203	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	204	+ @patch("mindie_llm.text_generator.adapter.generator_backend.time.sleep")
	205	+ @patch("torch_npu.npu.stop_device")
	206	+ def test_execute_recover_command_cmd_pause_engine_stop_failed(
	207	+ self, mock_stop_device, mock_sleep, mock_get_wrapper
	208	+ ):
	209	+ """Test execute_recover_command when stop_device fails."""
	210	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	211	+ mock_stop_device.return_value = 1 # failure
	212	+
	213	+ backend = GeneratorBackend(get_default_model_config())
	214	+ result = backend.execute_recover_command("CMD_PAUSE_ENGINE")
	215	+
	216	+ self.assertEqual(result["command_result"], 1)
	217	+ self.assertIn("Stop device failed", result["error_msg"])
	218	+
	219	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	220	+ @patch("mindie_llm.text_generator.adapter.generator_backend.time.sleep")
	221	+ @patch("torch_npu.npu.stop_device")
	222	+ def test_execute_recover_command_cmd_pause_engine_uce_reschedule(
	223	+ self, mock_stop_device, mock_sleep, mock_get_wrapper
	224	+ ):
	225	+ """Test execute_recover_command when UCE requires reschedule."""
	226	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	227	+ mock_stop_device.return_value = 0
	228	+
	229	+ backend = GeneratorBackend(get_default_model_config())
	230	+ backend._handle_uce_error = MagicMock(return_value=(1, "HBM uce address unknown"))
	231	+
	232	+ result = backend.execute_recover_command("CMD_PAUSE_ENGINE")
	233	+
	234	+ self.assertEqual(result["command_result"], 1)
	235	+ self.assertIn("HBM uce address unknown", result["error_msg"])
	236	+
	237	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	238	+ @patch("mindie_llm.text_generator.adapter.generator_backend.time.sleep")
	239	+ @patch("torch_npu.npu.stop_device")
	240	+ def test_execute_recover_command_cmd_pause_engine_uce_recovered(
	241	+ self, mock_stop_device, mock_sleep, mock_get_wrapper
	242	+ ):
	243	+ """Test execute_recover_command when UCE is recovered."""
	244	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	245	+ mock_stop_device.return_value = 0
	246	+
	247	+ backend = GeneratorBackend(get_default_model_config())
	248	+ backend._handle_uce_error = MagicMock(return_value=(2, ""))
	249	+
	250	+ result = backend.execute_recover_command("CMD_PAUSE_ENGINE")
	251	+
	252	+ self.assertEqual(result["command_result"], 0)
	253	+ self.assertEqual(result["error_msg"], "")
	254	+
	255	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	256	+ @patch("mindie_llm.text_generator.adapter.generator_backend.time.sleep")
	257	+ @patch("torch_npu.npu.stop_device")
	258	+ def test_execute_recover_command_cmd_pause_engine_force_stop_timeout(
	259	+ self, mock_stop_device, mock_sleep, mock_get_wrapper
	260	+ ):
	261	+ """Test execute_recover_command when force stop times out."""
	262	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	263	+ mock_stop_device.return_value = 0
	264	+
	265	+ backend = GeneratorBackend(get_default_model_config())
	266	+ backend._handle_uce_error = MagicMock(return_value=(0, ""))
	267	+ backend._wait_for_force_stop_exception = MagicMock(return_value=False)
	268	+
	269	+ result = backend.execute_recover_command("CMD_PAUSE_ENGINE")
	270	+
	271	+ self.assertEqual(result["command_result"], 1)
	272	+ self.assertIn("Timeout waiting for FORCE STOP exception", result["error_msg"])
	273	+
	274	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	275	+ @patch("mindie_llm.text_generator.adapter.generator_backend.time.sleep")
	276	+ @patch("torch_npu.npu.stop_device")
	277	+ def test_execute_recover_command_cmd_pause_engine_force_stop_success(
	278	+ self, mock_stop_device, mock_sleep, mock_get_wrapper
	279	+ ):
	280	+ """Test execute_recover_command when force stop succeeds."""
	281	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	282	+ mock_stop_device.return_value = 0
	283	+
	284	+ backend = GeneratorBackend(get_default_model_config())
	285	+ backend._handle_uce_error = MagicMock(return_value=(0, ""))
	286	+ backend._wait_for_force_stop_exception = MagicMock(return_value=True)
	287	+
	288	+ result = backend.execute_recover_command("CMD_PAUSE_ENGINE")
	289	+
	290	+ self.assertEqual(result["command_result"], 0)
	291	+ self.assertEqual(result["error_msg"], "")
	292	+
	293	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	294	+ def test_build_inputs(self, mock_get_wrapper):
	295	+ """Test build_inputs calls make_context for each conversation."""
	296	+ mock_wrapper = create_mock_model_wrapper()
	297	+ mock_wrapper.make_context = MagicMock(side_effect=[[1, 2], [3, 4, 5]])
	298	+ mock_get_wrapper.return_value = mock_wrapper
	299	+
	300	+ backend = GeneratorBackend(get_default_model_config())
	301	+ conversations = [[{"role": "user", "content": "hi"}], [{"role": "user", "content": "hello"}]]
	302	+
	303	+ result = backend.build_inputs(conversations)
	304	+
	305	+ self.assertEqual(result, [[1, 2], [3, 4, 5]])
	306	+ self.assertEqual(mock_wrapper.make_context.call_count, 2)
	307	+
	308	+ @patch(MOCKED_SAMPLER)
	309	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	310	+ def test_clear_cache(self, mock_get_wrapper, mock_sampler_cls):
	311	+ """Test clear_cache calls sampler.clear_cache."""
	312	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	313	+ mock_sampler = MagicMock()
	314	+ mock_sampler_cls.return_value = mock_sampler
	315	+ backend = GeneratorBackend(get_default_model_config())
	316	+
	317	+ result = backend.clear_cache([1, 2, 3])
	318	+ mock_sampler.clear_cache.assert_called_once()
	319	+ self.assertEqual(result, 1)
	320	+
	321	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	322	+ def test_update_config(self, mock_get_wrapper):
	323	+ """Test update_config updates config attributes."""
	324	+ mock_wrapper = create_mock_model_wrapper()
	325	+ mock_wrapper.config_dict = {'max_position_embeddings': 32768}
	326	+ mock_get_wrapper.return_value = mock_wrapper
	327	+
	328	+ backend = GeneratorBackend(get_default_model_config())
	329	+ backend.update_config({'max_position_embeddings': 8192})
	330	+
	331	+ self.assertEqual(backend.config.max_position_embeddings, 8192)
	332	+
	333	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	334	+ def test_forward(self, mock_get_wrapper):
	335	+ """Test forward delegates to model_wrapper.forward."""
	336	+ mock_wrapper = create_mock_model_wrapper()
	337	+ expected_result = torch.tensor([[0.5, 0.3, 0.2]])
	338	+ mock_wrapper.forward = MagicMock(return_value=expected_result)
	339	+ mock_get_wrapper.return_value = mock_wrapper
	340	+
	341	+ backend = GeneratorBackend(get_default_model_config())
	342	+ model_input = ModelInput(
	343	+ input_ids=np.array([1, 2, 3]),
	344	+ position_ids=np.array([0, 1, 2]),
	345	+ block_tables=np.array([[0]]),
	346	+ slots=np.array([0, 1, 2]),
	347	+ context_length=np.array([3]),
	348	+ max_seq_len=3,
	349	+ prefill_head_indices=np.array([2]),
	350	+ is_prefill=True,
	351	+ query_length=None,
	352	+ adapter_ids=None,
	353	+ dp_rank_ids=np.array([0]),
	354	+ )
	355	+
	356	+ result = backend.forward(model_input)
	357	+
	358	+ mock_wrapper.forward.assert_called_once_with(model_input)
	359	+ self.assertTrue(torch.equal(result, expected_result))
	360	+
	361	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	362	+ def test_sample_with_sampling_metadata(self, mock_get_wrapper):
	363	+ """Test sample with SamplingMetadata (non-deprecated path)."""
	364	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	365	+ backend = GeneratorBackend(get_default_model_config())
	366	+
	367	+ logits = torch.tensor([[0.1, 0.2, 0.7]])
	368	+ sampling_metadata = SamplingMetadata.from_numpy(
	369	+ batch_sequence_ids=[np.array([0])],
	370	+ is_prefill=True,
	371	+ to_tensor=lambda x: torch.tensor(x) if x is not None else None,
	372	+ )
	373	+
	374	+ mock_sampling_output = MagicMock()
	375	+ mock_sampling_output.token_ids = np.array([2])
	376	+ backend.sampler = MagicMock(return_value=mock_sampling_output)
	377	+
	378	+ output = backend.sample(logits, sampling_metadata)
	379	+ backend.sampler.assert_called_once_with(logits, sampling_metadata)
	380	+ self.assertEqual(output, mock_sampling_output)
	381	+
	382	+
	383	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	384	+ def test_execute_cmd_reinit_npu_raises(self, mock_get_wrapper):
	385	+ """Test _execute_cmd_reinit_npu raises NotImplementedError."""
	386	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	387	+ backend = GeneratorBackend(get_default_model_config())
	388	+
	389	+ with self.assertRaises(NotImplementedError) as cm:
	390	+ backend._execute_cmd_reinit_npu()
	391	+ self.assertIn("Subclasses must implement", str(cm.exception))
	392	+
	393	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	394	+ def test_wait_for_force_stop_exception_is_fault_device(self, mock_get_wrapper):
	395	+ """Test _wait_for_force_stop_exception when is_fault_device is True."""
	396	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	397	+ backend = GeneratorBackend(get_default_model_config())
	398	+ backend.is_fault_device = True
	399	+
	400	+ result = backend._wait_for_force_stop_exception()
	401	+ self.assertTrue(result)
	402	+
	403	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	404	+ def test_wait_for_force_stop_exception_detected(self, mock_get_wrapper):
	405	+ """Test _wait_for_force_stop_exception when event is set."""
	406	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	407	+ backend = GeneratorBackend(get_default_model_config())
	408	+ backend.force_stop_exception_occurred.set()
	409	+
	410	+ result = backend._wait_for_force_stop_exception()
	411	+ self.assertTrue(result)
	412	+
	413	+ @patch("torch.npu.check_uce_in_memory")
	414	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	415	+ def test_handle_uce_error_no_uce(self, mock_get_wrapper, mock_check_uce):
	416	+ """Test _handle_uce_error when no UCE error (res=0)."""
	417	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	418	+ mock_check_uce.return_value = 0
	419	+
	420	+ backend = GeneratorBackend(get_default_model_config())
	421	+ result, error_msg = backend._handle_uce_error()
	422	+
	423	+ self.assertEqual(result, 0)
	424	+ self.assertEqual(error_msg, "")
	425	+
	426	+ @patch("torch.npu.check_uce_in_memory")
	427	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	428	+ def test_handle_uce_error_unknown_addr(self, mock_get_wrapper, mock_check_uce):
	429	+ """Test _handle_uce_error when UCE address unknown (res=1)."""
	430	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	431	+ mock_check_uce.return_value = 1
	432	+
	433	+ backend = GeneratorBackend(get_default_model_config())
	434	+ result, error_msg = backend._handle_uce_error()
	435	+
	436	+ self.assertEqual(result, 1)
	437	+ self.assertIn("uce address unknown", error_msg)
	438	+
	439	+ @patch("torch.npu.check_uce_in_memory")
	440	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	441	+ def test_handle_uce_error_not_in_kvcache(self, mock_get_wrapper, mock_check_uce):
	442	+ """Test _handle_uce_error when UCE not in kvcache (res=2 or 3)."""
	443	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	444	+ mock_check_uce.return_value = 2
	445	+
	446	+ backend = GeneratorBackend(get_default_model_config())
	447	+ backend.cache_pool = MagicMock()
	448	+ backend.cache_pool.kvcache_settings = MagicMock()
	449	+ backend.cache_pool.kvcache_settings.num_layers = 1
	450	+ backend.cache_pool.npu_cache = [(torch.tensor([1, 2]), torch.tensor([3, 4]))]
	451	+ with patch("mindie_llm.text_generator.adapter.generator_backend.torch_npu.npu._get_uce_addr",
	452	+ return_value=[{"ptr": 99999, "size": 100}]):
	453	+ backend._check_and_recover_uce_in_kvcache = MagicMock(return_value=False)
	454	+
	455	+ result, error_msg = backend._handle_uce_error()
	456	+
	457	+ self.assertEqual(result, 1)
	458	+ self.assertIn("not overlap kvcache address", error_msg)
	459	+
	460	+ @patch("torch.npu.check_uce_in_memory")
	461	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	462	+ def test_handle_uce_error_recovered(self, mock_get_wrapper, mock_check_uce):
	463	+ """Test _handle_uce_error when UCE is recovered."""
	464	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	465	+ mock_check_uce.return_value = 2
	466	+
	467	+ backend = GeneratorBackend(get_default_model_config())
	468	+ backend.cache_pool = MagicMock()
	469	+ backend._check_and_recover_uce_in_kvcache = MagicMock(return_value=True)
	470	+
	471	+ result, error_msg = backend._handle_uce_error()
	472	+
	473	+ self.assertEqual(result, 2)
	474	+ self.assertEqual(error_msg, "")
	475	+
	476	+ @patch("mindie_llm.text_generator.adapter.generator_backend.torch_npu.npu._get_uce_addr")
	477	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	478	+ def test_check_and_recover_uce_in_kvcache_empty_list(self, mock_get_wrapper, mock_get_uce):
	479	+ """Test _check_and_recover_uce_in_kvcache with empty UCE list."""
	480	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	481	+ mock_get_uce.return_value = []
	482	+
	483	+ backend = GeneratorBackend(get_default_model_config())
	484	+ backend.cache_pool = MagicMock()
	485	+ backend.cache_pool.kvcache_settings = MagicMock()
	486	+ backend.cache_pool.kvcache_settings.num_layers = 1
	487	+ backend.cache_pool.npu_cache = [(torch.tensor([1]), torch.tensor([2]))]
	488	+
	489	+ result = backend._check_and_recover_uce_in_kvcache()
	490	+ self.assertFalse(result)
	491	+
	492	+ @patch("mindie_llm.text_generator.adapter.generator_backend.check_and_recover_uce_in_cache")
	493	+ @patch("mindie_llm.text_generator.adapter.generator_backend.torch_npu.npu._get_uce_addr")
	494	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	495	+ def test_check_and_recover_uce_in_kvcache_recovered(
	496	+ self, mock_get_wrapper, mock_get_uce, mock_check_recover
	497	+ ):
	498	+ """Test _check_and_recover_uce_in_kvcache when recovery succeeds."""
	499	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	500	+ mock_get_uce.return_value = [{"ptr": 100, "size": 50}]
	501	+ mock_check_recover.return_value = True
	502	+
	503	+ backend = GeneratorBackend(get_default_model_config())
	504	+ backend.cache_pool = MagicMock()
	505	+ backend.cache_pool.kvcache_settings = MagicMock()
	506	+ backend.cache_pool.kvcache_settings.num_layers = 1
	507	+ backend.cache_pool.npu_cache = [(torch.tensor([1]), torch.tensor([2]))]
	508	+
	509	+ result = backend._check_and_recover_uce_in_kvcache()
	510	+ self.assertTrue(result)
	511	+ mock_check_recover.assert_called()
	512	+
	513	+ @patch("mindie_llm.text_generator.adapter.generator_backend.check_and_recover_uce_in_cache")
	514	+ @patch("mindie_llm.text_generator.adapter.generator_backend.torch_npu.npu._get_uce_addr")
	515	+ @patch(MOCKED_GET_MODEL_WRAPPER)
	516	+ def test_check_and_recover_uce_in_kvcache_not_recovered(
	517	+ self, mock_get_wrapper, mock_get_uce, mock_check_recover
	518	+ ):
	519	+ """Test _check_and_recover_uce_in_kvcache when recovery fails."""
	520	+ mock_get_wrapper.return_value = create_mock_model_wrapper()
	521	+ mock_get_uce.return_value = [{"ptr": 99999, "size": 100}]
	522	+ mock_check_recover.return_value = False
	523	+
	524	+ backend = GeneratorBackend(get_default_model_config())
	525	+ backend.cache_pool = MagicMock()
	526	+ backend.cache_pool.kvcache_settings = MagicMock()
	527	+ backend.cache_pool.kvcache_settings.num_layers = 1
	528	+ backend.cache_pool.npu_cache = [(torch.tensor([1]), torch.tensor([2]))]
	529	+
	530	+ result = backend._check_and_recover_uce_in_kvcache()
	531	+ self.assertFalse(result)
	532	+
	533	+
	534	+if __name__ == "__main__":
	535	+ unittest.main()

Mtests/pythontest/npu/text_generator/adapter/test_generator_torch.py+5-5

	@@ -19,10 +19,10 @@ import numpy as np
19	from mindie_llm.text_generator.adapter.generator_torch import GeneratorTorch, reorder_array, reorder_tensor	19	from mindie_llm.text_generator.adapter.generator_torch import GeneratorTorch, reorder_array, reorder_tensor
20	from mindie_llm.text_generator.adapter.generator_torch import check_model_config	20	from mindie_llm.text_generator.adapter.generator_torch import check_model_config
21	from mindie_llm.text_generator.utils.model_input import ModelInput	21	from mindie_llm.text_generator.utils.model_input import ModelInput
22	-from mindie_llm.text_generator.adapter.generator_torch import (	22	+from mindie_llm.text_generator.adapter.recovery_utils import (
23	is_uce_error_addr_overlap_tensor_addr,	23	is_uce_error_addr_overlap_tensor_addr,
24	get_tensor_address_range,	24	get_tensor_address_range,
25	- check_and_recover_uce_in_cache	25	+ check_and_recover_uce_in_cache,
26	)	26	)
27		27
28	MOCKED_INIT_METHOD = "mindie_llm.text_generator.adapter.generator_torch.GeneratorTorch.__init__"	28	MOCKED_INIT_METHOD = "mindie_llm.text_generator.adapter.generator_torch.GeneratorTorch.__init__"
	@@ -1259,13 +1259,13 @@ class TestGeneratorTorch(unittest.TestCase):
1259	generator.npu_device_id = 0	1259	generator.npu_device_id = 0
1260		1260
1261	result, error_msg = generator._handle_uce_error()	1261	result, error_msg = generator._handle_uce_error()
1262	- self.assertEqual(result, 0)	1262	+ self.assertEqual(result, 2)
1263	self.assertEqual(error_msg, "")	1263	self.assertEqual(error_msg, "")
1264		1264
1265	addr = self.cache_pool.npu_cache[0][1].data_ptr()	1265	addr = self.cache_pool.npu_cache[0][1].data_ptr()
1266	mock_get_uce_addr.return_value = [{"ptr": addr, "size": 12}]	1266	mock_get_uce_addr.return_value = [{"ptr": addr, "size": 12}]
1267	result, error_msg = generator._handle_uce_error()	1267	result, error_msg = generator._handle_uce_error()
1268	- self.assertEqual(result, 0)	1268	+ self.assertEqual(result, 2)
1269	self.assertEqual(error_msg, "")	1269	self.assertEqual(error_msg, "")
1270		1270
1271	@patch("torch_npu.npu._recovery.update_npu_tensor_to_safe")	1271	@patch("torch_npu.npu._recovery.update_npu_tensor_to_safe")
	@@ -1288,7 +1288,7 @@ class TestGeneratorTorch(unittest.TestCase):
1288	self.assertEqual(addr_end, 1040)	1288	self.assertEqual(addr_end, 1040)
1289		1289
1290	@patch("torch_npu.npu._recovery.update_npu_tensor_to_safe")	1290	@patch("torch_npu.npu._recovery.update_npu_tensor_to_safe")
1291	- @patch("mindie_llm.text_generator.adapter.generator_torch.get_tensor_address_range")	1291	+ @patch("mindie_llm.text_generator.adapter.recovery_utils.get_tensor_address_range")
1292	def test_check_and_recover_uce_in_cache(self, mock_get_range, mock_update):	1292	def test_check_and_recover_uce_in_cache(self, mock_get_range, mock_update):
1293	# Mock tensor address range	1293	# Mock tensor address range
1294	mock_get_range.return_value = (100, 200)	1294	mock_get_range.return_value = (100, 200)

Mtests/pythontest/npu/text_generator/test_generator.py+122-26

	@@ -10,6 +10,7 @@
10	# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.	10	# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
11	# See the Mulan PSL v2 for more details.	11	# See the Mulan PSL v2 for more details.
12	import os	12	import os
		13	+import queue
13	import unittest	14	import unittest
14	import sys	15	import sys
15	from unittest.mock import MagicMock, patch	16	from unittest.mock import MagicMock, patch
	@@ -19,7 +20,7 @@ from ddt import ddt, data, unpack
19		20
20	from mindie_llm.utils.env import ENV	21	from mindie_llm.utils.env import ENV
21	from mindie_llm.utils.status import MindieLlmStatusCode	22	from mindie_llm.utils.status import MindieLlmStatusCode
22	-from mindie_llm.text_generator.generator import Generator, PDInterface, PDModelConfig	23	+from mindie_llm.text_generator.generator import Generator, PDInterface, PDModelConfig, STANDARD_TAG
23	from mindie_llm.text_generator.utils.generation_output import GenerationOutput	24	from mindie_llm.text_generator.utils.generation_output import GenerationOutput
24	from mindie_llm.text_generator.utils.request import Request	25	from mindie_llm.text_generator.utils.request import Request
25	from mindie_llm.text_generator.utils.input_metadata import InputMetadata, SAMPLING_DTYPE	26	from mindie_llm.text_generator.utils.input_metadata import InputMetadata, SAMPLING_DTYPE
	@@ -27,6 +28,7 @@ from mindie_llm.text_generator.adapter import generator_torch
27	from mindie_llm.text_generator.utils.generation_metadata import GenerationParams	28	from mindie_llm.text_generator.utils.generation_metadata import GenerationParams
28	from mindie_llm.connector.common.model_execute_data_pb2 import LoraOperationStatus	29	from mindie_llm.connector.common.model_execute_data_pb2 import LoraOperationStatus
29	from mindie_llm.modeling.model_wrapper.model_info import ModelInfo	30	from mindie_llm.modeling.model_wrapper.model_info import ModelInfo
		31	+from mindie_llm.utils.log.error_code import ErrorCode, ErrorCodeException
30		32
31	from tests.pythontest.npu import FakeModelRunner, FakeModelWrapper, FakeParallelInfo	33	from tests.pythontest.npu import FakeModelRunner, FakeModelWrapper, FakeParallelInfo
32		34
	@@ -489,21 +491,6 @@ class TestGenerator(unittest.TestCase):
489	generator.separate_deployment_worker = None	491	generator.separate_deployment_worker = None
490	ret = generator.unload_lora("fake_id")	492	ret = generator.unload_lora("fake_id")
491	self.assertEqual(ret, LoraOperationStatus.LORA_CMD_SUCCESS)	493	self.assertEqual(ret, LoraOperationStatus.LORA_CMD_SUCCESS)
492	-
493	-
494	- @patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)
495	- def test_execute_recover_command_non_atb_backend(self, _):
496	- """测试非atb后端不支持恢复命令"""
497	- generator = Generator(self.model_config)
498	- generator.separate_deployment_worker = None
499	- generator.backend_type = 'ms'
500	- generator.npu_device_id = 0
501	-
502	- result = generator.execute_recover_command("CMD_REINIT_NPU")
503	-
504	- self.assertEqual(result["command_result"], 1)
505	- self.assertIn("Recovery commands are only supported by 'atb' backend", result["error_msg"])
506	- self.assertEqual(result["npu_device_id"], 0)
507		494
508	@patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)	495	@patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)
509	@patch("mindie_llm.text_generator.generator.acl")	496	@patch("mindie_llm.text_generator.generator.acl")
	@@ -524,8 +511,6 @@ class TestGenerator(unittest.TestCase):
524	"npu_device_id": 0	511	"npu_device_id": 0
525	})	512	})
526	generator.model_wrapper = MagicMock()	513	generator.model_wrapper = MagicMock()
527	- generator.model_wrapper.resume_hccl_comm = MagicMock()
528	- mock_acl.rt.set_device = MagicMock()
529		514
530	result = generator.execute_recover_command("CMD_REINIT_NPU")	515	result = generator.execute_recover_command("CMD_REINIT_NPU")
531		516
	@@ -534,8 +519,6 @@ class TestGenerator(unittest.TestCase):
534	self.assertEqual(result["npu_device_id"], 0)	519	self.assertEqual(result["npu_device_id"], 0)
535	generator.infer_context.reset_all_context.assert_called_once()	520	generator.infer_context.reset_all_context.assert_called_once()
536	generator.generator_backend.execute_recover_command.assert_called_once_with("CMD_REINIT_NPU")	521	generator.generator_backend.execute_recover_command.assert_called_once_with("CMD_REINIT_NPU")
537	- mock_acl.rt.set_device.assert_called_once_with(0)
538	- generator.model_wrapper.resume_hccl_comm.assert_called_once()
539		522
540	@patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)	523	@patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)
541	def test_execute_recover_command_reinit_npu_backend_failure(self, _):	524	def test_execute_recover_command_reinit_npu_backend_failure(self, _):
	@@ -702,14 +685,42 @@ class TestGenerator(unittest.TestCase):
702	generator.generator_backend.execute_recover_command.assert_called_once_with("CMD_PAUSE_ENGINE")	685	generator.generator_backend.execute_recover_command.assert_called_once_with("CMD_PAUSE_ENGINE")
703		686
704	@patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)	687	@patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)
705	- def test_execute_recover_command_pause_engine_roce(self, _):	688	+ def test_execute_recover_command_pause_engine_backend_failure(self, _):
706	- """测试CMD_PAUSE_ENGINE_ROCE命令"""	689	+ """测试CMD_PAUSE_ENGINE命令后端执行失败时仍正确设置pause状态"""
707	generator = Generator(self.model_config)	690	generator = Generator(self.model_config)
		691	+ generator.separate_deployment_worker = None
708	generator.backend_type = 'atb'	692	generator.backend_type = 'atb'
709	generator.npu_device_id = 0	693	generator.npu_device_id = 0
710	generator.is_inference_pause = False	694	generator.is_inference_pause = False
711	- generator.plugin = MagicMock()	695	+ generator.plugin_manager = MagicMock()
712	- generator.plugin.is_inference_pause = False	696	+ generator.plugin_manager.is_inference_pause = False
		697	+ generator.generator_backend = MagicMock()
		698	+ generator.generator_backend.execute_recover_command = MagicMock(return_value={
		699	+ "command_result": 1,
		700	+ "error_msg": "Stop device failed",
		701	+ "npu_device_id": 0
		702	+ })
		703	+
		704	+ result = generator.execute_recover_command("CMD_PAUSE_ENGINE")
		705	+
		706	+ self.assertEqual(result["command_result"], 1)
		707	+ self.assertEqual(result["error_msg"], "Stop device failed")
		708	+ self.assertEqual(result["npu_device_id"], 0)
		709	+ # 即使后端失败，pause 状态应在调用 backend 前已设置
		710	+ self.assertTrue(generator.is_inference_pause)
		711	+ self.assertTrue(generator.plugin_manager.is_inference_pause)
		712	+ generator.generator_backend.execute_recover_command.assert_called_once_with("CMD_PAUSE_ENGINE")
		713	+
		714	+ @patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)
		715	+ def test_execute_recover_command_pause_engine_roce(self, _):
		716	+ """测试CMD_PAUSE_ENGINE_ROCE命令"""
		717	+ generator = Generator(self.model_config)
		718	+ generator.separate_deployment_worker = None
		719	+ generator.backend_type = 'atb'
		720	+ generator.npu_device_id = 0
		721	+ generator.is_inference_pause = False
		722	+ generator.plugin_manager = MagicMock()
		723	+ generator.plugin_manager.is_inference_pause = False
713		724
714	result = generator.execute_recover_command("CMD_PAUSE_ENGINE_ROCE")	725	result = generator.execute_recover_command("CMD_PAUSE_ENGINE_ROCE")
715		726
	@@ -717,7 +728,7 @@ class TestGenerator(unittest.TestCase):
717	self.assertEqual(result["error_msg"], "")	728	self.assertEqual(result["error_msg"], "")
718	self.assertEqual(result["npu_device_id"], 0)	729	self.assertEqual(result["npu_device_id"], 0)
719	self.assertTrue(generator.is_inference_pause)	730	self.assertTrue(generator.is_inference_pause)
720	- self.assertTrue(generator.plugin.is_inference_pause)	731	+ self.assertTrue(generator.plugin_manager.is_inference_pause)
721		732
722	@patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)	733	@patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)
723	def test_execute_recover_command_clear_transer(self, _):	734	def test_execute_recover_command_clear_transer(self, _):
	@@ -747,6 +758,91 @@ class TestGenerator(unittest.TestCase):
747	self.assertIn("Unknown recovery command", result["error_msg"])	758	self.assertIn("Unknown recovery command", result["error_msg"])
748	self.assertEqual(result["npu_device_id"], 0)	759	self.assertEqual(result["npu_device_id"], 0)
749		760
		761	+ @patch('mindie_llm.utils.prof.profiler.span_end')
		762	+ @patch('mindie_llm.utils.prof.profiler.span_attr')
		763	+ @patch('mindie_llm.utils.prof.profiler.span_start')
		764	+ @patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)
		765	+ def test_generate_token_sets_fault_device_when_exception_maps_to_error_code(
		766	+ self, _, mock_span_start, mock_span_attr, mock_span_end
		767	+ ):
		768	+ """异常信息命中 convert_exception_to_error_code 时设置 is_fault_device 并抛出 ErrorCodeException。"""
		769	+ mock_span_start.return_value = None
		770	+ generator = Generator(self.model_config)
		771	+ generator.pd_config = MagicMock()
		772	+ generator.pd_config.model_role = STANDARD_TAG
		773	+ generator.input_metadata_queue = queue.Queue()
		774	+ generator.rank = 0
		775	+ generator.async_inference = False
		776	+ generator.plugin_manager = MagicMock()
		777	+ generator.plugin_manager.generate_token.side_effect = RuntimeError(
		778	+ "backend reported MIE05E0000005 in stack"
		779	+ )
		780	+ generator.generator_backend = MagicMock()
		781	+ generator.generator_backend.is_fault_device = False
		782	+ im = MagicMock(spec=InputMetadata)
		783	+ im.batch_seq_len = np.array([0])
		784	+ im.is_prefill = False
		785	+ with self.assertRaises(ErrorCodeException) as cm:
		786	+ generator.generate_token(im, warmup=False)
		787	+ self.assertEqual(cm.exception.error_code, ErrorCode.TEXT_GENERATOR_OUT_OF_MEMORY)
		788	+ self.assertTrue(generator.generator_backend.is_fault_device)
		789	+
		790	+ @patch('mindie_llm.utils.prof.profiler.span_end')
		791	+ @patch('mindie_llm.utils.prof.profiler.span_attr')
		792	+ @patch('mindie_llm.utils.prof.profiler.span_start')
		793	+ @patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)
		794	+ def test_generate_token_notify_force_stop_when_inference_paused(
		795	+ self, _, mock_span_start, mock_span_attr, mock_span_end
		796	+ ):
		797	+ """推理暂停时 FORCE STOP 异常应调用 notify_force_stop_exception 并返回空 GenerationOutput。"""
		798	+ mock_span_start.return_value = None
		799	+ generator = Generator(self.model_config)
		800	+ generator.pd_config = MagicMock()
		801	+ generator.pd_config.model_role = STANDARD_TAG
		802	+ generator.input_metadata_queue = queue.Queue()
		803	+ generator.rank = 0
		804	+ generator.async_inference = False
		805	+ generator.is_inference_pause = True
		806	+ generator.plugin_manager = MagicMock()
		807	+ generator.plugin_manager.generate_token.side_effect = RuntimeError("User FORCE STOP request")
		808	+ generator.generator_backend = MagicMock()
		809	+ generator.generator_backend.notify_force_stop_exception = MagicMock()
		810	+ im = MagicMock(spec=InputMetadata)
		811	+ im.batch_seq_len = np.array([0])
		812	+ im.is_prefill = False
		813	+ out = generator.generate_token(im, warmup=False)
		814	+ self.assertIsInstance(out, GenerationOutput)
		815	+ self.assertEqual(out.sequence_ids.size, 0)
		816	+ generator.generator_backend.notify_force_stop_exception.assert_called_once()
		817	+
		818	+ @patch('mindie_llm.utils.prof.profiler.span_end')
		819	+ @patch('mindie_llm.utils.prof.profiler.span_attr')
		820	+ @patch('mindie_llm.utils.prof.profiler.span_start')
		821	+ @patch("mindie_llm.text_generator.generator.Generator.__init__", return_value=None)
		822	+ def test_generate_token_force_stop_reraises_when_not_paused(
		823	+ self, _, mock_span_start, mock_span_attr, mock_span_end
		824	+ ):
		825	+ """非暂停状态下 FORCE STOP 仍按未知异常向上抛出。"""
		826	+ mock_span_start.return_value = None
		827	+ generator = Generator(self.model_config)
		828	+ generator.pd_config = MagicMock()
		829	+ generator.pd_config.model_role = STANDARD_TAG
		830	+ generator.input_metadata_queue = queue.Queue()
		831	+ generator.rank = 0
		832	+ generator.async_inference = False
		833	+ generator.is_inference_pause = False
		834	+ generator.plugin_manager = MagicMock()
		835	+ err = RuntimeError("FORCE STOP abort")
		836	+ generator.plugin_manager.generate_token.side_effect = err
		837	+ generator.generator_backend = MagicMock()
		838	+ im = MagicMock(spec=InputMetadata)
		839	+ im.batch_seq_len = np.array([0])
		840	+ im.is_prefill = False
		841	+ with self.assertRaises(RuntimeError) as cm:
		842	+ generator.generate_token(im, warmup=False)
		843	+ self.assertIs(cm.exception, err)
		844	+ generator.generator_backend.notify_force_stop_exception.assert_not_called()
		845	+
750		846
751	class TestPDInterface(unittest.TestCase):	847	class TestPDInterface(unittest.TestCase):
752		848
	@@ -887,4 +983,4 @@ class TestPDInterface(unittest.TestCase):
887	self.pd_interface._init_sepd_engine()	983	self.pd_interface._init_sepd_engine()
888		984
889	if __name__ == "__main__":	985	if __name__ == "__main__":
890	- unittest.main()	986	+ unittest.main()