diff options
Diffstat (limited to 'tools')
| -rwxr-xr-x | tools/build.py | 5 | ||||
| -rwxr-xr-x | tools/ci_select.py | 162 | ||||
| -rw-r--r-- | tools/rtt.py | 727 |
3 files changed, 861 insertions, 33 deletions
diff --git a/tools/build.py b/tools/build.py index eeefca22d..aa8868cb8 100755 --- a/tools/build.py +++ b/tools/build.py @@ -34,6 +34,7 @@ ci_skip_boards = { 'adafruit_fruit_jam', 'adafruit_metro_rp2350', 'feather_rp2040_max3421', + 'pico2_etm_trace', 'pico_sdk', 'raspberry_pi_pico_w', ], @@ -356,11 +357,11 @@ def get_family_boards(family, one_random, one_first, examples=None, build_system # the WHOLE preferred list, in order - stopping at entry one would abandon a # curated list for the raw alphabetical order the moment its first board cannot # build the filter, which also moves the board the metrics baseline is keyed on + # the whole preferred list, in order. Unreachable-when-unfiltered: with + # examples is None, buildable() is True and the loop returns on entry one. for b in preferred_list: if buildable(b): return [b] - if preferred_list and examples is None: - return [preferred_list[0]] candidates = [b for b in all_boards if buildable(b)] or all_boards if one_first: return [candidates[0]] diff --git a/tools/ci_select.py b/tools/ci_select.py index 89a0d214c..1526f2064 100755 --- a/tools/ci_select.py +++ b/tools/ci_select.py @@ -13,6 +13,38 @@ JSON: full, boards (name -> 'all' | [tests]), families (bsp families the diff touches, including ones with no rig board - build-only consumers such as /pre-pr sample from these), args (hil_test.py args per config) and args_flasher (the same args split by each board's flasher, for CI legs that split one rig by flasher). + +THE RULE TABLE. First match wins; answers union per family (build) and per board +(HIL). A CARBON COPY of the table in the design spec above - edit both, or +TestRuleTableIsCarbonOfTheSpec fails. `FAM` = the families whose family.cmake +references the changed path (CMake only; make follows it). `DEV`/`HOST`/`DUAL`/ +`TYPEC`/`ALL` are the example role sets. The Build families column is PRE-PRUNE: +_prune_buildable then intersects each family with what it can actually build. + +| # | Changed path | Build families | Build examples | HIL boards → tests | +| 1 | `docs/`, `.claude/`, `*.md`, `*.rst`, `LICENSE` | — | — | — | +| 1b | `.gitignore`, `.clang-format`, `.idea/**`, `test/{fuzz,unit-test}/**`, `test/hil/test/**`, non-build `.github/**`, packaging manifests | — | — | — | +| 2 | `test/hil/**` (not `test/hil/test/**`) | — | — | all boards → all tests | +| 2b | `tools/metrics.py`, `.github/scripts/metrics_*.py` | `ALL` (unchanged — `tinyusb_metrics` runs `metrics.py` as a build target) | `ALL` | — (nothing on the rig runs it) | +| 3 | `src/portable/<port>/dcd_*`, `*_device.[ch]` | `FAM` | `DEV`+`DUAL` | `FAM`'s device-role boards → device+dual tests | +| 4 | `src/portable/<port>/hcd_*`, `*_host.[ch]` | `FAM` | `HOST`+`DUAL` | `FAM`'s host-role boards → host+dual tests | +| 5 | `src/portable/<port>/**` (anything else) | `FAM` | `ALL` | `FAM`'s boards → all their tests | +| 5b | `src/portable/<port>/**` where `FAM` is empty | — | — | — (empty resolves to nothing on BOTH axes) | +| 6 | `hw/bsp/<family>/**` | that family | `ALL` | that family's boards → all tests (a `boards/<board>/` path narrows to that board) | +| 7 | `hw/mcu/<vendor>/**` | `FAM` — empty resolves to nothing (maintainer ruling) | `ALL` | `FAM`'s boards → all tests; empty resolves to nothing (maintainer ruling) ⚠ *see below* | +| 8 | `src/class/<cls>/*_device.[ch]` | `ALL` | examples enabling `CFG_TUD_<CLS>` | device-role boards → HIL tests enabling `CFG_TUD_<CLS>` | +| 9 | `src/class/<cls>/*_host.[ch]` | `ALL` | examples enabling `CFG_TUH_<CLS>` | host-role boards → HIL tests enabling `CFG_TUH_<CLS>` | +| 10 | `src/class/<cls>/**` (shared header) | `ALL` | either, **plus include-edge classes** | both roles → same, plus include-edge classes | +| 11 | `src/device/**` | `ALL` | `DEV`+`DUAL` | device-role boards → device+dual tests | +| 12 | `src/host/**` | `ALL` | `HOST`+`DUAL` | host-role boards → host+dual tests | +| 12b | `src/typec/**` | `ALL` | examples enabling `CFG_TUC_ENABLED` | — (no rig board runs a typec test) | +| 13 | `examples/<role>/<name>/**` | `ALL` | just `<name>` | if `<name>` is a HIL test: all boards → that test; else nothing | +| 14 | `examples/device/board_test/**` | `ALL` | just `board_test` | all boards → all tests (HIL parking firmware) | +| 15 | `examples/build_system/**`, `examples/CMakeLists.txt`, `examples/<role>/CMakeLists.txt` | `ALL` | `ALL` | all boards → all tests | +| 16 | `src/common/`, `src/osal/`, `src/tusb.[ch]`, `src/tusb_option.h`, `tools/{build,build_utils,ci_select}.py`, `tools/cmake/**`, `src/CMakeLists.txt`, `src/tinyusb.mk`, `hw/bsp/{family_support.{cmake,mk},family_rules.mk,zephyr_board_aliases.cmake,board.c,board_api.h,ansi_escape.h}`, `.github/**`, `.circleci/**` | `ALL` | `ALL` | all boards → all tests | +| 16a | `lib/<name>/**` | `ALL` | examples whose own `CMakeLists.txt`/`Makefile` names `lib/<name>` | those examples that are HIL tests, on all boards; empty resolves to nothing | +| 16b | `tools/get_deps.py` | families whose `deps_mandatory`/`deps_optional` entries changed | `ALL` | those families' boards → all tests; a logic change, an `'all'` entry, no base content or a changed token naming no family → full | +| 17 | anything unclassified (no tracked file reaches this — TestNoTrackedFileIsUnclassified) | `ALL` | `ALL` | all boards → all tests (fail-open) | """ import argparse import ast @@ -53,7 +85,10 @@ def _read(path: str) -> str: _NONCODE_RE = re.compile( - r'^(docs/|\.claude/|.*\.(md|rst)$|LICENSE)') + # LICENSE is anchored and LICENSES/ named separately: a bare `LICENSE` alternative + # also swallowed anything merely STARTING with it (a future LICENSE_extra.c), + # which is the silent-under-selection direction + r'^(docs/|\.claude/|.*\.(md|rst)$|LICENSE$|LICENSES/)') # Repo metadata and tooling that no CI build reads. Enumerated rather than left to # rule 17, which widens BOTH axes: a PR touching only .gitignore and a README was # creating 74 cmake legs (each a runner doing checkout + toolchain + get_deps before @@ -73,7 +108,11 @@ _META_RE = re.compile( r'version\.yml$|SConscript$|' r'.*CMakePresets\.json$|hw/bsp/BoardPresets\.json$|examples/west\.yml$|' r'.*/[0-9]+-tinyusb[^/]*\.rules$|tools/usb_drivers/|tools/codespell/|' - r'test/(fuzz|unit-test)/|' + # test/hil/test/ holds the harness's own unit tests, not the harness: nothing on + # the rig runs them (pre-commit does, and build.yml runs test_ci_select.py as the + # gate before trusting a selection), so they cannot change what the rig does. + # The harness itself stays under _FULL_RE's test/hil/ prefix. + r'test/(fuzz|unit-test)/|test/hil/test/|' # .github, minus the build machinery named in _FULL_RE r'\.github/(FUNDING\.yml$|labeler\.yml$|membrowse_pr_message\.j2$|ISSUE_TEMPLATE/|' r'workflows/(cifuzz|claude|claude-code-review|labeler|membrowse-comment|' @@ -93,7 +132,11 @@ _METRICS_RE = re.compile( r'^(tools/metrics[^/]*\.py$|\.github/scripts/metrics_[^/]*\.py$)') _FULL_RE = re.compile( r'^(src/common/|src/osal/|src/tusb\.c$|src/tusb\.h$|src/tusb_option\.h$|' - r'test/hil/|\.github/workflows/build.*\.yml$|\.github/actions/|\.github/scripts/|' + # tools/rtt.py is part of the harness, not a standalone tool: hil_util imports it + # at module load, so a break in it breaks every rig run the same way a test/hil/ + # edit can (the pre-commit hil-test hook runs its unit tests for the same reason) + r'test/hil/|tools/rtt\.py$|' + r'\.github/workflows/build.*\.yml$|\.github/actions/|\.github/scripts/|' # generates the whole CircleCI matrix, same authority as .github/** r'\.circleci/|' # rule 16 says `tools/build*.py`; name the two siblings the glob implies. Both @@ -152,10 +195,20 @@ def board_tests(board: dict) -> list: return [x for x in run if x not in t.get('skip', [])] + +def _rg(repo_root: str, *parts: str) -> str: + """A glob pattern rooted at repo_root, with the ROOT escaped and the parts left as + patterns. The root is a filesystem path, not a pattern: a checkout at + /w/pr[1]/tinyusb (a worktree named after a PR, a CI workspace with brackets) makes + an unescaped '[1]' a character class that matches nothing, and every lookup below + then resolves to zero - families=0 instead of 30, i.e. the selector fails CLOSED + and the whole matrix compiles nothing while reporting green.""" + return os.path.join(glob.escape(repo_root), *parts) + # cached: called per changed file x roster board, and the tree doesn't change mid-run @functools.lru_cache(maxsize=None) def board_family(board_name: str, repo_root: str): - hits = glob.glob(os.path.join(repo_root, 'hw/bsp/*/boards', board_name)) + hits = glob.glob(_rg(repo_root, 'hw/bsp/*/boards', board_name)) return os.path.basename(os.path.dirname(os.path.dirname(hits[0]))) if hits else None @@ -263,10 +316,10 @@ def _family_file_texts(repo_root: str) -> tuple: CMakeLists.txt, read once. path_families is called per distinct directory in the diff and its own cache only helps repeats: a 6,000-file hw/mcu dep bump re-read these 84 files 99,892 times (2.2 s) before this.""" - bsp_root = os.path.join(repo_root, 'hw/bsp') + bsp_root = os.path.join(repo_root, 'hw/bsp') # escaped by _rg below out = [] - for f in sorted(glob.glob(os.path.join(bsp_root, '*/family.cmake')) + - glob.glob(os.path.join(bsp_root, '*/components/*/CMakeLists.txt'))): + for f in sorted(glob.glob(_rg(bsp_root, '*/family.cmake')) + + glob.glob(_rg(bsp_root, '*/components/*/CMakeLists.txt'))): try: out.append((os.path.relpath(f, bsp_root).split(os.sep, 1)[0], _read(f))) except OSError: @@ -386,7 +439,7 @@ def class_include_edges(repo_root: str) -> dict: Derived from the actual #include lines rather than a hand-written table so it cannot rot when a class picks up or drops a cross-class include.""" edges = {} - for f in sorted(glob.glob(os.path.join(repo_root, 'src/class/*/*.[ch]'))): + for f in sorted(glob.glob(_rg(repo_root, 'src/class/*/*.[ch]'))): cls = os.path.basename(os.path.dirname(f)) try: text = _read(f) @@ -470,11 +523,23 @@ def _class_roles(base: str) -> set: return {'device', 'host'} -def _config_enables(cfg_path: str, macros) -> bool: [email protected]_cache(maxsize=None) +def _config_text(cfg_path: str) -> str: + """An example's tusb_config.h, read once. Every class path re-asks the same 46 + configs on both axes, so the reads go up with the diff: 4,240 of the same 46 files + for a diff touching all of src/class (0.48s -> 0.13s), and they cannot change + mid-run. Cached here rather than on _config_enables so the macros argument stays an + ordinary list at every call site.""" try: with open(cfg_path, encoding='utf-8', errors='replace') as f: - text = f.read() + return f.read() except OSError: + return '' + + +def _config_enables(cfg_path: str, macros) -> bool: + text = _config_text(cfg_path) + if not text: return False for m in macros: for value in re.findall(_DEF_VALUE.format(m), text, re.M): @@ -511,10 +576,13 @@ def lib_examples(lib_name: str, repo_root: str) -> set: pat = re.compile(re.escape('lib/' + lib_name) + r'(?=[/\s"\')}]|$)', re.M) out = set() for ex in all_examples(repo_root): - for f in sorted(glob.glob(os.path.join(repo_root, 'examples', ex, '**', '*'), + # the two filenames directly: '**/*' enumerated 489 entries per lib against a + # clean tree to use 107, and grows without bound once `make BOARD=... all` has + # written examples/<role>/<name>/_build/ - which is where /pre-pr runs + for f in sorted(glob.glob(_rg(repo_root, 'examples', ex, '**', 'CMakeLists.txt'), + recursive=True) + + glob.glob(_rg(repo_root, 'examples', ex, '**', 'Makefile'), recursive=True)): - if os.path.basename(f) not in ('CMakeLists.txt', 'Makefile'): - continue try: text = _read(f) except OSError: @@ -580,7 +648,7 @@ def _classify_one(path, repo_root, roster_boards, extras: set, s: _Sel, if _NONCODE_RE.match(path) or _META_RE.match(path): s.reasons.append(f'{path}: non-code, no contribution') return - if _METRICS_RE.match(path): + if _METRICS_RE.match(path): # rule 2b s.reasons.append(f'{path}: build-size metrics tooling, no HIL contribution') return if _FULL_RE.match(path): @@ -700,6 +768,16 @@ def _classify_one(path, repo_root, roster_boards, extras: set, s: _Sel, # only the tests whose example builds the lib, and only those the rig runs tests = {e for e in lib_examples(lib, repo_root) if any(e in pool for pool in ALL_TESTS.values()) or e in extras} + if lib == 'SEGGER_RTT': + # no example names this lib, but a board whose roster entry says + # "logger": "rtt" (variant defines LOGGER=rtt) reads EVERY test's console + # through it -- a break here silently breaks all of that board's rows + rtt_boards = [b['name'] for b in roster_boards if b.get('logger') == 'rtt'] + if rtt_boards: + s.roles.update(('device', 'host')) + s.add(rtt_boards, 'all', + f'{path}: SEGGER_RTT is the rtt console on {rtt_boards} -> all tests') + return if not tests: s.reasons.append(f'{path}: lib {lib} used by no HIL test, no contribution') return @@ -903,7 +981,13 @@ def main(): print(f'ci_select[build]: {r}', file=sys.stderr) for r in s['reasons']: print(f'ci_select: {r}', file=sys.stderr) - print(json.dumps(s)) + # reasons go to stderr ONLY - they are a human diagnostic and no consumer reads them + # back. They are also ~97% of the payload (a whole-tree diff: 453 KB -> 12 KB), which + # build.yml re-parses with ci_set_matrix, hil_ci_set_matrix, an inline python and + # three jq calls. The in-process dicts still carry them, for the log and the tests. + out = {k: v for k, v in s.items() if k != 'reasons'} + out['build'] = {k: v for k, v in s['build'].items() if k != 'reasons'} + print(json.dumps(out)) # ------------------------------------------------------------- @@ -925,7 +1009,7 @@ def all_examples(repo_root: str) -> tuple: """Every examples/<role>/<name> with a CMakeLists.txt, as 'role/name'.""" out = [] for role in _EX_ROLES: - for d in sorted(glob.glob(os.path.join(repo_root, 'examples', role, '*/'))): + for d in sorted(glob.glob(_rg(repo_root, 'examples', role, '*/'))): if os.path.isfile(os.path.join(d, 'CMakeLists.txt')): out.append(f'{role}/{os.path.basename(d.rstrip(os.sep))}') return tuple(out) @@ -979,13 +1063,13 @@ class _BSel: def _classify_build_one(path, repo_root, s: _BSel, get_deps_families=None): base = os.path.basename(path) - if _NONCODE_RE.match(path) or _META_RE.match(path): # rule 1 + if _NONCODE_RE.match(path) or _META_RE.match(path): # rules 1, 1b s.reasons.append(f'{path}: non-code, no build contribution') return if re.match(r'test/hil/', path): # rule 2 s.reasons.append(f'{path}: HIL harness, no build contribution') return - if path == GET_DEPS_PATH: # get_deps rule + if path == GET_DEPS_PATH: # rule 16b if get_deps_families is None: s.force_full(f'{path}: dep changes not resolvable -> full build matrix') return @@ -1002,6 +1086,7 @@ def _classify_build_one(path, repo_root, s: _BSel, get_deps_families=None): roles = _port_roles(base) exs = 'all' if roles == {'device', 'host'} else \ role_examples(repo_root, tuple(roles) + ('dual',)) + # rule 5b: fams empty -> s.add iterates nothing -> no contribution s.add(fams, exs, f'{path}: port {port} -> families {sorted(fams)}') return if re.match(r'hw/bsp/[^/]+/', path): # rule 6 @@ -1064,13 +1149,17 @@ def _classify_build_one(path, repo_root, s: _BSel, get_deps_families=None): s.add(all_bsp_families(repo_root), exs, f'{path}: typec -> {sorted(exs)}') return m = re.match(r'lib/([^/]+)/', path) - if m: # lib rule + if m: # rule 16a lib = m.group(1) exs = lib_examples(lib, repo_root) if not exs: - # empty means empty: no example's build pulls this lib in, so no build - # compiles it (lib/SEGGER_RTT is only reached through LOGGER=rtt, which - # no CI build sets) + # empty means empty: no example's build pulls this lib in, so no MAIN- + # matrix build compiles it. (lib/SEGGER_RTT is reached through LOGGER=rtt, + # which the main matrix never sets; the hil-build legs set it only for + # roster boards whose variant defines carry it, via the HIL SEGGER_RTT rule. + # No committed CI roster has such a board yet, so a SEGGER_RTT edit is + # currently neither built nor HIL-tested by CI -- verify vendor bumps + # manually until a rig board adopts "logger": "rtt".) s.reasons.append(f'{path}: lib {lib} built by no example, no contribution') return s.add(all_bsp_families(repo_root), exs, f'{path}: lib {lib} -> {sorted(exs)}') @@ -1150,11 +1239,25 @@ def _prune_buildable(fams, fam_ex, repo_root): # for anything else spins up CI's most expensive leg to skip every example # it was given. Identical to the unfiltered list on all 81 other families. pool = set(build_py.get_examples(fam)) + + # asked per example instead of materialising the family's whole buildable + # list: skip_example is by far the hottest call in the selector, and every + # question below short-circuits (one cdc_device.c diff: 6,883 calls -> 1,889) + def can_build(ex): + # EITHER build system: this one list gates CircleCI's make legs too, and + # the two answer differently (build_utils.skip_example) + return ex in pool and any( + not build_utils.skip_example(ex, b) or + not build_utils.skip_example(ex, b, (), 'make') for b in boards) + + want = fam_ex.get(fam) try: - buildable = [e for e in allex if e in pool and - any(not build_utils.skip_example(e, b) or - not build_utils.skip_example(e, b, (), 'make') - for b in boards)] + if want is None: + kept = None if any(can_build(e) for e in allex) else [] + else: + kept = [e for e in want if can_build(e)] + if kept and not any(can_build(e) for e in allex if e not in want): + kept = None # already everything the family can build except OSError as e: # a family mid-bring-up (boards/ but no family.cmake/family.mk yet) # reads as unbuildable to the scrape; keep it rather than tracebacking @@ -1162,13 +1265,10 @@ def _prune_buildable(fams, fam_ex, repo_root): reasons.append(f'{fam}: mcu scrape unreadable ({e}), kept unfiltered') out_fams.append(fam) continue - want = fam_ex.get(fam) - have = set(buildable) - kept = buildable if want is None else [e for e in want if e in have] - if not kept: + if kept == []: continue # this diff builds nothing for this family out_fams.append(fam) - if set(kept) != set(buildable): + if kept is not None: out_ex[fam] = kept return out_fams, out_ex, reasons diff --git a/tools/rtt.py b/tools/rtt.py new file mode 100644 index 000000000..e3aef36f2 --- /dev/null +++ b/tools/rtt.py @@ -0,0 +1,727 @@ +#!/usr/bin/env python3 +"""RTT console/capture over a debug probe — importable classes + CLI (the rtt +skill's SKILL.md is the manual). + +Three routes (see the skill's transport matrix for which route a probe gets). +--backend is always explicit: + + J-Link route (console/capture, channel 0 only) + rtt.py --backend jlink --probe <sn> --device <JLINK_DEVICE> [--seconds N] [-i] + OpenOCD route (native probes: ST-Link/CMSIS-DAP; console/capture, any channel) + rtt.py --backend openocd [--probe <sn>] [--vid-pid "0xVVVV 0xPPPP"] \\ + --cfg "-f interface/stlink.cfg -f target/stm32h7x.cfg" \\ + (--elf <flashed.elf> | --addr 0x2000xxxx) [--channel N] [--seconds N] [-i] + [--reset-before-attach] # capture from the target's boot (SystemView) + Post-mortem ring dump (J-Link, no halt — debug-AP reads) + rtt.py --backend jlink --dump <out.bin> --probe <sn> --device <JLINK_DEVICE> \\ + (--elf <flashed.elf> | --addr 0x...) + +The probe is owned for the whole run: flash and reset BEFORE starting this, never +reset the target while it is attached. Pin the probe: rigs and benches run +several (jlink: --probe serial; openocd: --probe and/or --vid-pid). + +The classes (JlinkRtt for J-Link, OpenocdRtt for openocd-driven probes) expose +the slice of pyserial the HIL harness uses — read/in_waiting/write/close/timeout, +reset_input_buffer, context-manager use, plus an `eof` latch — and are imported +by test/hil/helper/hil_util.py, so this file is HARNESS-CRITICAL: a change here +is classified like a test/hil/ harness change (tools/ci_select.py) and runs the +console unit tests (pre-commit hil-test hook, test/hil/test/test_hil_rtt.py). +Stdlib only — hil_util imports this file, never the other way around. +""" +import argparse +import contextlib +import os +import re +import select +import shlex +import signal +import socket +import subprocess +import sys +import tempfile +import threading +import time + + +class RttError(RuntimeError): + """Every way a console can break: stall, closed, dead or reset server. + + A RuntimeError subclass so existing `except RuntimeError` callers keep working, + but named so the harness can tell a console failure from an unrelated + NotImplementedError / 'dictionary changed size during iteration' and stop + reporting harness bugs as board failures.""" + + +def _pos_float_env(name: str, default: float) -> float: + # mirrors hil_util.pos_float_env, including its rejection of inf/nan: an infinite + # write timeout is an unbounded write, the very thing this knob exists to bound + raw = os.environ.get(name) + if raw is None: + return default + try: + v = float(raw) + except ValueError: + print(f'warning: {name} is not a number; using {default}', file=sys.stderr, flush=True) + return default + if not (v > 0 and v < float('inf')): + print(f'warning: {name}={v} is not usable; using {default}', file=sys.stderr, flush=True) + return default + return v + + +# whole-call deadline for write() — same env knob as the harness's serial twin +RTT_WRITE_TIMEOUT = _pos_float_env('HIL_SERIAL_WRITE_TIMEOUT', 10) + +# J-Link Commander's telnet greeting, sent at connect BEFORE (or without) the control +# block being found: never target output. Three lines; the middle one is the PROBE +# MODEL string, which in libjlinkarm carries no 'SEGGER ' prefix (J-Link OH3, +# J-Trace H9, ...) though some builds do prefix it — match both shapes. Consumers +# judging "did the target speak" must strip these lines first. +RTT_BANNER_RE = re.compile(r'^(SEGGER J-|J-Link[ 0-9]|J-Trace[ 0-9]|Process:\s)') + + +def strip_banner(data: bytes, complete_only: bool = False) -> bytes: + """Target bytes only: drop the J-Link server banner lines and blanks. + + Both harness consumers (hil_test's device_info verdict, hil_pool_check's + aliveness score) must judge "did the target speak" through this one filter, + or the same byte stream scores differently per consumer. complete_only=True + additionally drops a trailing unterminated line — for poll loops judging a + growing buffer, where a banner FRAGMENT at a read boundary (b'SEGG', b'Proce') + would defeat the prefix regex and count as target output; the final verdict + after the window should pass complete_only=False to keep a genuine + unterminated tail.""" + lines = data.splitlines(keepends=False) + if complete_only and data and not data.endswith((b'\n', b'\r')) and lines: + lines = lines[:-1] + return b'\n'.join(l for l in lines + if l.strip() and not RTT_BANNER_RE.match(l.decode('utf-8', errors='ignore'))) + + +def free_ports(count: int) -> list: + """Bind ephemeral ports and hand back the numbers. Boards run in parallel, so the + RTT/GDB ports cannot be the SEGGER defaults or two boards collide. + + Known TOCTOU: the port is free when released here, but another process can claim + it before the server binds it. Accepted — the server binds the port itself, so + there is no fd to hand over. The post-connect re-poll catches the common outcome + (our server lost the bind and died); a foreign listener that stays alive is not + detectable here and would need the connected peer to be validated.""" + socks = [] + try: + for _ in range(count): + s = socket.socket() + s.bind(('127.0.0.1', 0)) + socks.append(s) + return [s.getsockname()[1] for s in socks] + finally: + for s in socks: + s.close() + + +def nm_rtt_addr(elf: str, nm: str = None) -> int: + """Control-block address from the FLASHED elf's symbol table. --addr is the way + out when nm cannot read the file (another architecture, no toolchain).""" + nm = nm or os.environ.get('RTT_NM', 'arm-none-eabi-nm') + try: + r = subprocess.run([nm, elf], capture_output=True, text=True, timeout=30) + except FileNotFoundError: + raise SystemExit(f'{nm} not on PATH — set RTT_NM=<your-nm>, or pass --addr') + except subprocess.TimeoutExpired: + raise SystemExit(f'{nm} did not finish reading {elf} in 30 s — pass --addr instead') + if r.returncode != 0: + raise SystemExit(f'{nm} could not read {elf}: {r.stderr.strip()[:200]}\n' + f'(wrong architecture? set RTT_NM=<your-nm>, or pass --addr)') + for line in r.stdout.splitlines(): + # "<addr> <type> _SEGGER_RTT": a defined data symbol only — an undefined one + # (" U _SEGGER_RTT") has no address and would int('U', 16) + m = re.match(r'^([0-9a-fA-F]+)\s+[bBdD]\s+_SEGGER_RTT$', line.strip()) + if m: + return int(m.group(1), 16) + raise SystemExit(f'no defined _SEGGER_RTT symbol in {elf} — was it built with LOGGER=rtt?') + + +class _SocketRtt: + """Shared console core: a TCP socket onto an RTT server owned by self._proc. + + Subclasses build their server argv and call _spawn() + _connect() in __init__. + One failure contract: RttError for every way the console can break (stall, + closed, dead server) — callers are written for exactly it. A dead or resetting + server LATCHES `eof` rather than raising from the read side, so read loops and + the harness's `assert not ser.eof` triage see it without an exception racing + them to a generic handler.""" + + server = 'RTT server' # for error messages + + def __init__(self, timeout: float = 0.1): + self.timeout = timeout + self._buf = b'' + self._eof = False + self._sock = None + self._proc = None + self._log = None + self._lock = threading.Lock() # _buf is touched by the CLI pump thread too + + def _spawn(self, cmd: list, stdin=None) -> None: + # server output spools to a temp file: a PIPE nobody drains blocks a + # single-threaded server once 64 KiB of log accumulates (openocd at + # polling_interval 1 against a resetting target fills that in minutes) and + # the console goes silent with no error; the file also feeds _server_tail + self._log = tempfile.NamedTemporaryFile(prefix='rtt-server-', suffix='.log') + try: + self._proc = subprocess.Popen(cmd, stdin=stdin, stdout=self._log, + stderr=subprocess.STDOUT, start_new_session=True) + except FileNotFoundError as e: + self.close() + raise RttError(f'RTT console: {e.filename or cmd[0]} not on PATH') from e + except BaseException: + # any other spawn failure (PermissionError...) must not leak the log fd + self.close() + raise + + def _connect(self, port: int) -> None: + try: + deadline = time.monotonic() + 15 + while time.monotonic() < deadline: + try: + self._sock = socket.create_connection(('127.0.0.1', port), timeout=2) + break + except OSError: + if self._proc.poll() is not None: + break + time.sleep(0.2) + if self._sock is None: + tail = self._server_tail() + self.close() + raise RttError(f'RTT console: {self.server} did not serve port {port}{tail}') + if self._proc.poll() is not None: + # the connect succeeded but our server is dead: a foreign process claimed + # the port in the free_ports window — refuse a console wired to a stranger + self.close() + raise RttError(f'RTT console: {self.server} died after connect (port {port} hijacked?)') + self._sock.setblocking(False) + except (KeyboardInterrupt, SystemExit): + # a signal mid-construction must not orphan the server we just spawned + self.close() + raise + + def _server_tail(self) -> str: + log = getattr(self, '_log', None) + if log: + with contextlib.suppress(OSError, ValueError): + with open(log.name, 'rb') as fh: + tail = fh.read()[-400:].decode(errors='replace') + if tail: + return ' — ' + tail + return '' + + def _drain(self) -> None: + # LATCH, never raise: a peer reset or a socket closed under us ends the + # stream exactly like an orderly EOF. Raising here raced the harness's + # `assert not ser.eof` triage into a generic handler that re-flashes the + # board, and leaked ConnectionResetError/ValueError to in_waiting callers. + # the WHOLE body under the lock, not just the append: the CLI's -i pump thread + # and the read loop drain the same socket concurrently, and recv->append being + # non-atomic let chunks land out of order (measured: transposed 64-byte + # segments in 3/6 stress trials) + try: + with self._lock: + while self._sock and select.select([self._sock], [], [], 0)[0]: + try: + chunk = self._sock.recv(65536) + except (BlockingIOError, InterruptedError): + return + if not chunk: + self._eof = True + return + self._buf += chunk + except (OSError, ValueError, TypeError, AttributeError): + self._eof = True + + @property + def eof(self) -> bool: + """True once the server hung up AND everything it sent has been read out.""" + if self._sock is None: + return True + self._drain() + return self._eof and not self._buf + + @property + def in_waiting(self) -> int: + if self._sock is None: + # pyserial raises on a closed port; answering "N bytes waiting" from a + # closed dead console would let a caller bug look like a healthy board + raise RttError('RTT console is closed') + self._drain() + return len(self._buf) + + def read(self, size: int = 1) -> bytes: + if size is None or size <= 0: + # pyserial's read(0) returns b'' and consumes nothing; a negative size + # must not silently hand over (or destroy) buffered bytes + return b'' + if self._sock is None: + raise RttError('RTT console is closed') + self._drain() + deadline = None if self.timeout is None else time.monotonic() + self.timeout + while (len(self._buf) < size and not self._eof + and (deadline is None or time.monotonic() < deadline)): + time.sleep(0.005) + self._drain() + if self._eof and len(self._buf) < size: + # dead server: pace the empty returns like a serial timeout would, so a + # caller's read loop cannot busy-spin at 100% CPU (416k empty reads/s + # measured unpaced). timeout=None deliberately diverges from pyserial's + # block-forever: the eof latch makes "server is gone" knowable, and an + # eternal block on it helps nobody -- paced empties + .eof is the contract. + pace = self.timeout if self.timeout is not None else 0.1 + remaining = (deadline - time.monotonic()) if deadline is not None else pace + time.sleep(max(0.0, min(remaining, pace))) + with self._lock: + out, self._buf = self._buf[:size], self._buf[size:] + return out + + def reset_input_buffer(self) -> None: + # pyserial surface: the host tests flush pre-reset backlog through this + if self._sock is None: + raise RttError('RTT console is closed') + self._drain() + with self._lock: + self._buf = b'' + + def write(self, data: bytes) -> int: + # select+send, not sendall(): the socket is non-blocking for reads, and sendall() + # on a non-blocking socket raises BlockingIOError as soon as the send buffer is + # full, with no count of what already went out -- a caller cannot resume without + # duplicating bytes. Same reason serial_write_all treats a short write as fatal. + sock = self._sock # snapshot: close() from another thread nulls the attribute + if sock is None: + raise RttError('RTT console is closed') + self._drain() + if self._eof: + # TCP accepts exactly one send after peer death — without this the bytes + # would "succeed" into the void and the read timeout gets blamed on the target + raise RttError(f'RTT console write to a dead server ({self.server} gone)') + sent = 0 + deadline = time.monotonic() + RTT_WRITE_TIMEOUT + while sent < len(data): + if time.monotonic() > deadline: + raise RttError(f'RTT console write stalled after {sent}/{len(data)} bytes') + try: + if not select.select([], [sock], [], 0.1)[1]: + continue + sent += sock.send(data[sent:]) + except (BlockingIOError, InterruptedError): + continue + except (OSError, ValueError, TypeError, AttributeError) as e: + # peer death (BrokenPipe/ConnectionReset) or the socket closed under us + # mid-call: keep the class's one failure contract + raise RttError(f'RTT console write failed after {sent}/{len(data)} bytes: {e}') from e + return sent + + def _gentle_stop(self, proc) -> None: + """Subclass hook: ask the server to exit before the group takedown.""" + + def close(self) -> None: + self._eof = True # latch: post-close eof reads True, like a hung-up server + if getattr(self, '_sock', None): + self._sock.close() + self._sock = None + with self._lock: + self._buf = b'' # pyserial contract: nothing is readable after close + proc = getattr(self, '_proc', None) + if proc: + if proc.poll() is None: + self._gentle_stop(proc) + with contextlib.suppress(subprocess.TimeoutExpired): + proc.wait(timeout=5) + if proc and proc.poll() is None: + # own session (start_new_session), so the group takedown gets the server and + # anything it spawned; leaving one alive would hold the probe for the next test + try: + os.killpg(proc.pid, signal.SIGTERM) + proc.wait(timeout=5) + except (ProcessLookupError, PermissionError): + pass + except subprocess.TimeoutExpired: + with contextlib.suppress(ProcessLookupError, PermissionError): + os.killpg(proc.pid, signal.SIGKILL) + # reap, or the server stays a zombie for the caller's lifetime + with contextlib.suppress(subprocess.TimeoutExpired): + proc.wait(timeout=2) + if proc: + for pipe in (proc.stdin, proc.stdout): + if pipe: + with contextlib.suppress(OSError, ValueError): + pipe.close() + # the server spool file: one fd plus a /tmp file per console, and the server + # grows it while alive -- GC is not a release policy on a rig + log = getattr(self, '_log', None) + if log: + with contextlib.suppress(OSError, ValueError): + log.close() + self._log = None + + # a console dropped without close() must not hold the probe for the process's life + def __enter__(self): + return self + + def __exit__(self, *exc): + self.close() + + def __del__(self): + with contextlib.suppress(Exception): + self.close() + + +class JlinkRtt(_SocketRtt): + """Bidirectional console over SEGGER RTT channel 0, for J-Link probes (the only + console on boards whose probe has no VCOM or whose BSP has no UART). + + J-Link Commander (JLinkExe) owns the probe and serves RTT channel 0 on + -RTTTelnetPort -- what JLinkRTTClient talks to, minus its banner. It keeps + hunting for the control block and streams whatever the buffer already holds, + where JLinkRTTLogger searches once when it attaches and gives up. It also + carries input, which the host tests that drive a menu need. + + The probe is held for as long as this is open, so flashing and resetting the + board must happen before it is created or after close(). Select the probe by + serial: rigs run more than one.""" + + server = 'JLinkExe' + + def __init__(self, board: dict, timeout: float = 0.1): + super().__init__(timeout) + flasher = board['flasher'] + args = shlex.split(flasher.get('args', '')) + if '-device' not in args: + # fail with the real cause now: JLinkExe without a device blocks prompting + # and would surface 15 s later as a misleading port error + raise RttError(f'RTT console: no -device in flasher args: {flasher.get("args")!r}') + port = free_ports(1)[0] + # defaults first, the roster's args after so they can override (-if jtag, + # -JLinkScriptFile, an explicit -speed). NOTE: hil_flash orders it the other + # way (roster args first, its own -if/-speed last, so ITS defaults win) -- + # a roster override honored here is ignored by flash/reset; align them if a + # roster ever carries such args. -ExitOnError makes a failed target connect + # EXIT Commander + # (a clean error with the log tail) instead of leaving a banner-only console + cmd = ['JLinkExe', '-USB', str(flasher['uid']), '-if', 'swd', + '-JTAGConf', '-1,-1', '-speed', 'auto', '-NoGui', '1', + '-ExitOnError', '1', '-AutoConnect', '1', + *args, '-RTTTelnetPort', str(port)] + # stdin stays open: Commander exits when it runs out of input; close() writes + # 'exit' there. + self._spawn(cmd, stdin=subprocess.PIPE) + self._connect(port) + + def _gentle_stop(self, proc) -> None: + with contextlib.suppress(OSError, ValueError): + proc.stdin.write(b'exit\n') + proc.stdin.flush() + # close our pipe end in its own suppress: a BrokenPipe on the write above must + # not skip it (the base close also closes it for the server-already-dead path) + with contextlib.suppress(OSError, ValueError): + proc.stdin.close() + + +class OpenocdRtt(_SocketRtt): + """The console surface over an openocd `rtt server` (native probes: + ST-Link/CMSIS-DAP — never point openocd at ea4088's LPC-Link2, measured to + knock that probe off USB; other J-Link-OB probes untested). + + Exact control-block address (never a full-RAM scan), polling_interval 1 + (default 100 ms polling loses most of a busy stream), attach WITHOUT reset — + flash and reset before starting; `rtt start` needs the block to exist. + reset_before_attach opts into an in-session reset for streams that only + decode from byte 0 (SystemView).""" + + server = 'openocd' + + def __init__(self, cfg: str, addr: int, channel: int, serial_no: str = None, + vid_pid: str = None, timeout: float = 0.1, reset_before_attach: bool = False): + super().__init__(timeout) + port = free_ports(1)[0] + # argv, never a shell string: cfg/serial/vid_pid come from roster JSON and the + # command line, and a '$', backtick or quote in any of them would otherwise be + # substituted by the shell or break out of it + cmd = ['openocd', '-c', 'tcl_port disabled', '-c', 'gdb_port disabled', + '-c', 'telnet_port disabled'] + # probe pin: vid_pid keeps discovery from opening foreign usbfs nodes (a + # wedged one hangs the open), serial disambiguates same-model probes — + # both before the -f scripts, like hil_flash does + if vid_pid: + if not re.fullmatch(r'0x[0-9a-fA-F]{1,4} 0x[0-9a-fA-F]{1,4}', vid_pid.strip()): + # openocd only WARNS and exits 0 on a malformed value, so the pin + # silently does not apply and discovery reopens every usbfs node -- + # the convoy hil_flash.valid_vid_pid exists to stop + raise RttError(f'--vid-pid must be "0xVVVV 0xPPPP", got {vid_pid!r}') + cmd += ['-c', f'adapter usb vid_pid {vid_pid.strip()}'] + if serial_no: + cmd += ['-c', f'adapter serial {serial_no}'] + cmd += shlex.split(cfg) + cmd += ['-c', 'init'] + # opt-in: reset the target INSIDE this session, give it 2 s to boot, THEN + # attach and drain. The order is forced: `rtt start` needs the control block + # to already exist in RAM (the firmware creates it at init), and attaching + # ahead of the reset would latch the PREVIOUS run's stale block. Byte 0 still + # reaches the consumer because NO_BLOCK_SKIP retains the ring's HEAD: a boot + # burst bigger than the ring loses its tail until the drain catches up, never + # its first bytes -- which is the part a boot-anchored decoder needs + # (SystemView's Init record, carrying the timestamp frequency, is emitted once + # at boot; a mid-flight attach yields a stream no decoder can lock onto; size + # BUFFER_SIZE_UP to the boot burst if the tail matters too). Costs the tool's + # usual no-reset invariant, and is unsafe on parts where an in-session reset + # leaves the core held (SAMD5x DSU) or perturbs the target (WCH SDI). + if reset_before_attach: + cmd += ['-c', 'reset run', '-c', 'sleep 2000'] + cmd += ['-c', f'rtt setup 0x{addr:x} 0x800 "SEGGER RTT"', + '-c', 'rtt polling_interval 1', '-c', 'rtt start', + '-c', f'rtt server start {port} {channel}'] + self._spawn(cmd) + self._connect(port) + + def _gentle_stop(self, proc) -> None: + # no stdin channel to ask openocd to exit, and it keeps its listener up after + # the client disconnects: go straight to the group takedown instead of blocking + # the base class's 5 s wait on a process that has no reason to leave + with contextlib.suppress(ProcessLookupError, PermissionError): + os.killpg(proc.pid, signal.SIGTERM) + + +def dump_ring(probe: str, device: str, addr: int, out_path: str, channel: int = 0) -> int: + """Post-mortem: read aUp[channel]'s ring over the debug AP (no halt) via JLinkExe. + NO_BLOCK_SKIP means an undrained ring holds the FIRST KB after boot, not the + tail — interpretation rules in the target-debug skill.""" + if re.search(r'[\s"\']', out_path): + raise SystemExit(f'--dump path must not contain whitespace or quotes: {out_path!r} ' + f'(it is spliced into a JLinkExe script line)') + # a stale file from an earlier run must not satisfy the success check below + with contextlib.suppress(OSError): + os.remove(out_path) + # SEGGER_RTT_CB: acID[16], MaxNumUpBuffers, MaxNumDownBuffers, then aUp[] at 0x18, + # each ring 6 words {sName, pBuffer, SizeOfBuffer, WrOff, RdOff, Flags}. Read the + # counts with the descriptor so an out-of-range channel is rejected instead of + # reading whatever RAM follows the array. + jlink = ['JLinkExe', '-USB', probe, '-device', device, '-if', 'swd', + '-speed', '4000', '-NoGui', '1', '-AutoConnect', '1'] + + def _jlink_run(script: str): + # same clean-exit contract as nm_rtt_addr/_spawn: a missing binary or a wedged + # probe must not reach the CLI as a traceback + try: + return subprocess.run(jlink, input=script, capture_output=True, text=True, timeout=60) + except FileNotFoundError: + raise SystemExit('JLinkExe not on PATH — the --dump route needs J-Link Commander') + except subprocess.TimeoutExpired: + raise SystemExit('JLinkExe did not finish in 60 s — probe wedged or target unreachable?') + + script = f'mem32 {addr + 0x10:#x}, 2\nmem32 {addr + 0x18 + channel * 24:#x}, 6\nexit\n' + r = _jlink_run(script) + words = [] + for line in r.stdout.splitlines(): + # UNANCHORED: when the script arrives on stdin, some JLinkExe versions glue + # the 'J-Link>' prompt onto the result line with no newline between + m = re.search(r'([0-9A-Fa-f]{8}) = ((?:[0-9A-Fa-f]{8} ?)+)$', line.strip()) + if m: + words += [int(w, 16) for w in m.group(2).split()] + if len(words) < 8: + print(r.stdout[-500:], file=sys.stderr) + raise SystemExit(f'could not read the aUp[{channel}] descriptor — wrong control block address?') + max_up = words[0] + if not 0 < max_up <= 32: + raise SystemExit(f'control block at {addr:#x} looks uninitialized ' + f'(MaxNumUpBuffers={max_up}) — the target has not written to RTT yet, ' + f'or the address is wrong') + if channel >= max_up: + raise SystemExit(f'--channel {channel}: this firmware has {max_up} up-buffer(s) (0..{max_up - 1})') + _, pbuf, size, wroff, rdoff, _ = words[2:8] + if not pbuf or not size: + raise SystemExit(f'up-buffer {channel} is not initialized (pBuffer={pbuf:#x} size={size}) — ' + f'the target has not written to it yet') + script = f'savebin {out_path}, {pbuf:#x}, {size:#x}\nexit\n' + _jlink_run(script) + # JLinkExe exits 0 even when a command inside its script fails, so the only proof + # savebin worked is the file itself: it must hold the WHOLE ring, since a read that + # dies partway (probe disconnect, unreadable address) still leaves a short file that + # would otherwise be reported as a complete dump. Removing it also keeps the + # invariant above -- no stale file can satisfy a later run's check. + got = os.path.getsize(out_path) if os.path.exists(out_path) else 0 + if got < size: + with contextlib.suppress(OSError): + os.remove(out_path) + if got == 0: + raise SystemExit(f'savebin produced no data at {out_path} — probe or address problem') + raise SystemExit(f'savebin wrote {got}/{size} B to {out_path} (truncated dump removed) ' + f'— probe or address problem') + print(f'ring: {size} B at {pbuf:#x}, WrOff={wroff:#x} RdOff={rdoff:#x} -> {out_path}\n' + f'valid bytes wrap at WrOff; default NO_BLOCK_SKIP holds the FIRST data after ' + f'boot, not the tail', file=sys.stderr) + return 0 + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument('--backend', choices=['jlink', 'openocd'], required=True, + help='transport route — explicit, no default (skill transport matrix)') + ap.add_argument('--probe', help='probe serial (JLinkExe -USB / openocd "adapter serial")') + ap.add_argument('--vid-pid', help='openocd probe pin by USB IDs, e.g. "0x2e8a 0x000c" ' + '(with or instead of --probe)') + ap.add_argument('--device', help='JLINK_DEVICE from board.cmake/family.cmake (jlink backend)') + ap.add_argument('--cfg', help='openocd -f/-c args, e.g. "-f interface/stlink.cfg -f target/stm32h7x.cfg"') + ap.add_argument('--elf', help='the FLASHED elf: exact _SEGGER_RTT address via nm (openocd/--dump)') + ap.add_argument('--addr', help='SEGGER RTT control block address (hex), instead of --elf') + ap.add_argument('--channel', type=int, default=0, help='up-buffer index (0 console, 1 SysView)') + ap.add_argument('--seconds', type=float, default=0, help='capture duration; 0 = until Ctrl-C/EOF') + ap.add_argument('-i', '--interactive', action='store_true', help='forward stdin to the target') + ap.add_argument('--reset-before-attach', action='store_true', + help='openocd: reset the target inside the capture session so the ' + 'server is draining when it boots (needed for streams that must ' + 'include the boot preamble, e.g. SystemView); unsafe on SAMD5x/WCH') + ap.add_argument('--dump', metavar='OUT.bin', + help='post-mortem ring dump (jlink backend; needs --elf or --addr)') + args = ap.parse_args() + + if args.seconds < 0 or args.seconds != args.seconds: # negative or nan + ap.error(f'--seconds must be >= 0 (0 = until Ctrl-C/EOF), got {args.seconds}') + if args.channel < 0: + # a negative index would walk backwards off aUp[] into the control-block + # header and read garbage as a descriptor + ap.error(f'--channel must be >= 0, got {args.channel}') + + def rtt_addr(): + if args.addr: + try: + return int(args.addr, 16) + except ValueError: + ap.error(f'--addr must be hex, got {args.addr!r}') + if args.elf: + return nm_rtt_addr(args.elf) + ap.error('need --elf (flashed elf, address via nm) or --addr') + + if args.backend == 'jlink': + if args.reset_before_attach: + ap.error('--reset-before-attach is openocd-only (the J-Link route attaches ' + 'to a running target; flash and reset before starting it)') + if args.channel and not args.dump: + # -RTTTelnetPort serves the Terminal buffer only; --dump can read any ring + ap.error('the jlink backend streams channel 0 only (use --backend openocd ' + 'for another channel, or --dump to read one)') + if args.vid_pid: + ap.error('--vid-pid is openocd-only; J-Link probes are selected by serial (--probe)') + if not (args.probe and args.device): + ap.error('the jlink backend needs --probe and --device') + elif not (args.probe or args.vid_pid): + ap.error('the openocd backend needs --probe and/or --vid-pid') + + if args.dump: + if args.backend != 'jlink': + ap.error('--dump uses the jlink backend (debug-AP reads via JLinkExe)') + return dump_ring(args.probe, args.device, rtt_addr(), args.dump, args.channel) + + # install BEFORE the console exists: an external `timeout`/kill during the + # up-to-15 s connect window must still reach the cleanup below, or the openocd + # route leaves a server holding the probe and the port (JLinkExe would exit on + # stdin EOF; openocd has no such channel and its own session shields it) + def _terminate(signum, _frame): + raise KeyboardInterrupt + for _sig in (signal.SIGTERM, signal.SIGHUP): + with contextlib.suppress(ValueError, OSError): + signal.signal(_sig, _terminate) + + try: + if args.backend == 'openocd': + if not args.cfg: + ap.error('--backend openocd needs --cfg') + con = OpenocdRtt(args.cfg, rtt_addr(), args.channel, + serial_no=args.probe, vid_pid=args.vid_pid, + reset_before_attach=args.reset_before_attach) + else: + con = JlinkRtt({'flasher': {'uid': args.probe, 'args': f'-device {args.device}'}}, + timeout=0.1) + except RttError as e: + print(e, file=sys.stderr) + return 1 + except KeyboardInterrupt: + return 130 # constructors clean up after themselves on the way out + + saw_output = threading.Event() + forwarded = threading.Event() + if args.interactive: + def pump_stdin(): + # Hold input until the capture side has seen TARGET output (or 5 s for a + # quiet firmware): the J-Link telnet route silently DROPS client bytes + # until Commander locates the control block, so input forwarded at attach + # vanishes (measured on the rig: instant 'ping' lost, delayed 'ping' + # echoed). The gate must ignore the server's own banner — it arrives at + # connect, BEFORE the block is found. Raw os.read, not sys.stdin.buffer: + # bytes with no newline wait, and no BufferedReader lock — a daemon + # thread blocked holding that lock at interpreter shutdown aborts + # CPython (_enter_buffered_busy). + saw_output.wait(5) + try: + while True: + data = os.read(0, 4096) + if not data: + return + con.write(data) + forwarded.set() + except (RttError, OSError, ValueError): + return # console closed/stalled/dead; capture side reports the state + threading.Thread(target=pump_stdin, daemon=True).start() + + deadline = time.monotonic() + args.seconds if args.seconds else None + rc = 0 + seen = b'' # pre-release accumulator for the banner check only + try: + while deadline is None or time.monotonic() < deadline: + try: + chunk = con.read(con.in_waiting or 1) + except RttError as e: + print(f'rtt: {e}', file=sys.stderr) + rc = 1 + break + if chunk: + if args.interactive and not saw_output.is_set(): + # target data = anything past the J-Link banner's final line + # ('Process: <name>'); the openocd server has no banner + seen = (seen + chunk)[-65536:] + if args.backend != 'jlink': + saw_output.set() + else: + i = seen.find(b'Process: ') + j = seen.find(b'\n', i) if i >= 0 else -1 + if j >= 0 and len(seen) > j + 1: + saw_output.set() + sys.stdout.buffer.write(chunk) + sys.stdout.buffer.flush() + elif con.eof: + print('rtt: server closed the connection', file=sys.stderr) + rc = 1 + break + except KeyboardInterrupt: + pass + except BrokenPipeError: + # downstream consumer (head/grep -m) closed the pipe: a normal way to end a + # capture, not an error. Point stdout at devnull so interpreter shutdown does + # not raise on the final implicit flush. + os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno()) + finally: + if args.interactive and not forwarded.is_set(): + # only claim what is true: the gate releases after 5 s and forwards anyway, + # so "never forwarded" must come from the forwarded flag, not the gate + print('rtt: -i stdin was never forwarded to the target (no input arrived, ' + 'or the console closed first)', file=sys.stderr) + if args.interactive and not saw_output.is_set(): + print('rtt: no target output within the window', file=sys.stderr) + # a late TERM landing during the up-to-12 s teardown must not skip the kill + # escalation and orphan the server -- cleanup is committed at this point + for _sig in (signal.SIGTERM, signal.SIGHUP): + with contextlib.suppress(ValueError, OSError): + signal.signal(_sig, signal.SIG_IGN) + con.close() + return rc + + +if __name__ == '__main__': + sys.exit(main()) |