Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/common/interface.py: 44%

1#!/usr/bin/env python3 

2# -*- coding: utf-8 -*- 

3# ---------------------------------------------------------------------------- 

4# Copyright (c) 2025 Huawei Technologies Co., Ltd. 

5# 

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

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

8# You may obtain a copy of the License at 

9# 

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

11# 

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

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

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

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

16# limitations under the License. 

17# ---------------------------------------------------------------------------- 

18 

19import ctypes 

20from threading import Thread 

21 

22from common import log_error, log_warning 

23from common import RetCode 

24from common.const import DETECT_DEFAULT_TIMEOUT, CPU_DETECT_ERROR_CODE_MIN, CPU_DETECT_ERROR_CODE_MAX, ScreenResult 

25from params import ParamDict 

26 

27ERROR_DEVICE_ID = -1 

28 

29 

30class AmlStressDtestctType(ctypes.Structure): 

31 _fields_ = [ 

32 ('STRESS_DETECT_ALL', ctypes.c_int), 

33 ('STRESS_DETECT_TYPE_MAX', ctypes.c_int) 

34 ] 

35 

36 

37class AmlStressDetectInfo(ctypes.Structure): 

38 _fields_ = [ 

39 ('type', AmlStressDtestctType) 

40 ] 

41 

42 

43def get_devices_master_id(device_obj, all_devices): 

44 if len(all_devices) == 1: 

45 return {all_devices[0]: all_devices[0]} # logic_id: master_id 

46 

47 ret = dict() 

48 for idx in all_devices: 

49 phy_id = device_obj.get_phyid_from_logicid(idx) 

50 if phy_id == RetCode.FAILED: 

51 ret[idx] = ERROR_DEVICE_ID 

52 continue 

53 master_id = device_obj.get_masterid_from_phyid(phy_id) 

54 if master_id == RetCode.FAILED: 

55 ret[idx] = ERROR_DEVICE_ID 

56 continue 

57 ret[idx] = master_id 

58 return ret 

59 

60 

61def run_stress_detect(device_id, device_obj, ret): 

62 info = AmlStressDetectInfo() 

63 info.type.STRESS_DETECT_ALL = 0 

64 

65 try: 

66 ret_code = device_obj.ascend_ml.AmlStressDetect(ctypes.c_int32(device_id), ctypes.byref(info)) 

67 except Exception as e: 

68 log_error(f"Run stress_detect failed, error_msg: {e}") 

69 ret_code = None 

70 

71 if ret_code == 0: 

72 ret[device_id] = ScreenResult.PASS.value 

73 else: 

74 ret[device_id] = ScreenResult.WARN.value 

75 return ret_code 

76 

77 

78def run_hbm(device_id, device_obj, ret): 

79 if device_id == ERROR_DEVICE_ID: 

80 ret[device_id] = [ScreenResult.WARN.value, "0"] 

81 return ret 

82 timeout = ParamDict().get_arg("timeout") 

83 if timeout is False: 

84 timeout = DETECT_DEFAULT_TIMEOUT 

85 

86 # 1: detect once, 0: detect by timeout 

87 hbm_type = 1 if timeout == 0 else 0 

88 

89 hbm_ecc_before = device_obj.get_ecc_isolated_page(device_id) 

90 if isinstance(hbm_ecc_before, int) and hbm_ecc_before < 0: 

91 device_obj.clear_ecc_isolated(device_id) 

92 hbm_ecc_before = 0 

93 try: 

94 ret_code = device_obj.ascend_ml.AmlHbmDetectWithType( 

95 ctypes.c_int32(device_id), ctypes.c_uint32(timeout), ctypes.c_uint32(hbm_type) 

96 ) 

97 except Exception as e: 

98 log_error(f"Run hbm detect failed, error_msg: {e}") 

99 ret_code = None 

100 hbm_ecc_after = device_obj.get_ecc_isolated_page(device_id) 

101 if hbm_ecc_before == "-" or hbm_ecc_after == "-": 

102 log_warning("The statistics of all uncorrectable ECC errors in the lifecycle cannot be queried.") 

103 hbm_ecc_count = "-" 

104 else: 

105 hbm_ecc_count = hbm_ecc_after - hbm_ecc_before 

106 if ret_code == 0: 

107 ret[device_id] = [ScreenResult.PASS.value, str(hbm_ecc_count)] 

108 else: 

109 ret[device_id] = [ScreenResult.WARN.value, "0"] 

110 return ret_code 

111 

112 

113def run_cpu(device_id, device_obj, ret): 

114 if device_id == ERROR_DEVICE_ID: 

115 ret[device_id] = ScreenResult.WARN.value 

116 return ret 

117 timeout = ParamDict().get_arg("timeout") 

118 if timeout is False: 

119 timeout = DETECT_DEFAULT_TIMEOUT 

120 

121 try: 

122 ret_code = device_obj.ascend_ml.AmlCpuDetect(ctypes.c_int32(device_id), ctypes.c_uint32(timeout)) 

123 except Exception as e: 

124 log_error(f"Run cpu detect failed, error_msg: {e}") 

125 ret_code = None 

126 

127 if ret_code == 0: 

128 ret[device_id] = ScreenResult.PASS.value 

129 elif ret_code and CPU_DETECT_ERROR_CODE_MIN <= ret_code <= CPU_DETECT_ERROR_CODE_MAX: 

130 ret[device_id] = ScreenResult.FAIL.value 

131 else: 

132 ret[device_id] = ScreenResult.WARN.value 

133 return ret_code 

134 

135 

136def run_diagnose(device_obj, diagnose_devices, run_mode): 

137 """Multi-thread parallel execution""" 

138 

139 threads = [] 

140 ret = {} 

141 

142 if run_mode == "hbm_detect": 

143 _target_func = run_hbm 

144 elif run_mode == "cpu_detect": 

145 _target_func = run_cpu 

146 else: 

147 _target_func = run_stress_detect 

148 

149 for device_id in diagnose_devices: 

150 # new thread 

151 t = Thread(target=_target_func, args=(device_id, device_obj, ret), daemon=True) 

152 t.start() 

153 threads.append(t) 

154 

155 # wait for all threads to end. 

156 for t in threads: 

157 t.join() 

158 

159 return ret