| #!/usr/bin/env python3 | |
| """Code size of TinyUSB examples. | |
| `report` builds the working tree in cmake-code-size/<board>/build and tabulates each elf's | |
| per-file section sizes in cmake-code-size/<board>/report[_<ex>].md. | |
| `diff` builds the base branch (master) and the current tree in | |
| cmake-code-size/<board>/{base,build}, pairs their elfs by (board, elf path) and reports | |
| each pair's per-file flash/RAM deltas in cmake-code-size/<board>/diff[_<ex>].md; with | |
| --combined, cmake-code-size/_combined/diff.md covers every board's pairs. | |
| The sizes come from --engine; each ENGINES entry maps (elf, filters) to | |
| {'files': {key: {'flash', 'ram'}}, 'all': {'flash', 'ram'}, 'sections': {key: {section: | |
| size}}, 'symbols': {key: {section: {name: size}}}}, keyed by the source path after the | |
| matching filter, the same for every engine. | |
| - membrowse: `membrowse report` symbols. Membrowse 1.2.9 truncates `source_file`, | |
| so paths come from `object_file`. | |
| - linkermap: input sections of the elf's GNU ld map, by object path; its symbols are | |
| those input sections. | |
| - bloaty: `bloaty -d compileunits,sections,symbols` VM sizes, by DWARF compile unit. | |
| Every engine takes flash/RAM from the elf's section and program headers | |
| (section_buckets()). | |
| Usage: | |
| python tools/code_size.py report -b raspberry_pi_pico -e device/cdc_msc --symbols | |
| python tools/code_size.py diff -b raspberry_pi_pico | |
| python tools/code_size.py diff -b raspberry_pi_pico -b raspberry_pi_pico2 | |
| python tools/code_size.py diff -b raspberry_pi_pico -f portable/raspberrypi | |
| python tools/code_size.py diff -b raspberry_pi_pico -e device/cdc_msc | |
| python tools/code_size.py diff -b raspberry_pi_pico -e device/cdc_msc --bloaty | |
| python tools/code_size.py diff -b raspberry_pi_pico --engine linkermap --json | |
| python tools/code_size.py diff --ci # CI-pinned boards, combined | |
| python tools/code_size.py diff -b raspberry_pi_pico -b raspberry_pi_pico2 --combined # combine listed boards | |
| """ | |
| import argparse | |
| import collections | |
| import concurrent.futures | |
| import csv | |
| import functools | |
| import glob | |
| import importlib.util | |
| import io | |
| import json | |
| import os | |
| import re | |
| import runpy | |
| import shlex | |
| import shutil | |
| import signal | |
| import struct | |
| import subprocess | |
| import sys | |
| import time | |
| import build_utils | |
| from membrowse_cli import extract_ld_scripts, extract_defsyms, link_command, report_inputs | |
| # resolved like tinyusb_src_filter(), so a symlinked checkout still matches its filter | |
| TINYUSB_ROOT = os.path.dirname(os.path.dirname(os.path.realpath(__file__))) | |
| CODE_SIZE_DIR = os.path.join(TINYUSB_ROOT, 'cmake-code-size') | |
| CI_PINNED_BOARDS = os.path.join(TINYUSB_ROOT, '.github', 'ci-pinned-boards.json') | |
| # a diff's side names when git cannot give their commit hashes | |
| SIDE_LABELS = ('base', 'new') | |
| _RAM_REGION_HINTS = ('ram', 'tcm', 'ddr') | |
| _FLASH_REGION_HINTS = ('flash', 'rom') | |
| def _find_ninja_build_dir(elf_path): | |
| """Nearest ancestor of `elf_path` containing build.ninja, or None.""" | |
| d = os.path.dirname(os.path.abspath(elf_path)) | |
| while True: | |
| if os.path.isfile(os.path.join(d, 'build.ninja')): | |
| return d | |
| parent = os.path.dirname(d) | |
| if parent == d: | |
| return None | |
| d = parent | |
| def _link_settings(elf_path): | |
| """Return the ELF's build dir, and the linker scripts and defsyms of its link.""" | |
| build_dir = _find_ninja_build_dir(elf_path) | |
| if build_dir is None: | |
| raise RuntimeError(f'no build.ninja found above {elf_path} - cannot ' | |
| f'determine its linker scripts') | |
| commands = link_command('ninja', build_dir, elf_path) | |
| ld_scripts = extract_ld_scripts(commands) | |
| if not ld_scripts: | |
| raise RuntimeError(f'no linker script found in the ninja build graph ' | |
| f'for {elf_path}') | |
| return build_dir, ld_scripts, extract_defsyms(commands) | |
| def report_for_elf(elf_path): | |
| """Run membrowse local report on one elf, return parsed JSON dict.""" | |
| build_dir, ld_scripts, defsyms = _link_settings(elf_path) | |
| cmd = (['membrowse', 'report'] + report_inputs(elf_path, ld_scripts, defsyms) | |
| + ['--json', '--all-symbols']) | |
| # from the link's working dir, as membrowse_cli.report() does | |
| r = subprocess.run(cmd, capture_output=True, text=True, cwd=build_dir) | |
| if r.returncode != 0: | |
| raise RuntimeError(f'membrowse report failed for {elf_path}: {r.stderr}') | |
| try: | |
| return json.loads(r.stdout) | |
| except json.JSONDecodeError as e: | |
| raise RuntimeError(f'malformed membrowse report for {elf_path}: {e}') from e | |
| def _classify_region(name): | |
| """Classify by region name; membrowse 1.2.9 reports parsed regions as UNKNOWN.""" | |
| n = (name or '').lower() | |
| if any(h in n for h in _FLASH_REGION_HINTS): | |
| return 'flash' | |
| if any(h in n for h in _RAM_REGION_HINTS): | |
| return 'ram' | |
| return None | |
| def _relative_key(src, filters): | |
| """Source path relative to the first matching filter, object suffix stripped, | |
| or None when no filter matches.""" | |
| for f in filters: | |
| idx = src.find(f) | |
| if idx < 0: | |
| continue | |
| key = src[idx + len(f):] | |
| for suffix in ('.obj', '.o'): | |
| if key.endswith(suffix): | |
| return key[:-len(suffix)] | |
| return key | |
| return None | |
| SHT_NOBITS = 8 | |
| SHT_ARRAYS = (14, 15, 16) # INIT_ARRAY, FINI_ARRAY, PREINIT_ARRAY | |
| SHF_WRITE, SHF_ALLOC = 0x1, 0x2 | |
| PT_LOAD = 1 | |
| _NOT_COUNTED = frozenset() | |
| def elf_layout(path): | |
| """Section headers [(name, type, flags, addr, size)] and PT_LOAD segments | |
| [(vaddr, paddr, memsz)] of an ELF32/64 file of either byte order. Raises | |
| RuntimeError for an unreadable or malformed file.""" | |
| try: | |
| with open(path, 'rb') as f: | |
| data = f.read() | |
| if data[:4] != b'\x7fELF': | |
| raise RuntimeError(f'{path} is not an ELF file') | |
| return _elf_layout(data) | |
| except (OSError, struct.error, IndexError, ValueError, UnicodeDecodeError) as e: | |
| raise RuntimeError(f'cannot read ELF headers of {path}: {e}') from e | |
| def _elf_layout(data): | |
| e = '<' if data[5] == 1 else '>' | |
| if data[4] == 2: | |
| phoff, shoff = struct.unpack_from(e + 'QQ', data, 0x20) | |
| phentsize, phnum, shentsize, shnum, shstrndx = struct.unpack_from(e + '5H', data, 0x36) | |
| phdr = lambda o: struct.unpack_from(e + 'II6Q', data, o) # noqa: E731 | |
| load = lambda p: (p[3], p[4], p[6]) # noqa: E731 | |
| shdr = lambda o: struct.unpack_from(e + 'IIQQQQIIQQ', data, o) # noqa: E731 | |
| else: | |
| phoff, shoff = struct.unpack_from(e + 'II', data, 0x1c) | |
| phentsize, phnum, shentsize, shnum, shstrndx = struct.unpack_from(e + '5H', data, 0x2a) | |
| phdr = lambda o: struct.unpack_from(e + '8I', data, o) # noqa: E731 | |
| load = lambda p: (p[2], p[3], p[5]) # noqa: E731 | |
| shdr = lambda o: struct.unpack_from(e + '10I', data, o) # noqa: E731 | |
| loads = [load(p) for p in (phdr(phoff + i * phentsize) for i in range(phnum)) if p[0] == PT_LOAD] | |
| headers = [shdr(shoff + i * shentsize) for i in range(shnum)] | |
| strtab = headers[shstrndx][4] | |
| def name(n): | |
| return data[strtab + n:data.index(b'\0', strtab + n)].decode() | |
| return [(name(h[0]), h[1], h[2], h[3], h[5]) for h in headers[1:]], loads | |
| def map_regions(map_path): | |
| """[(name, origin, length)] of a GNU ld map's Memory Configuration.""" | |
| if not os.path.isfile(map_path): | |
| raise RuntimeError(f'no linker map {map_path}') | |
| regions, inside = [], False | |
| with open(map_path, encoding='utf-8', errors='replace') as f: | |
| for line in f: | |
| if line.startswith('Memory Configuration'): | |
| inside = True | |
| elif line.startswith('Linker script and memory map'): | |
| break | |
| elif inside: | |
| m = re.match(r'(\S+)\s+0x([0-9a-fA-F]+)\s+0x([0-9a-fA-F]+)', line) | |
| if m and m.group(1) != '*default*': | |
| regions.append((m.group(1), int(m.group(2), 16), int(m.group(3), 16))) | |
| return regions | |
| def section_buckets(elf, map_path): | |
| """{section: flash/RAM bucket set} for every section of `elf`, empty when not | |
| allocated. A file-backed section loaded from elsewhere is a flash copy plus its | |
| run location; one running where it loads is flash unless writable, when the | |
| map region holding it decides. Raises RuntimeError rather than guess.""" | |
| sections, loads = elf_layout(elf) | |
| regions = map_regions(map_path) | |
| buckets = {} | |
| for name, sh_type, flags, addr, size in sections: | |
| if not flags & SHF_ALLOC: | |
| buckets[name] = _NOT_COUNTED | |
| continue | |
| seg = next((s for s in loads if s[0] <= addr < s[0] + s[2]), None) | |
| if seg is None: | |
| if size: | |
| raise RuntimeError(f'{elf}: allocated section {name} lies in no PT_LOAD segment') | |
| buckets[name] = _NOT_COUNTED | |
| continue | |
| if sh_type == SHT_NOBITS: | |
| buckets[name] = frozenset({'ram'}) | |
| continue | |
| kinds = {_classify_region(r) for r, origin, length in regions if origin <= addr < origin + length} | |
| if seg[1] + addr - seg[0] != addr: | |
| # a run address in flash is an alias (xmc4500 cached/uncached), not a RAM copy | |
| buckets[name] = frozenset({'flash'}) if kinds == {'flash'} else frozenset({'flash', 'ram'}) | |
| elif not flags & SHF_WRITE: | |
| buckets[name] = frozenset({'flash'}) | |
| elif len(kinds) == 1 and None not in kinds: | |
| buckets[name] = frozenset(kinds) # RAM-only images (raspberrypi_zero .data) | |
| elif sh_type in SHT_ARRAYS: | |
| buckets[name] = frozenset({'flash'}) # NXP .init_array in m_text | |
| else: | |
| raise RuntimeError(f'{elf}: writable section {name} at {addr:#x} runs where it ' | |
| f'loads, in no map region recognized as flash or RAM') | |
| return buckets | |
| class _Sizes: | |
| """Accumulates one elf's filtered per-file flash/RAM, section and symbol sizes, | |
| and its total flash/RAM.""" | |
| def __init__(self, filters): | |
| self.filters, self.files, self.all = filters, {}, {'flash': 0, 'ram': 0} | |
| self.sections, self.symbols = {}, {} | |
| def add(self, path, section, buckets, size, name): | |
| key = _relative_key(path, self.filters) if path else None | |
| for b in buckets: | |
| self.all[b] += size | |
| if key is not None: | |
| self.files.setdefault(key, {'flash': 0, 'ram': 0})[b] += size | |
| if key is not None and buckets: | |
| per_file = self.sections.setdefault(key, {}) | |
| per_file[section] = per_file.get(section, 0) + size | |
| per_section = self.symbols.setdefault(key, {}).setdefault(section, {}) | |
| per_section[name] = per_section.get(name, 0) + size | |
| def result(self): | |
| return {'files': self.files, 'all': self.all, 'sections': self.sections, | |
| 'symbols': self.symbols} | |
| def membrowse_sizes(elf, filters): | |
| """Symbols of `membrowse report`; linker-defined symbols without a section | |
| (`__StackLimit`) are not counted.""" | |
| buckets = section_buckets(elf, elf + '.map') | |
| sizes = _Sizes(filters) | |
| for sym in report_for_elf(elf).get('symbols', []): | |
| section = sym.get('section') | |
| if not sym.get('size') or not section: | |
| continue | |
| if section not in buckets: | |
| raise RuntimeError(f'membrowse symbol {sym.get("name")} is in section {section}, not in {elf}') | |
| sizes.add(sym.get('object_file') or sym.get('source_file'), section, buckets[section], sym['size'], | |
| sym.get('name')) | |
| return sizes.result() | |
| @functools.cache | |
| def _linkermap(): | |
| path = os.path.join(TINYUSB_ROOT, 'tools', 'linkermap', 'linkermap.py') | |
| if not os.path.isfile(path): | |
| raise FileNotFoundError(f'{path} not found - run `python3 tools/get_deps.py tools/linkermap`') | |
| spec = importlib.util.spec_from_file_location('linkermap', path) | |
| module = importlib.util.module_from_spec(spec) | |
| spec.loader.exec_module(module) | |
| return module | |
| def linkermap_sizes(elf, filters): | |
| """Input sections of `<elf>.map` by full object path; analyze_map() is not used, | |
| it keeps only four sections. An archive member | |
| (`lib.a(x.o)`) has no source dir, so it counts in 'all' only. The symbols are the | |
| input sections (`.text.foo`), not the labels inside one.""" | |
| map_path = elf + '.map' | |
| buckets = section_buckets(elf, map_path) | |
| with open(map_path, encoding='utf-8', errors='replace') as f: | |
| sections = _linkermap().parseSections(f) | |
| if sections is None: | |
| raise RuntimeError(f'{map_path}: no Memory Configuration, not a GNU ld map') | |
| sizes = _Sizes(filters) | |
| for out in sections: | |
| if not out.children: # memory regions | |
| continue | |
| if out.section not in buckets: | |
| raise RuntimeError(f'{map_path}: output section {out.section} is not in {elf}') | |
| for obj in out.children: | |
| sizes.add(obj.path[0], out.section, buckets[out.section], obj.size, obj.section) | |
| return sizes.result() | |
| def bloaty_sizes(elf, filters): | |
| """VM size per DWARF compile unit, section and symbol; bytes outside any section | |
| (`[LOAD #0 [RX]]` padding, loaded ELF headers) are not counted.""" | |
| r = subprocess.run(['bloaty', '--csv', '-n', '0', '--domain=vm', | |
| '-d', 'compileunits,sections,symbols', elf], capture_output=True, text=True) | |
| if r.returncode != 0: | |
| raise RuntimeError(f'bloaty failed for {elf}: {r.stderr.strip()}') | |
| rows = csv.DictReader(io.StringIO(r.stdout)) | |
| if not {'compileunits', 'sections', 'symbols', 'vmsize'} <= set(rows.fieldnames or ()): | |
| raise RuntimeError(f'unexpected bloaty csv columns for {elf}: {rows.fieldnames}') | |
| buckets = section_buckets(elf, elf + '.map') | |
| sizes = _Sizes(filters) | |
| for row in rows: | |
| section = row['sections'] | |
| try: | |
| size = int(row['vmsize']) | |
| except (TypeError, ValueError) as e: | |
| raise RuntimeError(f'malformed bloaty csv row for {elf}: {row}') from e | |
| if not size or not section or section.startswith('['): | |
| continue | |
| if section not in buckets: | |
| raise RuntimeError(f'bloaty section {section} is not in {elf}') | |
| unit = row['compileunits'] | |
| sizes.add(None if unit.startswith('[') else unit, section, buckets[section], size, row['symbols']) | |
| return sizes.result() | |
| # all_label: what the engine's 'all' total sums (totals of different engines are | |
| # not comparable); install: how to get its tool when `sizes` raises FileNotFoundError | |
| Engine = collections.namedtuple('Engine', 'sizes all_label install') | |
| ENGINES = { | |
| 'membrowse': Engine(membrowse_sizes, 'all symbols', '`pip install membrowse`'), | |
| 'linkermap': Engine(linkermap_sizes, 'all input sections', | |
| '`python3 tools/get_deps.py tools/linkermap`'), | |
| 'bloaty': Engine(bloaty_sizes, 'all accounted sections', | |
| 'bloaty on PATH (https://github.com/google/bloaty)'), | |
| } | |
| def engine_missing(engine): | |
| """Whether `engine`'s tool is absent; membrowse and bloaty are commands of that name.""" | |
| if engine == 'linkermap': | |
| try: | |
| _linkermap() | |
| except FileNotFoundError: | |
| return True | |
| return False | |
| return not shutil.which(engine) | |
| def _md_escape(text): | |
| """`text` safe in Markdown prose: gcc quotes as `name', which would open a code span.""" | |
| return text.replace('`', '\\`') | |
| def _fmt(delta): | |
| return f'+{delta}' if delta > 0 else str(delta) | |
| def md_table(header, rows, total=None): | |
| """Markdown table padded so its columns also line up as plain text: the first | |
| column left-aligned, the others right-aligned. Cells are strings; a `total` row | |
| follows the rows under a plain rule.""" | |
| body = rows + [total] if total else rows | |
| widths = [max(len(r[i]) for r in [header] + body) for i in range(len(header))] | |
| def line(cells): | |
| return '| ' + ' | '.join(c.ljust(w) if i == 0 else c.rjust(w) | |
| for i, (c, w) in enumerate(zip(cells, widths))) + ' |' | |
| sep = '|' + '|'.join('-' * (w + 2) if i == 0 else '-' * (w + 1) + ':' for i, w in enumerate(widths)) + '|' | |
| # only the header's rule is a delimiter; this one renders as a row, so colons would show as text | |
| rule = sep.replace(':', '-') | |
| return '\n'.join([line(header), sep] + [line(r) for r in rows] + ([rule, line(total)] if total else [])) | |
| def compare_reports(base_by_file, cur_by_file, labels=SIDE_LABELS): | |
| """Markdown per-file delta table; files sorted by |flash delta| desc, TOTAL over all files. | |
| Each region cell is `base โ new`, both values padded so they line up down the column; | |
| `labels` name the two sides in the header.""" | |
| zero = {'flash': 0, 'ram': 0} | |
| header = [f'File ({labels[0]} โ {labels[1]})', 'Flash', 'Flash ฮ', 'RAM', 'RAM ฮ'] | |
| rows = [(path, base_by_file.get(path, zero), cur_by_file.get(path, zero)) | |
| for path in sorted(set(base_by_file) | set(cur_by_file))] | |
| changed = sorted((r for r in rows if any(_delta(r[1], r[2]))), key=lambda r: -abs(_delta(r[1], r[2])[0])) | |
| shown = changed + [('TOTAL', _files_total(base_by_file), _files_total(cur_by_file))] | |
| bw = {k: max(len(str(b[k])) for _, b, _ in shown) for k in zero} | |
| cw = {k: max(len(str(c[k])) for _, _, c in shown) for k in zero} | |
| def cells(b, c): | |
| return [cell for k in zero for cell in (f'{b[k]:>{bw[k]}} โ {c[k]:>{cw[k]}}', _fmt(c[k] - b[k]))] | |
| *table, total_row = [[path, *cells(b, c)] for path, b, c in shown] | |
| if not changed and rows: | |
| table.append(['_no per-file changes_'] + [''] * (len(header) - 1)) | |
| return md_table(header, table, total_row) + '\n' | |
| def _section_deltas(b, c, path): | |
| """{section: ฮ} of one file.""" | |
| bs, cs = b['sections'].get(path, {}), c['sections'].get(path, {}) | |
| return {sec: cs.get(sec, 0) - bs.get(sec, 0) for sec in set(bs) | set(cs)} | |
| def _symbol_deltas(b, c, path): | |
| """{(section, name): ฮ} of one file's changed symbols.""" | |
| bs, cs = b['symbols'].get(path, {}), c['symbols'].get(path, {}) | |
| deltas = {} | |
| for sec in set(bs) | set(cs): | |
| bn, cn = bs.get(sec, {}), cs.get(sec, {}) | |
| for name in set(bn) | set(cn): | |
| if cn.get(name, 0) != bn.get(name, 0): | |
| deltas[(sec, name)] = cn.get(name, 0) - bn.get(name, 0) | |
| return deltas | |
| def _changed_files(b, c, symbols): | |
| """(path, section ฮs, symbol ฮs) of each file whose sections, or with `symbols` | |
| whose symbols, changed; a file's changes can cancel in flash/RAM.""" | |
| rows = [] | |
| for path in sorted(set(b['sections']) | set(c['sections'])): | |
| secs = _section_deltas(b, c, path) | |
| syms = _symbol_deltas(b, c, path) if symbols else {} | |
| if any(secs.values()) or syms: | |
| rows.append((path, secs, syms)) | |
| return rows | |
| def _row_label(symbols, engine): | |
| if not symbols: | |
| return 'File' | |
| return 'File / input section' if engine == 'linkermap' else 'File / symbol' | |
| def size_table(sizes, symbols, engine): | |
| """linkermap's table: a row per file, a column per output section, then size, % of | |
| the filtered total and TOTAL; with `symbols` each file's symbols follow it.""" | |
| by_file = sizes['sections'] | |
| cols = sorted({sec for secs in by_file.values() for sec in secs}, reverse=True) | |
| total = sum(sum(secs.values()) for secs in by_file.values()) | |
| def row(name, secs): | |
| size = sum(secs.values()) | |
| pct = f'{100 * size / total:.1f}%' if total else '-' | |
| return [name] + [str(secs.get(sec, 0)) for sec in cols] + [str(size), pct] | |
| lines = [] | |
| for path, secs in sorted(by_file.items(), key=lambda kv: (-sum(kv[1].values()), kv[0])): | |
| lines.append(row(path, secs)) | |
| if symbols: | |
| syms = [((sec, name), n) for sec, names in sizes['symbols'].get(path, {}).items() | |
| for name, n in names.items()] | |
| for (sec, name), n in sorted(syms, key=lambda kv: (-kv[1], kv[0])): | |
| lines.append(row(f'โ {name}', {sec: n})) | |
| total_row = row('TOTAL', {sec: sum(secs.get(sec, 0) for secs in by_file.values()) for sec in cols}) | |
| return md_table([_row_label(symbols, engine)] + cols + ['size', '%'], lines, total_row) + '\n' | |
| def render_report(sizes, engine, failures=(), boards=(), symbols=False): | |
| """Markdown report of one tree's elfs keyed by (board, elf path), None for a | |
| failed one; `failures` are (elf id, stage, message). One elf is inline, several | |
| get a summary table and each its own table in <details>.""" | |
| sized = {i: s for i, s in sizes.items() if s is not None} | |
| status = 'INCOMPLETE' if failures or not sized else 'complete' | |
| all_label = ENGINES[engine].all_label | |
| lines = [f'**Coverage ({status}, {engine}):** {len(sized)} of {len(sizes)} elfs sized'] | |
| if boards: | |
| lines.append('- boards: ' + ', '.join(f'`{b}`' for b in boards)) | |
| for elf_id, stage, message in failures: | |
| lines.append(f'- FAILED `{_label(elf_id)}` {stage}: {_md_escape(message)}') | |
| lines.append('') | |
| if not sized: | |
| return '\n'.join(lines + ['_no sized elfs_', '']) | |
| def totals(s): | |
| src = _files_total(s['files']) | |
| return (f'filtered Flash {src["flash"]}, RAM {src["ram"]}; ' | |
| f'{all_label} Flash {s["all"]["flash"]}, RAM {s["all"]["ram"]}') | |
| if len(sized) == 1: | |
| (elf_id, s), = sized.items() | |
| return '\n'.join(lines + [f'`{_label(elf_id)}` {totals(s)}', '', size_table(s, symbols, engine)]) | |
| lines.append(md_table(['Elf', 'filtered Flash', 'filtered RAM', f'{all_label} Flash', f'{all_label} RAM'], | |
| [[_label(i)] + [str(t[k]) for t in (_files_total(s['files']), s['all']) for k in ('flash', 'ram')] | |
| for i, s in sized.items()])) | |
| for elf_id, s in sized.items(): | |
| lines += ['', f'<details><summary>{_label(elf_id)}</summary>', '', | |
| f'{totals(s)}', '', size_table(s, symbols, engine), '</details>'] | |
| return '\n'.join(lines) + '\n' | |
| def delta_table(b, c, symbols, engine): | |
| """linkermap's table shape as deltas: a row per changed file, a column per changed | |
| output section, then size ฮ and TOTAL; with `symbols` each file's changed | |
| symbols follow it.""" | |
| rows = _changed_files(b, c, symbols) | |
| if not rows: | |
| return '_no section changes_\n' | |
| cols = sorted({sec for _, secs, _ in rows for sec, d in secs.items() if d} | |
| | {sec for _, _, syms in rows for sec, _ in syms}, reverse=True) | |
| label = _row_label(symbols, engine) | |
| def row(name, deltas): | |
| return [name] + [_fmt(deltas.get(sec, 0)) for sec in cols] + [_fmt(sum(deltas.values()))] | |
| lines = [] | |
| total = {} | |
| for path, secs, syms in sorted(rows, key=lambda r: (-abs(sum(r[1].values())), r[0])): | |
| lines.append(row(path, secs)) | |
| for (sec, name), d in sorted(syms.items(), key=lambda kv: (-abs(kv[1]), kv[0])): | |
| lines.append(row(f'โ {name}', {sec: d})) | |
| for sec, d in secs.items(): | |
| total[sec] = total.get(sec, 0) + d | |
| return md_table([label] + cols + ['size ฮ'], lines, row('TOTAL', total)) + '\n' | |
| def pair_elfs(base, cur): | |
| """Pair two {elf_id: elf_sizes, or None for a failed report} maps. | |
| Returns (pairs, base_only, cur_only): `pairs` maps each id with a report on | |
| both sides to (base, cur); ids on one side only are listed, never paired. | |
| """ | |
| pairs = {i: (base[i], cur[i]) for i in sorted(base.keys() & cur.keys()) | |
| if base[i] is not None and cur[i] is not None} | |
| return pairs, sorted(base.keys() - cur.keys()), sorted(cur.keys() - base.keys()) | |
| def _delta(b, c): | |
| return c['flash'] - b['flash'], c['ram'] - b['ram'] | |
| def _file_deltas(b, c): | |
| """{path: (flash ฮ, RAM ฮ)} over the union of both sides' files.""" | |
| zero = {'flash': 0, 'ram': 0} | |
| return {path: _delta(b['files'].get(path, zero), c['files'].get(path, zero)) | |
| for path in sorted(set(b['files']) | set(c['files']))} | |
| def _files_total(by_file): | |
| return {k: sum(f[k] for f in by_file.values()) for k in ('flash', 'ram')} | |
| def _label(elf_id): | |
| board, elf = elf_id | |
| return f'{board}: {elf}' if elf else board | |
| def _extreme(value_id): | |
| delta, elf_id = value_id | |
| return f'{_fmt(delta)} ({_label(elf_id)})' if delta else '0' | |
| def _file_stats(file_deltas): | |
| """Per file: pairs present/changed and the (ฮ, elf id) extremes of each | |
| metric, the first id winning ties. `file_deltas` maps sorted elf ids to | |
| _file_deltas().""" | |
| entries = {} | |
| for elf_id, deltas in file_deltas.items(): | |
| for path, d in deltas.items(): | |
| entries.setdefault(path, []).append((elf_id, d)) | |
| stats = {} | |
| for path, es in entries.items(): | |
| stats[path] = {'present': len(es), 'changed': sum(any(d) for _, d in es)} | |
| for i, k in enumerate(('flash', 'ram')): | |
| lo = min(es, key=lambda e: e[1][i]) | |
| hi = max(es, key=lambda e: e[1][i]) | |
| stats[path][k] = [(lo[1][i], lo[0]), (hi[1][i], hi[0])] | |
| return stats | |
| def _pair_tables(b, c, symbols, engine, labels): | |
| """A pair's Flash/RAM file table and its section delta table.""" | |
| return compare_reports(b['files'], c['files'], labels), delta_table(b, c, symbols, engine) | |
| def _pair_changed(b, c, symbols): | |
| return (any(_delta(b['all'], c['all'])) or any(any(d) for d in _file_deltas(b, c).values()) | |
| or bool(_changed_files(b, c, symbols))) | |
| def render_pairs(pairs, matched, engine, base_only=(), cur_only=(), failures=(), boards=(), symbols=False, | |
| labels=SIDE_LABELS): | |
| """Markdown report over paired elfs keyed by (board, elf path). | |
| Every statistic is over per-pair deltas, sized by `engine`. `matched` counts | |
| ids built on both sides, compared or not; | |
| `failures` are (elf id, side, stage, message), the elf path None for a | |
| board-level failure. Failures or unmatched elfs mark the report INCOMPLETE. | |
| `boards` lists the requested boards, so an unchanged or failed one is named. | |
| `symbols` adds each file's changed symbols to each pair's section table; `labels` | |
| name the base and current sides. | |
| """ | |
| file_deltas = {i: _file_deltas(b, c) for i, (b, c) in pairs.items()} | |
| changed = [i for i, (b, c) in pairs.items() if _pair_changed(b, c, symbols)] | |
| status = _status(failures, base_only, cur_only, pairs) | |
| all_label = ENGINES[engine].all_label | |
| lines = [f'**Coverage ({status}, {engine}):** {len(pairs)} of {matched} matched elf pairs ' | |
| f'compared, {len(changed)} changed'] | |
| if boards: | |
| lines.append('- boards: ' + ', '.join(f'`{b}`' for b in boards)) | |
| for label, ids in (('base-only', base_only), ('current-only', cur_only)): | |
| if ids: | |
| lines.append(f'- {label}: ' + ', '.join(f'`{_label(i)}`' for i in ids)) | |
| for elf_id, side, stage, message in failures: | |
| lines.append(f'- FAILED `{_label(elf_id)}` {side} {stage}: {_md_escape(message)}') | |
| lines.append('') | |
| if len(pairs) == 1: | |
| (elf_id, (b, c)), = pairs.items() | |
| df, dr = _delta(b['all'], c['all']) | |
| lines += [f'`{_label(elf_id)}` {all_label}: Flash ฮ {_fmt(df)}, RAM ฮ {_fmt(dr)}', '', | |
| *_pair_tables(b, c, symbols, engine, labels)] | |
| return '\n'.join(lines) | |
| if not pairs: | |
| return '\n'.join(lines + ['_no comparable pairs_', '']) | |
| if not changed: | |
| return '\n'.join(lines + ['_no changes_', '']) | |
| pair_rows = [] | |
| for elf_id in changed: | |
| b, c = pairs[elf_id] | |
| src = _delta(_files_total(b['files']), _files_total(c['files'])) | |
| syms = _delta(b['all'], c['all']) | |
| pair_rows.append([_label(elf_id)] + [_fmt(d) for d in src + syms]) | |
| lines.append(md_table(['Pair', 'filtered Flash ฮ', 'filtered RAM ฮ', f'{all_label} Flash ฮ', | |
| f'{all_label} RAM ฮ'], pair_rows)) | |
| stats = _file_stats(file_deltas) | |
| rows = sorted((p for p, s in stats.items() if s['changed']), | |
| key=lambda p: (-max(abs(v[0]) for k in ('flash', 'ram') for v in stats[p][k]), p)) | |
| lines += ['', '_Changed / present: pairs where the file\'s Flash or RAM total changed / compared pairs ' | |
| 'that contain it._', '', | |
| md_table(['File', 'Changed / present', 'Flash ฮ min', 'Flash ฮ max', 'RAM ฮ min', 'RAM ฮ max'], | |
| [[path, f'{stats[path]["changed"]}/{stats[path]["present"]}'] | |
| + [_extreme(v) for k in ('flash', 'ram') for v in stats[path][k]] for path in rows])] | |
| for elf_id in changed: | |
| b, c = pairs[elf_id] | |
| lines += ['', f'<details><summary>{_label(elf_id)}</summary>', '', | |
| *_pair_tables(b, c, symbols, engine, labels), '</details>'] | |
| return '\n'.join(lines) + '\n' | |
| def _status(failures, base_only, cur_only, pairs): | |
| return 'INCOMPLETE' if failures or base_only or cur_only or not pairs else 'complete' | |
| def _elf_record(elf_id): | |
| board, elf = elf_id | |
| return {'board': board, 'elf': elf} | |
| def _json_sizes(sizes, symbols): | |
| """An elf's sizes for JSON, symbols included only when `symbols` is set.""" | |
| return sizes if symbols else {k: v for k, v in sizes.items() if k != 'symbols'} | |
| def _filter_failure(engine): | |
| return (f'no {engine} sizes matched filters - check them, or a change in ' | |
| f'{engine} output broke its parsing (try another --engine to isolate)') | |
| def compare_sides(base, cur, engine, failures=(), boards=(), scope=None, symbols=False, labels=SIDE_LABELS): | |
| """Pair two sides and render them. Returns (md, failures, ok, data). | |
| `failures` comes back with a filter failure for `scope` added when no | |
| compared pair matched a file (wrong filters, or an engine output change | |
| broke its parsing); pass `scope=None` when each scope was already checked. | |
| `ok` is False on any failure or when nothing was compared. `data` is the | |
| report's raw paired sizes, for JSON, symbols included only when `symbols` is set. | |
| """ | |
| pairs, base_only, cur_only = pair_elfs(base, cur) | |
| failures = list(failures) | |
| if scope and pairs and not any(b['files'] or c['files'] for b, c in pairs.values()): | |
| failures.append((scope, 'both', 'filter', _filter_failure(engine))) | |
| md = render_pairs(pairs, len(base.keys() & cur.keys()), engine, base_only, cur_only, failures, boards, | |
| symbols, labels) | |
| data = { | |
| 'engine': engine, | |
| 'boards': list(boards), | |
| 'status': _status(failures, base_only, cur_only, pairs), | |
| 'pairs': [{**_elf_record(i), 'base': _json_sizes(b, symbols), 'current': _json_sizes(c, symbols)} | |
| for i, (b, c) in pairs.items()], | |
| 'base_only': [_elf_record(i) for i in base_only], | |
| 'current_only': [_elf_record(i) for i in cur_only], | |
| 'failures': [{**_elf_record(i), 'side': side, 'stage': stage, 'message': message} | |
| for i, side, stage, message in failures], | |
| } | |
| return md, failures, bool(pairs) and not failures, data | |
| def tinyusb_src_filter(checkout_dir): | |
| """Return a path-substring filter that uniquely matches TinyUSB stack source files | |
| in `checkout_dir`. The substring is the absolute path to the checkout's `src/` | |
| dir โ collision-free with vendored deps (pico-sdk, lwip, FreeRTOS, etc.) which | |
| live at unrelated paths.""" | |
| return os.path.realpath(os.path.join(checkout_dir, 'src')) + os.sep | |
| verbose = False | |
| TERMINATE_GRACE = 10 # s a timed-out command gets to exit on SIGTERM before SIGKILL | |
| KILL_DRAIN = 1 # s to read what a killed command left in its pipes | |
| def run(cmd, timeout=None): | |
| """Run a command. cmd must be a list (no shell=True). On `timeout`, SIGTERM the | |
| command, SIGKILL it if it outlives a grace period, and return a CompletedProcess | |
| with rc=124 instead of raising TimeoutExpired, so the caller can fall through to | |
| error reporting and worktree cleanup rather than crashing with a traceback. Any | |
| other exception, e.g. exit_on_termination()'s exit, stops the command the same way | |
| and is re-raised: Popen's exit would otherwise wait for it.""" | |
| if not isinstance(cmd, list): | |
| raise TypeError('run() requires a list, got str โ fix the caller') | |
| if verbose: | |
| print(f' $ {" ".join(shlex.quote(str(c)) for c in cmd)}') | |
| # the command stays in our process group, so Ctrl-C or a hangup reaches it directly | |
| with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, | |
| encoding='utf-8', errors='replace') as proc: | |
| try: | |
| out, err = proc.communicate(timeout=timeout) | |
| except BaseException as stopped: | |
| # SIGTERM lets ninja stop its jobs, which it runs in process groups of their own; | |
| # a descendant can outlive the SIGKILL holding the pipes open, so stop reading then | |
| for stop, wait in ((proc.terminate, TERMINATE_GRACE), (proc.kill, KILL_DRAIN)): | |
| stop() | |
| try: | |
| out, err = proc.communicate(timeout=wait) | |
| break | |
| except subprocess.TimeoutExpired as e: | |
| out, err = ((b or b'').decode(errors='replace') for b in (e.stdout, e.stderr)) | |
| if not isinstance(stopped, subprocess.TimeoutExpired): | |
| raise | |
| msg = f'Command timed out after {timeout}s: {" ".join(shlex.quote(str(c)) for c in cmd)}' | |
| return subprocess.CompletedProcess(cmd, 124, stdout=out, stderr=err + ('\n' if err else '') + msg) | |
| return subprocess.CompletedProcess(cmd, proc.returncode, stdout=out, stderr=err) | |
| def exit_on_termination(): | |
| """Exit on a SIGTERM or SIGHUP sent to this process alone as on Ctrl-C: through run()'s | |
| stop of its command and main()'s worktree removal, instead of orphaning the build.""" | |
| for signum in (signal.SIGTERM, signal.SIGHUP): | |
| signal.signal(signum, lambda sig, _frame: sys.exit(128 + sig)) | |
| def symlink_deps(main_root, worktree_dir): | |
| """Symlink each dependency the worktree's own tools/get_deps.py lists, when fetched in | |
| the main checkout: the worktree lacks these untracked dirs, and a base revision can | |
| name paths the current manifest has renamed or dropped.""" | |
| manifest = runpy.run_path(os.path.join(worktree_dir, 'tools', 'get_deps.py')) | |
| for rel in manifest['deps_all']: | |
| src = os.path.join(main_root, rel) | |
| dst = os.path.join(worktree_dir, rel) | |
| if os.path.isdir(src) and not os.path.lexists(dst): | |
| os.makedirs(os.path.dirname(dst), exist_ok=True) | |
| os.symlink(src, dst) | |
| def short_hash(checkout): | |
| """`checkout`'s short HEAD hash, `-dirty` when tracked files differ from it; None | |
| when git cannot tell. `--exclude='*'` keeps tag names out.""" | |
| ret = run(['git', '-C', checkout, 'describe', '--always', '--dirty', '--exclude=*']) | |
| return ret.stdout.strip() or None if ret.returncode == 0 else None | |
| def invalid_boards(boards): | |
| """Boards that are not a plain name: a board names a cmake-code-size/<board> dir that | |
| gets removed before the build can reject it, so `..` or a path must not pass.""" | |
| return [b for b in boards if not re.fullmatch(r'[A-Za-z0-9_-]+', b)] | |
| def ci_pinned_boards(): | |
| """Boards of .github/ci-pinned-boards.json: CI's membrowse set, which covers every | |
| dcd/hcd driver not waived in its `uncovered` list (drivers-coverage hook).""" | |
| with open(CI_PINNED_BOARDS) as f: | |
| return [entry['board'] for entry in json.load(f)['boards']] | |
| class Phase: | |
| """A progress line ` labelโฆ 1.2s`, or `FAILED after 1.2s`; with -v the commands | |
| print between the label and its time, so the time gets a line of its own.""" | |
| def __init__(self, label): | |
| self.label, self.start = label, time.monotonic() | |
| print(f' {label}โฆ', end='\n' if verbose else ' ', flush=True) | |
| def done(self, failed=False): | |
| took = f'{"FAILED after " if failed else ""}{time.monotonic() - self.start:.1f}s' | |
| print(f' {self.label}: {took}' if verbose else took) | |
| _NINJA_PROGRESS = re.compile(r'\[\d+/\d+\] ') | |
| def output_excerpt(ret, lines=20): | |
| """Up to `lines` of each nonempty stream of a failed command: from ninja's first | |
| `FAILED:` block when there is one, else the stream's tail. ninja reports a compile | |
| error on stdout, leaving stderr empty, and the jobs already running when it failed | |
| finish after it, so their `[n/m]` progress lines are dropped.""" | |
| def excerpt(stream): | |
| kept = [line for line in stream.rstrip().splitlines() if not _NINJA_PROGRESS.match(line)] | |
| first = next((i for i, line in enumerate(kept) if line.startswith('FAILED:')), None) | |
| return kept[-lines:] if first is None else kept[first:first + lines] | |
| return [line for stream in (ret.stdout, ret.stderr) if stream.strip() for line in excerpt(stream)] | |
| def build_error(ret, src_dir): | |
| """A failed build's first error for the report, from its whole output: the first | |
| diagnostic, else its last line that is not ninja's progress, `FAILED:` or stop line | |
| (its own `ninja: error:` stays), with paths relative to `src_dir`.""" | |
| out = '\n'.join(stream for stream in (ret.stdout, ret.stderr) if stream) | |
| lines = [line.strip() for line in out.splitlines() | |
| if line.strip() and not _NINJA_PROGRESS.match(line) | |
| and not line.startswith(('ninja: build stopped', 'ninja: Entering directory', 'FAILED: '))] | |
| error = build_utils.first_error(out) or (lines[-1] if lines else 'no output') | |
| return error.replace(src_dir.rstrip(os.sep) + os.sep, '') | |
| def build_board(src_dir, build_dir, board, example, label): | |
| """Configure and build examples for a board as a `label` progress phase, printing | |
| an excerpt of the output on failure. Returns None on success, else build_error(). | |
| When `example` is given, only that target is built (`ninja -C DIR NAME`), | |
| keeping single-example workflows fast. | |
| """ | |
| phase = Phase(label) | |
| os.makedirs(build_dir, exist_ok=True) | |
| ret = run(['cmake', '-B', build_dir, '-G', 'Ninja', | |
| f'-DBOARD={board}', '-DCMAKE_BUILD_TYPE=MinSizeRel', | |
| os.path.join(src_dir, 'examples')]) | |
| if ret.returncode == 0: | |
| # ninja itself, not `cmake --build`: cmake does not pass a timeout's SIGTERM on | |
| cmd = ['ninja', '-C', build_dir] | |
| if example: | |
| cmd.append(os.path.basename(example)) | |
| ret = run(cmd, timeout=600) | |
| failed = ret.returncode != 0 | |
| phase.done(failed=failed) | |
| if not failed: | |
| return None | |
| print('\n'.join(' ' + line for line in output_excerpt(ret))) | |
| return build_error(ret, src_dir) | |
| def report_path(board, example, kind='diff'): | |
| """A scope's report path without extension: cmake-code-size/<board>/<kind>[_<ex>].""" | |
| suffix = f'_{example.replace("/", "_")}' if example else '' | |
| return os.path.join(CODE_SIZE_DIR, board, f'{kind}{suffix}') | |
| def drop_stale_reports(boards, examples, kind): | |
| """Remove the reports a run will write before anything can fail: a run that stops | |
| early must not leave a previous run's report for a reader to take for this one's, | |
| since cmake-code-size/ is gitignored and persists. .json too: a run without --json | |
| must not leave an older one beside a newer .md.""" | |
| for board in boards: | |
| for example in examples: | |
| for ext in ('md', 'json'): | |
| stale_path = f'{report_path(board, example, kind)}.{ext}' | |
| if os.path.isfile(stale_path): | |
| os.remove(stale_path) | |
| def generate_sizes(build_dir, filters, example=None, engine='membrowse'): | |
| """Return (sizes, errors) for the scope's elfs. | |
| `sizes` maps each elf path relative to `build_dir` to its | |
| ENGINES[engine] sizes, or to None when that failed; `errors` | |
| lists (relative elf path, message), the path None when no elf can be | |
| sized at all. | |
| """ | |
| # escape the dir, not the wildcards: a checkout path is a path, not a pattern | |
| root = glob.escape(build_dir) | |
| # <role>/<example>/*.elf: deeper elfs are helpers, e.g. pico-sdk's bs2_default.elf | |
| elfs = sorted(glob.glob(f'{root}/{example or "*/*"}/*.elf')) | |
| if not elfs: | |
| return {}, [(None, f'no .elf files in {build_dir}')] | |
| sizer = ENGINES[engine].sizes | |
| def report(elf): | |
| try: | |
| return sizer(elf, filters), None | |
| except RuntimeError as e: | |
| return None, str(e) | |
| # report errors are caught here, not as tracebacks after both builds already ran | |
| try: | |
| with concurrent.futures.ThreadPoolExecutor() as pool: | |
| results = list(pool.map(report, elfs)) | |
| except FileNotFoundError as e: | |
| return {}, [(None, f'{engine} not found ({e}) - install {ENGINES[engine].install}, ' | |
| f'or pick another --engine')] | |
| sizes, errors = {}, [] | |
| for elf, (elf_sizes, error) in zip(elfs, results): | |
| rel = os.path.relpath(elf, build_dir) | |
| sizes[rel] = elf_sizes | |
| if error: | |
| errors.append((rel, error)) | |
| return sizes, errors | |
| def diff_summary(data, symbols): | |
| """Console lines of one diff from its JSON-shaped `data`: coverage, the single | |
| pair's filtered ฮ, unmatched elfs and failures.""" | |
| pairs = {(p['board'], p['elf']): (p['base'], p['current']) for p in data['pairs']} | |
| changed = sum(_pair_changed(b, c, symbols) for b, c in pairs.values()) | |
| line = f'{len(pairs)} pair{"" if len(pairs) == 1 else "s"}, {changed} changed' | |
| if len(pairs) == 1: | |
| (b, c), = pairs.values() | |
| df, dr = _delta(_files_total(b['files']), _files_total(c['files'])) | |
| line += f'; filtered Flash ฮ {_fmt(df)}, RAM ฮ {_fmt(dr)}' | |
| lines = [line if data['status'] == 'complete' else f'INCOMPLETE: {line}'] | |
| for key in ('base_only', 'current_only'): | |
| if data[key]: | |
| lines.append(f'{key.replace("_", "-")}: ' + ', '.join(_label((r['board'], r['elf'])) for r in data[key])) | |
| lines += [f'FAILED {_label((f["board"], f["elf"]))} {f["side"]} {f["stage"]}: {f["message"]}' | |
| for f in data['failures']] | |
| return lines | |
| def report_summary(sizes, failures, ok): | |
| """Console lines of one report: coverage, the single elf's filtered total, failures.""" | |
| sized = [s for s in sizes.values() if s is not None] | |
| line = f'{len(sized)} of {len(sizes)} elfs sized' | |
| if len(sized) == 1: | |
| src = _files_total(sized[0]['files']) | |
| line += f'; filtered Flash {src["flash"]}, RAM {src["ram"]}' | |
| lines = [line if ok else f'INCOMPLETE: {line}'] | |
| return lines + [f'FAILED {_label(i)} {stage}: {message}' for i, stage, message in failures] | |
| def print_result(lines, prefix='', tables=()): | |
| """A scope's result lines, then `tables` (Markdown, or an elf's label heading its | |
| tables), indented under its phases.""" | |
| for i, line in enumerate(lines): | |
| print(f' {prefix}{line}' if i == 0 else f' {line}') | |
| for table in tables: | |
| print('\n' + '\n'.join(f' {row}' for row in table.rstrip('\n').split('\n'))) | |
| if tables: | |
| print() | |
| def _shown(path): | |
| """`path` relative to the working directory when under it.""" | |
| rel = os.path.relpath(path) | |
| return path if rel.startswith('..') else rel | |
| def write_report(path, md, data=None): | |
| """Write a report to `path`.md, and `data` to `path`.json when given, printing | |
| their paths.""" | |
| with open(f'{path}.md', 'w') as f: | |
| f.write(md) | |
| print(f' report: {_shown(path)}.md') | |
| if data is not None: | |
| with open(f'{path}.json', 'w') as f: | |
| json.dump(data, f, indent=1, sort_keys=True) | |
| f.write('\n') | |
| print(f' json: {_shown(path)}.json') | |
| def _focused(boards, examples): | |
| """One board and one example: the console also gets that scope's tables.""" | |
| return len(boards) == 1 and bool(_single_example(examples)) | |
| def _labelled(tables_by_elf, elfs): | |
| """Console tables of {elf id: tables}, each elf's under its label when the scope | |
| has several `elfs`, shown or not.""" | |
| many = elfs > 1 | |
| return [t for elf_id, tables in tables_by_elf.items() | |
| for t in ([f'{_label(elf_id)}:'] if many else []) + list(tables)] | |
| def _single_example(examples): | |
| """The board header's ` / <example>` when one example is sized.""" | |
| return f' / {examples[0]}' if len(examples) == 1 and examples[0] else '' | |
| def _scope_label(examples, example): | |
| """A phase label's ` <example>` suffix, needed only when a board has several.""" | |
| return f' {example}' if example and len(examples) > 1 else '' | |
| def _build_failed(example, error): | |
| return f'build failed{f" ({example})" if example else ""}: {error}' | |
| def run_report(args): | |
| """`report`: build and size the working tree, one report per board and example.""" | |
| filters = args.filter or [tinyusb_src_filter(TINYUSB_ROOT)] | |
| examples = args.example or [None] | |
| drop_stale_reports(args.board, examples, 'report') | |
| focused = _focused(args.board, examples) | |
| print(f'report working tree ยท {args.engine}') | |
| failed = False | |
| for n, board in enumerate(args.board, 1): | |
| print(f'[{n}/{len(args.board)}] {board}{_single_example(examples)}') | |
| build = os.path.join(CODE_SIZE_DIR, board, 'build') | |
| shutil.rmtree(build, ignore_errors=True) | |
| # each scope is its own report, so one example's build failure spares the others | |
| for example in examples: | |
| scope = _scope_label(examples, example) | |
| sizes, failures = {}, [] | |
| error = build_board(TINYUSB_ROOT, build, board, example, f'build{scope}') | |
| if error: | |
| failures.append(((board, None), 'build', _build_failed(example, error))) | |
| else: | |
| phase = Phase(f'size{scope}') | |
| rel_sizes, errors = generate_sizes(build, filters, example, args.engine) | |
| sizes = {(board, rel): v for rel, v in rel_sizes.items()} | |
| failures += [((board, rel), 'report', msg) for rel, msg in errors] | |
| # per scope, as diff: an example need not link TinyUSB (board_test) | |
| sized = [s for s in sizes.values() if s is not None] | |
| if sized and not any(s['files'] for s in sized): | |
| failures.append(((board, None), 'filter', _filter_failure(args.engine))) | |
| phase.done(failed=bool(failures)) | |
| md = render_report(sizes, args.engine, failures, [board], args.symbols) | |
| ok = not failures and any(s is not None for s in sizes.values()) | |
| failed |= not ok | |
| data = None | |
| if args.json: | |
| data = {'engine': args.engine, 'boards': [board], 'filters': filters, | |
| 'status': 'complete' if ok else 'INCOMPLETE', | |
| 'elfs': [{**_elf_record(i), 'sizes': _json_sizes(s, args.symbols)} | |
| for i, s in sizes.items() if s is not None], | |
| 'failures': [{**_elf_record(i), 'stage': stage, 'message': message} | |
| for i, stage, message in failures]} | |
| tables = _labelled({i: [size_table(s, args.symbols, args.engine)] for i, s in sizes.items() if s}, | |
| len(sizes)) if focused else () | |
| print_result(report_summary(sizes, failures, ok), f'{example}: ' if scope else '', tables) | |
| write_report(report_path(board, example, 'report'), md, data) | |
| return 1 if failed else 0 | |
| def example_arg(value): | |
| example = value.rstrip('/') | |
| if not example: | |
| raise argparse.ArgumentTypeError(f'{value!r} names no example') | |
| return example | |
| def main(): | |
| global verbose | |
| common = argparse.ArgumentParser(add_help=False) | |
| common.add_argument('-b', '--board', action='append', default=[], | |
| help='Board name (repeatable). Required unless diff --ci is given.') | |
| common.add_argument('-f', '--filter', action='append', default=None, | |
| help='Path-substring filter (repeatable). When given, ' | |
| 'overrides the default and is applied to every build. ' | |
| 'Default: each build\'s own absolute <checkout>/src/ path, ' | |
| 'which uniquely matches TinyUSB stack code without colliding ' | |
| 'with vendored deps.') | |
| common.add_argument('-e', '--example', action='append', default=None, type=example_arg, | |
| help='Size specific example (repeatable, e.g. -e device/cdc_msc -e host/cdc_msc_hid)') | |
| common.add_argument('--engine', choices=sorted(ENGINES), | |
| default='membrowse', | |
| help='Where per-file sizes come from: membrowse (default, symbols with ' | |
| 'ELF-header flash/RAM), linkermap (the GNU ld map\'s input ' | |
| 'sections) or bloaty (DWARF compile units)') | |
| common.add_argument('--symbols', action='store_true', | |
| help='Show each file\'s symbols under it (linkermap: its input sections); ' | |
| 'with --json, include them in the JSON') | |
| common.add_argument('--json', action='store_true', | |
| help='Also write each report\'s sizes as .json next to its .md') | |
| common.add_argument('-v', '--verbose', action='store_true', | |
| help='Print build commands') | |
| top = argparse.ArgumentParser(description='Code size of TinyUSB examples') | |
| sub = top.add_subparsers(dest='command', required=True) | |
| report_parser = sub.add_parser('report', parents=[common], | |
| help='size the working tree (cmake-code-size/<board>/report*.md)') | |
| parser = sub.add_parser('diff', parents=[common], | |
| help='diff code size against a base ref (cmake-code-size/<board>/diff*.md)') | |
| parser.add_argument('--base-branch', default='master', | |
| help='Base branch to compare against (default: master)') | |
| parser.add_argument('--bloaty', action='store_true', | |
| help='Also print bloaty\'s section and symbol diff of each -e example ' | |
| '(console only, whatever --engine)') | |
| parser.add_argument('--ci', action='store_true', | |
| help='Add the CI-pinned boards (.github/ci-pinned-boards.json, covering ' | |
| 'every dcd/hcd driver not waived there). Implies --combined.') | |
| parser.add_argument('--combined', action='store_true', | |
| help='Also write one comparison over every board ' | |
| '(cmake-code-size/_combined/diff.md), in addition to per-board.') | |
| args = top.parse_args() | |
| verbose = args.verbose | |
| if engine_missing(args.engine): | |
| sub.choices[args.command].error(f'{args.engine} not found - install {ENGINES[args.engine].install}, ' | |
| f'or pick another --engine') | |
| if args.command == 'report': | |
| if not args.board: | |
| report_parser.error('at least one -b BOARD is required') | |
| if invalid := invalid_boards(args.board): | |
| report_parser.error(f'invalid board name: {", ".join(invalid)}') | |
| return run_report(args) | |
| if args.bloaty and not args.example: | |
| parser.error('--bloaty requires -e/--example') | |
| if args.bloaty and not shutil.which('bloaty'): | |
| parser.error('--bloaty requires bloaty on PATH') | |
| if args.ci: | |
| args.combined = True | |
| args.board = list(dict.fromkeys(args.board + ci_pinned_boards())) | |
| if not args.board: | |
| parser.error('at least one -b BOARD is required (or pass --ci)') | |
| if invalid := invalid_boards(args.board): | |
| parser.error(f'invalid board name: {", ".join(invalid)}') | |
| worktree_dir = os.path.join(CODE_SIZE_DIR, '_worktree') | |
| # Per-side filters: when no override is given, each build uses its own | |
| # absolute <checkout>/src/ path so we only match TinyUSB stack code from that | |
| # checkout (and never vendored-dep `src/` like pico-sdk/src/...). | |
| if args.filter: | |
| base_filters = cur_filters = list(args.filter) | |
| else: | |
| base_filters = [tinyusb_src_filter(worktree_dir)] | |
| cur_filters = [tinyusb_src_filter(TINYUSB_ROOT)] | |
| examples = args.example or [None] | |
| combined_dir = os.path.join(CODE_SIZE_DIR, '_combined') | |
| if args.combined: | |
| shutil.rmtree(combined_dir, ignore_errors=True) | |
| drop_stale_reports(args.board, examples, 'diff') | |
| if os.path.isdir(worktree_dir): | |
| # twice: a killed `worktree add` leaves it locked `initializing` | |
| run(['git', '-C', TINYUSB_ROOT, 'worktree', 'remove', '--force', '--force', worktree_dir]) | |
| failed = False | |
| try: | |
| # --detach: check out the ref at a detached HEAD instead of trying to claim the | |
| # branch. Lets us add a worktree of `master` even if master is already checked | |
| # out elsewhere (main repo, another worktree). | |
| ret = run(['git', '-C', TINYUSB_ROOT, 'worktree', 'add', '--detach', | |
| worktree_dir, args.base_branch]) | |
| if ret.returncode != 0: | |
| print(f'Error creating worktree: {ret.stderr}') | |
| sys.exit(1) | |
| symlink_deps(TINYUSB_ROOT, worktree_dir) | |
| # the commit actually built, which the ref may no longer name later | |
| base_sha = run(['git', '-C', worktree_dir, 'rev-parse', 'HEAD']).stdout.strip() | |
| current_rev = short_hash(TINYUSB_ROOT) | |
| labels = (short_hash(worktree_dir) or SIDE_LABELS[0], current_rev or SIDE_LABELS[1]) | |
| print(f'diff {args.base_branch} ({labels[0]}) vs working tree ({labels[1]}) ยท {args.engine}') | |
| focused = _focused(args.board, examples) | |
| def report_data(data): | |
| """The JSON report for --json, None without it.""" | |
| if not args.json: | |
| return None | |
| return {**data, 'base_ref': args.base_branch, 'base_sha': base_sha, 'current_rev': current_rev, | |
| 'filters': {'base': base_filters, 'current': cur_filters}} | |
| # --combined: every board's elf sizes and failures, paired at the end | |
| combined_sides = {'base': {}, 'current': {}} | |
| combined_failures = [] | |
| for n, board in enumerate(args.board, 1): | |
| print(f'[{n}/{len(args.board)}] {board}{_single_example(examples)}') | |
| board_dir = os.path.join(CODE_SIZE_DIR, board) | |
| base_build = os.path.join(board_dir, 'base') | |
| cur_build = os.path.join(board_dir, 'build') | |
| shutil.rmtree(base_build, ignore_errors=True) | |
| shutil.rmtree(cur_build, ignore_errors=True) | |
| build_failure = None | |
| for example in examples: | |
| scope = _scope_label(examples, example) | |
| for side, src, build, label in (('base', worktree_dir, base_build, f'build {args.base_branch}{scope}'), | |
| ('current', TINYUSB_ROOT, cur_build, f'build current{scope}')): | |
| error = build_board(src, build, board, example, label) | |
| if error: | |
| build_failure = ((board, None), side, 'build', _build_failed(example, error)) | |
| break | |
| if build_failure: | |
| break | |
| if build_failure: | |
| failed = True | |
| # still write each scope's report, with the build failure in it | |
| combined_failures.append(build_failure) | |
| for example in examples: | |
| scope = _scope_label(examples, example) | |
| sides = {'base': {}, 'current': {}} | |
| failures = [build_failure] if build_failure else [] | |
| if not build_failure: | |
| phase = Phase(f'size and compare{scope}') | |
| for side, build, filters in (('base', base_build, base_filters), | |
| ('current', cur_build, cur_filters)): | |
| sizes, errors = generate_sizes(build, filters, example, args.engine) | |
| sides[side] = {(board, rel): v for rel, v in sizes.items()} | |
| failures += [((board, rel), side, 'report', msg) for rel, msg in errors] | |
| md, failures, ok, data = compare_sides( | |
| sides['base'], sides['current'], args.engine, failures, [board], scope=(board, None), | |
| symbols=args.symbols, labels=labels) | |
| if not build_failure: | |
| phase.done(failed=not ok) | |
| failed |= not ok | |
| if not build_failure: # a build failure is recorded once, above | |
| for side, sizes in sides.items(): | |
| combined_sides[side].update(sizes) | |
| combined_failures += failures | |
| tables = _labelled({(p['board'], p['elf']): _pair_tables(p['base'], p['current'], args.symbols, | |
| args.engine, labels) | |
| for p in data['pairs'] if _pair_changed(p['base'], p['current'], args.symbols)}, | |
| len(data['pairs'])) if focused else () | |
| print_result(diff_summary(data, args.symbols), f'{example}: ' if scope else '', tables) | |
| write_report(report_path(board, example), md, report_data(data)) | |
| if args.bloaty and example and not build_failure: | |
| elf_name = os.path.basename(example) | |
| base_elf = os.path.join(base_build, example, f'{elf_name}.elf') | |
| cur_elf = os.path.join(cur_build, example, f'{elf_name}.elf') | |
| if os.path.exists(base_elf) and os.path.exists(cur_elf): | |
| # Bloaty expects one regex; OR-join all filters (current side | |
| # for the new ELF, base side for the base ELF). | |
| bloaty_regex = '(' + '|'.join( | |
| re.escape(f) for f in (cur_filters + base_filters) | |
| ) + ')' | |
| bloaty_common = ['bloaty', '--domain=vm', f'--source-filter={bloaty_regex}'] | |
| for title, options in (('sections', ['-d', 'compileunits,sections']), | |
| ('symbols', ['-d', 'compileunits,symbols', '-s', 'vm'])): | |
| print(f'--- bloaty {title} ---') | |
| ret = run(bloaty_common + options + [cur_elf, '--', base_elf]) | |
| if ret.returncode == 0: | |
| print(ret.stdout) | |
| else: | |
| print(f' bloaty FAILED (exit {ret.returncode})') | |
| print('\n'.join(' ' + line for line in output_excerpt(ret))) | |
| failed = True | |
| else: | |
| print(' bloaty: ELF not found') | |
| if args.combined: | |
| os.makedirs(combined_dir, exist_ok=True) | |
| print(f'combined ({len(args.board)} boards)') | |
| # every scope was filter-checked above, and its failures carried over | |
| md, _failures, ok, data = compare_sides( | |
| combined_sides['base'], combined_sides['current'], args.engine, | |
| combined_failures, args.board, symbols=args.symbols, labels=labels) | |
| failed |= not ok | |
| print_result(diff_summary(data, args.symbols)) | |
| write_report(os.path.join(combined_dir, 'diff'), md, report_data(data)) | |
| finally: | |
| # an add stopped by a signal leaves it too, locked `initializing` | |
| ret = run(['git', '-C', TINYUSB_ROOT, 'worktree', 'remove', '--force', '--force', worktree_dir]) \ | |
| if os.path.isdir(worktree_dir) else None | |
| if ret and ret.returncode != 0: | |
| print(f'Error removing worktree {worktree_dir}: {ret.stderr.strip()}') | |
| failed = True | |
| return 1 if failed else 0 | |
| if __name__ == '__main__': | |
| exit_on_termination() | |
| sys.exit(main()) |