Google GitGoogle Git
Sign in
pigweed/third_party/github/hathach/tinyusb/4c195b7b94f813a37b77b184bcae7b2b6ef2835e/./tools/code_size.py
blob: f09136e6e241f21776b7a015d6a8261eb3078ff8 [file]
#!/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 current tree in cmake-code-size/<board>/build and takes the base branch's
(master) sizes from CI's stored snapshots, else builds it in cmake-code-size/<board>/base
(--base-source); it pairs the 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 contextlib
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 (IDF_LD_SCRIPTS, NINJA_UTF8_REMEDY, 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')
WINDOWS = os.name == 'nt'
# a master run's snapshots are the same for every checkout; XDG ignores a relative XDG_CACHE_HOME
_cache_home = os.environ.get('LOCALAPPDATA' if WINDOWS else 'XDG_CACHE_HOME', '')
BASELINE_CACHE_DIR = os.path.join(_cache_home if os.path.isabs(_cache_home) else os.path.expanduser('~/.cache'),
'tinyusb', 'code-size-baseline')
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)
# ESP-IDF links its generated scripts by bare name via -L
idf_ld = [os.path.join(build_dir, p) for p in IDF_LD_SCRIPTS]
if any(map(os.path.isfile, idf_ld)):
if not all(map(os.path.isfile, idf_ld)):
raise RuntimeError(f'ESP-IDF build {build_dir} lacks one of {", ".join(idf_ld)}')
return build_dir, idf_ld, extract_defsyms(commands)
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."""
if WINDOWS:
src = src.replace('\\', '/')
for f in filters:
# Windows paths are case-insensitive; the key keeps the recorded case
m = re.search(re.escape(f), src, re.IGNORECASE if WINDOWS else 0)
if not m:
continue
key = src[m.end():]
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_path(elf):
"""The elf's GNU ld map: `<elf>.map`, or ESP-IDF's `<name>.map` beside it."""
idf = os.path.splitext(elf)[0] + '.map'
return idf if not os.path.isfile(elf + '.map') and os.path.isfile(idf) else elf + '.map'
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)
if any(s[0] == '.flash.appdesc' for s in sections):
return _esp_idf_buckets(sections)
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
def _esp_idf_buckets(sections):
"""section_buckets() of an ESP-IDF app (it has `.flash.appdesc`), whose ELF runs every
section where it loads. IDF's sections.ld names each output section in a flash-mapped region
`.flash*`; esptool's elf2image stores the rest in the image too, for the bootloader to load
into RAM (bin_image.py is_flash_addr())."""
buckets = {}
for name, sh_type, flags, _addr, _size in sections:
if not flags & SHF_ALLOC or (sh_type == SHT_NOBITS and name.startswith('.flash')):
buckets[name] = _NOT_COUNTED # reserved flash-window address space, not storage
elif sh_type == SHT_NOBITS:
buckets[name] = frozenset({'ram'})
elif name.startswith('.flash'):
buckets[name] = frozenset({'flash'})
else:
buckets[name] = frozenset({'flash', 'ram'})
return buckets
# CMake replaces the start of an over-long object name with its md5 (cmLocalGenerator.cxx,
# cmLocalGeneratorShortenObjectName): <target>.dir/<md5>/<rest of the source path>.obj
_SHORTENED_OBJECT_RE = re.compile(r'\.dir/[0-9a-f]{32}/')
# first output, rule and first input of a build.ninja edge; '$' escapes a path's ':' and ' '
_NINJA_EDGE_RE = re.compile(r'build ((?:[^$:\s]|\$.)+)(?: (?:[^$:]|\$.)*)?: \S+ ((?:[^$\s]|\$.)+)')
@functools.lru_cache(maxsize=None)
def _shortened_object_sources(build_dir):
"""{shortened object: source} of the compile edges in build_dir's build.ninja."""
path = os.path.join(build_dir, 'build.ninja')
sources = {}
try:
with open(path, encoding='utf-8') as f:
for line in f:
m = _NINJA_EDGE_RE.match(line)
if m and _SHORTENED_OBJECT_RE.search(m[1]):
obj, src = (re.sub(r'\$(.)', r'\1', p).replace('\\', '/') for p in m.groups())
sources[obj] = src
except OSError as e:
raise RuntimeError(f'cannot read {path}: {e}') from e
except UnicodeDecodeError as e:
raise RuntimeError(f'{path} is not UTF-8: {NINJA_UTF8_REMEDY}') from e
return sources
class _Sizes:
"""Accumulates one elf's filtered per-file flash/RAM, section and symbol sizes,
and its total flash/RAM."""
def __init__(self, filters, elf):
self.filters, self.elf, self.files, self.all = filters, elf, {}, {'flash': 0, 'ram': 0}
self.sections, self.symbols = {}, {}
def _shortened_key(self, path):
"""Key of the source build.ninja compiles to a CMake-shortened object; the object
path no longer holds the filtered dir."""
shortened = f'{path}: CMake shortened this object path to fit CMAKE_OBJECT_PATH_MAX (250 on Windows) and'
build_dir = _find_ninja_build_dir(self.elf)
if build_dir is None:
raise RuntimeError(f'{shortened} no build.ninja was found above {self.elf} to name its source - '
'use a shorter checkout path')
src = _shortened_object_sources(build_dir).get(path.replace('\\', '/'))
if not src:
raise RuntimeError(f'{shortened} the build.ninja above {self.elf} has no compile edge for it '
'to name its source - use a shorter checkout path')
return _relative_key(src, self.filters)
def add(self, path, section, buckets, size, name):
if path and WINDOWS and _SHORTENED_OBJECT_RE.search(path.replace('\\', '/')):
key = self._shortened_key(path) # a filter could match the kept tail of the source path
else:
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}
@functools.lru_cache(maxsize=None)
def _dwarf_sources(elf):
"""{basename: {full source path}} of the elf's DWARF compile units."""
try:
from elftools.elf.elffile import ELFFile # pyelftools, a membrowse dependency
except ImportError as e:
raise RuntimeError(f'pyelftools is needed to resolve {elf}\'s source paths ({e})')
sources = {}
with open(elf, 'rb') as f:
elffile = ELFFile(f)
if not elffile.has_dwarf_info():
return sources
for cu in elffile.get_dwarf_info().iter_CUs():
top = cu.get_top_DIE().attributes
if 'DW_AT_name' in top:
comp_dir = top['DW_AT_comp_dir'].value.decode() if 'DW_AT_comp_dir' in top else ''
path = os.path.normpath(os.path.join(comp_dir, top['DW_AT_name'].value.decode()))
sources.setdefault(os.path.basename(path), set()).add(path)
return sources
def _source_path(sym, elf, filters):
"""A symbol's source path. An archive member has no object path and membrowse 1.2.9
keeps only the basename of its source, so the DWARF compile unit of that name gives it
back (ESP-IDF links TinyUSB from an archive)."""
path = sym.get('object_file') or sym.get('source_file')
if not path or '/' in path:
return path
candidates = _dwarf_sources(elf).get(path, set())
if len(candidates) == 1:
return next(iter(candidates))
if any(_relative_key(c, filters) for c in candidates):
raise RuntimeError(f'{elf}: {path} names several compile units ({", ".join(sorted(candidates))}), '
f'one of them filtered; cannot tell which holds {sym.get("name")}')
return path
def membrowse_sizes(elf, filters):
"""Symbols of `membrowse report`; linker-defined symbols without a section
(`__StackLimit`) are not counted."""
buckets = section_buckets(elf, _map_path(elf))
sizes = _Sizes(filters, elf)
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(_source_path(sym, elf, filters), section, buckets[section], sym['size'], sym.get('name'))
return sizes.result()
@functools.cache
def _load_module(name, path):
"""The Python file at `path`, imported once as `name`."""
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
@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`')
return _load_module('linkermap', path)
def linkermap_sizes(elf, filters):
"""Input sections of the elf's 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 = _map_path(elf)
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, elf)
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, _map_path(elf))
sizes = _Sizes(filters, elf)
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')
MEMBROWSE_CI_VERSION = '1.2.12' # every CI install pins this; attribution can differ across versions
ENGINES = {
'membrowse': Engine(membrowse_sizes, 'all symbols', f'`pip install membrowse=={MEMBROWSE_CI_VERSION}`'),
'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)
_MD_CELL = str.maketrans({**{c: f'&#{ord(c)};' for c in '&<>|`\\[]\r\n'}, '@': '@\u200b'})
def _md_cell(text):
"""A table cell safe from names that would open HTML, a link or code span, end or split
the row or @mention someone (a zero-width space breaks the mention)."""
return text.translate(_MD_CELL)
def md_table(header, rows, total=None, *, left=(0,)):
"""Markdown table padded so its columns also line up as plain text: the `left`
columns left-aligned, the others right-aligned. Cells are strings; a `total` row
follows the rows under a plain rule."""
header = [_md_cell(c) for c in header]
rows = [[_md_cell(c) for c in r] for r in rows]
total = [_md_cell(c) for c in total] if total else total
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 in left else c.rjust(w)
for i, (c, w) in enumerate(zip(cells, widths))) + ' |'
sep = '|' + '|'.join('-' * (w + 2) if i in left 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, files_label='TinyUSB'):
"""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>; `files_label` names the
filtered files' totals."""
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'{files_label} 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', f'{files_label} Flash', f'{files_label} 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 _aligned_ranges(spans):
"""`min → max` cells of one right-aligned column, both numbers right-aligned so the arrows
line up when rendered: leading figure spaces (a digit's width, not collapsed by HTML) pad each,
U+2212 minus is as wide as `+` where `-` is narrower, and no-break spaces keep a cell on one
line in a table too wide for the page. Trailing padding would not do: a browser lets it hang."""
pad = '\u2007'
spans = [(_fmt(lo).replace('-', '\u2212'), _fmt(hi).replace('-', '\u2212')) for lo, hi in spans]
lo_w, hi_w = (max((len(s[i]) for s in spans), default=0) for i in (0, 1))
return [f'{lo.rjust(lo_w, pad)}\u00a0→\u00a0{hi.rjust(hi_w, pad)}' for lo, hi in spans]
def _comment_build(value_id, metric_changed):
"""A comment's build cell for one extreme: `board/example` when its Δ is nonzero, `—` for
a zero end of a changed metric, empty when the metric changed in no build."""
delta, (board, elf) = value_id
if delta:
return f'{board}/{os.path.splitext(os.path.basename(elf))[0]}'
return '—' if metric_changed else ''
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 _range(lo, hi):
return _fmt(lo) if lo == hi else f'{_fmt(lo)} → {_fmt(hi)}'
COMMENT_TERMS = {'base-only': 'missing in this PR', 'current-only': 'new in this PR (no master size yet)',
'base': 'master', 'current': 'PR'}
def _problems(base_only, cur_only, failures, terms=None):
"""Report lines for one-sided elfs and failures, each id named; `terms` rewords the report's words."""
word = lambda w: (terms or {}).get(w, w) # noqa: E731
lines = [f'- {word(label)}: ' + ', '.join(f'`{_label(i)}`' for i in ids)
for label, ids in (('base-only', base_only), ('current-only', cur_only)) if ids]
return lines + [f'- FAILED `{_label(elf_id)}` {word(side)} {stage}: {_md_escape(message)}'
for elf_id, side, stage, message in failures]
def _changed_paths(stats):
"""The changed files of _file_stats(), largest |Δ| first."""
return 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))
def render_comment(pairs, engine, base_only=(), cur_only=(), failures=(), symbols=False, files_label='TinyUSB'):
"""The PR comment's view as (body, footnote): a row per changed file with its min and max
Δ over the builds containing it and the build of each, so each changed driver shows however
heavy the others are. The footnote explains the whole-firmware totals, when the body has them."""
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)]
head = (f'{len(pairs)} builds on {len({board for board, _elf in pairs})} boards compared, '
f'{len(changed)} changed')
footnote = ''
if changed:
flash, ram = zip(*(_delta(pairs[i][0]['all'], pairs[i][1]['all']) for i in changed))
head += f'. Whole firmware¹ Flash Δ {_range(min(flash), max(flash))}, RAM Δ {_range(min(ram), max(ram))}'
footnote = f'\n¹ Sum of {ENGINES[engine].all_label} measured by {engine}, not the exact image size.\n'
problems = _problems(base_only, cur_only, failures, COMMENT_TERMS)
lines = [head, ''] + (['Not compared:', '', *problems, ''] if problems else [])
if not changed:
return '\n'.join(lines + ['_no changes_' if pairs else '_no comparable pairs_', '']), footnote
stats = _file_stats(file_deltas)
paths, metrics = _changed_paths(stats), ('flash', 'ram')
ranges = [_aligned_ranges([(stats[p][k][0][0], stats[p][k][1][0]) for p in paths]) for k in metrics]
rows = [[p, *cells] + [_comment_build(v, any(d for d, _i in stats[p][k])) for k in metrics for v in stats[p][k]]
for p, *cells in zip(paths, *ranges)]
if rows:
lines += [f'{files_label} file size: min → max change across builds containing the file, '
'and the build (board/example) of each; — marks a zero end.', '',
md_table(['File', 'Flash Δ', 'RAM Δ', 'Flash Δ min build', 'Flash Δ max build',
'RAM Δ min build', 'RAM Δ max build'], rows, left=(0, 3, 4, 5, 6))]
other = sum(not any(any(d) for d in file_deltas[i].values()) for i in changed)
if other:
lines += ['', f'{other} other build{"" if other == 1 else "s"} changed without a '
f'{files_label} file-size change.']
return '\n'.join(lines) + '\n', footnote
def render_pairs(pairs, matched, engine, base_only=(), cur_only=(), failures=(), boards=(), symbols=False,
labels=SIDE_LABELS, files_label='TinyUSB'):
"""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; `files_label` names the filtered files' totals.
"""
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))
lines += _problems(base_only, cur_only, failures) + ['']
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_', ''])
src = {i: _delta(_files_total(pairs[i][0]['files']), _files_total(pairs[i][1]['files'])) for i in changed}
pair_rows = [[_label(i)] + [_fmt(d) for d in src[i] + _delta(pairs[i][0]['all'], pairs[i][1]['all'])]
for i in changed]
lines.append(md_table(['Pair', f'{files_label} Flash Δ', f'{files_label} RAM Δ', f'{all_label} Flash Δ',
f'{all_label} RAM Δ'], pair_rows))
stats = _file_stats(file_deltas)
rows = _changed_paths(stats)
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,
files_label='TinyUSB'):
"""Pair two sides and render them. Returns (md, failures, ok, data, changed).
`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;
`changed` the pairs whose sizes changed, {elf id: (base, current)}.
`files_label` is render_pairs()'.
"""
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, files_label)
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],
}
changed = {i: (b, c) for i, (b, c) in pairs.items() if _pair_changed(b, c, symbols)}
return md, failures, bool(pairs) and not failures, data, changed
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 filter_arg(os.path.realpath(os.path.join(checkout_dir, 'src')) + os.sep)
def filter_arg(path):
"""A path filter in the form sizes are matched in: on Windows, past the drive with '/',
as CMake names an out-of-tree object D_/a/.../src/x.c.obj and DWARF D:/a/..."""
return os.path.splitdrive(path)[1].replace('\\', '/') if WINDOWS else path
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:
if WINDOWS: # terminate() and kill() stop only the command, not ninja's compilers
with contextlib.suppress(OSError, subprocess.SubprocessError):
subprocess.run(['taskkill', '/PID', str(proc.pid), '/T', '/F'],
capture_output=True, timeout=TERMINATE_GRACE)
# 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 name in ('SIGTERM', 'SIGHUP'): # Windows has no SIGHUP
if hasattr(signal, name):
signal.signal(getattr(signal, name), 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)
try:
os.symlink(src, dst, target_is_directory=True)
except OSError as e:
if getattr(e, 'winerror', None) == 1314: # ERROR_PRIVILEGE_NOT_HELD
sys.exit(f'Error: cannot symlink {rel} into the base worktree: enable Windows Developer '
'Mode, or use --base-source ci when CI stored a baseline')
raise
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 unknown_boards(boards):
"""Boards of no family in this checkout: build_utils would skip every example of one,
so a typo must fail instead."""
import build # tools/build.py; its import has no side effects
with contextlib.chdir(TINYUSB_ROOT): # build.py reads hw/bsp from the cwd
return [b for b in boards if build.find_family(b) is None]
def missing_examples(examples, sha=None):
"""Examples not named role/name (build.py's -e check), or with no examples/<example>
dir in this checkout nor in commit `sha` when given: one of either tree reaches the
report as one-sided, a typo must fail."""
import build # tools/build.py; its import has no side effects
return [e for e in examples if not build.EXAMPLE_RE.fullmatch(e)
or not os.path.isdir(os.path.join(TINYUSB_ROOT, 'examples', e))
and (sha is None or run(['git', '-C', TINYUSB_ROOT, 'cat-file', '-e', f'{sha}:examples/{e}']).returncode)]
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`, `FAILED after 1.2s` or `skipped: <why>`; 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, skipped=None):
took = (f'skipped: {skipped}' if skipped else
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')
root = src_dir.rstrip(os.sep)
for prefix in {root + os.sep, root.replace(os.sep, '/') + '/'}: # Windows: CMake passes the compiler D:/a/...
error = error.replace(prefix, '')
return error
ESP_IDF_IMAGE = 'espressif/idf:tinyusb' # .github/actions/setup_toolchain/espressif's tag
ESP_IDF_MISSING = (f'ESP-IDF: source $IDF_PATH/export.sh, or docker with {ESP_IDF_IMAGE} '
f'(docker tag the "esp-idf" image of .github/actions/setup_toolchain/toolchain.json as {ESP_IDF_IMAGE}, '
f'as CI does)')
def _link_hops(path):
"""The paths a symlink resolves through, its real path last."""
hops = []
while os.path.islink(path):
path = os.path.normpath(os.path.join(os.path.dirname(path), os.readlink(path)))
hops.append(path)
return hops
def is_espressif(board, src_dir=TINYUSB_ROOT):
return os.path.isdir(os.path.join(src_dir, 'hw', 'bsp', 'espressif', 'boards', board))
def _idf_image():
return bool(shutil.which('docker')) and run(['docker', 'image', 'inspect', ESP_IDF_IMAGE]).returncode == 0
def esp_without_idf(boards, examples=None, base_built=False):
"""The error for espressif `boards` no exported ESP-IDF or CI's image can build, else None.
A board builds nothing that skips each of `examples` here, unless `base_built` and one is
absent here, which that base may have."""
import build # tools/build.py; its import has no side effects
def builds(board):
kept, dropped = build.select_examples(board, examples, root=TINYUSB_ROOT)
return bool(kept) or base_built and build.NO_SUCH_EXAMPLE in dropped.values()
esp = [b for b in boards if is_espressif(b) and (not examples or builds(b))]
if esp and WINDOWS:
return f'{", ".join(esp)} need ESP-IDF, which code_size.py does not support on Windows'
if esp and not (shutil.which('idf.py') or _idf_image()):
return f'{", ".join(esp)} need {ESP_IDF_MISSING}'
return None
def _idf_command(src_dir, build_dir, name):
"""The argv that runs idf.py on `src_dir` into `build_dir`: an exported ESP-IDF, else CI's
image as this user in a container `name`, with both mounted at their own paths so the
elfs' DWARF matches the filters, and each symlinked dependency's real dir at every path
its link goes through (a base worktree's links point at the checkout's, which may point
at the main checkout's). None when neither is available."""
if shutil.which('idf.py'):
return ['idf.py']
if not _idf_image():
return None
# the image enables ccache; kept here it serves the next run (its default hash_dir keeps
# one tree's DWARF paths out of the other's objects)
cache = os.path.join(CODE_SIZE_DIR, '_ccache')
os.makedirs(cache, exist_ok=True)
mounts = {p: os.path.realpath(p) for p in map(os.path.abspath, (src_dir, build_dir, cache))}
for dep in runpy.run_path(os.path.join(src_dir, 'tools', 'get_deps.py'))['deps_all']:
path = os.path.join(src_dir, dep)
mounts.update((hop, os.path.realpath(path)) for hop in _link_hops(path))
cmd = ['docker', 'run', '--rm', '--name', name, '--user', f'{os.getuid()}:{os.getgid()}',
'-e', 'HOME=/tmp', '-e', f'CCACHE_DIR={os.path.abspath(cache)}']
for target, source in sorted(mounts.items()):
cmd += ['-v', f'{source}:{target}']
return cmd + [ESP_IDF_IMAGE, 'idf.py']
def _build_idf(src_dir, build_dir, board, example):
"""Build each ESP-IDF example project's app, as tools/build.py does (its bootloader is
not sized); the first failure stops."""
import build # tools/build.py; its import has no side effects
examples = build.select_examples(board, None if example is None else [example], root=src_dir)[0]
if not examples:
return subprocess.CompletedProcess([], 1, '', f'{board} builds no example')
container = f'tinyusb-code-size-{os.getpid()}'
idf = _idf_command(src_dir, build_dir, container)
if idf is None:
return subprocess.CompletedProcess([], 1, '', f'{board} needs {ESP_IDF_MISSING}')
for ex in examples:
ret = None
try:
ret = run(idf + ['-C', os.path.join(src_dir, 'examples', ex), '-B', os.path.join(build_dir, ex),
'-GNinja', f'-DBOARD={board}', 'app'], timeout=600)
finally:
# stopping the docker CLI leaves its container building; --rm only removes an exited one
if idf[0] == 'docker' and (ret is None or ret.returncode == 124):
run(['docker', 'rm', '-f', container])
if ret.returncode != 0:
break
return ret
class Skipped(str):
"""build_board()'s result for an example its tree does not build: why."""
class Absent(Skipped):
"""An example its tree does not have: a diff's other side may, so it never makes a scope skipped."""
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, Skipped when `example`
is one the tree does not build for the board, else build_error().
When `example` is given, only that target is built (`ninja -C DIR NAME`),
keeping single-example workflows fast. An espressif board builds each example as
its own ESP-IDF project.
"""
import build # tools/build.py; its import has no side effects
phase = Phase(label)
if example:
kept, dropped = build.select_examples(board, [example], root=src_dir)
if not kept:
skip = (Absent(f'{example} is not in this tree') if dropped[example] == build.NO_SUCH_EXAMPLE
else Skipped(f'{board} does not build {example}'))
phase.done(skipped=skip)
return skip
os.makedirs(build_dir, exist_ok=True)
if is_espressif(board, src_dir):
ret = _build_idf(src_dir, build_dir, board, example)
else:
name = example and os.path.basename(example)
# build.py always passes its default toolchain
ret = run(build.cmake_configure_cmd(board, build_dir, [f'-DTOOLCHAIN={build.DEFAULT_TOOLCHAIN}'],
os.path.join(src_dir, 'examples')))
# skip.txt/only.txt only mirror the CMake family filter: the configure is the truth
if ret.returncode == 0 and name and build.registered_examples(build_dir, [example]) == []:
skip = Skipped(f'{board} has no {name} target')
phase.done(skipped=skip)
return skip
if ret.returncode == 0:
# ninja itself, not `cmake --build`: cmake does not pass a timeout's SIGTERM on
ret = run(['ninja', '-C', build_dir] + ([name] if name else []), 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, and
# cmake's _deps/ holds fetched tools, e.g. picotool's enc_bootloader.elf
elfs = sorted(e for e in glob.glob(f'{root}/{example or "*/*"}/*.elf')
if not os.path.relpath(e, build_dir).startswith('_deps' + os.sep))
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).replace(os.sep, '/') # one elf id on every host
sizes[rel] = elf_sizes
if error:
errors.append((rel, error))
return sizes, errors
def size_tree(build_dir, board, example, filters, engine):
"""A built tree's scope sized: ({(board, elf): sizes}, [((board, elf), stage, message)]),
generate_sizes() keyed by board."""
sizes, errors = generate_sizes(build_dir, filters, example, engine)
return {(board, rel): v for rel, v in sizes.items()}, [((board, rel), 'report', msg) for rel, msg in errors]
def diff_summary(data, changed, files_label='TinyUSB'):
"""Console lines of one diff from its JSON-shaped `data` and its `changed` pair count:
coverage, the single pair's filtered Δ, unmatched elfs and failures."""
pairs = {(p['board'], p['elf']): (p['base'], p['current']) for p in data['pairs']}
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'; {files_label} 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, files_label='TinyUSB'):
"""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'; {files_label} 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."""
try:
rel = os.path.relpath(path)
except ValueError: # on another Windows drive
return 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."""
# a board's dir exists only once something built there
os.makedirs(os.path.dirname(path), exist_ok=True)
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)]
files_label = 'filtered' if args.filter else 'TinyUSB'
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 isinstance(error, Skipped):
continue
if error:
failures.append(((board, None), 'build', _build_failed(example, error)))
else:
phase = Phase(f'size{scope}')
sizes, failures = size_tree(build, board, example, filters, args.engine)
# 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, files_label)
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, files_label), f'{example}: ' if scope else '',
tables)
write_report(report_path(board, example, 'report'), md, data)
return 1 if failed else 0
def run_diff(args, requested_sha, unsupported):
"""`diff`: size the working tree against `requested_sha`, its CI snapshots or a local
build, one report per board and example, plus one over every board with --combined.
`unsupported` names the options that need a local base."""
worktree_dir = os.path.join(CODE_SIZE_DIR, '_worktree')
files_label = 'filtered' if args.filter else 'TinyUSB'
# 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')
try:
base_source, reason, baseline, snapshots = resolve_base(args.base_source, unsupported, args.baseline_repo,
requested_sha)
except BaselineUnavailable as e:
print(f'Error: CI base unavailable: {e} (--base-source local builds it)')
return 1
provenance = {'requested': args.base_source or 'default', 'effective': base_source, 'reason': reason,
'requested_sha': requested_sha, 'base_sha': baseline['sha'] if baseline else requested_sha,
'baseline': baseline}
if baseline:
subject = _commit_subject(baseline['sha'])
base_line = (f'Base: CI snapshots of {_clean(baseline["sha"][:10])}' + (f' "{subject}"' if subject else '')
+ (f', {baseline["note"]}' if baseline.get('note') else '')
+ f' {_clean(baseline.get("url"), 200)}')
else:
base_line = f'Base: local build of {requested_sha[:10]}' + (f' (CI base unavailable: {reason})' if reason else '')
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:
if base_source == 'local':
# --detach: check out the commit at a detached HEAD instead of trying to claim
# the branch, which may be checked out elsewhere (main repo, another worktree)
ret = run(['git', '-C', TINYUSB_ROOT, 'worktree', 'add', '--detach', worktree_dir, requested_sha])
if ret.returncode != 0:
print(f'Error creating worktree: {ret.stderr}')
sys.exit(1)
symlink_deps(TINYUSB_ROOT, worktree_dir)
else:
base_shards, base_failures = baseline_shards(snapshots, baseline, args.board)
# a failure of the downloaded run itself, not of one board, belongs in every report
run_failures = [f for f in base_failures if f[0][0] not in args.board]
membrowse_version = _membrowse_version()
base_sha = provenance['base_sha']
current_rev = short_hash(TINYUSB_ROOT)
base_label = short_hash(worktree_dir) if base_source == 'local' else base_sha[:10]
labels = (base_label 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}')
if base_source == 'ci' or reason:
print(f' {base_line}')
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,
'base_source': provenance,
'filters': {'base': base_filters if base_source == 'local' else None, 'current': cur_filters}}
def preface(warnings):
"""The report's base provenance and metadata warnings, above its tables."""
return '\n'.join([base_line] + [f'- {w}' for w in warnings]) + '\n\n'
def write_scope(board, example, sides, failures, warnings, symbols, phase=None):
"""Compare one scope's sides, print its result and write its report: (ok, failures)."""
scope = _scope_label(examples, example)
md, failures, ok, data, changed = compare_sides(
sides['base'], sides['current'], args.engine, failures, [board], scope=(board, None),
symbols=symbols, labels=labels, files_label=files_label)
md = preface(warnings) + md
data['warnings'] = warnings
if phase:
phase.done(failed=not ok)
tables = _labelled({i: _pair_tables(b, c, symbols, args.engine, labels) for i, (b, c) in changed.items()},
len(data['pairs'])) if focused else ()
print_result(diff_summary(data, len(changed), files_label), f'{example}: ' if scope else '', tables)
write_report(report_path(board, example), md, report_data(data))
return ok, failures
# --combined: every board's elf sizes and failures, paired at the end
combined_sides = {'base': {}, 'current': {}}
combined_failures = []
combined_warnings = []
combined_symbols = args.symbols
skipped_scopes = 0
# side: (checkout, filters, build phase label)
trees = {'base': (worktree_dir, base_filters, f'build {args.base_branch}'),
'current': (TINYUSB_ROOT, cur_filters, 'build current')}
local_sides = ['base', 'current'] if base_source == 'local' else ['current']
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)
build_dirs = {'base': os.path.join(board_dir, 'base'), 'current': os.path.join(board_dir, 'build')}
for build_dir in build_dirs.values():
shutil.rmtree(build_dir, ignore_errors=True)
build_failure, skipped = None, {}
for example in examples:
for side in local_sides:
src, _filters, label = trees[side]
error = build_board(src, build_dirs[side], board, example, label + _scope_label(examples, example))
if isinstance(error, Skipped):
skipped[side, example] = error
elif 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:
write_scope(board, example, {'base': {}, 'current': {}}, [build_failure], [], args.symbols)
continue
warnings, symbols, shard = [], args.symbols, base_shards.get(board) if base_source == 'ci' else None
# nothing built, no compiler to compare: every scope here is skipped or fails without one
if shard and not all(('current', e) in skipped for e in examples):
warnings = metadata_warnings(board, shard, _cmake_compiler(build_dirs['current']), membrowse_version)
if symbols and not all('symbols' in s for s in shard['elfs'].values()):
symbols = False
warnings.append(f'{board}: symbols omitted: the CI snapshot has none')
combined_symbols = combined_symbols and symbols
combined_warnings += warnings
for w in warnings:
print(f' WARNING {w}')
for example in examples:
all_skipped = all(type(skipped.get((side, example))) is Skipped for side in local_sides)
phase = None if all_skipped else Phase(f'size and compare{_scope_label(examples, example)}')
sides, failures = {'base': {}, 'current': {}}, []
for side in local_sides:
if (side, example) not in skipped:
_src, filters, _label = trees[side]
sides[side], errors = size_tree(build_dirs[side], board, example, filters, args.engine)
failures += [(i, side, stage, msg) for i, stage, msg in errors]
if base_source == 'ci':
failures += run_failures + [f for f in base_failures if f[0][0] == board]
cur_elfs = {elf: s for (_b, elf), s in sides['current'].items()}
if shard:
sides['base'], scoped, sides['current'], _out = _pair_base_shard(
board, shard, cur_elfs, [example] if example else None)
failures += scoped
else: # no usable baseline: nothing of this board is new, none compared
sides['current'] = {}
# a CI base proves the skip only when its leg selected the example
base_covers = base_source == 'local' or (
shard is not None and (shard['_examples'] is None or example in shard['_examples']))
if all_skipped and base_covers and not failures and not any(sides.values()):
skipped_scopes += 1
continue
ok, failures = write_scope(board, example, sides, failures, warnings, symbols, phase)
failed |= not ok
for side, sizes in sides.items():
combined_sides[side].update(sizes)
combined_failures += failures
if args.bloaty and example:
elf_name = os.path.basename(example)
base_elf = os.path.join(build_dirs['base'], example, f'{elf_name}.elf')
cur_elf = os.path.join(build_dirs['current'], 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 and skipped_scopes == len(args.board) * len(examples):
print('combined: all scopes skipped')
elif args.combined:
print(f'combined ({len(args.board)} boards)')
# every scope was filter-checked above, and its failures carried over
# a board's or the run's baseline failure is in each of its scopes' failures
md, _failures, ok, data, changed = compare_sides(
combined_sides['base'], combined_sides['current'], args.engine, list(dict.fromkeys(combined_failures)),
args.board, symbols=combined_symbols, labels=labels, files_label=files_label)
md = preface(combined_warnings) + md
data['warnings'] = combined_warnings
failed |= not ok
print_result(diff_summary(data, len(changed), files_label)
+ ([f' {len(combined_warnings)} metadata warnings, see the report'] if combined_warnings else []))
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
SNAPSHOT_SCHEMA = 1
_SHA_RE = re.compile(r'[0-9a-f]{40}')
def _cmake_compiler(build_dir):
"""{'id', 'version', 'name', 'build_type'} of a configured build dir, from CMake's own
compiler detection (not the host gcc), '' where CMake recorded none."""
info = {'id': '', 'version': '', 'name': '', 'build_type': ''}
found = glob.glob(os.path.join(glob.escape(build_dir), 'CMakeFiles', '*', 'CMakeCCompiler.cmake'))
if not found: # ESP-IDF: one project per <role>/<example>, all on the board's one toolchain
projects = sorted(glob.glob(os.path.join(glob.escape(build_dir), '*', '*', 'CMakeFiles')))
return _cmake_compiler(os.path.dirname(projects[0])) if projects else info
for path in found:
with open(path) as f:
text = f.read()
for key, var in (('id', 'CMAKE_C_COMPILER_ID'), ('version', 'CMAKE_C_COMPILER_VERSION'),
('name', 'CMAKE_C_COMPILER')):
m = re.search(rf'^set\({var} "([^"]*)"\)', text, re.M)
if m:
info[key] = m.group(1)
# without .exe, so a Windows build's name matches a Linux snapshot's
info['name'] = re.sub(r'\.exe$', '', os.path.basename(info['name']), flags=re.I)
cache = os.path.join(build_dir, 'CMakeCache.txt')
if os.path.isfile(cache):
with open(cache) as f:
m = re.search(r'^CMAKE_BUILD_TYPE:\w+=(.*)$', f.read(), re.M)
info['build_type'] = m.group(1) if m else ''
return info
def _git_shas(event):
"""(sha, base_sha, head_sha) of the checkout: a pull_request build checks out GitHub's
merge commit, whose first parent is the base branch tip it was built against and
second parent the PR head; any other build is its own base and head."""
ret = run(['git', '-C', TINYUSB_ROOT, 'rev-parse', 'HEAD', 'HEAD^1', 'HEAD^2']
if event == 'pull_request' else ['git', '-C', TINYUSB_ROOT, 'rev-parse', 'HEAD'])
shas = ret.stdout.split() if ret.returncode == 0 else []
if not shas or not all(_SHA_RE.fullmatch(s) for s in shas):
raise RuntimeError(f'cannot resolve the checkout commits for a {event} build: {ret.stderr.strip()}')
return (shas[0], shas[1], shas[2]) if len(shas) == 3 else (shas[0], shas[0], shas[0])
def _membrowse_version():
"""The installed membrowse's version, '' when it cannot be read."""
try:
from importlib.metadata import version
return version('membrowse')
except Exception:
return ''
def snapshot_boards(families, boards, examples, defines=()):
"""The CI-pinned boards a build leg produced, as the Membrowse upload step picks them:
each family's pinned boards that build one of `examples` (build.py's
resolve_ci_boards, boards-only), plus each `-b` board that is pinned. `defines` are
the leg's -D tokens, which can enable examples (MAX3421_HOST=1)."""
import build # tools/build.py; its import has no side effects
pinned = ci_pinned_boards()
# build.py skips a -b board that builds none of the examples, before configuring it
picked = [b for b in boards if b in pinned and build.builds_any(b, examples, defines)]
for family in families:
picked += build.resolve_ci_boards(CI_PINNED_BOARDS, family, True, examples, extra_defines=defines)
return list(dict.fromkeys(picked))
def board_snapshot(board, build_dir, examples, filters, defines=()):
"""(elfs, failures) of one board's build dir: {elf path: sizes} of every sized elf,
and [{'elf', 'stage', 'message'}]. A missing build dir, an example of the scope the
board builds but has no elf for, or an elf that failed to size is a failure, never a
zero; a board whose elfs all match no TinyUSB file fails the filter check."""
if not os.path.isdir(build_dir):
return {}, [{'elf': None, 'stage': 'build', 'message': f'no build dir {_shown(build_dir)}'}]
elfs, failures = {}, []
scopes = [e for e in dict.fromkeys(examples)
if not build_utils.skip_example(e, board, defines)] if examples else [None]
for example in scopes:
sizes, errors = generate_sizes(build_dir, filters, example)
for rel, s in sizes.items():
if s is not None and not _valid_sizes(s):
errors.append((rel, 'sizes are not integer byte counts of files, sections and symbols'))
elif s is not None:
elfs[rel] = s
failures += [{'elf': rel, 'stage': 'build' if rel is None else 'report',
'message': f'{example}: {msg}' if example and rel is None else msg}
for rel, msg in errors]
if elfs and not any(s['files'] for s in elfs.values()):
failures.append({'elf': None, 'stage': 'filter', 'message': _filter_failure('membrowse')})
return elfs, failures
def run_snapshot(args):
"""`snapshot`: size the CI-pinned boards a build leg already built, one
code-size-<board>.json each, for `compare` against another run's snapshots."""
os.chdir(TINYUSB_ROOT) # build_utils and build.py resolve hw/bsp and examples from the cwd
sha, base_sha, head_sha = _git_shas(args.event)
membrowse_version = _membrowse_version()
filters = args.filter or [tinyusb_src_filter(TINYUSB_ROOT)]
defines = tuple(args.define_symbol)
boards = snapshot_boards(args.families, args.board, args.example, defines)
# a variant leg (--build-name) reports its one board under the build name
names = {args.build_name or b: b for b in boards}
os.makedirs(args.output, exist_ok=True)
# written even with no board: `compare` tells a leg that ran from one that never did
with open(os.path.join(args.output, 'leg.json'), 'w') as f:
json.dump({'schema': SNAPSHOT_SCHEMA, 'examples': args.example, 'boards': list(names),
'build_outcome': args.build_outcome, 'sha': sha, 'base_sha': base_sha, 'head_sha': head_sha},
f, sort_keys=True)
if not boards:
print('snapshot: no CI-pinned board in this leg')
return 0
failed = False
for board, base_board in names.items():
build_dir = os.path.join(args.build_root, f'cmake-build-{board}')
elfs, failures = board_snapshot(base_board, build_dir, args.example, filters, defines)
# the leg's commits, examples and build outcome are in its leg.json
data = {'schema': SNAPSHOT_SCHEMA, 'board': board, 'engine': 'membrowse',
'membrowse_version': membrowse_version, 'compiler': _cmake_compiler(build_dir),
'elfs': {rel: _json_sizes(s, args.symbols) for rel, s in sorted(elfs.items())},
'failures': failures}
path = os.path.join(args.output, f'code-size-{board}.json')
with open(path, 'w') as f:
json.dump(data, f, sort_keys=True, separators=(',', ':'))
failed |= bool(failures)
print(f' {board}: {len(elfs)} elfs sized, {len(failures)} failures -> {_shown(path)}')
for fail in failures:
print(f' FAILED {fail["elf"] or board} {fail["stage"]}: {fail["message"]}')
return 1 if failed else 0
_NAME_RE = re.compile(r'[A-Za-z0-9_.+/-]+')
COMMENT_LIMIT = 60000 # GitHub caps a comment at 65536 characters
def _clean(text, limit=300):
"""Snapshot text for a report: a snapshot is PR-built data, so nothing in it may open
Markdown, HTML or an @mention."""
return re.sub(r'[^A-Za-z0-9_.,:;=()/+ -]', '?', str(text))[:limit]
def _is_count(value):
return isinstance(value, int) and not isinstance(value, bool)
def _region(value):
return isinstance(value, dict) and set(value) == {'flash', 'ram'} and all(map(_is_count, value.values()))
def _int_map(value, depth):
"""Whether `value` is `depth` levels of str-keyed dicts over byte counts."""
return isinstance(value, dict) and all(
isinstance(k, str) and (_int_map(v, depth - 1) if depth > 1 else _is_count(v)) for k, v in value.items())
def _valid_sizes(s):
return (isinstance(s, dict) and {'files', 'all', 'sections'} <= set(s) <= {'files', 'all', 'sections', 'symbols'}
and _region(s['all']) and isinstance(s['files'], dict)
and all(isinstance(k, str) and _region(v) for k, v in s['files'].items())
and _int_map(s['sections'], 2) and ('symbols' not in s or _int_map(s['symbols'], 3)))
def _is_name(value):
return isinstance(value, str) and bool(_NAME_RE.fullmatch(value)) and '..' not in value.split('/')
def _names(value, none_ok=False):
return (none_ok and value is None) or (isinstance(value, list) and all(map(_is_name, value)))
def _commits(data):
"""(sha, base_sha, head_sha), or None unless all are 40-hex."""
shas = tuple(data.get(k) for k in ('sha', 'base_sha', 'head_sha'))
return shas if all(isinstance(s, str) and _SHA_RE.fullmatch(s) for s in shas) else None
def _scope_error(data):
legs, fam = data.get('legs'), data.get('family_examples', {})
if not isinstance(data.get('code_changed'), bool):
return 'code_changed is not a boolean'
if not (isinstance(legs, list) and all(isinstance(leg, dict) and isinstance(leg.get('toolchain'), str)
and isinstance(leg.get('arg'), str) for leg in legs)):
return 'legs are not {toolchain, arg} strings'
if not (isinstance(fam, dict) and all(map(_names, fam.values()))):
return 'family_examples is not a map of example lists'
return None
def _leg_error(data):
if not isinstance(data.get('build_outcome'), str):
return 'build_outcome is not a string'
if not _names(data.get('boards')):
return 'boards is not a list of board names'
if 'examples' not in data or not _names(data['examples'], none_ok=True):
return 'examples is not a list of example names'
return None if _commits(data) else 'commit hashes are not 40-hex'
def _shard_error(data, filename):
board = data.get('board')
if not (_is_name(board) and filename == f'code-size-{board}.json'):
return 'board name does not match the file'
if data.get('engine') != 'membrowse':
return 'unsupported engine'
elfs, failures = data.get('elfs'), data.get('failures')
if not isinstance(elfs, dict) or not isinstance(failures, list):
return 'no elfs/failures'
for elf, sizes in elfs.items():
if not _is_name(elf):
return 'invalid elf path'
if not _valid_sizes(sizes):
return f'{elf}: sizes are not integer byte counts'
if not all(isinstance(f, dict) and 'elf' in f and (f['elf'] is None or _is_name(f['elf']))
and isinstance(f.get('stage'), str) and isinstance(f.get('message'), str) for f in failures):
return 'failures are not {elf, stage, message} records'
return None
def load_snapshots(root):
"""Every snapshot file under `root` (one directory per downloaded artifact):
{'scope': scope.json or None, 'legs': {artifact dir: leg.json}, 'shards': {board: shard},
'errors': [(file, why)]}. A shard records its artifact dir as '_leg'. An unusable file
is an error, never guessed at."""
run = {'scope': None, 'legs': {}, 'shards': {}, 'errors': []}
for dirpath, _dirs, files in sorted(os.walk(root)):
artifact = os.path.basename(dirpath)
for name in sorted(f for f in files if f.endswith('.json')):
rel = os.path.relpath(os.path.join(dirpath, name), root)
try:
with open(os.path.join(dirpath, name)) as f:
data = json.load(f)
except (OSError, ValueError):
run['errors'].append((rel, 'unreadable JSON'))
continue
if not isinstance(data, dict) or data.get('schema') != SNAPSHOT_SCHEMA:
why = 'unsupported snapshot schema'
elif name == 'scope.json':
why = _scope_error(data)
run['scope'] = None if why else data
elif name == 'leg.json':
why = _leg_error(data)
if not why:
run['legs'][artifact] = data
elif name.startswith('code-size-'):
why = _shard_error(data, name)
if not why and data['board'] in run['shards']:
why = f'second snapshot of {data["board"]}'
if not why:
run['shards'][data['board']] = {**data, '_leg': artifact}
else:
why = 'not a snapshot file'
if why:
run['errors'].append((rel, why))
return run
def _leg_examples(scope, arg):
"""The examples build_util.yml builds for a scope leg: its arg's own -e plus the scope's
family map entry for that arg, None for all."""
return sorted(set(re.findall(r' -e ([^ ]+)', arg) + scope.get('family_examples', {}).get(arg, []))) or None
def _in_scope(elf, examples):
"""Whether an elf path (<role>/<example>/<name>.elf) is one of `examples`, all when None."""
return examples is None or '/'.join(elf.split('/')[:2]) in examples
def _usable_shards(run, side, failures, boards=None):
"""`run`'s shards listed in the boards of their own leg, whose build succeeded, each given
its leg's examples as '_examples'. Returns ({board: shard}, {leg commits}); every
rejection is added to `failures`, only of `boards` when given."""
shards, commits = {}, {_commits(leg) for leg in run['legs'].values()}
for rel, why in run['errors']:
board = re.fullmatch(r'code-size-(.+)\.json', os.path.basename(rel))
if boards is None or not board or board.group(1) in boards:
failures.append(((_clean(rel), None), side, 'snapshot', _clean(why)))
for board, shard in sorted(run['shards'].items()):
if boards is not None and board not in boards:
continue
leg = run['legs'].get(shard['_leg'])
if leg is None or board not in leg['boards']:
failures.append(((board, None), side, 'snapshot', 'not a board of its leg\'s build'))
elif leg['build_outcome'] != 'success':
failures.append(((board, None), side, 'build', f'the build step ended {_clean(leg["build_outcome"])}'))
else:
shards[board] = {**shard, '_examples': leg['examples']}
for _artifact, leg in sorted(run['legs'].items()):
failures += [((board, None), side, 'snapshot', 'the leg left no snapshot of it') for board in leg['boards']
if board not in run['shards'] and (boards is None or board in boards)]
return shards, commits
def _scope_failures(scope, legs):
"""Legs of the scope that left no leg.json, or measured other examples than it selected,
and legs that are not of it."""
failures, expected = [], set()
for leg in scope['legs']:
tag = re.sub(r' -e [^ ]+', '', leg['arg']) # build_util.yml's ARTIFACT_TAG
artifact = f'code-size-{leg["toolchain"]}-{tag}'
expected.add(artifact)
if artifact not in legs:
failures.append(((_clean(artifact), None), 'current', 'snapshot', 'the build leg left no snapshot'))
elif (legs[artifact]['examples'] is not None # all examples covers any selection
and sorted(set(legs[artifact]['examples'])) != _leg_examples(scope, leg['arg'])):
failures.append(((_clean(artifact), None), 'current', 'scope', 'measured other examples than selected'))
failures += [((_clean(a), None), 'current', 'scope', 'not a leg of this run') for a in sorted(set(legs) - expected)]
return failures
def _base_commit_failure(commits, baseline):
"""The failure when a baseline run's snapshots are of several commits, or not of the
selected `baseline` ({'sha'}, or None for any), else None."""
shas = {c[0] for c in commits}
if len(shas) > 1 or (baseline and shas and shas != {baseline.get('sha')}):
return (('commits', None), 'base', 'snapshot', 'the snapshots are not of the selected baseline commit')
return None
def _pair_base_shard(board, base, cur_elfs, examples):
"""One board's current elfs ({elf: sizes}) against its baseline shard, within
`examples` (None for all): (base sizes, base failures, current sizes, baseline elfs
outside `examples`). A current elf whose baseline failed is left out: that failure is
the baseline's, never a new elf here."""
sizes, failures, failed, outside = {}, [], set(), 0
for f in base['failures']:
if f['elf'] is None or _in_scope(f['elf'], examples):
failures.append(((board, f['elf']), 'base', _clean(f['stage']), _clean(f['message'])))
failed.add(f['elf'])
for elf, s in base['elfs'].items():
if _in_scope(elf, examples):
sizes[(board, elf)] = s
else:
outside += 1
return sizes, failures, {(board, elf): s for elf, s in cur_elfs.items() if elf not in failed}, outside
def compare_runs(base, cur, baseline=None, symbols=False):
"""Compare two load_snapshots() runs, `cur` defining the scope. Returns
(full Markdown, comment Markdown, JSON data). `baseline` is code_size_ci.py's
description of the base run ({'sha', 'url', 'exact', 'note'}), or None. Anything that
leaves coverage short is a failure; a comparable difference is a note."""
notes, failures = [], []
scope = cur['scope']
if scope is None:
failures.append((('scope', None), 'current', 'snapshot', 'no usable scope manifest'))
elif not (scope['legs'] or cur['legs'] or cur['shards'] or cur['errors']):
md = 'Nothing to measure: no code change, or no CI-pinned build leg selected.\n'
return md, f'## Code size\n\n{md}', {'status': 'nothing measured'}
else:
failures += _scope_failures(scope, cur['legs'])
cur_shards, cur_commits = _usable_shards(cur, 'current', failures)
base_shards, base_commits = _usable_shards(base, 'base', failures, set(cur_shards))
labels = SIDE_LABELS
if len(cur_commits) > 1:
failures.append((('commits', None), 'current', 'snapshot', 'the snapshots are of different commits'))
cur_shards = {}
elif cur_commits:
(_sha, base_sha, head_sha), = cur_commits
labels = (base_sha[:10], head_sha[:10])
if commit_failure := _base_commit_failure(base_commits, baseline):
failures.append(commit_failure)
base_shards = {}
sides, no_baseline, outside = {'base': {}, 'current': {}}, [], 0
for board, shard in sorted(cur_shards.items()):
failures += [((board, f['elf']), 'current', _clean(f['stage']), _clean(f['message'])) for f in shard['failures']]
b = base_shards.get(board)
if b is None:
no_baseline.append(board)
failures.append(((board, None), 'base', 'snapshot', 'no baseline snapshot'))
continue
for key in ('membrowse_version', 'compiler'):
if b.get(key) != shard.get(key):
notes.append(f'{board}: {key} differs from the baseline ({_clean(b.get(key))} -> '
f'{_clean(shard.get(key))}): comparable, not exact')
base_sizes, base_failures, cur_sizes, out = _pair_base_shard(board, b, shard['elfs'], shard['_examples'])
sides['base'].update(base_sizes)
sides['current'].update(cur_sizes)
failures += base_failures
outside += out
if outside:
notes.append(f'{outside} baseline elfs are of examples this PR did not build (outside coverage)')
if symbols and not all('symbols' in s for side in sides.values() for s in side.values()):
symbols = False
notes.append('symbols omitted: some snapshots were taken without them')
head = []
if baseline:
kind = 'exact' if baseline.get('exact') else 'approximate'
head.append((f'Baseline: {_clean(baseline["sha"][:10])} ({kind}) {_clean(baseline.get("url"), 200)}'
if baseline.get('sha') else 'Baseline: unavailable')
+ (f' - {_clean(baseline["note"])}' if baseline.get('note') else ''))
head += [f'- {n}' for n in notes]
boards = sorted(cur_shards)
full, _fails, _ok, data, _changed = compare_sides(sides['base'], sides['current'], 'membrowse', failures, boards,
symbols=symbols, labels=labels)
pairs, base_only, cur_only = pair_elfs(sides['base'], sides['current'])
body, footnote = render_comment(pairs, 'membrowse', base_only, cur_only, failures, symbols)
preface = '\n'.join(head) + ('\n\n' if head else '')
comment = f'## Code size\n\n{preface}{body}'
truncated = '\n\n_Truncated: see the full report._\n'
if len(comment) + len(footnote) > COMMENT_LIMIT:
comment = comment[:COMMENT_LIMIT - len(footnote) - len(truncated)].rsplit('\n', 1)[0] + truncated
comment += footnote
data.update({'notes': notes, 'baseline': baseline, 'no_baseline': no_baseline, 'outside': outside})
return preface + full, comment, data
def run_compare(args):
"""`compare`: compare a run's snapshots against a baseline run's, no build."""
baseline = None
if args.baseline_info:
with open(args.baseline_info) as f:
baseline = json.load(f)
base = load_snapshots(args.base) # os.walk of a missing directory yields nothing
full, comment, data = compare_runs(base, load_snapshots(args.current), baseline, args.symbols)
os.makedirs(args.output, exist_ok=True)
write_report(os.path.join(args.output, 'code-size'), full, data)
with open(os.path.join(args.output, 'comment.md'), 'w') as f:
f.write(comment)
print(f' comment: {_shown(os.path.join(args.output, "comment.md"))}')
return 0
BASELINE_REPO = 'hathach/tinyusb'
DOWNLOAD_TIMEOUT = 600 # seconds for one run's code-size artifacts
class BaselineUnavailable(Exception):
"""The CI base cannot be had: no gh, no auth, no baseline run, a base off master, or a
failed transfer. Never a problem inside downloaded snapshots: those are failures."""
def _download_baseline(ci, repo, run_id):
"""The snapshot dir of run `run_id`'s code-size-* artifacts, downloaded once per run
attempt into the cache and published whole; a re-run during the download is retried once."""
attempt = ci.gh(f'repos/{repo}/actions/runs/{run_id}')['run_attempt']
for _ in range(2):
dest = os.path.join(BASELINE_CACHE_DIR, repo.replace('/', '_'), f'{run_id}-{attempt}')
if os.path.isfile(os.path.join(dest, 'manifest.json')):
return os.path.join(dest, 'snapshots')
tmp = f'{dest}.tmp-{os.getpid()}'
shutil.rmtree(tmp, ignore_errors=True)
try:
ret = run(['gh', 'run', 'download', str(run_id), '-R', repo, '-p', 'code-size-*',
'-D', os.path.join(tmp, 'snapshots')], timeout=DOWNLOAD_TIMEOUT)
if ret.returncode != 0:
raise BaselineUnavailable(f'gh run download {run_id}: {_clean(ret.stderr) or f"exit {ret.returncode}"}')
now = ci.gh(f'repos/{repo}/actions/runs/{run_id}')['run_attempt']
if now != attempt: # a re-run replaced artifacts mid-download: this copy is of neither attempt
attempt = now
continue
try:
with open(os.path.join(tmp, 'manifest.json'), 'w') as f:
json.dump({'repo': repo, 'run_id': run_id, 'run_attempt': attempt}, f, indent=1, sort_keys=True)
os.rename(tmp, dest)
except OSError as e:
if not os.path.isfile(os.path.join(dest, 'manifest.json')): # else another run published it first
raise BaselineUnavailable(f'caching run {run_id}: {e}') from e
return os.path.join(dest, 'snapshots')
finally:
shutil.rmtree(tmp, ignore_errors=True) # gone after a successful rename
raise BaselineUnavailable(f'run {run_id} was re-run during the download')
def ci_baseline(repo, sha):
"""(snapshot dir, baseline info) of `repo`'s stored master snapshots for `sha`, else
its nearest first-parent ancestor's (approximate), as code_size_ci.py picks a PR's.
Raises BaselineUnavailable."""
if not shutil.which('gh'):
raise BaselineUnavailable('gh not found')
sys.modules.setdefault('code_size', sys.modules[__name__]) # its `import code_size` is this module, not a copy
ci = _load_module('code_size_ci', os.path.join(TINYUSB_ROOT, '.github', 'scripts', 'code_size_ci.py'))
try:
info = ci.lookup_sha(repo, sha, 'yours')
if info['sha'] is None:
raise BaselineUnavailable(info['note'])
return _download_baseline(ci, repo, info['run_id']), info
except (RuntimeError, KeyError, TypeError, ValueError) as e: # gh auth, transport or an unexpected reply
raise BaselineUnavailable(f'the baseline lookup failed: {_clean(str(e))}') from e
except OSError as e: # gh could not be started, e.g. EMFILE
raise BaselineUnavailable(f'running gh failed: {_clean(e.strerror or e)}') from e
def resolve_base(base_source, unsupported, repo, sha):
"""A diff's base as (source, reason, baseline info, snapshots): the CI snapshots of
`sha`, unless `base_source` is 'local' or `unsupported` options need a local base. With
no `base_source`, an unavailable CI base falls back to a local build and `reason` says
why; with 'ci' it raises BaselineUnavailable."""
if base_source == 'local':
return 'local', None, None, None
if unsupported:
return 'local', f'{", ".join(unsupported)} needs a local base', None, None
print(f'CI base: looking up {repo} snapshots of {sha[:10]}…', flush=True)
try:
snapshot_dir, baseline = ci_baseline(repo, sha)
except BaselineUnavailable as e:
if base_source == 'ci':
raise
return 'local', str(e), None, None
return 'ci', None, baseline, load_snapshots(snapshot_dir)
def baseline_shards(snapshots, baseline, boards):
"""The base side of a diff from a downloaded run (load_snapshots()): ({board: shard},
failures). Only shards of `boards` from a successful leg of the
`baseline` commit count; anything else is a base failure, never a zero, and a failure
of the run itself (an unreadable file, other commits) is keyed by what failed, not a
board. Shards of other boards, e.g. -DMA variants, are excluded, never substituted."""
failures = []
shards, commits = _usable_shards(snapshots, 'base', failures, set(boards))
if commit_failure := _base_commit_failure(commits, baseline):
failures.append(commit_failure)
shards = {}
failures += [((board, None), 'base', 'snapshot', 'no baseline snapshot') for board in boards
if board not in shards and not any(f[0][0] == board for f in failures)]
return shards, failures
def _commit_subject(sha):
"""`sha`'s subject from the local history, '' when it is not there (not fetched)."""
ret = run(['git', '-C', TINYUSB_ROOT, 'log', '-1', '--format=%s', sha])
return ret.stdout.strip() if ret.returncode == 0 else ''
def metadata_warnings(board, shard, compiler, membrowse_version):
"""A board's warnings when its CI base was measured unlike its current build: the
board is still compared, the warning says what its deltas may include."""
base_c = shard.get('compiler') if isinstance(shard.get('compiler'), dict) else {}
warnings = []
unknown = [side for side, c in (('base', base_c), ('current', compiler))
if not all(c.get(k) for k in ('id', 'name', 'version'))]
if unknown:
warnings.append(f'{board}: compiler unknown on the {" and ".join(unknown)} side: '
'comparability cannot be established')
elif any(base_c.get(k) != compiler.get(k) for k in ('id', 'name', 'version', 'build_type')):
def show(c):
return ' '.join(_clean(c[k]) for k in ('id', 'name', 'version', 'build_type') if c.get(k))
warnings.append(f'{board}: base built with {show(base_c)}, current with {show(compiler)}: '
'deltas may include the toolchain change')
base_m = shard.get('membrowse_version')
unknown = [side for side, v in (('base', base_m), ('current', membrowse_version)) if not v]
if unknown:
warnings.append(f'{board}: membrowse version unknown on the {" and ".join(unknown)} side: '
'comparability cannot be established')
elif base_m != membrowse_version:
warnings.append(f'{board}: base measured with membrowse {_clean(base_m)}, current with '
f'{_clean(membrowse_version)}: per-file attribution may differ')
return warnings
def example_arg(value):
example = (value.replace('\\', '/') if WINDOWS else 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, type=filter_arg,
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.')
parser.add_argument('--base-source', choices=('ci', 'local'), default=None,
help='Where the base sizes come from. Default: try the stored CI snapshots of '
'--base-branch, fall back to a local build; ci: require CI; local: build '
'the base, no GitHub lookup.')
parser.add_argument('--baseline-repo', default=BASELINE_REPO,
help=f'Repository whose master snapshots are the CI base (default: {BASELINE_REPO})')
snap = sub.add_parser('snapshot', help='size the CI-pinned boards a CI build leg already built, '
'one code-size-<board>.json each (no build)')
snap.add_argument('families', nargs='*', default=[], help='Families the leg built (as tools/build.py)')
snap.add_argument('-b', '--board', action='append', default=[], help='Board the leg built (repeatable)')
snap.add_argument('-e', '--example', action='append', default=None, type=example_arg,
help='Examples the leg built (repeatable); omit when it built all')
snap.add_argument('--build-name', help='A variant leg\'s build name (as tools/build.py): its build dir, and '
'the board it reports as')
snap.add_argument('-D', '--define-symbol', action='append', default=[],
help='A variant leg\'s build-system define (as tools/build.py); it needs --build-name')
snap.add_argument('--cflag', action='append', default=[],
help='A variant leg\'s compiler flag (as tools/build.py); it needs --build-name')
snap.add_argument('-o', '--output', required=True, help='Directory for the code-size-<board>.json files')
snap.add_argument('--build-root', default=os.path.join(TINYUSB_ROOT, 'cmake-build'),
help='Where tools/build.py put cmake-build-<board> (default: cmake-build)')
snap.add_argument('--event', default=os.environ.get('GITHUB_EVENT_NAME', 'push'),
help='CI event; pull_request reads base/head from the merge commit '
'(default: $GITHUB_EVENT_NAME, else push)')
snap.add_argument('--build-outcome', default='success', help='The Build step\'s outcome, recorded as is')
snap.add_argument('-f', '--filter', action='append', default=None, type=filter_arg,
help='Path-substring filter (repeatable); default: this checkout\'s <checkout>/src/')
snap.add_argument('--symbols', action='store_true', help='Include each file\'s symbols')
snap.add_argument('-v', '--verbose', action='store_true', help='Print commands')
cmp = sub.add_parser('compare', help='compare a CI run\'s snapshots against a baseline run\'s '
'(code-size.md, code-size.json, comment.md; no build)')
cmp.add_argument('current', help='The run\'s downloaded artifacts, one directory each')
cmp.add_argument('base', help='The baseline run\'s downloaded artifacts; a missing directory is no baseline')
cmp.add_argument('-o', '--output', required=True, help='Directory for the reports')
cmp.add_argument('--baseline-info', help='JSON describing the baseline run: {sha, url, exact, note}')
cmp.add_argument('--symbols', action='store_true', help='Show each changed file\'s symbols in the full report')
args = top.parse_args()
verbose = getattr(args, 'verbose', False)
if args.command == 'compare':
return run_compare(args)
if args.command == 'snapshot':
# a leg whose build did not succeed may have skipped the membrowse install too;
# its snapshot still records that outcome, and any elf it left fails to size
if engine_missing('membrowse') and args.build_outcome == 'success':
snap.error(f'membrowse not found - install {ENGINES["membrowse"].install}')
if invalid := invalid_boards(args.board + ([args.build_name] if args.build_name else [])):
snap.error(f'invalid board name: {", ".join(invalid)}')
if args.build_name and (args.families or len(args.board) != 1):
snap.error('--build-name names exactly one -b board and no families')
if (args.define_symbol or args.cflag) and not args.build_name:
snap.error('-D or --cflag makes a variant, which needs --build-name to report under')
return run_snapshot(args)
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)}')
if unknown := unknown_boards(args.board):
report_parser.error(f'unknown board: {", ".join(unknown)}')
if missing := missing_examples(args.example or []):
report_parser.error(f'no such example: {", ".join(missing)}')
if error := esp_without_idf(args.board, args.example):
report_parser.error(error)
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)}')
if unknown := unknown_boards(args.board):
parser.error(f'unknown board: {", ".join(unknown)}')
# CI snapshots hold membrowse sizes under the TinyUSB filter, and no elf
unsupported = [opt for opt, used in (('-f', args.filter), ('--bloaty', args.bloaty),
(f'--engine {args.engine}', args.engine != 'membrowse')) if used]
# before any board builds: a --ci run would otherwise fail its espressif boards last
if error := esp_without_idf(args.board, args.example, args.base_source == 'local' or bool(unsupported)):
parser.error(error)
if args.base_source == 'ci' and unsupported:
parser.error(f'--base-source ci cannot be used with {", ".join(unsupported)}')
ret = run(['git', '-C', TINYUSB_ROOT, 'rev-parse', '--verify', '--quiet', f'{args.base_branch}^{{commit}}'])
if ret.returncode != 0:
parser.error(f'--base-branch {args.base_branch} names no commit')
requested_sha = ret.stdout.strip()
if missing := missing_examples(args.example or [], requested_sha):
parser.error(f'no such example here or in {args.base_branch}: {", ".join(missing)}')
return run_diff(args, requested_sha, unsupported)
if __name__ == '__main__':
exit_on_termination()
sys.exit(main())