##
## This file is part of the libsigrokdecode project.
##
## Copyright (C) 2015 Petteri Aimonen <jpa@sigrok.mail.kapsi.fi>
##
## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program; if not, see <http://www.gnu.org/licenses/>.
##
import sigrokdecode as srd
import subprocess
import re
# See ETMv3 Signal Protocol table 7-11: 'Encoding of Exception[8:0]'.
exc_names = [
'No exception', 'IRQ1', 'IRQ2', 'IRQ3', 'IRQ4', 'IRQ5', 'IRQ6', 'IRQ7',
'IRQ0', 'UsageFault', 'NMI', 'SVC', 'DebugMon', 'MemManage', 'PendSV',
'SysTick', 'Reserved', 'Reset', 'BusFault', 'Reserved', 'Reserved'
]
for i in range(8, 496):
exc_names.append('IRQ%d' % i)
def parse_varint(bytes_):
'''Parse an integer where the top bit is the continuation bit.
Returns value and number of parsed bytes.'''
v = 0
for i, b in enumerate(bytes_):
v |= (b & 0x7F) << (i * 7)
if b & 0x80 == 0:
return v, i+1
return v, len(bytes_)
def parse_uint(bytes_):
'''Parse little-endian integer.'''
v = 0
for i, b in enumerate(bytes_):
v |= b << (i * 8)
return v
def parse_exc_info(bytes_):
'''Parse exception information bytes from a branch packet.'''
if len(bytes_) < 1:
return None
excv, exclen = parse_varint(bytes_)
if bytes_[exclen - 1] & 0x80 != 0x00:
return None # Exception info not complete.
if exclen == 2 and excv & (1 << 13):
# Exception byte 1 was skipped, fix up the decoding.
excv = (excv & 0x7F) | ((excv & 0x3F80) << 7)
ns = excv & 1
exc = ((excv >> 1) & 0x0F) | ((excv >> 7) & 0x1F0)
cancel = (excv >> 5) & 1
altisa = (excv >> 6) & 1
hyp = (excv >> 12) & 1
resume = (excv >> 14) & 0x0F
return (ns, exc, cancel, altisa, hyp, resume)
def parse_branch_addr(bytes_, ref_addr, cpu_state, branch_enc):
'''Parse encoded branch address.
Returns addr, addrlen, cpu_state, exc_info.
Returns None if packet is not yet complete'''
addr, addrlen = parse_varint(bytes_)
if bytes_[addrlen - 1] & 0x80 != 0x00:
return None # Branch address not complete.
addr_bits = 7 * addrlen
have_exc_info = False
if branch_enc == 'original':
if addrlen == 5 and bytes_[4] & 0x40:
have_exc_info = True
elif branch_enc == 'alternative':
addr_bits -= 1 # Top bit of address indicates exc_info.
if addrlen >= 2 and addr & (1 << addr_bits):
have_exc_info = True
addr &= ~(1 << addr_bits)
exc_info = None
if have_exc_info:
exc_info = parse_exc_info(bytes_[addrlen:])
if exc_info is None:
return None # Exception info not complete.
if addrlen == 5:
# Possible change in CPU state.
if bytes_[4] & 0xB8 == 0x08:
cpu_state = 'arm'
elif bytes_[4] & 0xB0 == 0x10:
cpu_state = 'thumb'
elif bytes_[4] & 0xA0 == 0x20:
cpu_state = 'jazelle'
else:
raise NotImplementedError('Unhandled branch byte 4: 0x%02x' % bytes_[4])
# Shift the address according to current CPU state.
if cpu_state == 'arm':
addr = (addr & 0xFFFFFFFE) << 1
addr_bits += 1
elif cpu_state == 'thumb':
addr = addr & 0xFFFFFFFE
elif cpu_state == 'jazelle':
addr = (addr & 0xFFFFFFFFE) >> 1
addr_bits -= 1
else:
raise NotImplementedError('Unhandled state: ' + cpu_state)
# If the address wasn't full, fill in with the previous address.
if addrlen < 5:
addr |= ref_addr & (0xFFFFFFFF << addr_bits)
return addr, addrlen, cpu_state, exc_info
class Decoder(srd.Decoder):
api_version = 3
id = 'arm_etmv3'
name = 'ARM ETMv3'
longname = 'ARM Embedded Trace Macroblock v3'
desc = 'ARM ETM v3 instruction trace protocol.'
license = 'gplv2+'
inputs = ['uart']
outputs = []
tags = ['Debug/trace']
annotations = (
('trace', 'Trace info'),
('branch', 'Branch'),
('exception', 'Exception'),
('execution', 'Instruction execution'),
('data', 'Data access'),
('pc', 'Program counter'),
('instr_e', 'Executed instruction'),
('instr_n', 'Not executed instruction'),
('source', 'Source code'),
('location', 'Current location'),
('function', 'Current function'),
)
annotation_rows = (
('traces', 'Trace info', (0,)),
('flow', 'Code flow', (1, 2, 3,)),
('data-vals', 'Data access', (4,)),
('pc-vals', 'Program counters', (5,)),
('instructions', 'Instructions', (6, 7,)),
('sources', 'Source code', (8,)),
('locations', 'Current locations', (9,)),
('functions', 'Current functions', (10,)),
)
options = (
{'id': 'objdump', 'desc': 'objdump path',
'default': 'arm-none-eabi-objdump'},
{'id': 'objdump_opts', 'desc': 'objdump options',
'default': '-lSC'},
{'id': 'elffile', 'desc': '.elf path',
'default': ''},
{'id': 'branch_enc', 'desc': 'Branch encoding',
'default': 'alternative', 'values': ('alternative', 'original')},
)
def __init__(self):
self.reset()
def reset(self):
self.buf = []
self.syncbuf = []
self.prevsample = 0
self.last_branch = 0
self.cpu_state = 'arm'
self.current_pc = 0
self.current_loc = None
self.current_func = None
self.next_instr_lookup = {}
self.file_lookup = {}
self.func_lookup = {}
self.disasm_lookup = {}
self.source_lookup = {}
def start(self):
self.out_ann = self.register(srd.OUTPUT_ANN)
self.load_objdump()
def load_objdump(self):
'''Parse disassembly obtained from objdump into two tables:
next_instr_lookup: Find the next PC addr from current PC.
disasm_lookup: Find the instruction text from current PC.
source_lookup: Find the source code line from current PC.
'''
if not (self.options['objdump'] and self.options['elffile']):
return
opts = [self.options['objdump']]
opts += self.options['objdump_opts'].split()
opts += [self.options['elffile']]
try:
disasm = subprocess.check_output(opts)
except subprocess.CalledProcessError:
return
disasm = disasm.decode('utf-8', 'replace')
instpat = re.compile('\s*([0-9a-fA-F]+):\t+([0-9a-fA-F ]+)\t+([a-zA-Z][^;]+)\s*;?.*')
branchpat = re.compile('(b|bl|b..|bl..|cbnz|cbz)(?:\.[wn])?\s+(?:r[0-9]+,\s*)?([0-9a-fA-F]+)')
filepat = re.compile('[^\s]+[/\\\\]([a-zA-Z0-9._-]+:[0-9]+)(?:\s.*)?')
funcpat = re.compile('[0-9a-fA-F]+\s*<([^>]+)>:.*')
prev_src = ''
prev_file = ''
prev_func = ''
for line in disasm.split('\n'):
m = instpat.match(line)
if m:
addr = int(m.group(1), 16)
raw = m.group(2)
disas = m.group(3).strip().replace('\t', ' ')
self.disasm_lookup[addr] = disas
self.source_lookup[addr] = prev_src
self.file_lookup[addr] = prev_file
self.func_lookup[addr] = prev_func
# Next address in direct sequence.
ilen = len(raw.replace(' ', '')) // 2
next_n = addr + ilen
# Next address if branch is taken.
bm = branchpat.match(disas)
if bm:
next_e = int(bm.group(2), 16)
else:
next_e = next_n
self.next_instr_lookup[addr] = (next_n, next_e)
else:
m = funcpat.match(line)
if m:
prev_func = m.group(1)
prev_src = None
else:
m = filepat.match(line)
if m:
prev_file = m.group(1)
prev_src = None
else:
prev_src = line.strip()
def flush_current_loc(self):
if self.current_loc is not None:
ss, es, loc, src = self.current_loc
if loc:
self.put(ss, es, self.out_ann, [9, [loc]])
if src:
self.put(ss, es, self.out_ann, [8, [src]])
self.current_loc = None
def flush_current_func(self):
if self.current_func is not None:
ss, es, func = self.current_func
if func:
self.put(ss, es, self.out_ann, [10, [func]])
self.current_func = None
def instructions_executed(self, exec_status):
'''Advance program counter based on executed instructions.
Argument is a list of False for not executed and True for executed
instructions.
'''
if len(exec_status) == 0:
return
tdelta = max(1, (self.prevsample - self.startsample) / len(exec_status))
for i, exec_status in enumerate(exec_status):
pc = self.current_pc
default_next = pc + 2 if self.cpu_state == 'thumb' else pc + 4
target_n, target_e = self.next_instr_lookup.get(pc, (default_next, default_next))
ss = self.startsample + round(tdelta * i)
es = self.startsample + round(tdelta * (i+1))
self.put(ss, es, self.out_ann,
[5, ['PC 0x%08x' % pc, '0x%08x' % pc, '%08x' % pc]])
new_loc = self.file_lookup.get(pc)
new_src = self.source_lookup.get(pc)
new_dis = self.disasm_lookup.get(pc)
new_func = self.func_lookup.get(pc)
# Report source line only when it changes.
if self.current_loc is not None:
if new_loc != self.current_loc[2] or new_src != self.current_loc[3]:
self.flush_current_loc()
if self.current_loc is None:
self.current_loc = [ss, es, new_loc, new_src]
else:
self.current_loc[1] = es
# Report function name only when it changes.
if self.current_func is not None:
if new_func != self.current_func[2]:
self.flush_current_func()
if self.current_func is None:
self.current_func = [ss, es, new_func]
else:
self.current_func[1] = es
# Report instruction every time.
if new_dis:
if exec_status:
a = [6, ['Executed: ' + new_dis, new_dis, new_dis.split()[0]]]
else:
a = [7, ['Not executed: ' + new_dis, new_dis, new_dis.split()[0]]]
self.put(ss, es, self.out_ann, a)
if exec_status:
self.current_pc = target_e
else:
self.current_pc = target_n
def get_packet_type(self, byte):
'''Identify packet type based on its first byte.
See ARM IHI0014Q section "ETMv3 Signal Protocol" "Packet Types"
'''
if byte & 0x01 == 0x01:
return 'branch'
elif byte == 0x00:
return 'a_sync'
elif byte == 0x04:
return 'cyclecount'
elif byte == 0x08:
return 'i_sync'
elif byte == 0x0C:
return 'trigger'
elif byte & 0xF3 in (0x20, 0x40, 0x60):
return 'ooo_data'
elif byte == 0x50:
return 'store_failed'
elif byte == 0x70:
return 'i_sync'
elif byte & 0xDF in (0x54, 0x58, 0x5C):
return 'ooo_place'
elif byte == 0x3C:
return 'vmid'
elif byte & 0xD3 == 0x02:
return 'data'
elif byte & 0xFB == 0x42:
return 'timestamp'
elif byte == 0x62:
return 'data_suppressed'
elif byte == 0x66:
return 'ignore'
elif byte & 0xEF == 0x6A:
return 'value_not_traced'
elif byte == 0x6E:
return 'context_id'
elif byte == 0x76:
return 'exception_exit'
elif byte == 0x7E:
return 'exception_entry'
elif byte & 0x81 == 0x80:
return 'p_header'
else:
return 'unknown'
def fallback(self, buf):
ptype = self.get_packet_type(buf[0])
return [0, ['Unhandled ' + ptype + ': ' + ' '.join(['%02x' % b for b in buf])]]
def handle_a_sync(self, buf):
if buf[-1] == 0x80:
return [0, ['Synchronization']]
def handle_exception_exit(self, buf):
return [2, ['Exception exit']]
def handle_exception_entry(self, buf):
return [2, ['Exception entry']]
def handle_i_sync(self, buf):
contextid_bytes = 0 # This is the default ETM config.
if len(buf) < 6:
return None # Packet definitely not full yet.
if buf[0] == 0x08: # No cycle count.
cyclecount = None
idx = 1 + contextid_bytes # Index to info byte.
elif buf[0] == 0x70: # With cycle count.
cyclecount, cyclen = parse_varint(buf[1:6])
idx = 1 + cyclen + contextid_bytes
if len(buf) <= idx + 4:
return None
infobyte = buf[idx]
addr = parse_uint(buf[idx+1:idx+5])
reasoncode = (infobyte >> 5) & 3
reason = ('Periodic', 'Tracing enabled', 'After overflow', 'Exit from debug')[reasoncode]
jazelle = (infobyte >> 4) & 1
nonsec = (infobyte >> 3) & 1
altisa = (infobyte >> 2) & 1
hypervisor = (infobyte >> 1) & 1
thumb = addr & 1
addr &= 0xFFFFFFFE
if reasoncode == 0 and self.current_pc != addr:
self.put(self.startsample, self.prevsample, self.out_ann,
[0, ['WARN: Unexpected PC change 0x%08x -> 0x%08x' % \
(self.current_pc, addr)]])
elif reasoncode != 0:
# Reset location when the trace has been interrupted.
self.flush_current_loc()
self.flush_current_func()
self.last_branch = addr
self.current_pc = addr
if jazelle:
self.cpu_state = 'jazelle'
elif thumb:
self.cpu_state = 'thumb'
else:
self.cpu_state = 'arm'
cycstr = ''
if cyclecount is not None:
cycstr = ', cyclecount %d' % cyclecount
if infobyte & 0x80: # LSIP packet
self.put(self.startsample, self.prevsample, self.out_ann,
[0, ['WARN: LSIP I-Sync packet not implemented']])
return [0, ['I-Sync: %s, PC 0x%08x, %s state%s' % \
(reason, addr, self.cpu_state, cycstr), \
'I-Sync: %s 0x%08x' % (reason, addr)]]
def handle_trigger(self, buf):
return [0, ['Trigger event', 'Trigger']]
def handle_p_header(self, buf):
# Only non cycle-accurate mode supported.
if buf[0] & 0x83 == 0x80:
n = (buf[0] >> 6) & 1
e = (buf[0] >> 2) & 15
self.instructions_executed([1] * e + [0] * n)
if n:
return [3, ['%d instructions executed, %d skipped due to ' \
'condition codes' % (e, n),
'%d ins exec, %d skipped' % (e, n),
'%dE,%dN' % (e, n)]]
else:
return [3, ['%d instructions executed' % e,
'%d ins exec' % e, '%dE' % e]]
elif buf[0] & 0xF3 == 0x82:
i1 = (buf[0] >> 3) & 1
i2 = (buf[0] >> 2) & 1
self.instructions_executed([not i1, not i2])
txt1 = ('executed', 'skipped')
txt2 = ('E', 'S')
return [3, ['Instruction 1 %s, instruction 2 %s' % (txt1[i1], txt1[i2]),
'I1 %s, I2 %s' % (txt2[i1], txt2[i2]),
'%s,%s' % (txt2[i1], txt2[i2])]]
else:
return self.fallback(buf)
def handle_branch(self, buf):
if buf[-1] & 0x80 != 0x00:
return None # Not complete yet.
brinfo = parse_branch_addr(buf, self.last_branch, self.cpu_state,
self.options['branch_enc'])
if brinfo is None:
return None # Not complete yet.
addr, addrlen, cpu_state, exc_info = brinfo
self.last_branch = addr
self.current_pc = addr
txt = ''
if cpu_state != self.cpu_state:
txt += ', to %s state' % cpu_state
self.cpu_state = cpu_state
annidx = 1
if exc_info:
annidx = 2
ns, exc, cancel, altisa, hyp, resume = exc_info
if ns:
txt += ', to non-secure state'
if exc:
if exc < len(exc_names):
txt += ', exception %s' % exc_names[exc]
else:
txt += ', exception 0x%02x' % exc
if cancel:
txt += ', instr cancelled'
if altisa:
txt += ', to AltISA'
if hyp:
txt += ', to hypervisor'
if resume:
txt += ', instr resume 0x%02x' % resume
return [annidx, ['Branch to 0x%08x%s' % (addr, txt),
'B 0x%08x%s' % (addr, txt)]]
def decode(self, ss, es, data):
ptype, rxtx, pdata = data
if ptype != 'DATA':
return
# Reset packet if there is a long pause between bytes.
# This helps getting the initial synchronization.
self.byte_len = es - ss
if ss - self.prevsample > 16 * self.byte_len:
self.flush_current_loc()
self.flush_current_func()
self.buf = []
self.prevsample = es
self.buf.append(pdata[0])
# Store the start time of the packet.
if len(self.buf) == 1:
self.startsample = ss
# Keep separate buffer for detection of sync packets.
# Sync packets override everything else, so that we can regain sync
# even if some packets are corrupted.
self.syncbuf = self.syncbuf[-4:] + [pdata[0]]
if self.syncbuf == [0x00, 0x00, 0x00, 0x00, 0x80]:
self.buf = self.syncbuf
self.syncbuf = []
# See if it is ready to be decoded.
ptype = self.get_packet_type(self.buf[0])
if hasattr(self, 'handle_' + ptype):
func = getattr(self, 'handle_' + ptype)
data = func(self.buf)
else:
data = self.fallback(self.buf)
if data is not None:
if data:
self.put(self.startsample, es, self.out_ann, data)
self.buf = []