diff options
Diffstat (limited to 'tools')
| -rwxr-xr-x | tools/build.py | 1 | ||||
| -rwxr-xr-x | tools/ci_select.py | 26 | ||||
| -rw-r--r-- | tools/rtt.py | 727 |
3 files changed, 750 insertions, 4 deletions
diff --git a/tools/build.py b/tools/build.py index 0bb366e3d..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', ], diff --git a/tools/ci_select.py b/tools/ci_select.py index ca9d54c27..1526f2064 100755 --- a/tools/ci_select.py +++ b/tools/ci_select.py @@ -132,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 @@ -764,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 @@ -1139,9 +1153,13 @@ def _classify_build_one(path, repo_root, s: _BSel, get_deps_families=None): 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)}') 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()) |