Coverage for /opt/cloud/slavespace/usr1/096471637100f3de0fcfc01072822a80/ut/src/asys/analyze/coredump_analyze.py: 88%
253 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-14 17:42 +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 re
20import subprocess
21from platform import machine
22from multiprocessing import Pool, cpu_count, Manager
24from common import log_error, log_warning
25from common.task_common import is_hexadecimal, int_to_hex
26from common.const import REG_OFF, REG_THREAD, REG_STACK, ADDR_LEN_HEX, ADDR_BIT_LEN, GDB_LAYER_MAX
27from params import ParamDict
30def _get_reg_info_cmd():
31 sys_type = machine()
32 if sys_type == "x86_64":
33 reg_cmd = f"info reg rbp rsp rip\n"
34 elif sys_type == "aarch64":
35 reg_cmd = f"info reg x29 sp pc\n"
36 else:
37 reg_cmd = ""
38 return reg_cmd
41def thread_stacks_reg_info(cmd, thread, stacks, queue_reg_info):
42 """
43 get thread all stacks reg value
44 """
45 thread_id = thread.split(" ")[1]
46 gdb_process = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
47 bufsize=1, encoding='utf-8', errors='ignore', text=True)
48 _reg_info = {thread: dict()}
49 gdb_process.stdin.write(f"thread {thread_id}\n")
51 reg_cmd = _get_reg_info_cmd()
52 if not reg_cmd:
53 return False
55 for stack in stacks:
56 stack_id = stack.split(" ")[0][1:]
57 reg_list = [None, None, None]
58 gdb_process.stdin.write(f"frame {stack_id}\n")
59 gdb_process.stdin.write(reg_cmd)
60 while True:
61 line = gdb_process.stdout.readline().rstrip()
62 gdb_process.stdout.flush()
63 data_list = line.strip("\n").strip(" ").split()
64 if line.startswith("(gdb) rbp ") or line.startswith("(gdb) x29 "):
65 reg_list[0] = data_list[2]
66 continue
67 if line.startswith("rsp ") or line.startswith("sp "):
68 reg_list[1] = data_list[1]
69 continue
70 if line.startswith("rip ") or line.startswith("pc "):
71 reg_list[2] = data_list[1]
73 if all(reg_list):
74 line_num = f"#{stack_id}"
75 # #0 -> #00
76 if len(line_num) == 2:
77 line_num = f"#0{stack_id}"
78 _reg_info[thread][line_num] = reg_list
79 break
80 gdb_process.stdin.write("quit\n")
81 gdb_process.stdin.write("y\n")
83 queue_reg_info.put(_reg_info)
84 return True
87class CoreDump:
88 def __init__(self, exe_file, core_file, symbol, output):
89 self.exe_file = exe_file
90 self.core_file = core_file
91 self.symbol = symbol
92 self.output = output
93 self.bt_info = dict()
94 self.map_info = list()
95 self.map_str = "[maps]\n"
96 self.reg_level = ParamDict().get_arg("reg")
98 @staticmethod
99 def check_map_line(data_list):
100 """
101 Check whether the data row is in the mapping format.
102 """
103 if len(data_list) < 5:
104 return False
105 if not (is_hexadecimal(data_list[0]) and is_hexadecimal(data_list[1]) and is_hexadecimal(data_list[2])):
106 return False
107 return True
109 @staticmethod
110 def _get_gdb_cmd(exe_file, core_file):
111 return ["gdb", exe_file, core_file]
113 def collect_info(self, thread_name, data_list, line, before_line):
114 """
115 Collect stack information and map table information.
116 """
117 if re.match(r'#(\d+)', data_list[0]) and thread_name:
118 if self.bt_info.get(thread_name):
119 self.bt_info[thread_name].append(line)
120 else:
121 self.bt_info[thread_name] = [line]
122 elif data_list[0] == "Start" and data_list[-1] == "objfile":
123 self.map_str += (line[6:] + "\n")
124 elif self.check_map_line(data_list):
125 self.map_str += (line[6:] + "\n")
126 if data_list[-1] != before_line:
127 self.map_info.append([int_to_hex(data_list[0]), int_to_hex(data_list[1]), data_list[-1]])
128 else:
129 if int_to_hex(data_list[0]) < self.map_info[-1][0]:
130 self.map_info[-1][0] = int_to_hex(data_list[0])
131 if int_to_hex(data_list[1]) > self.map_info[-1][1]:
132 self.map_info[-1][1] = int_to_hex(data_list[1])
133 before_line = data_list[-1]
134 return before_line
136 def view_map(self, stack_txt, bt_line, reg_info):
137 """
138 Obtain the stack start address and dynamic library from the mapping table.
139 """
140 bt_list = bt_line.strip("\n").strip(" ").split()
141 # #0 -> #00
142 if len(bt_list[0]) == 2:
143 bt_list[0] = f"#0{bt_list[0][1]}"
144 address = int_to_hex(bt_list[1])
145 for values in self.map_info:
146 start_address, end_address, key = values
147 if int(start_address, ADDR_BIT_LEN) < int(address, ADDR_BIT_LEN) < int(end_address, ADDR_BIT_LEN):
148 stack_txt += f"{bt_list[0]} 0x{address[2:].rjust(ADDR_BIT_LEN, '0')} " \
149 f"0x{start_address[2:].rjust(ADDR_BIT_LEN, '0')} {key}\n"
150 stack_txt = self._stack_add_reg(stack_txt, bt_list[0], reg_info)
151 return stack_txt
152 if self.symbol:
153 stack_txt += " ".join(bt_list)
154 stack_txt += "\n"
155 stack_txt = self._stack_add_reg(stack_txt, bt_list[0], reg_info)
157 return stack_txt
159 @staticmethod
160 def _get_reg_str(start_str, reg_values):
161 if not reg_values or len(reg_values) != 3:
162 return ""
164 reg_cmd = _get_reg_info_cmd()
165 line_start = " " * (len(start_str) + 1)
166 if "x29 sp pc" in reg_cmd:
167 reg_str = f"{line_start}fp = {reg_values[0]} sp = {reg_values[1]}\n{line_start}pc = {reg_values[2]}\n"
168 elif "rbp rsp rip" in reg_cmd:
169 reg_str = f"{line_start}rbp = {reg_values[0]} rsp = {reg_values[1]}\n{line_start}rip = {reg_values[2]}\n"
170 else:
171 reg_str = ""
172 return reg_str
174 def _stack_add_reg(self, stack_txt, stack_id, reg_info):
175 # stack_txt add stack reg_data
176 if not reg_info or isinstance(reg_info, list):
177 return stack_txt
178 reg_values = reg_info.get(stack_id)
179 stack_txt += self._get_reg_str(stack_id, reg_values)
181 return stack_txt
183 def parse_stackcore(self, stack_txt, bt_lines, reg_info=None):
184 if self.reg_level == REG_THREAD:
185 stack_txt += self._get_reg_str("", reg_info)
186 for bt_line in bt_lines:
187 bt_list = bt_line.strip("\n").strip(" ").split()
188 # #0 -> #00
189 if len(bt_list[0]) == 2:
190 bt_list[0] = f"#0{bt_list[0][1]}"
191 if self.symbol and "in ??" not in bt_line:
192 stack_txt += " ".join(bt_list)
193 stack_txt += "\n"
194 stack_txt = self._stack_add_reg(stack_txt, bt_list[0], reg_info)
195 continue
196 if not is_hexadecimal(bt_list[1]):
197 stack_txt += f"{bt_list[0]} {' ' * ADDR_LEN_HEX} {' ' * ADDR_LEN_HEX} Ignore\n"
198 continue
199 stack_txt = self.view_map(stack_txt, bt_line, reg_info)
200 return stack_txt
202 def _get_reg_info_level_stack(self):
203 queue_reg_info = Manager().Queue()
204 cmd = self._get_gdb_cmd(self.exe_file, self.core_file)
205 p = Pool(cpu_count() - 1)
206 for thread, stacks in self.bt_info.items():
207 p.apply_async(thread_stacks_reg_info, args=(cmd, thread, stacks, queue_reg_info))
208 p.close()
209 p.join()
211 threads_stacks_reg_info = dict()
212 while queue_reg_info.qsize() > 0:
213 for thread, stacks in queue_reg_info.get().items():
214 threads_stacks_reg_info[thread] = stacks
215 return threads_stacks_reg_info
217 def _get_reg_info_level_thread(self):
218 gdb_process = subprocess.Popen(self._get_gdb_cmd(self.exe_file, self.core_file),
219 stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
220 encoding='utf-8', errors='ignore', text=True, bufsize=1)
221 _reg_info = dict()
222 reg_cmd = _get_reg_info_cmd()
223 if not reg_cmd:
224 return _reg_info
226 for thread in self.bt_info.keys():
227 thread_id = thread.split(" ")[1]
228 gdb_process.stdin.write(f"thread {thread_id}\n")
229 gdb_process.stdin.write(reg_cmd)
230 reg_list = [None, None, None]
231 while True:
232 line = gdb_process.stdout.readline().rstrip()
233 gdb_process.stdout.flush()
234 data_list = line.strip("\n").strip(" ").split()
235 if line.startswith("(gdb) rbp ") or line.startswith("(gdb) x29 "):
236 reg_list[0] = data_list[2]
237 continue
238 if line.startswith("rsp ") or line.startswith("sp "):
239 reg_list[1] = data_list[1]
240 continue
241 if line.startswith("rip ") or line.startswith("pc "):
242 reg_list[2] = data_list[1]
244 if all(reg_list):
245 _reg_info[thread] = reg_list
246 break
247 gdb_process.stdin.write("quit\n")
248 gdb_process.stdin.write("y\n")
250 return _reg_info
252 def get_threads_stacks_reg_info(self):
253 if self.reg_level == REG_OFF:
254 return {}
255 elif self.reg_level == REG_THREAD:
256 return self._get_reg_info_level_thread()
257 elif self.reg_level == REG_STACK:
258 return self._get_reg_info_level_stack()
259 else:
260 return {}
262 def start_gdb(self, stack_txt):
263 try:
264 gdb_process = subprocess.Popen(self._get_gdb_cmd(self.exe_file, self.core_file),
265 stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
266 encoding='utf-8', errors='ignore', text=True)
267 except FileNotFoundError:
268 log_error("Failed to obtain the core dump information.")
269 return stack_txt, 0
270 gdb_process.stdin.write("info inferiors\n")
271 gdb_process.stdin.write("info sharedlibrary\n")
272 gdb_process.stdin.write(f"thread apply all bt {GDB_LAYER_MAX}\n")
273 gdb_process.stdin.write("info proc mappings\n")
274 gdb_process.stdin.write("quit\n")
275 gdb_process.stdin.write("y\n")
276 console_output, _ = gdb_process.communicate()
277 out_lines = console_output.split("\n")
278 thread_name = None
279 pid = 0
280 before_line = ""
281 crash_info = {"crash reason": "", "crash pid": "", "crash tid": ""}
282 for line in out_lines:
283 if f"{self.exe_file}: No such file or directory" in line:
284 log_error(line)
285 return stack_txt, pid
286 if "core file may not match specified executable file" in line:
287 log_warning("Core file may not match specified executable file")
288 continue
289 if line.startswith("Program terminated"):
290 crash_info["crash reason"] = f"{line.split('signal')[-1].strip(' ').split(',')[0]}"
291 continue
292 data_list = line.strip("\n").strip(" ").split()
293 if "Current thread" in line:
294 crash_info["crash tid"] = re.match(r"(\d+)", data_list[-1])[0]
295 continue
296 if data_list and data_list[0] == "No":
297 log_warning(f'Could not load shared library symbols for "{data_list[-1]}", parsing errors may occur.')
298 continue
299 if len(data_list) < 2:
300 continue
301 if line.startswith("*") and "process" == data_list[2]:
302 pid = data_list[3]
303 continue
304 if line.startswith("Thread") and "LWP" in line:
305 tid = re.search(r"LWP (\d+)", line).group(1)
306 thread_name = f"Thread {data_list[1]} ({tid})"
307 continue
308 before_line = self.collect_info(thread_name, data_list, line, before_line)
309 stack_txt, pid = self._process_stack_txt(crash_info, pid, stack_txt)
310 return stack_txt, pid
312 def _process_stack_txt(self, crash_info, pid, stack_txt):
313 if not self.map_info or not self.bt_info:
314 log_error("Failed to obtain the core dump information.")
315 return stack_txt, pid
317 crash_info["crash pid"] = pid
319 for crash_key, crash_value in crash_info.items():
320 stack_txt += f"{crash_key}: {crash_value}\n"
322 stack_txt += "\n"
323 stack_txt += "[stack]\n"
324 threads_stacks_reg_info = self.get_threads_stacks_reg_info()
325 for stack_name, bt_lines in self.bt_info.items():
326 stack_txt += (stack_name + "\n")
327 stacks_reg_info = threads_stacks_reg_info.get(stack_name)
328 stack_txt = self.parse_stackcore(stack_txt, bt_lines, stacks_reg_info)
329 stack_txt += "\n"
330 stack_txt += self.map_str
331 return stack_txt, pid