diff --git a/.github/workflows/card_stall_native_artifact.yaml b/.github/workflows/card_stall_native_artifact.yaml new file mode 100644 index 0000000000..804e751def --- /dev/null +++ b/.github/workflows/card_stall_native_artifact.yaml @@ -0,0 +1,89 @@ +name: Card stall native artifact qualification + +on: + push: + branches: [diagnostics/card-stall-build-20261012] + paths: + - .github/workflows/card_stall_native_artifact.yaml + - tools/ci/card_stall_native_artifact/** + - msgq_repo/** + - openpilot/selfdrive/car/card.py + - openpilot/starpilot/card_stall_timing.py + - openpilot/starpilot/aol/tests/test_card_stall_timing.py + +permissions: + contents: read + +concurrency: + group: card-stall-native-artifact-${{ github.ref }} + cancel-in-progress: false + +jobs: + native_arm64: + if: github.repository == 'firestar5683/StarPilot' && github.ref == 'refs/heads/diagnostics/card-stall-build-20261012' + runs-on: ubuntu-24.04-arm + timeout-minutes: 25 + env: + PYTHONNOUSERSITE: '1' + UV_NO_PROGRESS: '1' + ARTIFACT_DIR: ${{ runner.temp }}/card-stall-native-artifact + steps: + - name: Fetch exact public source with no archive fallback + shell: bash + env: + GIT_TERMINAL_PROMPT: '0' + GIT_LFS_SKIP_SMUDGE: '1' + GIT_CONFIG_NOSYSTEM: '1' + GIT_CONFIG_GLOBAL: /dev/null + run: | + set -euo pipefail + test "$GITHUB_REPOSITORY" = firestar5683/StarPilot + test "$GITHUB_REF" = refs/heads/diagnostics/card-stall-build-20261012 + [[ "$GITHUB_SHA" =~ ^[0-9a-f]{40}$ ]] + python3 -c 'import re, subprocess; v = subprocess.check_output(["git", "version"], text=True); m = re.search(r"(\d+)\.(\d+)\.(\d+)", v); assert m and tuple(map(int, m.groups())) >= (2, 28, 0), v; print(v.strip())' + test ! -e .git + git init + git remote add origin https://github.com/firestar5683/StarPilot.git + git config remote.origin.promisor true + git config remote.origin.partialclonefilter blob:none + git sparse-checkout init --no-cone + printf '%s\n' '/msgq_repo/' '/openpilot/selfdrive/car/card.py' '/openpilot/starpilot/card_stall_timing.py' '/openpilot/starpilot/aol/tests/test_card_stall_timing.py' '/tools/ci/card_stall_native_artifact/' '/uv.lock' '/pyproject.toml' '/.github/workflows/card_stall_native_artifact.yaml' | git sparse-checkout set --no-cone --stdin + git -c protocol.version=2 fetch --filter=blob:none --depth=2 origin "$GITHUB_SHA" + test "$(git rev-parse FETCH_HEAD)" = "$GITHUB_SHA" + test "$(git rev-parse FETCH_HEAD^)" = e8389b0b7529bad402f1f4e5fbd672ff13d9d382 + git checkout --detach "$GITHUB_SHA" + test ! -e openpilot/selfdrive/modeld/models + - name: Verify frozen source and baseline artifacts + run: python3 tools/ci/card_stall_native_artifact/verify_source.py "$ARTIFACT_DIR/provenance" + - name: Install bounded native build prerequisites + run: | + sudo apt-get update + sudo apt-get install -y --no-install-recommends gcc-13 g++-13 binutils libzmq5 + - name: Install pinned Python environment manager + uses: astral-sh/setup-uv@e92bafb6253dcd438e0484186d7669ea7a8ca1cc # v6.4.3 + with: + version: '0.12.5' + enable-cache: false + - name: Qualify source, build and execute native ARM tests + shell: bash + env: + EXPECTED_BASE_COMMIT: e8389b0b7529bad402f1f4e5fbd672ff13d9d382 + EXPECTED_SOURCE_COMMIT: ${{ github.sha }} + EXPECTED_GCC_VERSION: '13.3.0' + GCC: /usr/bin/gcc-13 + GXX: /usr/bin/g++-13 + run: | + uv python install 3.12.13 + uv venv --python 3.12.13 "$RUNNER_TEMP/card-stall-native-venv" + export PYTHON="$RUNNER_TEMP/card-stall-native-venv/bin/python" + uv pip install --python "$PYTHON" --require-hashes -r tools/ci/card_stall_native_artifact/requirements.lock + bash tools/ci/card_stall_native_artifact/build_native.sh "$PWD" "$ARTIFACT_DIR/provenance/baseline-ipc_pyx.so" "$ARTIFACT_DIR/build" + - name: Retain artifact and complete qualification evidence + if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2 + with: + name: card-stall-native-arm64-${{ github.sha }} + path: ${{ env.ARTIFACT_DIR }}/ + if-no-files-found: error + retention-days: 7 + compression-level: 6 diff --git a/msgq_repo/msgq/ipc_pyx.pyx b/msgq_repo/msgq/ipc_pyx.pyx index b16e9702e9..13585c54c5 100644 --- a/msgq_repo/msgq/ipc_pyx.pyx +++ b/msgq_repo/msgq/ipc_pyx.pyx @@ -7,9 +7,49 @@ from libcpp.vector cimport vector from libcpp cimport bool from libc cimport errno from libc.string cimport strerror +from libc.stdint cimport int64_t, uint64_t from cython.operator import dereference +cdef extern from "Python.h": + unsigned long PyThread_get_thread_ident() noexcept + + +cdef extern from *: + """ + #include + #include + #include + + // Diagnostics must neither raise nor disturb the backend's errno. + static inline void msgq_receive_clock_pair(int64_t *wall, int64_t *cpu) { + const int saved_errno = errno; + struct timespec ts; + *wall = clock_gettime(CLOCK_MONOTONIC, &ts) == 0 + ? (int64_t)ts.tv_sec * 1000000000LL + ts.tv_nsec : -1; + *cpu = clock_gettime(CLOCK_THREAD_CPUTIME_ID, &ts) == 0 + ? (int64_t)ts.tv_sec * 1000000000LL + ts.tv_nsec : -1; + errno = saved_errno; + } + """ + void msgq_receive_clock_pair(int64_t *wall, int64_t *cpu) noexcept nogil + + +cdef struct ReceiveTiming: + uint64_t sequence + bool non_blocking + bool returned_message + size_t byte_length + int64_t before_wall + int64_t before_cpu + int64_t backend_return_wall + int64_t backend_return_cpu + int64_t gil_reacquired_wall + int64_t gil_reacquired_cpu + int64_t python_return_wall + int64_t python_return_cpu + + from .ipc cimport Context as cppContext from .ipc cimport SubSocket as cppSubSocket from .ipc cimport PubSocket as cppPubSocket @@ -161,6 +201,14 @@ cdef class Poller: cdef class SubSocket: cdef cppSubSocket * socket cdef bool is_owner + cdef bool receive_timing_enabled + cdef unsigned long receive_timing_owner + cdef ReceiveTiming receive_timing_ring[64] + cdef unsigned int receive_timing_next + cdef unsigned int receive_timing_count + cdef uint64_t receive_timing_sequence + cdef uint64_t receive_timing_omitted + cdef uint64_t receive_timing_generation def __cinit__(self): with nogil: @@ -198,8 +246,87 @@ cdef class SubSocket: with nogil: self.socket.setTimeout(timeout) + cdef check_receive_timing_owner(self): + if self.receive_timing_owner != 0 and self.receive_timing_owner != PyThread_get_thread_ident(): + raise RuntimeError("receive timing must be managed by the thread that enabled it") + + def enable_receive_timing(self, bool enabled): + """Opt in on the receiving thread; disabling preserves previous records.""" + self.check_receive_timing_owner() + if enabled: + self.receive_timing_owner = PyThread_get_thread_ident() + self.receive_timing_enabled = enabled + + def reset_receive_timing(self): + """Explicitly clear records and counts, without changing enablement.""" + self.check_receive_timing_owner() + self.receive_timing_next = 0 + self.receive_timing_count = 0 + self.receive_timing_sequence = 0 + self.receive_timing_omitted = 0 + self.receive_timing_generation += 1 + + def take_receive_timing(self): + """Return (oldest-to-newest records, overwritten count), without clearing. + + Each record is (sequence, non_blocking, returned_message, byte_length, + before_mono_ns, before_thread_cpu_ns, backend_return_mono_ns, + backend_return_thread_cpu_ns, gil_reacquired_mono_ns, + gil_reacquired_thread_cpu_ns, python_return_mono_ns, + python_return_thread_cpu_ns). Failed clocks are -1. The final interval + includes the bytes copy and message cleanup; gil_handoff includes scheduling. + """ + self.check_receive_timing_owner() + cdef unsigned int i + cdef unsigned int start = (self.receive_timing_next + 64 - self.receive_timing_count) % 64 + cdef ReceiveTiming *record + records = [] + for i in range(self.receive_timing_count): + record = &self.receive_timing_ring[(start + i) % 64] + records.append((record.sequence, record.non_blocking, record.returned_message, record.byte_length, + record.before_wall, record.before_cpu, record.backend_return_wall, record.backend_return_cpu, + record.gil_reacquired_wall, record.gil_reacquired_cpu, record.python_return_wall, record.python_return_cpu)) + return tuple(records), self.receive_timing_omitted + + cdef receive_timed(self, bool non_blocking): + cdef cppMessage *msg + cdef ReceiveTiming record + cdef uint64_t generation = self.receive_timing_generation + record.non_blocking = non_blocking + with nogil: + msgq_receive_clock_pair(&record.before_wall, &record.before_cpu) + msg = self.socket.receive(non_blocking) + msgq_receive_clock_pair(&record.backend_return_wall, &record.backend_return_cpu) + msgq_receive_clock_pair(&record.gil_reacquired_wall, &record.gil_reacquired_cpu) + + record.returned_message = msg != NULL + record.byte_length = 0 + if msg == NULL: + m = None + else: + record.byte_length = msg.getSize() + m = msg.getData()[:record.byte_length] + with nogil: + del msg + + msgq_receive_clock_pair(&record.python_return_wall, &record.python_return_cpu) + # All shared ring access holds the GIL. Only this stack-local record is + # written while nogil; no Python object is created for the receipt here. + if generation == self.receive_timing_generation: + self.receive_timing_sequence += 1 + record.sequence = self.receive_timing_sequence + self.receive_timing_ring[self.receive_timing_next] = record + self.receive_timing_next = (self.receive_timing_next + 1) % 64 + if self.receive_timing_count < 64: + self.receive_timing_count += 1 + else: + self.receive_timing_omitted += 1 + return m + def receive(self, bool non_blocking=False): cdef cppMessage *msg + if self.receive_timing_enabled and self.receive_timing_owner == PyThread_get_thread_ident(): + return self.receive_timed(non_blocking) with nogil: msg = self.socket.receive(non_blocking) diff --git a/msgq_repo/msgq/tests/test_receive_timing.py b/msgq_repo/msgq/tests/test_receive_timing.py new file mode 100644 index 0000000000..dd3bd288be --- /dev/null +++ b/msgq_repo/msgq/tests/test_receive_timing.py @@ -0,0 +1,131 @@ +import ctypes +import threading +import time +import uuid + +import msgq +import pytest + + +def sockets(timeout=10): + endpoint = f"receive_timing_{uuid.uuid4().hex}" + pub = msgq.pub_sock(endpoint) + sub = msgq.sub_sock(endpoint, timeout=timeout) + return pub, sub + + +def assert_clocks(record): + assert len(record) == 12 + assert all(t >= 0 for t in record[4:]) + assert list(record[4::2]) == sorted(record[4::2]) + assert list(record[5::2]) == sorted(record[5::2]) + + +def test_receive_timing_opt_in_and_preserves_messages(): + pub, sub = sockets() + for payload in (b"x", b"unchanged\x00message"): + pub.send(payload) + assert sub.receive() == payload + assert sub.receive(non_blocking=True) is None + assert sub.take_receive_timing() == ((), 0) + + sub.enable_receive_timing(True) + for payload in (b"x", b"unchanged\x00message"): + pub.send(payload) + assert sub.receive() == payload + assert sub.receive(non_blocking=True) is None + assert sub.receive() is None + records, omitted = sub.take_receive_timing() + assert omitted == 0 + assert [r[:4] for r in records] == [(1, False, True, 1), (2, False, True, 17), + (3, True, False, 0), (4, False, False, 0)] + for record in records: + assert_clocks(record) + assert sub.take_receive_timing() == (records, omitted) + + sub.enable_receive_timing(False) + pub.send(b"disabled again") + assert sub.receive() == b"disabled again" + assert sub.receive(non_blocking=True) is None + assert sub.take_receive_timing() == (records, omitted) + sub.reset_receive_timing() + assert sub.take_receive_timing() == ((), 0) + + +def test_receive_timing_bounded_ring_and_explicit_reset(): + _, sub = sockets() + sub.enable_receive_timing(True) + for _ in range(150): + assert sub.receive(non_blocking=True) is None + records, omitted = sub.take_receive_timing() + assert len(records) == 64 + assert omitted == 86 + assert [r[0] for r in records] == list(range(87, 151)) + sub.reset_receive_timing() + assert sub.take_receive_timing() == ((), 0) + assert sub.receive(non_blocking=True) is None + assert sub.take_receive_timing()[0][0][0] == 1 + + +def test_receive_timing_thread_owner_and_socket_isolation(): + _, sub = sockets() + _, other_sub = sockets() + sub.enable_receive_timing(True) + failures = [] + + def other_thread(): + try: + # The receive itself remains valid and uninstrumented on other threads. + assert sub.receive(non_blocking=True) is None + for op in (sub.take_receive_timing, sub.reset_receive_timing, + lambda: sub.enable_receive_timing(False), lambda: sub.enable_receive_timing(True)): + with pytest.raises(RuntimeError, match="thread that enabled"): + op() + except BaseException as exc: + failures.append(exc) + + worker = threading.Thread(target=other_thread) + worker.start() + worker.join(timeout=3) + assert not worker.is_alive() + assert not failures + assert sub.take_receive_timing() == ((), 0) + assert sub.receive(non_blocking=True) is None + assert len(sub.take_receive_timing()[0]) == 1 + assert other_sub.receive(non_blocking=True) is None + assert other_sub.take_receive_timing() == ((), 0) + + +def test_receive_timing_exposes_gil_handoff_delay(): + # PyDLL deliberately retains the main thread's GIL while libc sleeps. + # The receiving thread's native timeout can expire independently of it. + libc = ctypes.PyDLL(None) + libc.usleep.argtypes = [ctypes.c_uint] + libc.usleep.restype = ctypes.c_int + ready = threading.Event() + results = [] + failures = [] + + def receiver(): + try: + _, sub = sockets(timeout=40) + sub.enable_receive_timing(True) + ready.set() + assert sub.receive() is None + results.append(sub.take_receive_timing()[0][0]) + except BaseException as exc: + failures.append(exc) + + worker = threading.Thread(target=receiver) + worker.start() + assert ready.wait(timeout=3) + time.sleep(0.01) + assert libc.usleep(150_000) == 0 + worker.join(timeout=3) + assert not worker.is_alive() + assert not failures + record = results[0] + assert_clocks(record) + assert record[6] - record[4] >= 30_000_000 + assert record[8] - record[6] >= 50_000_000 + assert record[9] - record[7] < 20_000_000 diff --git a/openpilot/selfdrive/car/card.py b/openpilot/selfdrive/car/card.py index 4b6cb75c4c..cc44d2370f 100755 --- a/openpilot/selfdrive/car/card.py +++ b/openpilot/selfdrive/car/card.py @@ -13,6 +13,7 @@ from opendbc.car.structs import car from openpilot.common.params import Params from openpilot.starpilot.card_loop_timing import CardLoopTiming +from openpilot.starpilot.card_stall_timing import CardStallTiming from openpilot.common.realtime import config_realtime_process, drop_realtime, set_core_affinity, Priority, Ratekeeper from openpilot.common.swaglog import cloudlog, ForwardingHandler @@ -442,6 +443,12 @@ class Car: tracker.claim(gesture, now_ns, key) def timing_mark(self, name): + try: + stall = getattr(self, 'stall_timing', None) + if stall is not None: + stall.mark(name) + except Exception: + pass # Observations must never interrupt control or its fail-closed paths. timing = getattr(self, 'loop_timing', None) if timing is not None and getattr(timing, 'detailed', True): timing.mark(name) @@ -455,6 +462,12 @@ class Car: can_received_ns = time.monotonic_ns() can_list = can_capnp_to_list(can_strs) self.timing_mark('can_decode_end') + try: + stall = getattr(self, 'stall_timing', None) + if stall is not None: + stall.observe_can(can_strs, can_list) + except Exception: + pass # Update carState from CAN CS = self.CI.update(can_list) @@ -920,6 +933,12 @@ class Car: self.timing_mark('car_state_send_start') self.pm.send('carState', cs_send) self.timing_mark('car_state_send_end') + try: + stall = getattr(self, 'stall_timing', None) + if stall is not None: + stall.published(int(cs_send.logMonoTime), self.slc_producer_session) + except Exception: + pass self.timing_source_car_state_ns = int(cs_send.logMonoTime) timing = getattr(self, 'loop_timing', None) if timing is not None: @@ -1402,6 +1421,13 @@ class Car: set_core_affinity([0, 1, 2, 3]) next_cruise_read = 0.0 while not evt.is_set(): + try: + stall = getattr(self, 'stall_timing', None) + if stall is not None: + stall.worker_mark('settings_start') + stall.flush(cloudlog.event) + except Exception: + pass timing = getattr(self, 'loop_timing', None) report = timing.take_report() if timing is not None else None if report is not None: @@ -1418,9 +1444,20 @@ class Car: self.slc_requested_configuration = self.read_slc_configuration() self.v_cruise_helper.intervals = read_cruise_intervals(self.params, pcm_cruise=self.CP.pcmCruise) next_cruise_read = now + 1.0 + try: + if stall is not None: + stall.worker_mark('settings_end') + except Exception: + pass time.sleep(0.1) def card_thread(self): + self.stall_timing = None + try: + if self.aol_replay and not REPLAY and ioniq6_long_eligible(self.CP): + self.stall_timing = CardStallTiming(self.can_sock) + except Exception: + pass self.loop_timing = (CardLoopTiming(detailed=os.getenv('CARD_PHASE_TIMING') == '1') if self.aol_replay and not REPLAY else None) self.timing_thread_id = threading.get_native_id() @@ -1429,6 +1466,11 @@ class Car: try: t.start() while True: + try: + if self.stall_timing is not None: + self.stall_timing.begin() + except Exception: + pass timing = self.loop_timing if timing is not None and timing.detailed: timing.begin() @@ -1436,6 +1478,11 @@ class Car: self.timing_mark('startup_check_end') self.step() self.rk.monitor_time() + try: + if self.stall_timing is not None: + self.stall_timing.finish() + except Exception: + pass if timing is not None and timing.detailed: timing.finish({'source_car_state_ns': getattr(self, 'timing_source_car_state_ns', 0), 'source_control_ns': int(self.sm.logMonoTime['carControl']), diff --git a/openpilot/starpilot/aol/tests/test_card_stall_timing.py b/openpilot/starpilot/aol/tests/test_card_stall_timing.py new file mode 100644 index 0000000000..399ffb8e86 --- /dev/null +++ b/openpilot/starpilot/aol/tests/test_card_stall_timing.py @@ -0,0 +1,446 @@ +"""Bounded, exception-isolated Card observations never grant authority.""" +from contextlib import ExitStack +from types import SimpleNamespace +from typing import Any + +import pytest + +from openpilot.starpilot.card_stall_timing import CardStallTiming + +BASE = 10_000_000_000 +MS = 1_000_000 + + +class Clock: + def __init__(self): + self.wall = BASE + self.cpu = 0 + self.calls = 0 + + def __call__(self): + self.calls += 1 + return self.wall, self.cpu + + def advance(self, delta, cpu=0): + self.wall += delta + self.cpu += cpu + + +class Socket: + def __init__(self): + self.enabled = False + self.takes = 0 + + def enable_receive_timing(self, enabled): + self.enabled = enabled + + def take_receive_timing(self): + self.takes += 1 + return (), 0 + + +def recorder(): + clock = Clock() + return CardStallTiming(Socket(), clock=clock), clock + + +def frame(timing, clock, delta, source=None, session='producer'): + timing.begin() + clock.advance(delta) + timing.mark('car_state_send_end') + timing.published(clock.wall if source is None else source, session) + timing.finish() + + +def report(timing): + result = [] + timing.last_attempt = None + timing.flush(lambda *args, **kwargs: result.append(kwargs)) + assert len(result) == 1 + return result[0] + + +def test_cross_loop_tail_and_head_gap_is_captured_before_loop_end(): + timing, clock = recorder() + timing.begin() + clock.advance(MS) + timing.published(clock.wall, 'producer') + clock.advance(19 * MS) + timing.mark('controller_apply_end') + timing.finish() + timing.begin() + clock.advance(18 * MS, cpu=2 * MS) + timing.mark('submaster_end') + timing.published(clock.wall, 'producer') + result = report(timing) + assert result['source_gap_ns'] == result['send_gap_ns'] == 37 * MS + assert result['previous_publication_frame_retained'] and not result['current_frame_complete'] + assert result['previous_frame'][-1][0] == 'loop_end' + assert result['current_frame'][-1][0] == 'submaster_end' + assert result['current_frame'][-1][2] - result['current_frame'][0][2] == 2 * MS + saved = result['current_frame'] + for _ in range(4): + frame(timing, clock, MS) + assert result['current_frame'] == saved and timing.socket.takes == 1 + + +@pytest.mark.parametrize('extra, emitted', [(0, False), (1, True)]) +def test_strict_threshold(extra, emitted): + timing, clock = recorder() + frame(timing, clock, MS) + frame(timing, clock, 30 * MS + extra) + assert bool(timing.pending) == emitted + + +@pytest.mark.parametrize('source_gap, send_gap', [(37 * MS, 21 * MS), (14 * MS, 33 * MS)]) +def test_source_and_send_clocks_remain_separate(source_gap, send_gap): + timing, clock = recorder() + frame(timing, clock, MS, source=BASE) + frame(timing, clock, send_gap, source=BASE + source_gap) + result = report(timing) + assert result['source_gap_ns'] == source_gap and result['send_gap_ns'] == send_gap + assert result['source_clock'] == 'CLOCK_MONOTONIC' and result['clock'] == 'CLOCK_MONOTONIC' + assert len(result['clock_mapping_mono_boot_mono']) == 3 + + +def test_sessions_missing_sources_and_clock_reversal_reset_baseline(): + timing, clock = recorder() + frame(timing, clock, MS) + frame(timing, clock, 40 * MS, session='replacement') + frame(timing, clock, 40 * MS, source=0, session='replacement') + frame(timing, clock, 40 * MS, session='replacement') + clock.wall -= 100 * MS + frame(timing, clock, MS, session='replacement') + assert not timing.pending + frame(timing, clock, 31 * MS, session='replacement') + assert report(timing)['source_gap_ns'] == 31 * MS + + +def test_bounds_throttle_queue_full_and_quiet_loss_summary(): + timing, clock = recorder() + frame(timing, clock, MS) + frame(timing, clock, 31 * MS) + frame(timing, clock, 31 * MS) + frame(timing, clock, timing.INTERVAL_NS) + assert len(timing.pending) == 1 + assert timing.late_count == 3 and timing.throttled_count == 1 and timing.queue_full_count == 1 + for _ in range(100): + timing.mark('can_receive_end') + assert max(timing.lengths) == timing.MAX_MARKS and timing.omitted_marks > 0 + result = report(timing) + assert result['counters']['queue_full_count'] == 1 + timing.errors += 1 + result = report(timing) + assert 'source_gap_ns' not in result and result['counters']['capture_errors'] == 1 + assert len(timing.frames) == 2 and all(len(f) == timing.MAX_MARKS for f in timing.frames) + + +def test_nonpublishing_frames_and_worker_snapshot_are_explicit(): + timing, clock = recorder() + frame(timing, clock, MS) + timing.worker_mark('settings_start') + for _ in range(3): + timing.begin() + clock.advance(10 * MS) + timing.finish() + frame(timing, clock, MS) + old = timing.worker + timing.worker_mark('settings_end') + result = report(timing) + assert result['nonpublishing_frames'] == 3 and not result['previous_publication_frame_retained'] + assert result['worker_progress'] == old + + +def test_missing_native_api_is_reported_and_python_capture_remains_honest(): + clock = Clock() + timing = CardStallTiming(object(), clock=clock) + assert not timing.native_available + result = report(timing) + assert not result['native_available'] and result['error'] + frame(timing, clock, MS) + frame(timing, clock, 31 * MS) + result = report(timing) + assert result['native'] is None and not result['native_available'] + + +def test_clock_snapshot_and_logging_failures_do_not_escape(monkeypatch): + timing, clock = recorder() + def fail(*args, **kwargs): + raise RuntimeError('observer failure') + timing.clock = fail + timing.begin() + timing.mark('can_receive_end') + timing.finish() + timing.published(BASE, 'producer') + timing.worker_mark('settings_start') + assert timing.errors == 5 + timing.clock = clock + frame(timing, clock, MS) + timing.socket.take_receive_timing = fail + frame(timing, clock, 31 * MS) + assert timing.errors == 6 + monkeypatch.setattr('openpilot.starpilot.card_stall_timing.time.monotonic_ns', lambda: clock.wall) + for _ in range(100): + timing.flush(fail) + assert timing.log_errors == 1 + clock.advance(timing.INTERVAL_NS) + timing.flush(fail) + assert timing.log_errors == 2 + + +def test_unselected_marks_do_not_read_clocks(): + timing, clock = recorder() + before = clock.calls + for _ in range(100): + timing.mark('unselected') + assert clock.calls == before + + +@pytest.mark.parametrize('full_op', [False, True]) +def test_actual_card_publication_bytes_and_order_are_unchanged(full_op): + from openpilot.common.prefix import OpenpilotPrefix + from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture + case = fixture.IoniqAvailabilityTests(methodName='runTest') + with ExitStack() as cleanup, OpenpilotPrefix(): + cleanup.callback(case.doCleanups) + sd, _, step, daemon, _ = case._ioniq_availability_callers(full_op=full_op) + axis, _, command = step(2) + assert axis.lateralActive and command.latActive and axis.longitudinalActive == full_op + state = fixture.car_state() + daemon.car_params_published = True + daemon.wheel_commands = () + daemon.initialized_prev = True + daemon.timing_thread_id = 7 + initial_sequence = daemon.aol_sequence + results = [] + for active in (False, True): + timing, clock = recorder() + clock.wall = BASE + 20 * MS + daemon.stall_timing = timing if active else None + daemon.aol_sequence = initial_sequence + output = [] + daemon.pm = SimpleNamespace(send=lambda name, msg, output=output: output.append((name, msg.to_bytes()))) + timing.begin() + daemon.state_publish(state, None) + timing.finish() + results.append(output) + if active: + assert timing.last_publication[1] == BASE + 20 * MS + assert results[0] == results[1] + services = [name for name, _ in results[1]] + assert services.index('carOutput') < services.index('aolIntentWire') < services.index('carState') + assert sd.aol_axis_decision.lateral_active and sd.aol_axis_decision.longitudinal_active == full_op + + +def test_failed_send_does_not_advance_successful_publication_baseline(): + from openpilot.common.prefix import OpenpilotPrefix + from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture + case = fixture.IoniqAvailabilityTests(methodName='runTest') + with ExitStack() as cleanup, OpenpilotPrefix(): + cleanup.callback(case.doCleanups) + _, _, _, daemon, _ = case._ioniq_availability_callers(full_op=False) + timing, clock = recorder() + daemon.stall_timing = timing + daemon.car_params_published = True + daemon.timing_thread_id = 7 + timing.begin() + def send(name, unused): + if name == 'carState': + raise RuntimeError('send failed') + daemon.pm = SimpleNamespace(send=send) + with pytest.raises(RuntimeError, match='send failed'): + daemon.state_publish(fixture.car_state(), None) + assert timing.last_publication is None and not timing.pending + + + + +@pytest.mark.parametrize('replay, aol, long_profile, expected', [ + (False, True, True, True), (True, True, True, False), + (False, False, True, False), (False, True, False, False), +]) +def test_actual_card_thread_gate_uses_configured_profile_not_active_long(replay, aol, long_profile, expected, monkeypatch): + import threading + from openpilot.selfdrive.car import card as module + from openpilot.selfdrive.car.tests.test_hyundai_aol import candidate + class StopLoop(Exception): + pass + class Worker: + def __init__(self, **unused): + pass + def start(self): + pass + def join(self): + pass + _, cp = candidate(True) + cp.safetyConfigs[-1].safetyParam |= 0x0800 + if not long_profile: + cp.openpilotLongitudinalControl = False + daemon: Any = module.Car.__new__(module.Car) + daemon.CP, daemon.aol_replay = cp, aol + daemon.can_sock = Socket() + daemon.ioniq6_keepalive = None + daemon.vehicle_startup = SimpleNamespace(check=lambda: None) + def stop(): + raise StopLoop + daemon.step = stop + monkeypatch.setattr(module, 'REPLAY', replay) + monkeypatch.setattr(module, 'threading', SimpleNamespace(Thread=Worker, Event=threading.Event, get_native_id=lambda: 7)) + with pytest.raises(StopLoop): + daemon.card_thread() + assert (daemon.stall_timing is not None) == expected + + +def test_existing_card_worker_continues_settings_after_diagnostic_logger_error(monkeypatch): + import threading + from openpilot.selfdrive.car import card as module + timing, clock = recorder() + frame(timing, clock, MS) + frame(timing, clock, 31 * MS) + daemon: Any = module.Car.__new__(module.Car) + daemon.stall_timing, daemon.loop_timing = timing, None + daemon.CP = SimpleNamespace(openpilotLongitudinalControl=True, pcmCruise=False) + daemon.aol_card_intent = SimpleNamespace(settings=None) + daemon.v_cruise_helper = SimpleNamespace(intervals=None) + reads = [] + daemon.params = SimpleNamespace(get_bool=lambda key: reads.append(key) or False) + daemon.read_slc_configuration = lambda: 'configuration' + stopped = threading.Event() + def fail(*unused, **kwargs): + raise RuntimeError('diagnostic logger unavailable') + monkeypatch.setattr(module, 'cloudlog', SimpleNamespace(event=fail)) + monkeypatch.setattr(module, 'drop_realtime', lambda: None) + monkeypatch.setattr(module, 'set_core_affinity', lambda cores: None) + monkeypatch.setattr(module, 'read_settings', lambda params: 'aol') + monkeypatch.setattr(module, 'read_cruise_intervals', lambda *args, **kwargs: 'intervals') + monkeypatch.setattr(module, 'time', SimpleNamespace(monotonic=lambda: 10., sleep=lambda duration: stopped.set())) + daemon.params_thread(stopped) + assert reads == ['IsMetric', 'ExperimentalMode'] + assert daemon.aol_card_intent.settings == 'aol' and daemon.v_cruise_helper.intervals == 'intervals' + assert timing.log_errors == 1 and timing.worker[0] == 'settings_end' + + + + +@pytest.mark.parametrize('hook', ['mark', 'published']) +def test_added_publication_hooks_cannot_stop_original_outputs(hook): + from openpilot.common.prefix import OpenpilotPrefix + from openpilot.starpilot.aol.tests import test_ioniq_availability as fixture + case = fixture.IoniqAvailabilityTests(methodName='runTest') + def fail(*args, **kwargs): + raise RuntimeError('injected observer failure') + with ExitStack() as cleanup, OpenpilotPrefix(): + cleanup.callback(case.doCleanups) + _, _, _, daemon, _ = case._ioniq_availability_callers(full_op=False) + daemon.car_params_published = True + daemon.wheel_commands = () + daemon.stall_timing = SimpleNamespace(mark=lambda name: None, published=lambda *args: None) + setattr(daemon.stall_timing, hook, fail) + output = [] + daemon.pm = SimpleNamespace(send=lambda name, msg: output.append(name)) + daemon.state_publish(fixture.car_state(), None) + assert 'carState' in output and 'carOutput' in output + + +@pytest.mark.parametrize('hook', ['begin', 'finish', 'constructor']) +def test_added_loop_hooks_cannot_interrupt_control_order(hook, monkeypatch): + import threading + from openpilot.selfdrive.car import card as module + class StopLoop(Exception): + pass + class Worker: + def __init__(self, **kwargs): + pass + def start(self): + pass + def join(self): + pass + trace = [] + daemon: Any = module.Car.__new__(module.Car) + daemon.aol_replay = True + daemon.ioniq6_keepalive = None + daemon.CP = object() + daemon.can_sock = Socket() + daemon.vehicle_startup = SimpleNamespace(check=lambda: trace.append('check')) + daemon.step = lambda: trace.append('step') + def monitor(): + trace.append('monitor') + if trace.count('monitor') == 2: + raise StopLoop + daemon.rk = SimpleNamespace(monitor_time=monitor) + def fail(*args): + raise RuntimeError('observer failure') + timing = SimpleNamespace(begin=lambda: None, finish=lambda: None, mark=lambda name: None) + if hook != 'constructor': + setattr(timing, hook, fail) + monkeypatch.setattr(module, 'REPLAY', False) + monkeypatch.setattr(module, 'ioniq6_long_eligible', lambda cp: True) + monkeypatch.setattr(module, 'CardStallTiming', fail if hook == 'constructor' else lambda socket: timing) + monkeypatch.setattr(module, 'CardLoopTiming', lambda **kwargs: SimpleNamespace(detailed=False)) + monkeypatch.setattr(module, 'threading', SimpleNamespace(Thread=Worker, Event=threading.Event, get_native_id=lambda: 7)) + with pytest.raises(StopLoop): + daemon.card_thread() + assert trace == ['check', 'step', 'monitor'] * 2 + + +@pytest.mark.parametrize('hook', ['worker_mark', 'flush']) +def test_worker_hooks_cannot_stop_settings(hook, monkeypatch): + import threading + from openpilot.selfdrive.car import card as module + daemon: Any = module.Car.__new__(module.Car) + def fail(*args): + raise RuntimeError('observer failure') + daemon.stall_timing = SimpleNamespace(worker_mark=lambda *args: None, flush=lambda *args: None) + setattr(daemon.stall_timing, hook, fail) + daemon.loop_timing = None + daemon.CP = SimpleNamespace(openpilotLongitudinalControl=False, pcmCruise=False) + daemon.aol_card_intent = None + daemon.v_cruise_helper = SimpleNamespace(intervals=None) + reads = [] + daemon.params = SimpleNamespace(get_bool=lambda key: reads.append(key) or False) + daemon.read_slc_configuration = lambda: 'configuration' + stopped = threading.Event() + monkeypatch.setattr(module, 'drop_realtime', lambda: None) + monkeypatch.setattr(module, 'set_core_affinity', lambda cores: None) + monkeypatch.setattr(module, 'read_cruise_intervals', lambda *args, **kwargs: 'intervals') + monkeypatch.setattr(module, 'time', SimpleNamespace(monotonic=lambda: 10., sleep=lambda duration: stopped.set())) + daemon.params_thread(stopped) + assert reads == ['IsMetric', 'ExperimentalMode'] and daemon.v_cruise_helper.intervals == 'intervals' + + +def test_can_summary_is_bounded_and_detached_without_payload_copy(): + timing, clock = recorder() + timing.begin() + raw = [b'can'] * 100 + decoded = [(i + BASE, [(1, b'abc', 0)] * 3) for i in range(100)] + timing.observe_can(raw, decoded) + assert timing.can_summaries[timing.index] == (100, 64, 192, 192, BASE, BASE + 63) + timing.published(BASE, 'producer') + timing.finish() + frame(timing, clock, 31 * MS) + result = report(timing) + raw.clear() + decoded.clear() + assert result['previous_can'] == (100, 64, 192, 192, BASE, BASE + 63) + assert result['can_source_clock'] == 'CLOCK_BOOTTIME' + + +def test_actual_receive_observer_failure_preserves_can_and_original_interface_error(monkeypatch): + from openpilot.selfdrive.car import card as module + daemon: Any = module.Car.__new__(module.Car) + daemon.can_sock = object() + raw = [b'raw'] + decoded = [(BASE, [(1, b'data', 0)])] + seen = [] + def observer_fail(*args): + raise RuntimeError('observer failure') + def original_update(batch): + seen.append(batch) + raise ValueError('original interface exception') + daemon.stall_timing = SimpleNamespace(mark=lambda name: None, observe_can=observer_fail) + daemon.CI = SimpleNamespace(update=original_update) + monkeypatch.setattr(module.messaging, 'drain_sock_raw', lambda socket, **kwargs: raw) + monkeypatch.setattr(module, 'can_capnp_to_list', lambda batch: decoded if batch is raw else None) + with pytest.raises(ValueError, match='original interface exception'): + daemon.state_update() + assert seen == [decoded] and seen[0] is decoded diff --git a/openpilot/starpilot/card_stall_timing.py b/openpilot/starpilot/card_stall_timing.py new file mode 100644 index 0000000000..9bb2e6cdad --- /dev/null +++ b/openpilot/starpilot/card_stall_timing.py @@ -0,0 +1,185 @@ +"""Temporary, bounded observations of Card publication gaps. + +This observer never supplies control data. MONOTONIC and thread CPU are paired +sequential samples, not atomic observations. Backend-return to GIL-reacquired +also includes scheduler delay; it does not prove contention on Python's lock. +""" +from collections import deque +import threading +import time + + +class CardStallTiming: + LIMIT_NS = 30_000_000 + INTERVAL_NS = 5_000_000_000 + MAX_MARKS = 32 + MARKS = frozenset(('can_receive_start', 'can_receive_end', 'can_decode_end', 'interface_update_end', + 'radar_update_start', 'radar_update_end', 'submaster_end', 'permission_and_cruise_end', + 'conditional_params_end', 'wheel_params_start', 'wheel_params_end', 'state_update_end', + 'startup_maintain_end', 'car_state_send_start', 'car_state_send_end', 'state_publish_end', + 'controller_apply_start', 'controller_apply_end', 'sendcan_start', 'sendcan_end', + 'controls_update_end')) + + def __init__(self, socket, *, clock=None): + self.clock = clock or self._clock + self.socket = socket + self.frames = [[None] * self.MAX_MARKS for _ in range(2)] + self.lengths = [0, 0] + self.can_summaries = [None, None] + self.index = 0 + self.generation = 0 + self.previous = False + self.last_publication = None + self.last_capture = None + self.last_attempt = None + self.pending = deque(maxlen=1) + self.late_count = 0 + self.throttled_count = 0 + self.queue_full_count = 0 + self.omitted_marks = 0 + self.errors = 0 + self.log_errors = 0 + self.worker = None + self.thread_id = threading.get_native_id() + self.native_available = False + try: + socket.enable_receive_timing(True) + self.native_available = True + except Exception: + self.errors += 1 + self.status_pending = True + self.last_counters = None + + @staticmethod + def _clock(): + return time.monotonic_ns(), time.thread_time_ns() + + def begin(self): + try: + self.generation += 1 + self.index = 1 - self.index + self.previous = self.lengths[1 - self.index] > 0 + self.lengths[self.index] = 0 + self.can_summaries[self.index] = None + self._mark('loop_start') + except Exception: + self.errors += 1 + + def _mark(self, name): + length = self.lengths[self.index] + if length == self.MAX_MARKS: + self.omitted_marks += 1 + return + self.frames[self.index][length] = (name, *self.clock()) + self.lengths[self.index] = length + 1 + + def mark(self, name): + try: + if name in self.MARKS: + self._mark(name) + except Exception: + self.errors += 1 + + def observe_can(self, raw, decoded): + try: + # Bound observer work even if the ordinary drain returned an unusually + # large batch. No additional envelope decode or CAN payload copying. + count = min(len(raw), len(decoded), 64) + frames = byte_count = 0 + first = last = None + for i in range(count): + stamp, messages = decoded[i] + frames += len(messages) + byte_count += len(raw[i]) + first = stamp if first is None else min(first, stamp) + last = stamp if last is None else max(last, stamp) + self.can_summaries[self.index] = (len(raw), count, frames, byte_count, first, last) + except Exception: + self.errors += 1 + + def finish(self): + try: + self._mark('loop_end') + except Exception: + self.errors += 1 + + def published(self, source_ns, session): + try: + now, cpu = self.clock() + last = self.last_publication + self.last_publication = (source_ns, now, session, self.generation) + if source_ns <= 0 or last is None or last[0] <= 0 or last[2] != session or source_ns <= last[0] or now <= last[1]: + return + source_gap, send_gap = source_ns - last[0], now - last[1] + if max(source_gap, send_gap) <= self.LIMIT_NS: + return + self.late_count += 1 + if self.last_capture is not None and now - self.last_capture < self.INTERVAL_NS: + self.throttled_count += 1 + return + if self.pending: + self.queue_full_count += 1 + return + self.last_capture = now + # Snapshot only on a bounded late event. No formatting, logging, Params, + # file I/O or locks on Card; the sole consumer runs in params_thread. + previous = tuple(self.frames[1 - self.index][:self.lengths[1 - self.index]]) if self.previous else () + current = tuple(self.frames[self.index][:self.lengths[self.index]]) + native = self.socket.take_receive_timing() if self.native_available else None + clock_mapping = (time.monotonic_ns(), time.clock_gettime_ns(time.CLOCK_BOOTTIME), time.monotonic_ns()) + self.pending.append((source_ns, now, cpu, session, last, source_gap, send_gap, + previous, current, native, self.worker, self.generation, clock_mapping, + self.can_summaries[1 - self.index], self.can_summaries[self.index])) + except Exception: + self.errors += 1 + + def worker_mark(self, phase): + try: + self.worker = (phase, threading.get_native_id(), *self.clock()) + except Exception: + self.errors += 1 + + def flush(self, emit): + """Called only by the existing non-realtime worker, at most once per 5s. + + This Python work still uses the GIL. Native ring/phase sizes and report + frequency bound it; formatting is not evidence of zero observer effect. + """ + try: + now = time.monotonic_ns() + if self.last_attempt is not None and now - self.last_attempt < self.INTERVAL_NS: + return + counter_values = (self.late_count, self.throttled_count, self.queue_full_count, self.omitted_marks, self.errors, self.log_errors) + if not self.pending and not self.status_pending and counter_values == self.last_counters: + return + self.last_attempt = now + self.last_counters = counter_values + status = self.status_pending + self.status_pending = False + sample = self.pending.popleft() if self.pending else None + counters = {'late_count': self.late_count, 'throttled_count': self.throttled_count, + 'queue_full_count': self.queue_full_count, 'omitted_marks': self.omitted_marks, + 'capture_errors': self.errors, 'log_errors': self.log_errors} + report = {'version': 1, 'native_available': self.native_available, 'thread_id': self.thread_id, + 'clock': 'CLOCK_MONOTONIC', 'source_clock': 'CLOCK_MONOTONIC', 'cpu_clock': 'CLOCK_THREAD_CPUTIME_ID', + 'native_fields': ('sequence', 'non_blocking', 'returned_message', 'byte_length', 'before_wall', 'before_cpu', + 'backend_return_wall', 'backend_return_cpu', 'gil_reacquired_wall', + 'gil_reacquired_cpu', 'python_return_wall', 'python_return_cpu'), + 'counters': counters, 'status': status} + if sample is not None: + source, send, cpu, session, last, source_gap, send_gap, previous, current, native, worker, generation, mapping, prev_can, current_can = sample + report.update(source_ns=source, send_return_observed_ns=send, send_thread_cpu_ns=cpu, + producer_session=session, previous_publication=last, + source_gap_ns=source_gap, send_gap_ns=send_gap, previous_frame=previous, + current_frame=current, current_frame_complete=False, native=native, + native_capacity=64, native_omitted_is_cumulative=True, generation=generation, + previous_publication_frame_retained=last[3] == generation - 1, + nonpublishing_frames=max(0, generation - last[3] - 1), + clock_mapping_mono_boot_mono=mapping, + can_summary_fields=('total_envelopes', 'summarized_envelopes', 'summarized_frames', + 'summarized_bytes', 'min_source_ns', 'max_source_ns'), + can_source_clock='CLOCK_BOOTTIME', previous_can=prev_can, current_can=current_can, + worker_progress=worker) + emit('card.stall_timing', error=sample is not None or not self.native_available, **report) + except Exception: + self.log_errors += 1 diff --git a/tools/ci/card_stall_native_artifact/build_native.py b/tools/ci/card_stall_native_artifact/build_native.py new file mode 100644 index 0000000000..b8270cb74f --- /dev/null +++ b/tools/ci/card_stall_native_artifact/build_native.py @@ -0,0 +1,177 @@ +#!/usr/bin/env python3 +"""Isolated native msgq build. Never modifies source checkout or build receipts.""" +import argparse +import hashlib +import importlib.metadata +import json +import os +from pathlib import Path +import platform +import shutil +import subprocess +import sys +import sysconfig +import tempfile + +sys.dont_write_bytecode = True + +from validate_elf import BASELINE_SHA256 + +HERE = Path(__file__).resolve().parent +PINNED = {'Cython': '3.3.0', 'SCons': '4.11.1', 'pytest': '8.4.2', 'iniconfig': '2.3.0', + 'packaging': '26.3', 'pluggy': '1.6.0', 'Pygments': '2.21.0'} + + +def check(condition, message): + if not condition: + raise RuntimeError(message) + + +def capture(*args, cwd=None): + return subprocess.check_output(args, cwd=cwd, text=True).strip() + + +def sha(path): + return hashlib.sha256(Path(path).read_bytes()).hexdigest() + + +def manifest(root): + return {str(p.relative_to(root)): {'size': p.stat().st_size, 'sha256': sha(p)} + for p in sorted(root.rglob('*')) if p.is_file() and '__pycache__' not in p.parts} + + +def bind_committed_sources(source, revision, build, source_manifest): + tree = subprocess.check_output(['git', 'ls-tree', '-rz', revision, '--', 'msgq_repo'], cwd=source) + committed = {} + for row in tree.split(b'\0'): + if row: + metadata, path = row.split(b'\t', 1) + mode, kind, blob = metadata.decode().split() + committed[path.decode()] = (mode, kind, blob) + for relative, info in source_manifest.items(): + original = 'msgq_repo/'+relative + check(original in committed, f'Uncommitted build input: {original}') + mode, kind, blob = committed[original] + check(kind == 'blob' and mode in ('100644', '100755'), f'Unsupported build input type: {original}') + copied = build/relative + check(not copied.is_symlink(), f'Symlink build input is not permitted: {original}') + check(('100755' if copied.stat().st_mode & 0o111 else '100644') == mode, f'Build input mode differs from commit: {original}') + data = subprocess.check_output(['git', 'cat-file', 'blob', blob], cwd=source) + check(copied.read_bytes() == data, f'Build input bytes differ from committed blob: {original}') + info.update(git_path=original, git_mode=mode, git_blob=blob) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('source_root', type=Path) + parser.add_argument('baseline_so', type=Path) + parser.add_argument('output_dir', type=Path) + args = parser.parse_args() + source, baseline, output = (p.resolve() for p in (args.source_root, args.baseline_so, args.output_dir)) + check(platform.machine() == 'aarch64' and platform.system() == 'Linux', 'Actual native Linux AArch64 required') + check(sys.implementation.name == 'cpython' and sys.version_info[:3] == (3, 12, 13), 'Actual CPython 3.12.13 required') + check(os.environ.get('GITHUB_ACTIONS') == 'true' and os.environ.get('CI') == 'true', 'This helper is scoped to the explicitly approved CI job') + check(os.environ.get('RUNNER_ARCH') == 'ARM64' and sys.flags.optimize == 0, 'Native ARM64 runner and non-optimized Python required') + check(not Path('/AGNOS').exists(), 'Do not run this helper on a device') + check(not output.is_relative_to(source) and not source.is_relative_to(output), 'Output and source must be separate trees') + check(not output.exists() or not any(output.iterdir()), 'Output directory must be new or empty') + check(sha(baseline) == BASELINE_SHA256, 'Wrong reviewed baseline library') + source_commit = capture('git', 'rev-parse', 'HEAD', cwd=source) + check(source_commit == os.environ['EXPECTED_SOURCE_COMMIT'], 'Source commit differs from approved SHA') + base_commit = os.environ['EXPECTED_BASE_COMMIT'] + check(base_commit == 'e8389b0b7529bad402f1f4e5fbd672ff13d9d382', 'Unreviewed baseline commit') + check(len(base_commit) == 40 and capture('git', 'rev-parse', base_commit+'^{commit}', cwd=source) == base_commit, 'Full baseline commit must be available locally') + subprocess.run(['git', 'merge-base', '--is-ancestor', base_commit, source_commit], cwd=source, check=True) + check(not capture('git', 'status', '--porcelain', '--untracked-files=all', cwd=source), 'Source checkout must be clean, including untracked paths') + versions = {name: importlib.metadata.version(name) for name in PINNED} + check(versions == PINNED, f'Build/test packages differ from reviewed uv.lock pins: {versions}') + compiler = {key: Path(os.environ.get(key, '/usr/bin/'+name+'-13')).resolve() for key, name in [('GCC', 'gcc'), ('GXX', 'g++')]} + expected_gcc = os.environ['EXPECTED_GCC_VERSION'] + check(expected_gcc.split('.')[0] == '13', 'Only explicitly reviewed GCC 13 is allowed') + for path in compiler.values(): + check(capture(str(path), '-dumpfullversion') == expected_gcc, f'Unapproved compiler version: {path}') + check(capture(str(path), '-dumpmachine') == 'aarch64-linux-gnu', f'Compiler is not native generic AArch64: {path}') + output.mkdir(parents=True, exist_ok=True) + artifacts = output/'artifacts' + artifacts.mkdir() + toolchain = { + 'qualification': 'CI ARM host artifact; not deployed or device-qualified', + 'source_commit': source_commit, 'base_commit': base_commit, 'baseline_sha256': sha(baseline), + 'python': sys.version, 'python_executable': str(Path(sys.executable).resolve()), 'python_sha256': sha(Path(sys.executable).resolve()), + 'python_sysconfig': {key: sysconfig.get_config_var(key) for key in ('SOABI', 'MULTIARCH', 'CONFIG_ARGS', 'CC', 'CFLAGS', 'LDFLAGS')}, + 'packages': versions, 'platform': platform.platform(), 'os_release': Path('/etc/os-release').read_text(), + 'compilers': {key: {'path': str(path), 'sha256': sha(path), 'version': capture(str(path), '--version')} for key, path in compiler.items()}, + 'compiler_packages': capture('dpkg-query', '-W', '-f=${Package}\t${Version}\t${Architecture}\n', 'gcc-13', 'g++-13', 'libstdc++6', 'libc6', 'binutils'), + 'binutils': capture('readelf', '--version'), 'isa_flags': ['-march=armv8-a', '-mtune=generic'], + 'cxx_compatibility_flags': ['-fno-sized-deallocation'], + 'requirements_sha256': sha(HERE/'requirements.lock'), 'uv_lock_sha256': sha(source/'uv.lock'), + } + libstdcxx = Path(capture(str(compiler['GXX']), '-print-file-name=libstdc++.so')).resolve() + toolchain['compiler_libstdcxx'] = {'path': str(libstdcxx), 'sha256': sha(libstdcxx)} + (artifacts/'toolchain.json').write_text(json.dumps(toolchain, indent=2)+'\n') + build = output/'build' + shutil.copytree(source/'msgq_repo', build, ignore=shutil.ignore_patterns('.git', '*.so', '*.o', '*.a', '*.pyc', '__pycache__', '.sconsign*', '*_pyx.cpp')) + source_manifest = manifest(build) + bind_committed_sources(source, source_commit, build, source_manifest) + (artifacts/'source_files.json').write_text(json.dumps(source_manifest, indent=2)+'\n') + baseline_root = output/'baseline' + (baseline_root/'msgq/visionipc').mkdir(parents=True) + (baseline_root/'msgq/tests').mkdir() + baseline_files = ['msgq/__init__.py', 'msgq/visionipc/__init__.py', 'msgq/visionipc/visionipc_pyx.so', + 'msgq/tests/__init__.py', 'msgq/tests/test_fake.py', 'msgq/tests/test_messaging.py', 'msgq/tests/test_poller.py'] + for relative in baseline_files: + data = subprocess.check_output(['git', 'show', base_commit+':msgq_repo/'+relative], cwd=source) + (baseline_root/relative).write_bytes(data) + tracked_baseline = subprocess.check_output(['git', 'show', base_commit+':msgq_repo/msgq/ipc_pyx.so'], cwd=source) + check(hashlib.sha256(tracked_baseline).hexdigest() == BASELINE_SHA256, 'Baseline commit does not contain reviewed ipc_pyx bytes') + shutil.copy2(baseline, baseline_root/'msgq/ipc_pyx.so') + companion_relative = Path('msgq/visionipc/visionipc_pyx.so') + companion_sha = sha(baseline_root/companion_relative) + check(companion_sha == 'd9aee015a9ca9d6d8f4dc54c5bca6a8f3ab24ad9a42477561ea7cc32eebff5a4', 'Unreviewed companion baseline') + check(sha(source/'msgq_repo'/companion_relative) == companion_sha, 'Source checkout changed shipped visionipc companion') + shutil.copy2(baseline_root/companion_relative, build/companion_relative) + (artifacts/'unchanged_companion.json').write_text(json.dumps({'git_blob': capture('git', 'rev-parse', base_commit+':msgq_repo/'+str(companion_relative), cwd=source), 'sha256': companion_sha}, indent=2)+'\n') + wrappers = output/'compiler-bin' + wrappers.mkdir() + for name in ('cc', 'gcc', 'c++', 'g++'): + (wrappers/name).symlink_to(HERE/'compiler_wrapper.py') + env = dict(os.environ, PATH=str(wrappers)+os.pathsep+str(Path(sys.executable).parent)+os.pathsep+os.environ['PATH'], + PYTHONNOUSERSITE='1', PYTHONDONTWRITEBYTECODE='1', PYTEST_DISABLE_PLUGIN_AUTOLOAD='1', PYTHONPATH=str(build), + ARM_CI_GCC=str(compiler['GCC']), ARM_CI_GXX=str(compiler['GXX']), + ARM_CI_COMPILER_LOG=str(artifacts/'compiler_commands.jsonl'), + ARM_CI_BASELINE_ROOT=str(baseline_root), ARM_CI_CANDIDATE_ROOT=str(build), + ARM_CI_COMPANION_SHA256=companion_sha, ARM_CI_PAIR_REPORT=str(artifacts/'api_and_companion.json')) + env.pop('CEREAL_FAKE', None) + env.pop('SCONSFLAGS', None) + env.pop('LD_PRELOAD', None) + env.pop('PYTHONOPTIMIZE', None) + + def run_logged(name, command, cwd, command_env=env): + with (artifacts/(name+'.log')).open('w') as log: + log.write(json.dumps({'cwd': str(cwd), 'argv': command})+'\n') + log.flush() + subprocess.run(command, cwd=cwd, env=command_env, stdout=log, stderr=subprocess.STDOUT, check=True, timeout=300) + + run_logged('build', [sys.executable, '-m', 'SCons', '--minimal', '--cache-disable', '-j2', 'msgq/ipc_pyx.so'], build) + candidate = build/'msgq/ipc_pyx.so' + run_logged('static_abi', [sys.executable, str(HERE/'validate_elf.py'), '--baseline', str(baseline), '--candidate', str(candidate), '--report', str(artifacts/'elf_comparison.json')], output) + for name, package in [('baseline', baseline_root), ('candidate', build)]: + with tempfile.TemporaryDirectory(prefix='msgq_armci_', dir='/dev/shm') as namespace: + test_env = dict(env, PYTHONPATH=str(package), OPENPILOT_PREFIX=Path(namespace).name.removeprefix('msgq_')) + run_logged(name+'_tests', [sys.executable, '-m', 'pytest', '--noconftest', '--rootdir', str(output), '-p', 'no:cacheprovider', '-o', 'addopts=', '-q', '--junitxml', str(artifacts/(name+'_tests.xml')), str(package/'msgq/tests')], output, test_env) + run_logged('mixed_version_tests', [sys.executable, '-m', 'pytest', '--noconftest', '--rootdir', str(output), '-p', 'no:cacheprovider', '-o', 'addopts=', '-q', '--junitxml', str(artifacts/'mixed_version_tests.xml'), str(HERE/'test_native_pair.py')], output) + generated = {str(p.relative_to(build)): {'size': p.stat().st_size, 'sha256': sha(p)} for p in sorted(build.rglob('*')) + if p.is_file() and (p.suffix in ('.o', '.a', '.so') or p.name == 'ipc_pyx.cpp')} + (artifacts/'generated_files.json').write_text(json.dumps(generated, indent=2)+'\n') + shutil.copy2(candidate, artifacts/'ipc_pyx.so') + shutil.copy2(build/'msgq/ipc_pyx.cpp', artifacts/'ipc_pyx.cpp') + check(all(sha(build/p) == info['sha256'] for p, info in source_manifest.items()), 'Build modified an input source file') + check(not capture('git', 'status', '--porcelain', '--untracked-files=all', cwd=source), 'Build changed checkout, including untracked paths') + (artifacts/'RESULT.json').write_text(json.dumps({'status': 'native_arm_ci_checks_passed', 'candidate_sha256': sha(candidate), + 'device_qualification': 'NOT ESTABLISHED; explicit artifact/toolchain/runtime review required', + 'receipts_modified': False, 'deployed': False}, indent=2)+'\n') + print(artifacts) + + +if __name__ == '__main__': + main() diff --git a/tools/ci/card_stall_native_artifact/build_native.sh b/tools/ci/card_stall_native_artifact/build_native.sh new file mode 100755 index 0000000000..53aec07059 --- /dev/null +++ b/tools/ci/card_stall_native_artifact/build_native.sh @@ -0,0 +1,9 @@ +#!/usr/bin/env bash +# Artifact-only preparation. Run only after explicit CI execution approval. +set -euo pipefail +export PYTHONDONTWRITEBYTECODE=1 +: "${PYTHON:?Set the absolute native ARM CPython 3.12.13 interpreter}" +: "${EXPECTED_SOURCE_COMMIT:?Set the exact approved diagnostics commit SHA}" +: "${EXPECTED_BASE_COMMIT:?Set the exact reviewed e838 baseline commit SHA}" +: "${EXPECTED_GCC_VERSION:?Set the explicitly reviewed GCC 13 full version}" +exec "$PYTHON" "$(dirname "$0")/build_native.py" "$@" diff --git a/tools/ci/card_stall_native_artifact/compiler_wrapper.py b/tools/ci/card_stall_native_artifact/compiler_wrapper.py new file mode 100755 index 0000000000..a6cde445bb --- /dev/null +++ b/tools/ci/card_stall_native_artifact/compiler_wrapper.py @@ -0,0 +1,20 @@ +#!/usr/bin/env python3 +"""Use only the preflight-approved GCC 13, recording effective ISA flags.""" +import json +import os +from pathlib import Path +import sys + +name = Path(sys.argv[0]).name +compiler = os.environ['ARM_CI_GXX' if name in ('g++', 'c++') else 'ARM_CI_GCC'] +compatibility_flags = ['-fno-sized-deallocation'] if name in ('g++', 'c++') else [] +command = [compiler, '-march=armv8-a', '-mtune=generic', *compatibility_flags, *sys.argv[1:]] +if any(flag == '-fsized-deallocation' or flag.startswith('-mcpu=') or (flag.startswith('-march=') and flag != '-march=armv8-a') or + (flag.startswith('-mtune=') and flag != '-mtune=generic') for flag in sys.argv[1:]): + raise SystemExit('CPU-specific compiler flags are prohibited for this artifact') +fd = os.open(os.environ['ARM_CI_COMPILER_LOG'], os.O_APPEND | os.O_CREAT | os.O_WRONLY, 0o600) +try: + os.write(fd, (json.dumps({'cwd': os.getcwd(), 'argv': command})+'\n').encode()) +finally: + os.close(fd) +os.execv(compiler, command) diff --git a/tools/ci/card_stall_native_artifact/frozen-source.json b/tools/ci/card_stall_native_artifact/frozen-source.json new file mode 100644 index 0000000000..5d16addc70 --- /dev/null +++ b/tools/ci/card_stall_native_artifact/frozen-source.json @@ -0,0 +1,58 @@ +{ + "parent_commit": "e8389b0b7529bad402f1f4e5fbd672ff13d9d382", + "expected_branch": "diagnostics/card-stall-build-20261012", + "source_whitelist": [ + { + "path": "openpilot/selfdrive/car/card.py", + "sha256": "41694a8ca800b0c850cafd18cf902f83a7155d4c15405d0a44bbde3efd8269a2", + "git_mode": "100755", + "baseline_sha256": "14e2dcc0aa4e4a945677a41b35dd599ea6d9f8bf7c6a449e74b362ef9203045f" + }, + { + "path": "openpilot/starpilot/card_stall_timing.py", + "sha256": "268d67c8ebdf07a83c254fe4cde04bb3fb56f5cd5bc793da57b6e8c2370a8063", + "git_mode": "100644", + "baseline_sha256": null + }, + { + "path": "openpilot/starpilot/aol/tests/test_card_stall_timing.py", + "sha256": "dc56cd86346da6cd790475d055c9f873663ce5342d46cbabda196aa358efccf6", + "git_mode": "100644", + "baseline_sha256": null + }, + { + "path": "msgq_repo/msgq/ipc_pyx.pyx", + "sha256": "344491d95253e677b1190d7d9440edbb75d2067bb2c41e1919a21a244566d4ef", + "git_mode": "100644", + "baseline_sha256": "02cf21d0232967eed7bd68ed0209e55f5f1e3edef3675671bc6524ca64976fa4" + }, + { + "path": "msgq_repo/msgq/tests/test_receive_timing.py", + "sha256": "88b7f736c7cbd3d0d0c29634df35112f9eca8f10236a3561afce6683e88f7006", + "git_mode": "100644", + "baseline_sha256": null + } + ], + "baseline_native": { + "path": "msgq_repo/msgq/ipc_pyx.so", + "git_blob": "3453001a9e283d60ae4489607beafaa6776acd1e", + "sha256": "5a50085f727e3b22dcf596cb42be351066b0dcf7db7866fe04be40e69fe57133", + "git_mode": "100755" + }, + "python_version": "3.12.13", + "cython_version": "3.3.0", + "compiler_family_major": "GCC 13", + "generic_isa": "armv8-a", + "max_symbol_versions": { + "GLIBC": "2.34", + "GLIBCXX": "3.4.32", + "CXXABI": "1.3", + "GCC": "3.0" + }, + "baseline_companion": { + "path": "msgq_repo/msgq/visionipc/visionipc_pyx.so", + "git_blob": "b4f95da4ec84c15a515d56dea0f028c37674d083", + "sha256": "d9aee015a9ca9d6d8f4dc54c5bca6a8f3ab24ad9a42477561ea7cc32eebff5a4", + "git_mode": "100755" + } +} diff --git a/tools/ci/card_stall_native_artifact/requirements.lock b/tools/ci/card_stall_native_artifact/requirements.lock new file mode 100644 index 0000000000..654d497a10 --- /dev/null +++ b/tools/ci/card_stall_native_artifact/requirements.lock @@ -0,0 +1,10 @@ +# Minimal ARM build/test closure pinned to repository uv.lock. +# Install with: python -m pip install --require-hashes -r requirements.lock +# No project extras, model dependencies, or editable workspace installation. +cython==3.3.0 --hash=sha256:7e321ae700995a16dc3055ada06ffb8d61e1a7434e5d0e811547a45ac1015ebd --hash=sha256:9b24b5c8cd536946b62086fcafee6d5509d3f549f72d553d2336af87ffbe0da1 +scons==4.11.1 --hash=sha256:454cef364348053422696e3d2ecb4fa593c96a624f955842eaaea64f95c8d11d +pytest==8.4.2 --hash=sha256:872f880de3fc3a5bdc88a11b39c9710c3497a547cfa9320bc3c5e62fbf272e79 +iniconfig==2.3.0 --hash=sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12 +packaging==26.3 --hash=sha256:d7193f7c8e4e93f444fde0262bf90af30e16fa0ad0ad44cb553c87339b23cd1c +pluggy==1.6.0 --hash=sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746 +pygments==2.21.0 --hash=sha256:2363c69b61c4a97c838da3b130dcd6468f4848992b21a82f2a63ec34377137d9 diff --git a/tools/ci/card_stall_native_artifact/test_native_pair.py b/tools/ci/card_stall_native_artifact/test_native_pair.py new file mode 100644 index 0000000000..a7b6cff27b --- /dev/null +++ b/tools/ci/card_stall_native_artifact/test_native_pair.py @@ -0,0 +1,147 @@ +"""Run on native ARM CPython 3.12.13; never substitute a rebuilt companion.""" +import hashlib +import json +import os +from pathlib import Path +import platform +import select +import subprocess +import sys +import tempfile +import uuid + +import pytest + +PAYLOADS = [b'x', b'unchanged\x00binary', bytes(range(256)), b'z'*4096] * 32 + + +def exact_runtime(): + assert platform.machine() == 'aarch64', platform.machine() + assert sys.implementation.name == 'cpython' and sys.version_info[:3] == (3, 12, 13), sys.version + + +def child_env(root, prefix): + env = dict(os.environ, PYTHONPATH=str(root), PYTHONNOUSERSITE='1', OPENPILOT_PREFIX=prefix) + env.pop('CEREAL_FAKE', None) + return env + + +def read_line(process, timeout=15): + if not select.select([process.stdout], [], [], timeout)[0]: + raise TimeoutError('native IPC subprocess did not produce its expected line') + line = process.stdout.readline() + if not line: + raise RuntimeError(f'native IPC subprocess exited before its expected line: {process.poll()}') + return line.decode().rstrip('\n') + + +def roots(): + return {name: Path(os.environ['ARM_CI_'+name.upper()+'_ROOT']) for name in ('baseline', 'candidate')} + + +def test_public_api_and_unchanged_companion(): + exact_runtime() + results = {} + for name, root in roots().items(): + result = subprocess.run([sys.executable, __file__, 'api'], env=child_env(root, 'api_'+uuid.uuid4().hex), + check=True, text=True, capture_output=True, timeout=20) + results[name] = json.loads(result.stdout) + assert results[name]['companion_sha256'] == os.environ['ARM_CI_COMPANION_SHA256'] + assert Path(results[name]['ipc_path']).resolve() == (root/'msgq/ipc_pyx.so').resolve() + assert Path(results[name]['companion_path']).resolve() == (root/'msgq/visionipc/visionipc_pyx.so').resolve() + baseline, candidate = results['baseline'], results['candidate'] + for name, methods in baseline['api'].items(): + assert set(methods) <= set(candidate['api'][name]), name + assert {'enable_receive_timing', 'reset_receive_timing', 'take_receive_timing'} <= set(candidate['api']['SubSocket']) + Path(os.environ['ARM_CI_PAIR_REPORT']).write_text(json.dumps(results, indent=2)+'\n') + + +@pytest.mark.parametrize('publisher,subscriber,timing', [ + ('baseline', 'baseline', False), ('baseline', 'candidate', False), + ('baseline', 'candidate', True), ('candidate', 'baseline', False), ('candidate', 'candidate', True), +]) +def test_interprocess_payload_equivalence(publisher, subscriber, timing): + exact_runtime() + packages = roots() + with tempfile.TemporaryDirectory(prefix='msgq_armci_', dir='/dev/shm') as namespace: + prefix = Path(namespace).name.removeprefix('msgq_') + endpoint = 'abi_'+uuid.uuid4().hex + processes = [] + try: + def start(role, root, enabled): + process = subprocess.Popen([sys.executable, __file__, role, endpoint, str(int(enabled))], + env=child_env(root, prefix), stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, bufsize=0) + processes.append(process) + assert read_line(process) == 'ready' + return process + pub = start('publish', packages[publisher], False) + sub = start('subscribe', packages[subscriber], timing) + pub.stdin.write(b'go\n') + pub.stdin.flush() + result = json.loads(read_line(sub)) + assert result['sha256'] == hashlib.sha256(b''.join(PAYLOADS)).hexdigest() + assert result['count'] == len(PAYLOADS) + assert result['bytes'] == sum(map(len, PAYLOADS)) + assert sub.wait(timeout=10) == 0 + pub.stdin.write(b'done\n') + pub.stdin.flush() + assert pub.wait(timeout=10) == 0 + finally: + for process in processes: + if process.poll() is None: + process.kill() + process.wait(timeout=10) + stderr = process.stderr.read().decode(errors='replace') + if stderr: + print(stderr, file=sys.stderr) + + +def child(): + exact_runtime() + import msgq + import msgq.ipc_pyx + role = sys.argv[1] + if role == 'api': + # A missing dependency is a blocking import error, never a reason to build + # or swap this companion, whose reviewed bytes must remain unchanged. + import msgq.visionipc.visionipc_pyx as companion + names = ('Context', 'SubSocket', 'PubSocket', 'Poller', 'Event', 'SocketEventHandle') + print(json.dumps({'api': {n: sorted(x for x in dir(getattr(msgq.ipc_pyx, n)) if not x.startswith('_')) for n in names}, + 'ipc_path': msgq.ipc_pyx.__file__, 'companion_path': companion.__file__, + 'companion_sha256': hashlib.sha256(Path(companion.__file__).read_bytes()).hexdigest(), + 'python': sys.version, 'machine': platform.machine()}), flush=True) + return + endpoint, timing = sys.argv[2], bool(int(sys.argv[3])) + if role == 'publish': + sock = msgq.pub_sock(endpoint) + print('ready', flush=True) + assert input() == 'go' + for payload in PAYLOADS: + sock.send(payload) + assert input() == 'done' + elif role == 'subscribe': + sock = msgq.sub_sock(endpoint, timeout=2000) + if timing: + sock.enable_receive_timing(True) + print('ready', flush=True) + received = [sock.receive() for _ in PAYLOADS] + assert received == PAYLOADS + assert sock.receive(non_blocking=True) is None + if hasattr(sock, 'take_receive_timing'): + records, omitted = sock.take_receive_timing() + if timing: + assert len(records) == 64 and omitted == len(PAYLOADS)+1-64 + assert records[-1][:4] == (len(PAYLOADS)+1, True, False, 0) + assert all(len(r) == 12 and min(r[4:]) >= 0 for r in records) + assert all(list(r[4::2]) == sorted(r[4::2]) and list(r[5::2]) == sorted(r[5::2]) for r in records) + else: + assert (records, omitted) == ((), 0) + print(json.dumps({'count': len(received), 'bytes': sum(map(len, received)), + 'sha256': hashlib.sha256(b''.join(received)).hexdigest()}), flush=True) + else: + raise ValueError(role) + + +if __name__ == '__main__': + child() diff --git a/tools/ci/card_stall_native_artifact/validate_elf.py b/tools/ci/card_stall_native_artifact/validate_elf.py new file mode 100644 index 0000000000..a0a4744b12 --- /dev/null +++ b/tools/ci/card_stall_native_artifact/validate_elf.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""Static, fail-closed comparison. Does not execute either shared library.""" +import argparse +import hashlib +import json +from pathlib import Path +import re +import struct +import subprocess + +BASELINE_SHA256 = '5a50085f727e3b22dcf596cb42be351066b0dcf7db7866fe04be40e69fe57133' + + +def command(*args): + return subprocess.check_output(args, text=True) + + +def inspect(path): + data = path.read_bytes() + if len(data) < 64 or data[:6] != b'\x7fELF\x02\x01' or struct.unpack_from('= 2: + symbols[fields[0]] = fields[1] + names = sorted(symbols) + demangled = subprocess.check_output(['c++filt'], input='\n'.join(names)+'\n', text=True).splitlines() + undefined = command('nm', '-D', '--undefined-only', '--format=posix', str(path)) + version_names = sorted(set(re.findall(r'Name: ([A-Za-z0-9_.]+)', versions))) + return { + 'path': str(path), 'sha256': hashlib.sha256(data).hexdigest(), 'size': len(data), + 'needed': sorted(re.findall(r'\(NEEDED\).*?\[(.*?)\]', dynamic)), + 'rpaths': re.findall(r'\((?:RUNPATH|RPATH)\).*?\[(.*?)\]', dynamic), + 'versions': version_names, 'defined_symbols': dict(zip(names, demangled, strict=True)), + 'undefined_symbols': undefined.splitlines(), + } + + +def version_max(info, namespace): + values = [tuple(map(int, name[len(namespace)+1:].split('.'))) for name in info['versions'] if re.fullmatch(re.escape(namespace)+r'_[0-9]+(?:\.[0-9]+)*', name)] + return max(values, default=(0,)) + + +def project_symbol(name): + classes = r'(?:Context|Message|SubSocket|PubSocket|Poller|MSGQSubSocket|MSGQPoller|FakePoller|FakeSubSocket|SocketEventHandle|Event)' + return bool(re.match(r'^(?:PyInit_ipc_pyx$|msgq_|messaging_use_fake\(|sigusr2_handler\(|event_state_shm_mmap\(|' + + classes+r'(?:::|<)|(?:typeinfo(?: name)?|vtable) for '+classes+r'(?:$|<))', name)) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--baseline', type=Path, required=True) + parser.add_argument('--candidate', type=Path, required=True) + parser.add_argument('--report', type=Path, required=True) + args = parser.parse_args() + baseline, candidate = inspect(args.baseline), inspect(args.candidate) + errors = [] + if baseline['sha256'] != BASELINE_SHA256: + errors.append('baseline hash does not match the reviewed e838 tracked artifact') + unexpected = sorted(set(candidate['needed']) - set(baseline['needed'])) + if unexpected: + errors.append(f'new dynamic dependencies: {unexpected}') + # Standalone linking needs no runtime search path, so reject all rather than + # maintaining a fragile allow-list of possible build-host paths. + if candidate['rpaths']: + errors.append(f'candidate must not embed RPATH/RUNPATH: {candidate["rpaths"]}') + for namespace, ceiling in {'GLIBC': (2, 34), 'GLIBCXX': (3, 4, 32), 'CXXABI': version_max(baseline, 'CXXABI'), 'GCC': version_max(baseline, 'GCC')}.items(): + ceiling = min(ceiling, version_max(baseline, namespace)) + if version_max(candidate, namespace) > ceiling: + errors.append(f'{namespace} requirement exceeds {ceiling}') + nonnumeric_versions = [v for v in candidate['versions'] if not re.fullmatch(r'[A-Z][A-Z0-9]*_[0-9]+(?:\.[0-9]+)*', v)] + if nonnumeric_versions: + errors.append(f'unreviewed nonnumeric symbol versions: {nonnumeric_versions}') + unexpected_versions = {v.split('_', 1)[0] for v in candidate['versions']} - {v.split('_', 1)[0] for v in baseline['versions']} + if unexpected_versions: + errors.append(f'new symbol-version namespace: {sorted(unexpected_versions)}') + bsyms, csyms = baseline['defined_symbols'], candidate['defined_symbols'] + missing_project = {n: d for n, d in bsyms.items() if project_symbol(d) and n not in csyms} + if missing_project: + errors.append(f'missing existing msgq/extension exported symbols: {missing_project}') + report = { + 'qualification': 'Static ABI boundary checks only; native tests and deployment review remain separate.', + 'baseline': baseline, 'candidate': candidate, 'errors': errors, + 'removed_exports': {n: bsyms[n] for n in sorted(set(bsyms)-set(csyms))}, + 'added_exports': {n: csyms[n] for n in sorted(set(csyms)-set(bsyms))}, + } + args.report.write_text(json.dumps(report, indent=2, sort_keys=True)+'\n') + if errors: + raise SystemExit('\n'.join(errors)) + print('AArch64 ELF, dependency, search-path, symbol-version and existing msgq-export checks passed.') + + +if __name__ == '__main__': + main() diff --git a/tools/ci/card_stall_native_artifact/verify_source.py b/tools/ci/card_stall_native_artifact/verify_source.py new file mode 100644 index 0000000000..a9b79cc402 --- /dev/null +++ b/tools/ci/card_stall_native_artifact/verify_source.py @@ -0,0 +1,57 @@ +#!/usr/bin/env python3 +"""Read-only source/manifest gate for the isolated native-artifact branch.""" +import hashlib +import json +import os +from pathlib import Path +import subprocess +import sys + +root = Path.cwd().resolve() +output = Path(sys.argv[1]).resolve() +output.mkdir(parents=True, exist_ok=False) +manifest = json.loads(Path(__file__).with_name('frozen-source.json').read_text()) + +def git(*args): + return subprocess.check_output(['git', '-c', 'core.hooksPath=/dev/null', *args], cwd=root) + +def digest(raw): + return hashlib.sha256(raw).hexdigest() + +head = git('rev-parse', 'HEAD').decode().strip() +assert head == os.environ['GITHUB_SHA'], 'checkout is not the exact event commit' +assert os.environ['GITHUB_REF'] == 'refs/heads/' + manifest['expected_branch'] +assert os.environ['GITHUB_REPOSITORY'] == 'firestar5683/StarPilot' +base = manifest['parent_commit'] +git('cat-file', '-e', base + '^{commit}') +changed = git('diff', '--name-only', base, head).decode().splitlines() +source_paths = {row['path'] for row in manifest['source_whitelist']} +allowed = lambda p: p in source_paths or p == '.github/workflows/card_stall_native_artifact.yaml' or p.startswith('tools/ci/card_stall_native_artifact/') +assert all(allowed(p) for p in changed), changed +assert source_paths <= set(changed) +rows = [] +for entry in manifest['source_whitelist']: + path = entry['path'] + data = (root / path).read_bytes() + assert digest(data) == entry['sha256'], path + tree = git('ls-tree', head, '--', path).decode().strip().split() + assert tree[0] == entry['git_mode'], (path, tree) + assert digest(git('show', head + ':' + path)) == entry['sha256'] + rows.append({'path': path, 'sha256': digest(data), 'git_mode': tree[0], 'git_blob': tree[2]}) +for key in ('baseline_native', 'baseline_companion'): + entry = manifest[key] + data = git('show', base + ':' + entry['path']) + tree = git('ls-tree', base, '--', entry['path']).decode().strip().split() + assert tree[0] == entry['git_mode'] and tree[2] == entry['git_blob'] + assert digest(data) == entry['sha256'] + assert (root / entry['path']).read_bytes() == data, 'unexpected already-replaced native artifact' + (output / ('baseline-ipc_pyx.so' if key == 'baseline_native' else 'baseline-visionipc_pyx.so')).write_bytes(data) +assert not (root / 'openpilot/selfdrive/modeld/models').exists(), 'model payload checkout is outside this job' +source_tree = git('ls-tree', '-rz', '--full-tree', head) +(output / 'source-tree.git.bin').write_bytes(source_tree) +report = {'head': head, 'base': base, 'tree': git('rev-parse', 'HEAD^{tree}').decode().strip(), + 'changed_paths': changed, 'source_tree_sha256': digest(source_tree), 'frozen_sources': rows, + 'baseline_native': manifest['baseline_native'], 'baseline_companion': manifest['baseline_companion'], + 'workflow': '.github/workflows/card_stall_native_artifact.yaml', 'source_only_sparse_checkout': True} +(output / 'source.json').write_text(json.dumps(report, indent=2) + '\n') +print(json.dumps(report, indent=2))