This is an automated email from the ASF dual-hosted git repository.
xiaoxiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-nuttx.git
The following commit(s) were added to refs/heads/master by this push:
new e597f38 tools:use dump log creat a gdbsever to debug
e597f38 is described below
commit e597f3812d6987506f9c009f35fb99b67a1add2c
Author: anjiahao <[email protected]>
AuthorDate: Mon Feb 28 16:03:43 2022 +0800
tools:use dump log creat a gdbsever to debug
this is a tool,when crash help you to debug.
Signed-off-by: anjiahao <[email protected]>
---
tools/minidumpserver.py | 520 ++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 520 insertions(+)
diff --git a/tools/minidumpserver.py b/tools/minidumpserver.py
new file mode 100755
index 0000000..9800708
--- /dev/null
+++ b/tools/minidumpserver.py
@@ -0,0 +1,520 @@
+#!/usr/bin/env python3
+# tools/minidumpserver.py
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership. The
+# ASF licenses this file to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance with the
+# License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+import argparse
+import binascii
+import logging
+import os
+import re
+import socket
+import struct
+import sys
+
+import elftools
+from elftools.elf.elffile import ELFFile
+
+# ELF section flags
+SHF_WRITE = 0x1
+SHF_ALLOC = 0x2
+SHF_EXEC = 0x4
+SHF_WRITE_ALLOC = SHF_WRITE | SHF_ALLOC
+SHF_ALLOC_EXEC = SHF_ALLOC | SHF_EXEC
+
+
+logger = logging.getLogger()
+
+
+class dump_elf_file:
+ """
+ Class to parse ELF file for memory content in various sections.
+ There are read-only sections (e.g. text and rodata) where
+ the memory content does not need to be dumped via coredump
+ and can be retrived from the ELF file.
+ """
+
+ def __init__(self, elffile):
+ self.elffile = elffile
+ self.fd = None
+ self.elf = None
+ self.memories = list()
+
+ def open(self):
+ self.fd = open(self.elffile, "rb")
+ self.elf = ELFFile(self.fd)
+
+ def close(self):
+ self.fd.close()
+
+ def parse(self):
+ if self.fd is None:
+ self.open()
+
+ for section in self.elf.iter_sections():
+ # REALLY NEED to match exact type as all other sections
+ # (debug, text, etc.) are descendants where
+ # isinstance() would match.
+ if (
+ type(section) is not elftools.elf.sections.Section
+ ): # pylint: disable=unidiomatic-typecheck
+ continue
+
+ size = section["sh_size"]
+ flags = section["sh_flags"]
+ start = section["sh_addr"]
+ end = start + size - 1
+
+ store = False
+ desc = "?"
+
+ if section["sh_type"] == "SHT_PROGBITS":
+ if (flags & SHF_ALLOC_EXEC) == SHF_ALLOC_EXEC:
+ # Text section
+ store = True
+ desc = "text"
+ elif (flags & SHF_WRITE_ALLOC) == SHF_WRITE_ALLOC:
+ # Data section
+ #
+ # Running app changes the content so no need
+ # to store
+ pass
+ elif (flags & SHF_ALLOC) == SHF_ALLOC:
+ # Read only data section
+ store = True
+ desc = "read-only data"
+
+ if store:
+ memory = {"start": start, "end": end, "data": section.data()}
+ logger.info(
+ "ELF Section: 0x%x to 0x%x of size %d (%s)"
+ % (memory["start"], memory["end"], len(memory["data"]),
desc)
+ )
+
+ self.memories.append(memory)
+
+ return True
+
+
+reg_table = {
+ "arm": {
+ "R0": 0,
+ "R1": 1,
+ "R2": 2,
+ "R3": 3,
+ "R4": 4,
+ "R5": 5,
+ "R6": 6,
+ "FP": 7,
+ "R8": 8,
+ "SB": 9,
+ "SL": 10,
+ "R11": 11,
+ "IP": 12,
+ "SP": 13,
+ "LR": 14,
+ "PC": 15,
+ "xPSR": 16,
+ },
+ "riscv": {
+ "ZERO": 0,
+ "RA": 1,
+ "SP": 2,
+ "GP": 3,
+ "TP": 4,
+ "T0": 5,
+ "T1": 6,
+ "T2": 7,
+ "FP": 8,
+ "S1": 9,
+ "A0": 10,
+ "A1": 11,
+ "A2": 12,
+ "A3": 13,
+ "A4": 14,
+ "A5": 15,
+ "A6": 16,
+ "A7": 17,
+ "S2": 18,
+ "S3": 19,
+ "S4": 20,
+ "S5": 21,
+ "S6": 22,
+ "S7": 23,
+ "S8": 24,
+ "S9": 25,
+ "S10": 26,
+ "S11": 27,
+ "T3": 28,
+ "T4": 29,
+ "T5": 30,
+ "T6": 31,
+ "PC": 32,
+ },
+}
+
+
+class dump_log_file:
+ def __init__(self, logfile):
+ self.logfile = logfile
+ self.fd = None
+ self.arch = ""
+ self.registers = []
+ self.memories = list()
+
+ def open(self):
+ self.fd = open(self.logfile, "r")
+
+ def close(self):
+ self.fd.closeself()
+
+ def parse(self):
+ data = bytes()
+ start = 0
+ if self.fd is None:
+ self.open()
+ while 1:
+ line = self.fd.readline()
+ if line == "":
+ break
+
+ tmp = re.search(r"([^ ]*)_registerdump:?", line)
+ if tmp is not None:
+ # find arch
+ self.arch = tmp.group(1)
+ if self.arch not in reg_table:
+ logger.error("%s not supported" % (self.arch))
+ # init register list
+ if len(self.registers) == 0:
+ for x in range(len(reg_table[self.arch])):
+ self.registers.append(b"x")
+
+ # find register value
+ line = line[tmp.span()[1] :]
+ line = line.replace("\n", " ")
+ while 1:
+ tmp = re.search("([^ ]+):", line)
+ if tmp is None:
+ break
+ register = tmp.group(1)
+ line = line[tmp.span()[1] :]
+ tmp = re.search("([0-9a-fA-F]+) ", line)
+ if tmp is None:
+ break
+ if register in reg_table[self.arch].keys():
+ self.registers[reg_table[self.arch][register]] = int(
+ "0x" + tmp.group().replace(" ", ""), 16
+ )
+ line = line[tmp.span()[1] :]
+ continue
+
+ tmp = re.search("_stackdump:", line)
+ if tmp is not None:
+ # find stackdump
+ line = line[tmp.span()[1] :]
+ tmp = re.search("([0-9a-fA-F]+):", line)
+ if tmp is not None:
+ line_start = int("0x" + tmp.group()[:-1], 16)
+
+ if start + len(data) != line_start:
+ # stack is not contiguous
+ if len(data) == 0:
+ start = line_start
+ else:
+ memory = {
+ "start": start,
+ "end": start + len(data),
+ "data": data,
+ }
+ self.memories.append(memory)
+ data = b""
+ start = line_start
+
+ line = line[tmp.span()[1] :]
+ line = line.replace("\n", " ")
+
+ while 1:
+ # record stack value
+ tmp = re.search(" ([0-9a-fA-F]+)", line)
+ if tmp is None:
+ break
+ data = data + struct.pack(
+ "<I", int("0x" + tmp.group().replace(" ", ""), 16)
+ )
+ line = line[tmp.span()[1] :]
+
+ if len(data):
+ memory = {"start": start, "end": start + len(data), "data": data}
+ self.memories.append(memory)
+
+
+GDB_SIGNAL_DEFAULT = 7
+
+
+class gdb_stub:
+ def __init__(self, logfile, elffile):
+ self.logfile = logfile
+ self.elffile = elffile
+ self.socket = None
+ self.gdb_signal = GDB_SIGNAL_DEFAULT
+ self.mem_regions = self.elffile.memories + self.logfile.memories
+
+ def get_gdb_packet(self):
+ socket = self.socket
+ if socket is None:
+ return None
+
+ data = b""
+ checksum = 0
+ # Wait for '$'
+ while True:
+ ch = socket.recv(1)
+ if ch == b"$":
+ break
+
+ # Get a full packet
+ while True:
+ ch = socket.recv(1)
+ if ch == b"#":
+ # End of packet
+ break
+
+ checksum += ord(ch)
+ data += ch
+
+ # Get checksum (2-bytes)
+ ch = socket.recv(2)
+ in_chksum = ord(binascii.unhexlify(ch))
+
+ logger.debug(f"Received GDB packet: {data}")
+
+ if (checksum % 256) == in_chksum:
+ # ACK
+ logger.debug("ACK")
+ socket.send(b"+")
+
+ return data
+ else:
+ # NACK
+ logger.debug(f"NACK (checksum {in_chksum} != {checksum}")
+ socket.send(b"-")
+
+ return None
+
+ def put_gdb_packet(self, data):
+ socket = self.socket
+ if socket is None:
+ return
+
+ checksum = 0
+ for d in data:
+ checksum += d
+
+ pkt = b"$" + data + b"#"
+
+ checksum = checksum % 256
+ pkt += format(checksum, "02X").encode()
+
+ logger.debug(f"Sending GDB packet: {pkt}")
+
+ socket.send(pkt)
+
+ def handle_signal_query_packet(self):
+ # the '?' packet
+ pkt = b"S"
+ pkt += format(self.gdb_signal, "02X").encode()
+
+ self.put_gdb_packet(pkt)
+
+ def handle_register_group_read_packet(self):
+ reg_fmt = "<I"
+ pkt = b""
+
+ for reg in self.logfile.registers:
+ if reg != b"x":
+ bval = struct.pack(reg_fmt, reg)
+ pkt += binascii.hexlify(bval)
+ else:
+ # Register not in coredump -> unknown value
+ # Send in "xxxxxxxx"
+ pkt += b"x" * 8
+
+ self.put_gdb_packet(pkt)
+
+ def handle_register_single_read_packet(self, pkt):
+ # Mark registers as "<unavailable>".
+ # 'p' packets are usually used for registers
+ # other than the general ones (e.g. eax, ebx)
+ # so we can safely reply "xxxxxxxx" here.
+ self.put_gdb_packet(b"x" * 8)
+
+ def handle_register_group_write_packet(self):
+ # the 'G' packet for writing to a group of registers
+ #
+ # We don't support writing so return error
+ self.put_gdb_packet(b"E01")
+
+ def handle_register_single_write_packet(self, pkt):
+ # the 'P' packet for writing to registers
+ #
+ # We don't support writing so return error
+ self.put_gdb_packet(b"E01")
+
+ def handle_memory_read_packet(self, pkt):
+ # the 'm' packet for reading memory: m<addr>,<len>
+
+ def get_mem_region(addr):
+ for r in self.mem_regions:
+ if r["start"] <= addr <= r["end"]:
+ return r
+
+ return None
+
+ # extract address and length from packet
+ # and convert them into usable integer values
+ addr, length = pkt[1:].split(b",")
+ s_addr = int(b"0x" + addr, 16)
+ length = int(b"0x" + length, 16)
+
+ # FIXME: Need more efficient way of extracting memory content
+ remaining = length
+ addr = s_addr
+ barray = b""
+ r = get_mem_region(addr)
+ while remaining > 0:
+ if r is None:
+ barray = None
+ break
+
+ if addr > r["end"]:
+ r = get_mem_region(addr)
+ continue
+
+ offset = addr - r["start"]
+ barray += r["data"][offset:offset + 1]
+
+ addr += 1
+ remaining -= 1
+
+ if barray is not None:
+ pkt = binascii.hexlify(barray)
+ self.put_gdb_packet(pkt)
+ else:
+ self.put_gdb_packet(b"E01")
+
+ def handle_memory_write_packet(self, pkt):
+ # the 'M' packet for writing to memory
+ #
+ # We don't support writing so return error
+ self.put_gdb_packet(b"E02")
+
+ def handle_general_query_packet(self, pkt):
+ self.put_gdb_packet(b"")
+
+ def run(self, socket):
+ self.socket = socket
+
+ while True:
+ pkt = self.get_gdb_packet()
+ if pkt is None:
+ continue
+
+ pkt_type = pkt[0:1]
+ logger.debug(f"Got packet type: {pkt_type}")
+
+ if pkt_type == b"?":
+ self.handle_signal_query_packet()
+ elif pkt_type in (b"C", b"S"):
+ # Continue/stepping execution, which is not supported.
+ # So signal exception again
+ self.handle_signal_query_packet()
+ elif pkt_type == b"g":
+ self.handle_register_group_read_packet()
+ elif pkt_type == b"G":
+ self.handle_register_group_write_packet()
+ elif pkt_type == b"p":
+ self.handle_register_single_read_packet(pkt)
+ elif pkt_type == b"P":
+ self.handle_register_single_write_packet(pkt)
+ elif pkt_type == b"m":
+ self.handle_memory_read_packet(pkt)
+ elif pkt_type == b"M":
+ self.handle_memory_write_packet(pkt)
+ elif pkt_type == b"q":
+ self.handle_general_query_packet(pkt)
+ elif pkt_type == b"k":
+ # GDB quits
+ break
+ else:
+ self.put_gdb_packet(b"")
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser()
+
+ parser.add_argument("-e", "--elffile", required=True, help="elffile")
+
+ parser.add_argument("-l", "--logfile", required=True, help="logfile")
+
+ parser.add_argument("-p", "--port", help="gdbport", type=int, default=1234)
+
+ parser.add_argument("--debug", action="store_true", default=False)
+
+ args = parser.parse_args()
+
+ if not os.path.isfile(args.elffile):
+ logger.error(f"Cannot find file {args.elffile}, exiting...")
+ sys.exit(1)
+
+ if not os.path.isfile(args.logfile):
+ logger.error(f"Cannot find file {args.logfile}, exiting...")
+ sys.exit(1)
+
+ if args.debug:
+ logger.setLevel(logging.DEBUG)
+ else:
+ logger.setLevel(logging.INFO)
+
+ log = dump_log_file(args.logfile)
+ log.parse()
+ elf = dump_elf_file(args.elffile)
+ elf.parse()
+
+ gdbstub = gdb_stub(log, elf)
+ logging.basicConfig(format="[%(levelname)s][%(name)s] %(message)s")
+
+ gdbserver = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
+ # Reuse address so we don't have to wait for socket to be
+ # close before we can bind to the port again
+ gdbserver.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+
+ gdbserver.bind(("", args.port))
+ gdbserver.listen(1)
+
+ logger.info(f"Waiting GDB connection on port {args.port} ...")
+
+ conn, remote = gdbserver.accept()
+
+ if conn:
+ logger.info(f"Accepted GDB connection from {remote}")
+
+ gdbstub.run(conn)
+
+ conn.close()
+
+ gdbserver.close()
