Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/profiling/asys_profiling.py: 98%

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 os 

20import shlex 

21import subprocess 

22from datetime import datetime, timezone 

23 

24from common import log_error, log_info, log_debug 

25from common.const import MAX_PERIOD, UNKNOWN, LP_MODE_NO, LP_MODE_AIC, LP_MODE_LP 

26from params import ParamDict 

27from common import AsysProfilingSupportedChip, ChipHandler 

28 

29support_profiling_list = set(["aicore", "dvpp", "os", "link", "memory", "power"]) 

30 

31 

32class AsysProfiling(): 

33 def __init__(self): 

34 self.device_id = int(ParamDict().get_arg("device_id")) 

35 self.period = ParamDict().get_arg("period", 0) 

36 self.output_path = ParamDict().get_arg("output") 

37 self.run_modes = ParamDict().get_arg("run_mode") 

38 self.aic_metrics = ParamDict().get_arg("aic_metrics", "PipeUtilization") 

39 self.lp_mode = LP_MODE_NO 

40 

41 @staticmethod 

42 def concat_dvpp(cmd): 

43 concat_cmd = "--dvpp-profiling=on" 

44 log_debug("Concat dvpp command.") 

45 out_cmd = cmd + " " + concat_cmd 

46 return out_cmd 

47 

48 @staticmethod 

49 def concat_link(cmd): 

50 concat_cmd = "--sys-interconnection-profiling=on --sys-io-profiling=on" 

51 log_debug("Concat link command.") 

52 out_cmd = cmd + " " + concat_cmd 

53 return out_cmd 

54 

55 @staticmethod 

56 def concat_memory(cmd): 

57 concat_cmd = "--sys-hardware-mem=on --llc-profiling=read" 

58 log_debug("Concat memory command.") 

59 out_cmd = cmd + " " + concat_cmd 

60 return out_cmd 

61 

62 @staticmethod 

63 def concat_os(cmd): 

64 concat_cmd = "--sys-profiling=on" 

65 log_debug("Concat os command.") 

66 out_cmd = cmd + " " + concat_cmd 

67 return out_cmd 

68 

69 def concat_power(self, cmd): 

70 out_cmd = cmd 

71 if self.lp_mode == LP_MODE_AIC: 

72 concat_cmd = "--ai-core=on" 

73 out_cmd += " " + concat_cmd 

74 elif self.lp_mode == LP_MODE_LP: 

75 concat_cmd = "--sys-lp=on" 

76 out_cmd += " " + concat_cmd 

77 return out_cmd 

78 

79 def concat_aicore(self, cmd): 

80 concat_cmd = f"--ai-core=on --aic-mode=sample-based --aic-metrics={self.aic_metrics}" 

81 log_debug("Concat aicore command.") 

82 out_cmd = cmd + " " + concat_cmd 

83 return out_cmd 

84 

85 def run(self): 

86 if not self._check_param(): 

87 return False 

88 log_info(f"Run mode is {self.run_modes}.") 

89 cmd = (f"msprof --output={shlex.quote(self.output_path)} --sys-period={str(self.period)} " 

90 f"--sys-devices={self.device_id} ") 

91 for run_mode in self.run_modes: 

92 func_name = "concat_" + run_mode 

93 func = getattr(self, func_name, None) 

94 if func: 

95 cmd = func(cmd) 

96 ret = self._run_cmd(cmd) 

97 return ret 

98 

99 def _check_param(self): 

100 if self.period < 1 or self.period > MAX_PERIOD: 

101 log_error("Period is invalid, it should be in [1,2592000].") 

102 return False 

103 

104 profiling_supported = AsysProfilingSupportedChip() 

105 ret, _chip_info = profiling_supported.get_supported_chip_info(self.device_id) 

106 if not ret: 

107 if _chip_info == UNKNOWN: 

108 log_error(f"device id {self.device_id} is invalid, please enter a valid value.") 

109 return False 

110 else: 

111 log_error(f"The profiling command does not support {_chip_info}.") 

112 return False 

113 

114 self.run_modes = set(self.run_modes.split(',')) 

115 if not self.run_modes.issubset(support_profiling_list): 

116 log_error("Run mode type is unsupported, run 'asys profiling -h' to get help.") 

117 return False 

118 

119 handler = ChipHandler().get_handler(_chip_info) 

120 if handler is None: 

121 log_error(f"{_chip_info} is not supported.") 

122 return False 

123 if 'dvpp' in self.run_modes and not getattr(handler, "support_dvpp", lambda: True)(): 

124 log_error(f"{_chip_info} does not support dvpp profiling.") 

125 return False 

126 if handler.need_lp_param(): 

127 self.lp_mode = LP_MODE_LP 

128 elif 'aicore' not in self.run_modes: 

129 self.lp_mode = LP_MODE_AIC 

130 

131 utc_dt = datetime.now(timezone.utc) # UTC time 

132 dir_name = utc_dt.astimezone().strftime('%Y%m%d%H%M%S%f')[:-3] 

133 if not self.output_path: 

134 self.output_path = f"asys_profiling_result_{dir_name}" 

135 else: 

136 self.output_path = os.path.join(self.output_path, f"asys_profiling_result_{dir_name}") 

137 return True 

138 

139 def _run_cmd(self, cmd): 

140 log_info(f"Start run: {cmd}, please wait about {self.period} seconds.") 

141 # Run msprof command 

142 ret = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, encoding='utf-8', env=os.environ) 

143 if ret.returncode == 0: 

144 log_info(f"Succeeded in running profiling.\n{ret.stdout}") 

145 else: 

146 log_error(f"Failed to run profiling.\n{ret.stdout}") 

147 return False 

148 return True