From: Pengfei Li <[email protected]> Add stackmap_dump.py to decode native-endian version 1 stack_map_bin streams from a file or stdin and print text or JSON.
Parse incrementally with bounded reads. Validate the header count, require stack depths from 1 through 64, reject trailing bytes, and report malformed input without a traceback. --top rejects negative values, keeps zero as unlimited, and selects a stable top N with an O(N) heap instead of materializing and sorting the whole input. Normal text output is emitted record by record. JSON output is staged in a SpooledTemporaryFile so malformed later records or symbolization failures leave stdout empty. The spool keeps up to 1 MiB in memory and then spills to disk. With --vmlinux, addr2line receives addresses on stdin in batches bounded by both 128 records and 128 unique addresses. This avoids ARG_MAX and bounds retained records even when addresses repeat. Document that KASLR-slid core addresses and module addresses are unsupported. Install the script from the tools/tracing install target and add parser, CLI, streaming, memory-bound and error-path tests. Report totals as stack records rather than strict unique stacks. Signed-off-by: Pengfei Li <[email protected]> --- tools/tracing/Makefile | 17 +- tools/tracing/stackmap_dump.py | 354 +++++++++++++++ tools/tracing/tests/test_stackmap_dump.py | 498 ++++++++++++++++++++++ 3 files changed, 866 insertions(+), 3 deletions(-) create mode 100755 tools/tracing/stackmap_dump.py create mode 100644 tools/tracing/tests/test_stackmap_dump.py diff --git a/tools/tracing/Makefile b/tools/tracing/Makefile index 95e485f12d97..8c2be033e360 100644 --- a/tools/tracing/Makefile +++ b/tools/tracing/Makefile @@ -1,11 +1,22 @@ # SPDX-License-Identifier: GPL-2.0 include ../scripts/Makefile.include +INSTALL ?= install +BINDIR ?= /usr/bin +PYTHON ?= python3 + all: latency rtla +check: + $(PYTHON) -m unittest discover -s tests -v + clean: latency_clean rtla_clean -install: latency_install rtla_install +install: latency_install rtla_install stackmap_install + +stackmap_install: + $(call QUIET_INSTALL,stackmap_dump.py)$(INSTALL) -D -m 755 stackmap_dump.py \ + $(DESTDIR)$(BINDIR)/stackmap_dump.py latency: $(call descend,latency) @@ -25,5 +36,5 @@ rtla_install: rtla_clean: $(call descend,rtla,clean) -.PHONY: all install clean latency latency_install latency_clean \ - rtla rtla_install rtla_clean +.PHONY: all check install clean stackmap_install latency latency_install \ + latency_clean rtla rtla_install rtla_clean diff --git a/tools/tracing/stackmap_dump.py b/tools/tracing/stackmap_dump.py new file mode 100755 index 000000000000..cf461b49cee7 --- /dev/null +++ b/tools/tracing/stackmap_dump.py @@ -0,0 +1,354 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: GPL-2.0 +""" +stackmap_dump.py - Parse and display ftrace stack_map_bin binary export. + +Usage: + # Read straight from the device over a pipe + adb shell cat /sys/kernel/debug/tracing/stack_map_bin | stackmap_dump.py + + # Or pull the file first and parse it + adb pull /sys/kernel/debug/tracing/stack_map_bin /tmp/stack_map.bin + python3 stackmap_dump.py /tmp/stack_map.bin + + # With vmlinux when addresses already match its link-time range + # (for example, a nokaslr core kernel; modules are not supported) + python3 stackmap_dump.py /tmp/stack_map.bin --vmlinux vmlinux + + # JSON output for tooling + python3 stackmap_dump.py /tmp/stack_map.bin --json +""" + +import argparse +import heapq +import io +import json +import struct +import subprocess +import sys +import tempfile + +MAGIC = 0x46534D42 # 'FSMB' +HEADER_SIZE = 16 # 4 x u32 +ENTRY_SIZE = 16 # 4 x u32 +MAX_STACK_DEPTH_V1 = 64 +MAX_STACKS_V1 = 1 << 18 +SYMBOL_BATCH_RECORDS = 128 +SYMBOL_BATCH_ADDRESSES = 128 +SYMBOL_MAX_BATCHES = 64 +SYMBOL_MAX_ADDRESSES = 8192 +JSON_SPOOL_MEMORY_LIMIT = 1 << 20 +COPY_BUFFER_SIZE = 64 << 10 + +# __ftrace_trace_stack() replaces trampoline addresses with this marker +# before storing the stack, so the binary export carries it verbatim. +FTRACE_TRAMPOLINE_MARKER = 0x7fffffff +TRAMPOLINE_LABEL = '[FTRACE TRAMPOLINE]' + + +def detect_endianness(data): + """Detect byte order from magic number in header.""" + if len(data) < 4: + raise ValueError("File too small") + magic_le = struct.unpack_from('<I', data, 0)[0] + if magic_le == MAGIC: + return '<' + magic_be = struct.unpack_from('>I', data, 0)[0] + if magic_be == MAGIC: + return '>' + raise ValueError(f"Bad magic: 0x{magic_le:08x} (neither LE nor BE)") + + +def batch_addr2line(vmlinux, addrs): + """Resolve multiple addresses in one addr2line invocation.""" + if not addrs: + return {} + try: + # Use stdin to avoid ARG_MAX with large address sets. + stdin = '\n'.join(hex(a) for a in addrs) + '\n' + result = subprocess.run( + ['addr2line', '-f', '-e', vmlinux], + input=stdin, capture_output=True, text=True, timeout=60 + ) + except subprocess.TimeoutExpired as error: + raise RuntimeError("addr2line failed: timed out") from error + except FileNotFoundError as error: + raise RuntimeError("addr2line failed: executable not found") from error + except OSError as error: + detail = error.strerror or str(error) + raise RuntimeError(f"addr2line failed: {detail}") from error + + if result.returncode: + detail = result.stderr.strip() or f"exit status {result.returncode}" + raise RuntimeError(f"addr2line failed: {detail}") + + lines = result.stdout.splitlines() + expected_lines = len(addrs) * 2 + if (len(lines) != expected_lines or + any(not line for line in lines)): + raise RuntimeError( + "addr2line returned malformed output: " + f"expected {expected_lines} non-empty lines, got {len(lines)}") + + symbols = {} + for i, addr in enumerate(addrs): + function = lines[i * 2] + if function != '??': + symbols[addr] = function + return symbols + + +def read_exact(stream, size, error_message): + """Read exactly size bytes from a stream that may return short reads.""" + chunks = [] + remaining = size + while remaining: + chunk = stream.read(remaining) + if not chunk: + raise ValueError(error_message) + chunks.append(chunk) + remaining -= len(chunk) + return b''.join(chunks) + + +def parse_stackmap_stream(stream): + """Yield (stack_id, ref_count, ips) tuples from a binary stream.""" + header = read_exact(stream, HEADER_SIZE, "File too small for header") + endian = detect_endianness(header) + header_fmt = f'{endian}IIII' + entry_fmt = f'{endian}IIII' + + _magic, version, nr_stacks, _reserved = struct.unpack(header_fmt, header) + if version != 1: + raise ValueError(f"Unsupported version: {version}") + if nr_stacks > MAX_STACKS_V1: + raise ValueError( + f"Invalid stack count {nr_stacks}: version 1 maximum is " + f"{MAX_STACKS_V1}") + + for _ in range(nr_stacks): + entry = read_exact(stream, ENTRY_SIZE, + "Truncated stack entry header") + stack_id, nr, ref_count, _reserved = struct.unpack(entry_fmt, entry) + if nr == 0: + raise ValueError( + "Invalid stack depth 0: version 1 minimum is 1") + if nr > MAX_STACK_DEPTH_V1: + raise ValueError( + f"Invalid stack depth {nr}: version 1 maximum is " + f"{MAX_STACK_DEPTH_V1}") + + ips_data = read_exact( + stream, nr * 8, + f"Truncated stack IP data for stack_id {stack_id}") + ips = struct.unpack(f'{endian}{nr}Q', ips_data) + yield stack_id, ref_count, list(ips) + + if stream.read(1): + raise ValueError(f"Trailing data after {nr_stacks} stack records") + + +def parse_stackmap_bin(data): + """Parse in-memory binary data using the streaming parser.""" + return list(parse_stackmap_stream(io.BytesIO(data))) + + +def non_negative_int(value): + number = int(value) + if number < 0: + raise argparse.ArgumentTypeError("--top must be non-negative") + return number + + +def select_top(records, limit): + """Select the largest ref_counts stably using O(limit) space.""" + heap = [] + for order, record in enumerate(records): + item = (record[1], -order, order, record) + if len(heap) < limit: + heapq.heappush(heap, item) + elif item[:2] > heap[0][:2]: + heapq.heapreplace(heap, item) + + heap.sort(key=lambda item: (-item[0], item[2])) + return [item[3] for item in heap] + + +def iter_symbol_batches(records): + """Group records without exceeding the unique-address limit.""" + batch = [] + addresses = set() + for record in records: + _stack_id, _ref_count, ips = record + record_addresses = { + ip for ip in ips if ip != FTRACE_TRAMPOLINE_MARKER + } + if (batch and + (len(batch) >= SYMBOL_BATCH_RECORDS or + len(addresses | record_addresses) > + SYMBOL_BATCH_ADDRESSES)): + yield batch, addresses + batch.clear() + addresses.clear() + batch.append(record) + addresses.update(record_addresses) + if batch: + yield batch, addresses + + +class RecordRenderer: + """Write records incrementally without changing text or JSON formats.""" + + def __init__(self, output, json_output, include_symbols): + self.output = output + self.json_output = json_output + self.include_symbols = include_symbols + self.count = 0 + + @staticmethod + def terminal_safe(text): + """Escape terminal controls while preserving printable Unicode.""" + escaped = [] + for character in text: + codepoint = ord(character) + if character.isprintable(): + escaped.append(character) + elif codepoint <= 0xff: + escaped.append(f'\\x{codepoint:02x}') + elif codepoint <= 0xffff: + escaped.append(f'\\u{codepoint:04x}') + else: + escaped.append(f'\\U{codepoint:08x}') + return ''.join(escaped) + + @staticmethod + def render_ip(ip, symbols): + if ip == FTRACE_TRAMPOLINE_MARKER: + return TRAMPOLINE_LABEL + return symbols.get(ip, f'0x{ip:x}') + + def emit(self, record, symbols): + stack_id, ref_count, ips = record + if self.json_output: + entry = { + 'stack_id': stack_id, + 'ref_count': ref_count, + 'ips': [f'0x{ip:x}' for ip in ips] + } + if self.include_symbols: + entry['symbols'] = [self.render_ip(ip, symbols) + for ip in ips] + encoded = json.dumps(entry, indent=2) + indented = '\n'.join(f' {line}' for line in encoded.splitlines()) + if self.count == 0: + self.output.write('[\n') + else: + self.output.write(',\n') + self.output.write(indented) + else: + self.output.write( + f"stack_id {stack_id} [ref {ref_count}, depth {len(ips)}]\n") + for index, ip in enumerate(ips): + if ip == FTRACE_TRAMPOLINE_MARKER: + self.output.write( + f" [{index}] {TRAMPOLINE_LABEL}\n") + continue + symbol = symbols.get(ip, '') + if symbol: + symbol = f' {self.terminal_safe(symbol)}' + self.output.write(f" [{index}] 0x{ip:x}{symbol}\n") + self.output.write('\n') + self.count += 1 + + def finish(self): + if not self.json_output: + return + if self.count: + self.output.write('\n]\n') + else: + self.output.write('[]\n') + + +def render_records(records, args, output): + renderer = RecordRenderer(output, args.json, bool(args.vmlinux)) + if args.vmlinux: + batch_count = 0 + address_count = 0 + for batch, addresses in iter_symbol_batches(records): + batch_count += 1 + address_count += len(addresses) + if (batch_count > SYMBOL_MAX_BATCHES or + address_count > SYMBOL_MAX_ADDRESSES): + raise RuntimeError( + "symbolization work limit exceeded: " + f"maximum {SYMBOL_MAX_BATCHES} batches and " + f"{SYMBOL_MAX_ADDRESSES} addresses") + symbols = batch_addr2line(args.vmlinux, list(addresses)) + for record in batch: + renderer.emit(record, symbols) + else: + for record in records: + renderer.emit(record, {}) + renderer.finish() + return renderer.count + + +def main(argv=None, input_file=None, output=None, error=None): + parser = argparse.ArgumentParser(description='Parse ftrace stack_map_bin') + parser.add_argument('file', nargs='?', default='-', + help="Path to stack_map_bin file, or '-' for stdin " + "(the default)") + parser.add_argument( + '--vmlinux', + help=('Path to vmlinux for symbol resolution; addresses must already ' + 'match vmlinux (for example, nokaslr). KASLR runtime addresses ' + 'and module addresses are not supported. Symbolization is ' + 'limited to 8192 addresses in 64 batches')) + parser.add_argument('--json', action='store_true', help='JSON output') + parser.add_argument('--top', type=non_negative_int, default=0, + help='Show only top N stacks by ref_count; 0 is unlimited') + args = parser.parse_args(argv) + + output = output if output is not None else sys.stdout + error = error if error is not None else sys.stderr + stream = input_file + close_stream = False + + try: + # stdin accepts a stack_map_bin stream directly from a pipe. + if stream is None: + if args.file == '-': + stream = sys.stdin.buffer + else: + stream = open(args.file, 'rb') + close_stream = True + + records = parse_stackmap_stream(stream) + if args.top: + records = select_top(records, args.top) + if args.json: + with tempfile.SpooledTemporaryFile( + max_size=JSON_SPOOL_MEMORY_LIMIT, mode='w+', + encoding='utf-8') as staged_output: + count = render_records(records, args, staged_output) + staged_output.seek(0) + while True: + chunk = staged_output.read(COPY_BUFFER_SIZE) + if not chunk: + break + output.write(chunk) + else: + count = render_records(records, args, output) + except (ValueError, OSError, RuntimeError) as exception: + print(f"error: {exception}", file=error) + return 1 + finally: + if close_stream: + stream.close() + + print(f"Total: {count} stack records", file=error) + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/tools/tracing/tests/test_stackmap_dump.py b/tools/tracing/tests/test_stackmap_dump.py new file mode 100644 index 000000000000..03769cbc190e --- /dev/null +++ b/tools/tracing/tests/test_stackmap_dump.py @@ -0,0 +1,498 @@ +#!/usr/bin/env python3 +# SPDX-License-Identifier: GPL-2.0 + +import importlib.util +import io +import json +import os +from pathlib import Path +import struct +import subprocess +import sys +import tempfile +import unittest +from unittest import mock + +SCRIPT = Path(__file__).resolve().parents[1] / "stackmap_dump.py" +SPEC = importlib.util.spec_from_file_location("stackmap_dump", SCRIPT) +stackmap_dump = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(stackmap_dump) + + +def image(endian="<", version=1, entries=()): + data = bytearray(struct.pack(f"{endian}IIII", stackmap_dump.MAGIC, + version, len(entries), 0)) + for stack_id, ref_count, ips in entries: + data += struct.pack(f"{endian}IIII", stack_id, len(ips), + ref_count, 0) + data += struct.pack(f"{endian}{len(ips)}Q", *ips) + return bytes(data) + + +class ShortReadInput(io.BytesIO): + def read(self, size=-1): + if size < 0: + raise AssertionError("unbounded read() is not allowed") + if size > stackmap_dump.MAX_STACK_DEPTH_V1 * 8: + raise AssertionError("read() exceeds one version 1 stack") + return super().read(min(size, 3)) + + +class StreamingGuardInput(io.BytesIO): + def __init__(self, data, first_record_end, output): + super().__init__(data) + self.first_record_end = first_record_end + self.output = output + + def read(self, size=-1): + if (self.tell() >= self.first_record_end and + "stack_id 1" not in self.output.getvalue()): + raise AssertionError("next record read before current output") + return super().read(size) + + +class TrackedRecord: + live_count = 0 + max_live_count = 0 + + def __init__(self, stack_id, ip=None): + self.stack_id = stack_id + self.ip = 0x1000 + stack_id if ip is None else ip + type(self).live_count += 1 + type(self).max_live_count = max(type(self).max_live_count, + type(self).live_count) + + def __iter__(self): + return iter((self.stack_id, 1, [self.ip])) + + def __del__(self): + type(self).live_count -= 1 + + +class StackmapParserTest(unittest.TestCase): + def test_little_endian_entry(self): + raw = image(entries=[(7, 3, [0x1234, 0x7fffffff])]) + self.assertEqual(list(stackmap_dump.parse_stackmap_bin(raw)), + [(7, 3, [0x1234, 0x7fffffff])]) + + def test_big_endian_entry(self): + raw = image(">", entries=[(9, 2, [0xabcdef])]) + self.assertEqual(list(stackmap_dump.parse_stackmap_bin(raw)), + [(9, 2, [0xabcdef])]) + + def test_bad_magic(self): + with self.assertRaisesRegex(ValueError, "Bad magic"): + list(stackmap_dump.parse_stackmap_bin(b"BAD!" + bytes(12))) + + def test_unknown_version(self): + with self.assertRaisesRegex(ValueError, "Unsupported version"): + list(stackmap_dump.parse_stackmap_bin(image(version=2))) + + def test_truncated_header(self): + with self.assertRaisesRegex(ValueError, "too small for header"): + list(stackmap_dump.parse_stackmap_bin(bytes(15))) + + def test_truncated_entry_header(self): + raw = struct.pack("<IIII", stackmap_dump.MAGIC, 1, 1, 0) + with self.assertRaisesRegex(ValueError, "Truncated stack entry header"): + list(stackmap_dump.parse_stackmap_bin(raw)) + + def test_truncated_ip_data(self): + raw = image(entries=[(1, 1, [0x1234])])[:-1] + with self.assertRaisesRegex(ValueError, "Truncated stack IP data"): + list(stackmap_dump.parse_stackmap_bin(raw)) + + def test_rejects_records_beyond_header_count(self): + raw = image(entries=[(1, 1, [0x1234])]) + raw = bytearray(raw) + struct.pack_into("<I", raw, 8, 0) + with self.assertRaisesRegex(ValueError, "Trailing data"): + stackmap_dump.parse_stackmap_bin(raw) + + def test_cli_reports_malformed_input_without_traceback(self): + result = subprocess.run([sys.executable, str(SCRIPT), "-"], + input=b"bad", capture_output=True) + stderr = result.stderr.decode() + self.assertNotEqual(result.returncode, 0) + self.assertIn("error: File too small for header", stderr) + self.assertNotIn("Traceback", stderr) + + def test_cli_reports_missing_file_without_traceback(self): + missing = SCRIPT.parent / "does-not-exist.stackmap" + result = subprocess.run([sys.executable, str(SCRIPT), str(missing)], + capture_output=True, text=True) + self.assertNotEqual(result.returncode, 0) + self.assertIn("error:", result.stderr) + self.assertNotIn("Traceback", result.stderr) + + def test_cli_reports_addr2line_failure(self): + raw = image(entries=[(1, 2, [0x1234])]) + with tempfile.TemporaryDirectory() as temp_dir: + addr2line = Path(temp_dir) / "addr2line" + addr2line.write_text( + "#!/bin/sh\necho controlled addr2line failure >&2\nexit 23\n", + encoding="utf-8") + addr2line.chmod(0o755) + env = os.environ.copy() + env["PATH"] = temp_dir + result = subprocess.run( + [sys.executable, str(SCRIPT), "-", "--vmlinux", "vmlinux"], + input=raw, capture_output=True, env=env) + stderr = result.stderr.decode() + self.assertNotEqual(result.returncode, 0) + self.assertIn( + "error: addr2line failed: controlled addr2line failure", stderr) + self.assertNotIn("Traceback", stderr) + + def test_cli_reports_truncated_addr2line_output(self): + raw = image(entries=[(1, 2, [0x1234])]) + with tempfile.TemporaryDirectory() as temp_dir: + addr2line = Path(temp_dir) / "addr2line" + addr2line.write_text( + "#!/bin/sh\nprintf 'resolved_function\\n'\n", + encoding="utf-8") + addr2line.chmod(0o755) + env = os.environ.copy() + env["PATH"] = temp_dir + result = subprocess.run( + [sys.executable, str(SCRIPT), "-", "--vmlinux", "vmlinux"], + input=raw, capture_output=True, env=env) + stderr = result.stderr.decode() + self.assertNotEqual(result.returncode, 0) + self.assertIn("error: addr2line returned malformed output", stderr) + self.assertNotIn("Traceback", stderr) + + def test_cli_accepts_unresolved_addr2line_output(self): + raw = image(entries=[(1, 2, [0x1234])]) + with tempfile.TemporaryDirectory() as temp_dir: + addr2line = Path(temp_dir) / "addr2line" + addr2line.write_text( + "#!/bin/sh\nprintf '??\\n??:0\\n'\n", + encoding="utf-8") + addr2line.chmod(0o755) + env = os.environ.copy() + env["PATH"] = temp_dir + result = subprocess.run( + [sys.executable, str(SCRIPT), "-", "--vmlinux", "vmlinux"], + input=raw, capture_output=True, env=env) + self.assertEqual(result.returncode, 0, result.stderr.decode()) + self.assertIn(b"[0] 0x1234", result.stdout) + + def test_cli_rejects_trailing_data(self): + raw = image(entries=[(1, 1, [0x1234])]) + b"junk" + result = subprocess.run([sys.executable, str(SCRIPT), "-"], + input=raw, capture_output=True) + self.assertNotEqual(result.returncode, 0) + self.assertIn(b"error: Trailing data", result.stderr) + + def test_cli_accepts_valid_input(self): + raw = image(entries=[(1, 2, [0x1234])]) + result = subprocess.run([sys.executable, str(SCRIPT), "-", "--json"], + input=raw, capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr.decode()) + self.assertIn(b'"stack_id": 1', result.stdout) + + def test_stream_parser_handles_short_reads(self): + raw = image(entries=[(7, 3, [0x1234, 0x5678])]) + self.assertEqual(list(stackmap_dump.parse_stackmap_stream( + ShortReadInput(raw))), [(7, 3, [0x1234, 0x5678])]) + + def test_rejects_stack_depth_above_version_1_limit(self): + raw = (struct.pack("<IIII", stackmap_dump.MAGIC, 1, 1, 0) + + struct.pack("<IIII", 7, 65, 1, 0)) + with self.assertRaisesRegex(ValueError, "stack depth.*64"): + list(stackmap_dump.parse_stackmap_stream(io.BytesIO(raw))) + + def test_accepts_stack_depth_at_version_1_limit(self): + ips = list(range(stackmap_dump.MAX_STACK_DEPTH_V1)) + raw = image(entries=[(7, 1, ips)]) + self.assertEqual(stackmap_dump.parse_stackmap_bin(raw), [(7, 1, ips)]) + + def test_rejects_stack_count_above_version_1_limit(self): + raw = struct.pack("<IIII", stackmap_dump.MAGIC, 1, + (1 << 18) + 1, 0) + with self.assertRaisesRegex(ValueError, "stack count.*262144"): + list(stackmap_dump.parse_stackmap_stream(io.BytesIO(raw))) + + def test_accepts_stack_count_at_version_1_limit(self): + raw = struct.pack("<IIII", stackmap_dump.MAGIC, 1, 1 << 18, 0) + with self.assertRaisesRegex(ValueError, + "Truncated stack entry header"): + list(stackmap_dump.parse_stackmap_stream(io.BytesIO(raw))) + + def test_cli_rejects_negative_top(self): + raw = image(entries=[(1, 2, [0x1234])]) + result = subprocess.run( + [sys.executable, str(SCRIPT), "-", "--top", "-1"], + input=raw, capture_output=True) + self.assertNotEqual(result.returncode, 0) + self.assertIn(b"--top must be non-negative", result.stderr) + + def test_cli_top_zero_remains_unlimited(self): + raw = image(entries=[(1, 2, [0x1000]), (2, 1, [0x2000])]) + result = subprocess.run( + [sys.executable, str(SCRIPT), "-", "--json", "--top", "0"], + input=raw, capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr.decode()) + self.assertEqual([entry["stack_id"] for entry in + json.loads(result.stdout)], [1, 2]) + + def test_cli_top_n_is_descending_and_stable_for_ties(self): + raw = image(entries=[(1, 3, [0x1000]), + (2, 9, [0x2000]), + (3, 9, [0x3000]), + (4, 9, [0x4000]), + (5, 1, [0x5000])]) + result = subprocess.run( + [sys.executable, str(SCRIPT), "-", "--json", "--top", "2"], + input=raw, capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr.decode()) + self.assertEqual([(entry["stack_id"], entry["ref_count"]) + for entry in json.loads(result.stdout)], + [(2, 9), (3, 9)]) + + def test_rejects_zero_stack_depth(self): + raw = (struct.pack("<IIII", stackmap_dump.MAGIC, 1, 1, 0) + + struct.pack("<IIII", 7, 0, 1, 0)) + with self.assertRaisesRegex(ValueError, "stack depth.*minimum is 1"): + list(stackmap_dump.parse_stackmap_stream(io.BytesIO(raw))) + + def test_cli_empty_text_output_format(self): + result = subprocess.run([sys.executable, str(SCRIPT), "-"], + input=image(), capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr.decode()) + self.assertEqual(result.stdout, b"") + self.assertEqual(result.stderr, b"Total: 0 stack records\n") + + def test_cli_empty_json_output_format(self): + result = subprocess.run([sys.executable, str(SCRIPT), "-", "--json"], + input=image(), capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr.decode()) + self.assertEqual(result.stdout, b"[]\n") + self.assertEqual(result.stderr, b"Total: 0 stack records\n") + + def test_cli_multi_record_text_output_format(self): + raw = image(entries=[(7, 3, [0x1234, + stackmap_dump.FTRACE_TRAMPOLINE_MARKER]), + (9, 1, [0xabcd])]) + result = subprocess.run([sys.executable, str(SCRIPT), "-"], + input=raw, capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr.decode()) + self.assertEqual( + result.stdout.decode(), + "stack_id 7 [ref 3, depth 2]\n" + " [0] 0x1234\n" + " [1] [FTRACE TRAMPOLINE]\n\n" + "stack_id 9 [ref 1, depth 1]\n" + " [0] 0xabcd\n\n") + self.assertEqual(result.stderr, b"Total: 2 stack records\n") + + def test_cli_multi_record_json_output_format(self): + raw = image(entries=[(7, 3, [0x1234]), (9, 1, [0xabcd])]) + result = subprocess.run([sys.executable, str(SCRIPT), "-", "--json"], + input=raw, capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr.decode()) + expected = json.dumps([ + {"stack_id": 7, "ref_count": 3, "ips": ["0x1234"]}, + {"stack_id": 9, "ref_count": 1, "ips": ["0xabcd"]}, + ], indent=2) + "\n" + self.assertEqual(result.stdout.decode(), expected) + self.assertEqual(result.stderr, b"Total: 2 stack records\n") + + def test_vmlinux_text_output_format(self): + raw = image(entries=[(7, 3, [ + 0x1000, 0x2000, stackmap_dump.FTRACE_TRAMPOLINE_MARKER])]) + output = io.StringIO() + error = io.StringIO() + with mock.patch.object(stackmap_dump, "batch_addr2line", + return_value={0x1000: "resolved"}): + result = stackmap_dump.main( + ["--vmlinux", "vmlinux"], input_file=io.BytesIO(raw), + output=output, error=error) + + self.assertEqual(result, 0, error.getvalue()) + self.assertEqual( + output.getvalue(), + "stack_id 7 [ref 3, depth 3]\n" + " [0] 0x1000 resolved\n" + " [1] 0x2000\n" + " [2] [FTRACE TRAMPOLINE]\n\n") + self.assertEqual(error.getvalue(), "Total: 1 stack records\n") + + def test_vmlinux_json_output_format(self): + raw = image(entries=[(7, 3, [ + 0x1000, 0x2000, stackmap_dump.FTRACE_TRAMPOLINE_MARKER])]) + output = io.StringIO() + error = io.StringIO() + with mock.patch.object(stackmap_dump, "batch_addr2line", + return_value={0x1000: "resolved"}): + result = stackmap_dump.main( + ["--json", "--vmlinux", "vmlinux"], + input_file=io.BytesIO(raw), output=output, error=error) + + expected = json.dumps([{ + "stack_id": 7, + "ref_count": 3, + "ips": ["0x1000", "0x2000", "0x7fffffff"], + "symbols": ["resolved", "0x2000", "[FTRACE TRAMPOLINE]"], + }], indent=2) + "\n" + self.assertEqual(result, 0, error.getvalue()) + self.assertEqual(output.getvalue(), expected) + self.assertEqual(error.getvalue(), "Total: 1 stack records\n") + + def test_json_malformed_later_record_leaves_stdout_empty(self): + raw = image(entries=[(1, 2, [0x1000]), (2, 1, [0x2000])])[:-1] + output = io.StringIO() + error = io.StringIO() + + result = stackmap_dump.main(["--json"], input_file=io.BytesIO(raw), + output=output, error=error) + + self.assertEqual(result, 1) + self.assertEqual(output.getvalue(), "") + self.assertIn("error: Truncated stack IP data", error.getvalue()) + + def test_json_later_addr2line_failure_leaves_stdout_empty(self): + raw = image(entries=[(stack_id, 1, [0x1000 + stack_id]) + for stack_id in range(129)]) + output = io.StringIO() + error = io.StringIO() + with mock.patch.object( + stackmap_dump, "batch_addr2line", + side_effect=[{}, RuntimeError("addr2line failed: later batch")]): + result = stackmap_dump.main( + ["--json", "--vmlinux", "vmlinux"], + input_file=io.BytesIO(raw), output=output, error=error) + + self.assertEqual(result, 1) + self.assertEqual(output.getvalue(), "") + self.assertEqual(error.getvalue(), + "error: addr2line failed: later batch\n") + + def test_text_streams_before_reading_the_next_record(self): + raw = image(entries=[(1, 2, [0x1000]), (2, 1, [0x2000])]) + output = io.StringIO() + first_record_end = (stackmap_dump.HEADER_SIZE + + stackmap_dump.ENTRY_SIZE + 8) + input_file = StreamingGuardInput(raw, first_record_end, output) + error = io.StringIO() + + result = stackmap_dump.main([], input_file=input_file, + output=output, error=error) + + self.assertEqual(result, 0, error.getvalue()) + self.assertIn("stack_id 2", output.getvalue()) + + def test_json_input_uses_only_bounded_reads(self): + raw = image(entries=[(1, 2, [0x1000]), (2, 1, [0x2000])]) + output = io.StringIO() + error = io.StringIO() + + result = stackmap_dump.main( + ["--json"], input_file=ShortReadInput(raw), + output=output, error=error) + + self.assertEqual(result, 0, error.getvalue()) + self.assertEqual([entry["stack_id"] + for entry in json.loads(output.getvalue())], [1, 2]) + + def test_json_does_not_materialize_all_records(self): + TrackedRecord.live_count = 0 + TrackedRecord.max_live_count = 0 + + def records(_stream): + for stack_id in range(1000): + yield TrackedRecord(stack_id) + + output = io.StringIO() + with mock.patch.object(stackmap_dump, "parse_stackmap_stream", + side_effect=records): + result = stackmap_dump.main( + ["--json"], input_file=io.BytesIO(), output=output, + error=io.StringIO()) + + self.assertEqual(result, 0) + self.assertLessEqual(TrackedRecord.max_live_count, 2) + self.assertEqual(len(json.loads(output.getvalue())), 1000) + + def test_vmlinux_symbolization_limits_unique_addresses_per_call(self): + raw = image(entries=[ + (stack_id, 1, [0x10000 + stack_id * 4 + frame + for frame in range(4)]) + for stack_id in range(50)]) + calls = [] + + def resolve(_vmlinux, addrs): + calls.append(list(addrs)) + return {} + + with mock.patch.object(stackmap_dump, "batch_addr2line", + side_effect=resolve): + result = stackmap_dump.main( + ["--vmlinux", "vmlinux"], input_file=io.BytesIO(raw), + output=io.StringIO(), error=io.StringIO()) + + self.assertEqual(result, 0) + self.assertLessEqual(max(map(len, calls)), 128) + self.assertEqual(stackmap_dump.SYMBOL_BATCH_ADDRESSES, 128) + + def test_vmlinux_duplicate_addresses_keep_record_batches_bounded(self): + TrackedRecord.live_count = 0 + TrackedRecord.max_live_count = 0 + + def records(_stream): + for stack_id in range(1000): + yield TrackedRecord(stack_id, ip=0x1000) + + with mock.patch.object(stackmap_dump, "parse_stackmap_stream", + side_effect=records), mock.patch.object( + stackmap_dump, "batch_addr2line", + return_value={}): + result = stackmap_dump.main( + ["--json", "--vmlinux", "vmlinux"], + input_file=io.BytesIO(), output=io.StringIO(), + error=io.StringIO()) + + self.assertEqual(result, 0) + self.assertLessEqual(TrackedRecord.max_live_count, 130) + + def test_vmlinux_rejects_aggregate_symbolization_work_above_limit(self): + raw = image(entries=[(stack_id, 1, [0x10000 + stack_id]) + for stack_id in range(8193)]) + error = io.StringIO() + with mock.patch.object(stackmap_dump, "batch_addr2line", + return_value={}): + result = stackmap_dump.main( + ["--vmlinux", "vmlinux"], input_file=io.BytesIO(raw), + output=io.StringIO(), error=error) + + self.assertNotEqual(result, 0) + self.assertIn("symbolization work limit", error.getvalue()) + + def test_vmlinux_text_output_escapes_terminal_control_characters(self): + raw = image(entries=[(7, 3, [0x1000])]) + output = io.StringIO() + with mock.patch.object( + stackmap_dump, "batch_addr2line", + return_value={0x1000: "name\x1b[2J\rspoof\tend"}): + result = stackmap_dump.main( + ["--vmlinux", "vmlinux"], input_file=io.BytesIO(raw), + output=output, error=io.StringIO()) + + self.assertEqual(result, 0) + self.assertIn(r"name\x1b[2J\x0dspoof\x09end", output.getvalue()) + self.assertNotIn("\x1b", output.getvalue()) + self.assertNotIn("\r", output.getvalue()) + self.assertNotIn("\t", output.getvalue()) + + def test_help_states_vmlinux_kaslr_and_work_limits(self): + result = subprocess.run([sys.executable, str(SCRIPT), "--help"], + capture_output=True, text=True, check=True) + self.assertIn("KASLR", result.stdout) + self.assertIn("module", result.stdout) + self.assertIn("8192", result.stdout) + self.assertIn("64 batches", result.stdout) + + +if __name__ == "__main__": + unittest.main() -- 2.34.1
