Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/collect/stackcore/stackcore_collect.py: 91%
213 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-19 17:46 +0800
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# ----------------------------------------------------------------------------
19import os
20import re
21import subprocess
22import stat
23from threading import Thread, Lock
25from common import FileOperate as f
26from common import log_error, log_warning
27from common.cmd_run import check_command, run_linux_cmd
28from common.task_common import out_progress_bar, str_to_hex, is_hexadecimal
29from common.const import ADDR_LEN_HEX
32class ParseStackCore:
33 def __init__(self, symbol, file=None):
34 self.file = file
35 self.symbol_path = symbol
36 self.__readelf = "readelf"
37 self.__addr2line = "addr2line"
38 self.output_logs = {}
39 self.maps_addr_binary_path = {}
40 self.lock = Lock()
42 def check_tool_exists(self):
43 if not check_command(self.__readelf):
44 log_error("The readelf tool does not exist. Install it before using it.")
45 return False
46 if not check_command(self.__addr2line):
47 log_error("The addr2line tool does not exist. Install it before using it.")
48 return False
49 return True
51 @staticmethod
52 def write_res_file(file_name, file_lines):
53 try:
54 with open(file_name, "w") as fw:
55 fw.writelines(file_lines)
56 except Exception as e:
57 log_error(e)
58 return False
59 return True
61 def error(self, msg):
62 if self.file:
63 log_error(msg)
65 def warning(self, so_name, msg):
66 if self.file:
67 self.lock.acquire() # Exclusive Locking
68 error_info = self.output_logs.get(so_name)
69 if not (error_info and msg == error_info):
70 log_warning(msg)
71 self.output_logs[so_name] = msg
72 self.lock.release() # Unlock
74 def get_source_location(self, so_name, address):
75 """Run the addr2line command to obtain the function name and the line where the function is located."""
76 try:
77 output = subprocess.check_output([self.__addr2line, hex(address), '-e', so_name, '-f', '-C', '-s', '-i'],
78 stderr=subprocess.STDOUT)
79 output_lines = output.decode().strip().split("\n")
80 result_lines = []
81 for line in output_lines:
82 if line.startswith(self.__addr2line):
83 warning_info = f"{so_name} {line.split(':')[-1]}"
84 self.warning(so_name, warning_info)
85 continue
86 result_lines.append(line)
87 return result_lines
88 except Exception as e:
89 self.warning(so_name, f"{so_name} is not permitted to read.")
90 return []
92 def file_lines_add_stack_num(self, file_lines):
93 # stack add line num
94 file_lines_with_stack_num = []
95 stack_num = 0
96 for line in file_lines:
98 if line.endswith("Ignore\n"):
99 stack_str = f"#0{stack_num}" if stack_num < 10 else f"#{stack_num}"
100 file_lines_with_stack_num.append(f"{stack_str} {' ' * ADDR_LEN_HEX} Ignore\n")
101 stack_num += 1
102 continue
103 if line.startswith("Thread "):
104 stack_num = 0
105 if not line.startswith("### "):
106 file_lines_with_stack_num.append(line)
107 continue
109 for _line in line.split("\n"):
110 if not _line:
111 continue
112 stack_str = f"#0{stack_num}" if stack_num < 10 else f"#{stack_num}"
113 file_lines_with_stack_num.append(f"{stack_str}{_line[3:]}\n")
114 stack_num += 1
115 return file_lines_with_stack_num
117 def _get_line_with_addr2line(self, binary_path, address, stack_addr, so_name):
118 all_func = self.get_source_location(binary_path, address)
119 file_line = ""
120 if not all_func:
121 return file_line
122 for i in range(0, len(all_func), 2):
123 func_name, func_file = all_func[i], all_func[i + 1]
124 if i == 0:
125 file_line += f"### {stack_addr} {func_name} in {func_file} from {so_name}\n"
126 else:
127 file_line += f"### {' ' * len(stack_addr)} {func_name} in {func_file} from {so_name}\n"
128 return file_line
130 def parse_line(self, index, line, file_lines):
131 line_num, stack_addr, delta_addr, binary_path = line.strip("\n").split()[:4]
132 so_name = os.path.basename(binary_path)
133 # Obtain the actual binary file.
134 if self.symbol_path:
135 for path in self.symbol_path:
136 so_path = os.path.join(path, so_name)
137 binary_path = ""
138 if os.path.exists(so_path):
139 binary_path = so_path
140 break
141 else:
142 maps_binary_path = self.maps_addr_binary_path.get(str_to_hex(delta_addr))
143 if maps_binary_path:
144 binary_path = maps_binary_path
146 # if it does not exist or is not a file
147 if binary_path == "" or not os.path.exists(binary_path):
148 is_file = False
149 else:
150 _mode = os.stat(binary_path).st_mode
151 is_file = any([os.path.isfile(binary_path), stat.S_ISBLK(_mode), stat.S_ISCHR(_mode), stat.S_ISSOCK(_mode)])
152 if not is_file:
153 warning_info = f"{so_name} not found in symbol_path directory." if self.symbol_path \
154 else f"{binary_path} is not exists."
155 self.warning(so_name, warning_info)
156 file_lines[index] = line.replace(line_num, "###")
157 return False
158 if run_linux_cmd(f"{self.__readelf} -h {binary_path} | grep EXEC"):
159 address = str_to_hex(stack_addr)
160 else:
161 address = str_to_hex(stack_addr) - str_to_hex(delta_addr)
163 line_with_addr = self._get_line_with_addr2line(binary_path, address, stack_addr, so_name)
164 file_lines[index] = line_with_addr if line_with_addr else line.replace(line_num, "###")
165 return True
167 def set_maps_addr_binary_path(self, file_lines):
168 if self.symbol_path:
169 return
170 start_up = False
171 for line in file_lines:
172 if line.startswith("["):
173 # get [maps] info
174 if not start_up and line.startswith("[map"):
175 start_up = True
176 else:
177 start_up = False
178 continue
179 if not start_up:
180 continue
181 line_list = [i.strip() for i in line.strip("\n").split(" ") if i.strip()]
182 if len(line_list) != 6 or not line_list[-1].startswith("/"):
183 continue
184 addr, _, _, _, _, binary_path = line_list
185 start_addr = addr.split("-")[0]
186 if not is_hexadecimal(start_addr):
187 continue
188 start_addr_int = str_to_hex(start_addr)
189 self.maps_addr_binary_path[start_addr_int] = binary_path
191 def start_parse_file(self, stackcore_file, count=0):
192 """Parsing a single file"""
193 stackcore_file_name = stackcore_file.split(os.sep)[-1]
194 if not stackcore_file_name.startswith("stackcore"):
195 log_error(f"The {stackcore_file} file is not in stackcore format.")
196 return False
197 # Check whether the readelf and addr2line tools exist.
198 if not self.check_tool_exists():
199 return False
200 try:
201 with open(stackcore_file, "r") as fp:
202 file_lines = fp.readlines()
203 except Exception as e:
204 self.error(e)
205 return False
206 if not file_lines:
207 self.error(f"The {stackcore_file_name} file is empty.")
208 return False
210 # not symbol_path, get binary_path from maps
211 self.set_maps_addr_binary_path(file_lines)
213 start_up = False
214 threads = []
215 if self.file:
216 count = len(file_lines)
217 for index, line in enumerate(file_lines):
218 if self.file:
219 out_progress_bar(count, index)
220 if line.startswith("["):
221 # If [stack] is found, the parsing of the next line starts. Otherwise, the parsing ends.
222 if not start_up and line.startswith("[stack]"):
223 start_up = True
224 else:
225 start_up = False
226 continue
227 line = re.sub(r" +", " ", line.strip().strip("\n"))
228 # If it is not started and is not in stackcore format, it is not processed.
229 if not start_up or not re.match("#[0-9]+?", line) or len(line.split()) < 4:
230 continue
231 line_num, stack_addr, delta_addr, binary_path = line.split()[:4]
232 if not (is_hexadecimal(stack_addr) and is_hexadecimal(delta_addr)):
233 continue
234 t = Thread(target=self.parse_line, args=(index, line, file_lines), daemon=True)
235 t.start()
236 threads.append(t)
237 # wait for all threads to end.
238 for t in threads:
239 t.join()
240 file_lines = self.file_lines_add_stack_num(file_lines)
241 return self.write_res_file(stackcore_file, file_lines)
243 def save_file_result(self, stackcore_file, count, num, results):
244 ret = self.start_parse_file(stackcore_file, count)
245 out_progress_bar(count, num)
246 if not ret:
247 log_error(f"Failed to analyze the '{stackcore_file}' file.")
248 results.append(ret)
250 def run(self, stack_core_path, count=0):
251 stackcore_dirs = f.walk_dir(stack_core_path)
252 if not stackcore_dirs:
253 return False
254 if not self.check_tool_exists():
255 return False
256 num = 0
257 threads = []
258 results = []
259 for dirs, _, files in stackcore_dirs:
260 for file in files:
261 stackcore_file = os.path.join(dirs, file)
262 num += 1
263 t = Thread(target=self.save_file_result, args=(stackcore_file, count, num, results), daemon=True)
264 t.start()
265 threads.append(t)
266 # wait for all threads to end.
267 for t in threads:
268 t.join()
269 out_progress_bar(count, count)
270 if not self.symbol_path:
271 return any(results)
272 return True