kopia lustrzana https://github.com/espressif/esp-idf
341 wiersze
12 KiB
Python
341 wiersze
12 KiB
Python
#!/usr/bin/env python
|
|
# coding=utf-8
|
|
#
|
|
# This script decodes Xtensa CPU trace dumps. It allows tracing the program
|
|
# execution at instruction level.
|
|
#
|
|
# Some trivia about the Xtensa CPU trace (TRAX):
|
|
# TRAX format mostly follows the IEEE-ISTO 5001-2003 (Nexus) standard.
|
|
# The following Nexus Program Trace messages are implemented by TRAX:
|
|
# - Indirect Branch Message
|
|
# - Syncronization Message
|
|
# - Indirect Branch with Synchronization Message
|
|
# - Correlation Message
|
|
# TRAX outputs compressed traces with 2 MSEO bits (LSB) and 6 MDO bits (MSB),
|
|
# packed into a byte. MSEO bits are used to split the stream into packets and messages,
|
|
# and MDO bits carry the actual data of the messages. Each message may contain multiple packets.
|
|
#
|
|
# This script can be used standalone, or loaded into GDB.
|
|
# When used standalone, it dumps the list of trace messages to stdout.
|
|
# When used from GDB, it also invokes GDB command to dump the list of assembly
|
|
# instructions corresponding to each of the messages.
|
|
#
|
|
# Standalone usage:
|
|
# traceparse.py <dump_file>
|
|
#
|
|
# Usage from GDB:
|
|
# xtensa-esp32-elf-gdb -n --batch program.elf -x gdbinit
|
|
# with the following gdbinit script:
|
|
# set pagination off
|
|
# set confirm off
|
|
# add-symbol-file rom.elf <address of ROM .text section>
|
|
# source traceparse.py
|
|
# python parse_and_dump("/path/to/dump_file")
|
|
#
|
|
# Loading the ROM code is optional; if not loaded, disassembly for ROM sections of code
|
|
# will be missing.
|
|
#
|
|
###
|
|
# Copyright 2020 Espressif Systems (Shanghai) PTE LTD
|
|
#
|
|
# Licensed 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.
|
|
from __future__ import print_function
|
|
|
|
import sys
|
|
|
|
# Check if loaded into GDB
|
|
try:
|
|
assert gdb.__name__ == 'gdb' # type: ignore
|
|
WITH_GDB = True
|
|
except NameError:
|
|
WITH_GDB = False
|
|
|
|
# MSEO bit masks:
|
|
MSEO_PKTEND = 1 << 0 # bit 0: indicates the last byte of a packet
|
|
MSEO_MSGEND = 1 << 1 # bit 1: indicates the last byte of the message
|
|
|
|
# Message types. The type is stored in the first 6 MDO bits or the first packet.
|
|
TVAL_INDBR = 4 # Indirect branch
|
|
TVAL_INDBRSYNC = 12 # Indirect branch w/ synchronisation
|
|
TVAL_SYNC = 9 # Synchronisation msg
|
|
TVAL_CORR = 33 # Correlation message
|
|
|
|
|
|
class TraxPacket(object):
|
|
def __init__(self, data, truncated=False):
|
|
self.data = data
|
|
self.size_bytes = len(data)
|
|
self.truncated = truncated
|
|
|
|
def get_bits(self, start, count=0):
|
|
"""
|
|
Extract data bits from the packet
|
|
:param start: offset, in bits, of the part to be extracted
|
|
:param count: number of bits to extract; if omitted or zero,
|
|
extracts until the end of the packet
|
|
:return: integer containing the extracted bits
|
|
"""
|
|
start_byte = start // 6
|
|
if count <= 0:
|
|
# all remaining bits
|
|
count = len(self.data) * 6 - start
|
|
bits_remaining = count
|
|
result = 0
|
|
shift = 0
|
|
for i, b in enumerate(self.data[start_byte:]):
|
|
# which bit in the byte is the starting bit
|
|
if i == 0:
|
|
# at start_byte: take the offset into account
|
|
start_bit = 2 + (start % 6)
|
|
else:
|
|
# every other byte: start after MSEO bits
|
|
start_bit = 2
|
|
# how many bits do we need to copy from this byte
|
|
cnt_bits = min(bits_remaining, 8 - start_bit)
|
|
mask = (2 ** cnt_bits) - 1
|
|
# take this many bits after the start_bit
|
|
bits = (b >> start_bit) & mask
|
|
# add these bits to the result
|
|
result |= bits << shift
|
|
# update the remaining bit count
|
|
shift += cnt_bits
|
|
bits_remaining -= cnt_bits
|
|
if bits_remaining == 0:
|
|
break
|
|
return result
|
|
|
|
def __str__(self):
|
|
return '%d byte packet%s' % (self.size_bytes, ' (truncated)' if self.truncated else '')
|
|
|
|
|
|
class TraxMessage(object):
|
|
def __init__(self, packets, truncated=False):
|
|
"""
|
|
Create and parse a TRAX message from packets
|
|
:param packets: list of TraxPacket objects, must not be empty
|
|
:param truncated: whether the message was truncated in the stream
|
|
"""
|
|
assert len(packets) > 0
|
|
self.packets = packets
|
|
self.truncated = truncated
|
|
if truncated:
|
|
self.msg_type = None
|
|
else:
|
|
self.msg_type = self._get_type()
|
|
|
|
# Start and end of the instruction range corresponding to this message
|
|
self.pc_start = 0 # inclusive
|
|
self.pc_end = 0 # not inclusive
|
|
self.pc_target = 0 # PC of the next range
|
|
self.is_exception = False # whether the message indicates an exception
|
|
self.is_correlation = False # whether this is a correlation message
|
|
|
|
# message-specific fields
|
|
self.icnt = 0
|
|
self.uaddr = 0
|
|
self.dcont = 0
|
|
|
|
# decode the fields
|
|
if not truncated:
|
|
self._decode()
|
|
|
|
def _get_type(self):
|
|
"""
|
|
:return: Message type, one of TVAL_XXX values
|
|
"""
|
|
return self.packets[0].get_bits(0, 6)
|
|
|
|
def _decode(self):
|
|
""" Parse the packets and fill in the message-specific fields """
|
|
if self.msg_type == TVAL_INDBR:
|
|
self.icnt = self.packets[0].get_bits(7, -1)
|
|
self.btype = self.packets[0].get_bits(6, 1)
|
|
self.uaddr = self.packets[1].get_bits(0)
|
|
self.is_exception = self.btype > 0
|
|
elif self.msg_type == TVAL_INDBRSYNC:
|
|
self.icnt = self.packets[0].get_bits(8, -1)
|
|
self.btype = self.packets[0].get_bits(7, 1)
|
|
self.pc_target = self.packets[1].get_bits(0)
|
|
self.dcont = self.packets[0].get_bits(6, 1)
|
|
self.is_exception = self.btype > 0
|
|
elif self.msg_type == TVAL_SYNC:
|
|
self.icnt = self.packets[0].get_bits(7, -1)
|
|
self.dcont = self.packets[0].get_bits(6, 1)
|
|
self.pc_target = self.packets[1].get_bits(0)
|
|
elif self.msg_type == TVAL_CORR:
|
|
self.icnt = self.packets[0].get_bits(12, -1)
|
|
self.is_correlation = True
|
|
else:
|
|
raise NotImplementedError('Unknown message type (%d)' % self.msg_type)
|
|
|
|
def process_forward(self, cur_pc):
|
|
"""
|
|
Given the target PC known from the previous message, determine
|
|
the PC range corresponding to the current message.
|
|
:param cur_pc: previous known PC
|
|
:return: target PC after the current message
|
|
"""
|
|
assert not self.truncated
|
|
|
|
next_pc = cur_pc
|
|
if self.msg_type == TVAL_INDBR:
|
|
next_pc = cur_pc ^ self.uaddr
|
|
self.pc_target = next_pc
|
|
self.pc_start = cur_pc
|
|
self.pc_end = self.pc_start + self.icnt + 1
|
|
if self.msg_type == TVAL_INDBRSYNC:
|
|
next_pc = self.pc_target
|
|
self.pc_start = cur_pc
|
|
self.pc_end = cur_pc + self.icnt + 1
|
|
if self.msg_type == TVAL_SYNC:
|
|
next_pc = self.pc_target
|
|
self.pc_start = next_pc - self.icnt
|
|
self.pc_end = next_pc + 1
|
|
if self.msg_type == TVAL_CORR:
|
|
pass
|
|
return next_pc
|
|
|
|
def process_backward(self, cur_pc):
|
|
"""
|
|
Given the address of the PC known from the _next_ message, determine
|
|
the PC range corresponding to the current message.
|
|
:param cur_pc: next known PC
|
|
:return: target PC of the _previous_ message
|
|
"""
|
|
assert not self.truncated
|
|
# Backward pass is only used to resolve addresses of messages
|
|
# up to the first SYNC/INDBRSYNC message.
|
|
# SYNC/INDBRSYNC messages are only handled in the forward pass.
|
|
assert self.msg_type != TVAL_INDBRSYNC
|
|
assert self.msg_type != TVAL_SYNC
|
|
|
|
prev_pc = cur_pc
|
|
self.pc_target = cur_pc
|
|
if self.msg_type == TVAL_INDBR:
|
|
prev_pc ^= self.uaddr
|
|
self.pc_start = prev_pc
|
|
self.pc_end = prev_pc + self.icnt + 1
|
|
if self.msg_type == TVAL_CORR:
|
|
pass
|
|
return prev_pc
|
|
|
|
def __str__(self):
|
|
desc = 'Unknown (%d)' % self.msg_type
|
|
extra = ''
|
|
if self.truncated:
|
|
desc = 'Truncated'
|
|
if self.msg_type == TVAL_INDBR:
|
|
desc = 'Indirect branch'
|
|
extra = ', icnt=%d, uaddr=0x%x, exc=%d' % (self.icnt, self.uaddr, self.is_exception)
|
|
if self.msg_type == TVAL_INDBRSYNC:
|
|
desc = 'Indirect branch w/sync'
|
|
extra = ', icnt=%d, dcont=%d, exc=%d' % (self.icnt, self.dcont, self.is_exception)
|
|
if self.msg_type == TVAL_SYNC:
|
|
desc = 'Synchronization'
|
|
extra = ', icnt=%d, dcont=%d' % (self.icnt, self.dcont)
|
|
if self.msg_type == TVAL_CORR:
|
|
desc = 'Correlation'
|
|
extra = ', icnt=%d' % self.icnt
|
|
return '%s message, %d packets, PC range 0x%08x - 0x%08x, target PC 0x%08x' % (
|
|
desc, len(self.packets), self.pc_start, self.pc_end, self.pc_target) + extra
|
|
|
|
|
|
def load_messages(data):
|
|
"""
|
|
Decodes TRAX data and resolves PC ranges.
|
|
:param data: input data, bytes
|
|
:return: list of TraxMessage objects
|
|
"""
|
|
messages = []
|
|
packets = []
|
|
packet_start = 0
|
|
msg_cnt = 0
|
|
pkt_cnt = 0
|
|
|
|
# Iterate over the input data, splitting bytes into packets and messages
|
|
for i, b in enumerate(data):
|
|
if (b & MSEO_MSGEND) and not (b & MSEO_PKTEND):
|
|
raise AssertionError('Invalid MSEO bits in b=0x%x. Not a TRAX dump?' % b)
|
|
|
|
if b & MSEO_PKTEND:
|
|
pkt_cnt += 1
|
|
packets.append(TraxPacket(data[packet_start:i + 1], packet_start == 0))
|
|
packet_start = i + 1
|
|
|
|
if b & MSEO_MSGEND:
|
|
msg_cnt += 1
|
|
try:
|
|
messages.append(TraxMessage(packets, len(messages) == 0))
|
|
except NotImplementedError as e:
|
|
sys.stderr.write('Failed to parse message #%03d (at %d bytes): %s\n' % (msg_cnt, i, str(e)))
|
|
packets = []
|
|
|
|
# Resolve PC ranges of messages.
|
|
# Forward pass: skip messages until a message with known PC,
|
|
# i.e. a SYNC/INDBRSYNC message. Process all messages following it.
|
|
pc = 0
|
|
first_sync_index = -1
|
|
for i, m in enumerate(messages):
|
|
if pc == 0 and m.pc_target == 0:
|
|
continue
|
|
if first_sync_index < 0:
|
|
first_sync_index = i
|
|
pc = m.process_forward(pc)
|
|
|
|
# Now process the skipped messages in the reverse direction,
|
|
# starting from the first message with known PC.
|
|
pc = messages[first_sync_index].pc_start
|
|
for m in reversed(messages[0:first_sync_index]):
|
|
if m.truncated:
|
|
break
|
|
pc = m.process_backward(pc)
|
|
|
|
return messages
|
|
|
|
|
|
def parse_and_dump(filename, disassemble=WITH_GDB):
|
|
"""
|
|
Decode TRAX data from a file, print out the messages.
|
|
:param filename: file to load the dump from
|
|
:param disassemble: if True, print disassembly of PC ranges
|
|
"""
|
|
with open(filename, 'rb') as f:
|
|
data = f.read()
|
|
|
|
messages = load_messages(data)
|
|
sys.stderr.write('Loaded %d messages in %d bytes\n' % (len(messages), len(data)))
|
|
|
|
for i, m in enumerate(messages):
|
|
if m.truncated:
|
|
continue
|
|
print('%04d: %s' % (i, str(m)))
|
|
if m.is_exception:
|
|
print('*** Exception occurred ***')
|
|
if disassemble and WITH_GDB:
|
|
try:
|
|
gdb.execute('disassemble 0x%08x, 0x%08x' % (m.pc_start, m.pc_end)) # noqa: F821
|
|
except gdb.MemoryError: # noqa: F821
|
|
print('Failed to disassemble from 0x%08x to 0x%08x' % (m.pc_start, m.pc_end))
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 2:
|
|
sys.stderr.write('Usage: %s <dump_file>\n')
|
|
raise SystemExit(1)
|
|
|
|
parse_and_dump(sys.argv[1])
|
|
|
|
|
|
if __name__ == '__main__' and not WITH_GDB:
|
|
main()
|