# This work is licensed under the terms of the GNU GPL, version 2
# or later. See the COPYING file in the top-level directory.
+import atexit
import gdb
+import os
+import pty
+import re
+import struct
+import textwrap
+
+from collections import OrderedDict
+from copy import deepcopy
VOID_PTR = gdb.lookup_type('void').pointer()
+# Registers in the same order they're present in ELF coredump file.
+# See asm/ptrace.h
+PT_REGS = ['r15', 'r14', 'r13', 'r12', 'rbp', 'rbx', 'r11', 'r10', 'r9',
+ 'r8', 'rax', 'rcx', 'rdx', 'rsi', 'rdi', 'orig_rax', 'rip', 'cs',
+ 'eflags', 'rsp', 'ss']
+
+coredump = None
+
+
+class Coredump:
+ _ptregs_suff = '.ptregs'
+
+ def __init__(self, coredump, executable):
+ atexit.register(self._cleanup)
+
+ self.coredump = coredump
+ self.executable = executable
+ self._ptregs_blob = coredump + self._ptregs_suff
+ self._dirty = False
+
+ with open(coredump, 'rb') as f:
+ while f.read(4) != b'CORE':
+ pass
+ gdb.write(f'core file {coredump}: found "CORE" at 0x{f.tell():x}\n')
+
+ # Looking for struct elf_prstatus and pr_reg field in it (an array
+ # of general purpose registers). See sys/procfs.h.
+
+ # lseek(f.fileno(), 4, SEEK_CUR): go to elf_prstatus
+ f.seek(4, 1)
+
+ # lseek(f.fileno(), 112, SEEK_CUR):
+ # offsetof(struct elf_prstatus, pr_reg)
+ f.seek(112, 1)
+
+ self._ptregs_offset = f.tell()
+
+ # If binary blob with the name /path/to/coredump + '.ptregs'
+ # exists, that means proper cleanup didn't happen during previous
+ # GDB session with the same coredump, and registers in the dump
+ # itself might've remained patched. Thus we restore original
+ # registers values from this blob
+ if os.path.exists(self._ptregs_blob):
+ with open(self._ptregs_blob, 'rb') as b:
+ orig_ptregs_bytes = b.read()
+ self._dirty = True
+ else:
+ orig_ptregs_bytes = f.read(len(PT_REGS) * 8)
+
+ values = struct.unpack(f"={len(PT_REGS)}q", orig_ptregs_bytes)
+ self._orig_ptregs = OrderedDict(zip(PT_REGS, values))
+
+ if not os.path.exists(self._ptregs_blob):
+ gdb.write(f'saving original pt_regs in {self._ptregs_blob}\n')
+ with open(self._ptregs_blob, 'wb') as b:
+ b.write(orig_ptregs_bytes)
+
+ gdb.write('\n')
+
+ def patch_regs(self, regs):
+ # Set dirty flag early on to make sure regs are restored upon cleanup
+ self._dirty = True
+
+ gdb.write(f'patching core file {self.coredump}\n')
+ patched_ptregs = deepcopy(self._orig_ptregs)
+ int_regs = {k: int(v) for k, v in regs.items()}
+ patched_ptregs.update(int_regs)
+
+ with open(self.coredump, 'ab') as f:
+ gdb.write(f'assume pt_regs at 0x{self._ptregs_offset:x}\n')
+ f.seek(self._ptregs_offset, 0)
+ gdb.write('writing regs:\n')
+ for reg in self._orig_ptregs.keys():
+ if reg in int_regs:
+ gdb.write(f" {reg}: {int_regs[reg]:#16x}\n")
+ f.write(struct.pack(f"={len(PT_REGS)}q", *patched_ptregs.values()))
+
+ gdb.write('\n')
+
+ def restore_regs(self):
+ if not self._dirty:
+ return
+
+ gdb.write(f'\nrestoring original regs in core file {self.coredump}\n')
+ with open(self.coredump, 'ab') as f:
+ gdb.write(f'assume pt_regs at 0x{self._ptregs_offset:x}\n')
+ f.seek(self._ptregs_offset, 0)
+ f.write(struct.pack(f"={len(PT_REGS)}q",
+ *self._orig_ptregs.values()))
+
+ self._dirty = False
+ gdb.write('\n')
+
+ def _cleanup(self):
+ if os.path.exists(self._ptregs_blob):
+ self.restore_regs()
+ gdb.write(f'\nremoving saved pt_regs file {self._ptregs_blob}\n')
+ os.unlink(self._ptregs_blob)
+
+
def pthread_self():
'''Fetch the base address of TLS.'''
return gdb.parse_and_eval("$fs_base")
return f"{func_str} at {path}:{line}"
+def run_with_pty(cmd):
+ # Create a PTY pair
+ master_fd, slave_fd = pty.openpty()
+
+ pid = os.fork()
+ if pid == 0: # Child
+ os.close(master_fd)
+ # Attach stdin/stdout/stderr to the PTY slave side
+ os.dup2(slave_fd, 0)
+ os.dup2(slave_fd, 1)
+ os.dup2(slave_fd, 2)
+ os.close(slave_fd)
+ os.execvp("gdb", cmd) # Runs gdb and doesn't return
+
+ # Parent
+ os.close(slave_fd)
+
+ output = bytearray()
+ try:
+ while True:
+ data = os.read(master_fd, 65536)
+ if not data:
+ break
+ output.extend(data)
+ except OSError: # in case subprocess exits and we get EBADF on read()
+ pass
+ finally:
+ try:
+ os.close(master_fd)
+ except OSError: # in case we get EBADF on close()
+ pass
+
+ # Wait for child to finish (reap zombie)
+ os.waitpid(pid, 0)
+
+ return output.decode('utf-8')
+
+def dump_backtrace_patched(regs):
+ cmd = ['gdb', '-batch',
+ '-ex', 'set debuginfod enabled off',
+ '-ex', 'set complaints 0',
+ '-ex', 'set style enabled on',
+ '-ex', 'python print("----split----")',
+ '-ex', 'bt', coredump.executable, coredump.coredump]
+
+ coredump.patch_regs(regs)
+ out = run_with_pty(cmd).split('----split----')[1]
+ gdb.write(out)
+
def dump_backtrace(regs):
'''
Backtrace dump with raw registers, mimic GDB command 'bt'.
selected_frame.select()
-def bt_jmpbuf(jmpbuf):
+def bt_jmpbuf(jmpbuf, detailed=False):
'''Backtrace a jmpbuf'''
regs = get_jmpbuf_regs(jmpbuf)
try:
# but only works with live sessions.
dump_backtrace_live(regs)
except:
- # If above doesn't work, fallback to poor man's unwind
- dump_backtrace(regs)
+ if detailed:
+ # Obtain detailed trace by patching regs in copied coredump
+ dump_backtrace_patched(regs)
+ else:
+ # If above doesn't work, fallback to poor man's unwind
+ dump_backtrace(regs)
def co_cast(co):
return co.cast(gdb.lookup_type('CoroutineUContext').pointer())
coroutine_pointer = co_cast(co)
return coroutine_pointer['env']['__jmpbuf']
+def init_coredump():
+ global coredump
+
+ files = gdb.execute('info files', False, True).split('\n')
+
+ if not 'core dump' in files[1]:
+ return False
+
+ core_path = re.search("`(.*)'", files[2]).group(1)
+ exec_path = re.match('^Symbols from "(.*)".$', files[0]).group(1)
+
+ if coredump is None:
+ coredump = Coredump(core_path, exec_path)
+
+ return True
class CoroutineCommand(gdb.Command):
- '''Display coroutine backtrace'''
+ __doc__ = textwrap.dedent("""\
+ Display coroutine backtrace
+
+ Usage: qemu coroutine COROPTR [--detailed]
+ Show backtrace for a coroutine specified by COROPTR
+
+ --detailed obtain detailed trace by copying coredump, patching
+ regs in it, and runing gdb subprocess to get
+ backtrace from the patched coredump
+ """)
+
def __init__(self):
gdb.Command.__init__(self, 'qemu coroutine', gdb.COMMAND_DATA,
gdb.COMPLETE_NONE)
+ def _usage(self):
+ gdb.write('usage: qemu coroutine <coroutine-pointer> [--detailed]\n')
+ return
+
def invoke(self, arg, from_tty):
argv = gdb.string_to_argv(arg)
- if len(argv) != 1:
- gdb.write('usage: qemu coroutine <coroutine-pointer>\n')
+ argc = len(argv)
+ if argc == 0 or argc > 2 or (argc == 2 and argv[1] != '--detailed'):
+ return self._usage()
+ detailed = True if argc == 2 else False
+
+ is_coredump = init_coredump()
+ if detailed and not is_coredump:
+ gdb.write('--detailed is only valid when debugging core dumps\n')
return
- bt_jmpbuf(coroutine_to_jmpbuf(gdb.parse_and_eval(argv[0])))
+ try:
+ bt_jmpbuf(coroutine_to_jmpbuf(gdb.parse_and_eval(argv[0])),
+ detailed=detailed)
+ finally:
+ coredump.restore_regs()
class CoroutineBt(gdb.Command):
- '''Display backtrace including coroutine switches'''
+ __doc__ = textwrap.dedent("""\
+ Display backtrace including coroutine switches
+
+ Usage: qemu bt [--detailed]
+
+ --detailed obtain detailed trace by copying coredump, patching
+ regs in it, and runing gdb subprocess to get
+ backtrace from the patched coredump
+ """)
+
def __init__(self):
gdb.Command.__init__(self, 'qemu bt', gdb.COMMAND_STACK,
gdb.COMPLETE_NONE)
+ def _usage(self):
+ gdb.write('usage: qemu bt [--detailed]\n')
+ return
+
def invoke(self, arg, from_tty):
+ argv = gdb.string_to_argv(arg)
+ argc = len(argv)
+ if argc > 1 or (argc == 1 and argv[0] != '--detailed'):
+ return self._usage()
+ detailed = True if argc == 1 else False
+
+ is_coredump = init_coredump()
+ if detailed and not is_coredump:
+ gdb.write('--detailed is only valid when debugging core dumps\n')
+ return
gdb.execute("bt")
if co_ptr == False:
return
- while True:
- co = co_cast(co_ptr)
- co_ptr = co["base"]["caller"]
- if co_ptr == 0:
- break
- gdb.write("Coroutine at " + str(co_ptr) + ":\n")
- bt_jmpbuf(coroutine_to_jmpbuf(co_ptr))
+ try:
+ while True:
+ co = co_cast(co_ptr)
+ co_ptr = co["base"]["caller"]
+ if co_ptr == 0:
+ break
+ gdb.write("\nCoroutine at " + str(co_ptr) + ":\n")
+ bt_jmpbuf(coroutine_to_jmpbuf(co_ptr), detailed=detailed)
+ finally:
+ coredump.restore_regs()
class CoroutineSPFunction(gdb.Function):
def __init__(self):
projects / thirdparty / qemu.git / commitdiff
