Coverage for  / home / jenkins / .local / lib / python3.10 / site-packages / hyper_parallel / trainer / callbacks / stress_detect_callback.py: 100%

1# Copyright 2026 Huawei Technologies Co., Ltd 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

7# http://www.apache.org/licenses/LICENSE-2.0 

8# 

9# Unless required by applicable law or agreed to in writing, software 

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14# ============================================================================ 

15"""Periodic online NPU hardware precision stress detection.""" 

16 

17from typing import Any 

18 

19import torch # pylint: disable=forbidden-backend-import 

20 

21from hyper_parallel.trainer.runtime.device import get_device_type 

22from hyper_parallel.trainer.callbacks.base import Callback, TrainerState 

23from hyper_parallel.tools.logging import get_logger 

24 

25 

26logger = get_logger("HP") 

27HARDWARE_FAILURE_CODE = 2 

28 

29 

30class StressDetectCallback(Callback): 

31 """Run hardware precision stress detection between optimizer steps on every rank. 

32 

33 ``stress_detect.op_stress_detect`` configures this callback. 

34 ``detection_interval`` sets the cadence in completed optimizer steps; zero 

35 disables detection. ``num_detections`` sets the number of consecutive checks 

36 in each round. Detection uses the current NPU and requires the memory headroom 

37 specified by torch_npu. ``detect_type`` selects ``aic``, ``hccs``, or ``all``; 

38 ``all`` runs the configured number of checks for each type, AIC first. 

39 ``exit_training_on_sdc`` raises on precision failure when enabled; otherwise 

40 the failure is logged at ERROR and training continues. 

41 """ 

42 

43 def __init__(self, trainer: Any) -> None: 

44 """Read detection settings and resolve the optional NPU API when enabled. 

45 

46 Args: 

47 trainer: Trainer that owns the callback lifecycle. 

48 

49 Raises: 

50 RuntimeError: If enabled on a non-NPU device or without the API. 

51 """ 

52 super().__init__(trainer) 

53 config = trainer.config.stress_detect.op_stress_detect 

54 self.detection_interval = config.detection_interval 

55 self.num_detections = config.num_detections 

56 self.detect_type = config.detect_type 

57 self.exit_training_on_sdc = config.exit_training_on_sdc 

58 if self.detection_interval == 0: 

59 return 

60 if get_device_type() != "npu": 

61 raise RuntimeError("StressDetectCallback requires an NPU device when enabled") 

62 # Device discovery imports the optional torch_npu backend and registers torch.npu. 

63 if not callable(getattr(getattr(torch, "npu", None), "stress_detect", None)): 

64 raise RuntimeError("StressDetectCallback requires torch.npu.stress_detect") 

65 

66 def on_step_end(self, state: TrainerState, **kwargs: Any) -> None: # pylint: disable=arguments-differ 

67 """Synchronize training work and run a detection round at the configured step. 

68 

69 Args: 

70 state: Training state containing the completed optimizer step count. 

71 **kwargs: Unused callback context. 

72 

73 Raises: 

74 RuntimeError: If configured to stop on precision failure, or torch_npu 

75 raises an error such as voltage recovery failure. 

76 """ 

77 del kwargs 

78 if self.detection_interval == 0 or state.global_step <= 0: 

79 return 

80 if state.global_step % self.detection_interval != 0: 

81 return 

82 torch.npu.synchronize() 

83 logger.info("Start hardware precision stress detection at step %s", state.global_step) 

84 detect_types = ("aic", "hccs") if self.detect_type == "all" else (self.detect_type,) 

85 for detect_type in detect_types: 

86 for _ in range(self.num_detections): 

87 result = torch.npu.stress_detect(detect_type=detect_type) 

88 self.log_stress_detect_result( 

89 result, detect_type=detect_type, exit_training_on_sdc=self.exit_training_on_sdc, 

90 ) 

91 

92 @staticmethod 

93 def log_stress_detect_result( 

94 result: int, detect_type: str = "aic", exit_training_on_sdc: bool = False, 

95 ) -> None: 

96 """Report one completed detection without waiting for the remaining checks. 

97 

98 Args: 

99 result: Return code from one completed detection. 

100 detect_type: Detection type included in the result messages. 

101 exit_training_on_sdc: Whether hardware precision failure must raise. 

102 

103 Raises: 

104 RuntimeError: If precision failure is detected and exit_training_on_sdc is true. 

105 """ 

106 if result == 0: 

107 logger.info("Stress detection passed (%s)", detect_type) 

108 elif result == HARDWARE_FAILURE_CODE: 

109 message = ( 

110 f"Stress detection completed but found a hardware precision failure (possible SDC): " 

111 f"detect_type={detect_type}, return_code={result}" 

112 ) 

113 logger.error(message) 

114 if exit_training_on_sdc: 

115 raise RuntimeError(message) 

116 else: 

117 reason = "test case execution failed" if result == 1 else "unrecognized return status" 

118 logger.warning( 

119 "Stress detection did not produce a precision verdict (%s): detect_type=%s, return_code=%s", 

120 reason, detect_type, result, 

121 )