From c924a262282d4c2698c3cf3dbd044a32caafb600 Mon Sep 17 00:00:00 2001 From: firestarsdog <229254897+firestarsdog@users.noreply.github.com> Date: Thu, 15 Oct 2026 01:24:07 -0400 Subject: [PATCH] Add Replay compositor trace --- openpilot/starpilot/replay/render.py | 26 +- .../starpilot/replay/tests/test_render.py | 173 ++++++++- tools/diagnostics/README.md | 46 +++ tools/diagnostics/replay_render.py | 330 ++++++++++++++++++ tools/diagnostics/tests/test_replay_render.py | 236 +++++++++++++ 5 files changed, 802 insertions(+), 9 deletions(-) create mode 100644 tools/diagnostics/replay_render.py create mode 100644 tools/diagnostics/tests/test_replay_render.py diff --git a/openpilot/starpilot/replay/render.py b/openpilot/starpilot/replay/render.py index f4e11abd78..d57666a52d 100644 --- a/openpilot/starpilot/replay/render.py +++ b/openpilot/starpilot/replay/render.py @@ -507,13 +507,28 @@ class Compositor: self.render_receipts = {} self.last_route_frame = None - def draw(self, epoch: int, position: float, now_ns: int): + def _draw_rejection(self, epoch: int): sample = self.ui.replay_sample - if sample is None or sample.now_ns is None: + if sample is None or sample.now_ns is None or not sample.valid: + return 'invalid_clock' + if sample.epoch != epoch: + return 'wrong_epoch' + if (epoch, sample.now_ns) == self.last_route_frame: + return 'unchanged_clock' + # OnroadView skips the camera layer when unavailable, so this draw cannot publish. + if not (self._primary_matches() or self._optional_placeholder_ready()): + return 'camera_not_ready' + return None + + def _prepare(self): + for prepare in tuple(self.gui._render_prepare): + prepare() + + def draw(self, epoch: int, position: float, now_ns: int): + if self._draw_rejection(epoch) is not None: return False + sample = self.ui.replay_sample key = epoch, sample.now_ns - if key == self.last_route_frame: - return False from openpilot.starpilot.system.android_auto import gpu_nv12 from openpilot.starpilot.ui.shell import ShellMode rl = self.rl @@ -521,8 +536,7 @@ class Compositor: self.placement.maintain(now_ns / 1e9) snapshot = self.adapter.build(ShellMode.ONROAD) self.primary_ready = False - for prepare in tuple(self.gui._render_prepare): - prepare() + self._prepare() rl.begin_texture_mode(self.content) try: rl.clear_background(rl.BLACK) diff --git a/openpilot/starpilot/replay/tests/test_render.py b/openpilot/starpilot/replay/tests/test_render.py index 8d2afb74f7..af8c3c9de1 100644 --- a/openpilot/starpilot/replay/tests/test_render.py +++ b/openpilot/starpilot/replay/tests/test_render.py @@ -1,5 +1,5 @@ from contextlib import nullcontext -from dataclasses import dataclass +from dataclasses import dataclass, replace from types import SimpleNamespace from typing import Any from unittest.mock import Mock @@ -9,6 +9,7 @@ import pytest from openpilot.starpilot.replay import render from openpilot.starpilot.replay.render import Compositor, Engine, camera_options, frame_matches, parse_wall_offset from openpilot.starpilot.ui.presentation import Profile +from openpilot.tools.replay.display_clock import ClockSample def test_camera_selection_uses_selected_sources_and_current_preferences(): @@ -52,7 +53,7 @@ def test_camera_matches_recorded_identity_across_distinct_clock_domains(): def test_paused_same_route_clock_never_advances_filters_or_gpu(): core: Any = object.__new__(Compositor) - core.ui = SimpleNamespace(replay_sample=SimpleNamespace(now_ns=100)) + core.ui = SimpleNamespace(replay_sample=ClockSample(100, 100, 100, epoch=2, paused=True, valid=True)) core.last_route_frame = (2, 100) core.context = Mock() assert core.draw(2, 0, 9_000_000_000) is False @@ -76,7 +77,7 @@ def test_recorded_camera_preview_does_not_require_initial_device_state(frame_rea viewport_width: int = 0 core: Any = object.__new__(Compositor) - core.ui = SimpleNamespace(replay_sample=SimpleNamespace(now_ns=100), + core.ui = SimpleNamespace(replay_sample=ClockSample(100, 100, 100, epoch=2, valid=True), sm=SimpleNamespace(seen={'extrinsicsCalibration': False})) core.last_route_frame = None core.context = core.placement = Mock() @@ -95,12 +96,178 @@ def test_recorded_camera_preview_does_not_require_initial_device_state(frame_rea core.converter = Mock() core.readback = Mock() assert core.draw(2, 0, 9_000_000_000) is frame_ready + if not frame_ready: + core.context.begin_frame.assert_not_called() + core.adapter.build.assert_not_called() + core.view.render.assert_not_called() + core.readback.start.assert_not_called() + return drawn = core.view.render.call_args.args[0] assert drawn.camera_available is frame_ready assert core.adapter.build.return_value.onroad.camera_available is False assert core.readback.start.call_count == int(frame_ready) +@pytest.fixture(params=(Profile.LARGE, Profile.COMPACT)) +def draw_core(request): + from openpilot.cereal.visionipc import VisionStreamType + from openpilot.starpilot.ui.onroad_state import OnroadState, SpeedLimitObservation + + class Messages(dict): + pass + + idx = SimpleNamespace(frameId=5, timestampSof=9_000_000_000, timestampEof=9_010_000_000) + sm = Messages(narrowRoadEncodeIdx=idx, wideRoadEncodeIdx=idx) + sm.seen = dict.fromkeys(('narrowRoadEncodeIdx', 'wideRoadEncodeIdx', 'modelV2', 'radarState', 'carParams', + 'carState', 'extrinsicsCalibration'), True) + sm.valid = dict(sm.seen) + sm.logMonoTime = dict.fromkeys(sm, 9_020_000_000) + sm.logMonoTime['extrinsicsCalibration'] = 1_000_000_000 + sm.recv_frame = dict.fromkeys(sm.seen, 2) + sm.updated = {'modelV2': True} + client = Mock(frame_id=5, timestamp_sof=idx.timestampSof, timestamp_eof=idx.timestampEof) + frame = SimpleNamespace(frame_id=5) + client.recv.return_value = frame + core: Any = object.__new__(Compositor) + core.profile = request.param + core.wide = core.cabin = False + core.ui = SimpleNamespace(sm=sm, replay_sample=ClockSample(1_000_000_000, 9_050_000_000, 100, epoch=2, valid=True)) + core.camera = Mock(client=client, frame=frame, stream_type=VisionStreamType.VISION_STREAM_NARROW_ROAD) + core.camera._ensure_connection.return_value = True + core.last_route_frame = None + core.render_receipts = {'modelV2': 1} + core.context, core.placement, core.adapter = Mock(), Mock(), Mock() + core.adapter.build.return_value = SimpleNamespace(onroad=OnroadState(True, False, 15., 80., SpeedLimitObservation())) + core.rl, core.content = Mock(), Mock() + core.gui = SimpleNamespace(_render_prepare=[Mock()], _frame=0) + core.scale = 1 + core.view, core.converter, core.readback = Mock(), Mock(), Mock() + core.output = None + core.camera.render_camera_model_layer.side_effect = lambda *args, **kwargs: core._receive_primary() + def render_view(state): + # The real OnroadView enters its camera layer only with this availability flag. + if state.camera_available: + core._camera_layer(Mock(), state) + core.view.render.side_effect = render_view + return core + + +@pytest.mark.parametrize('change', ('ahead', 'behind', 'missing_frame', 'timestamp', 'stale_state', 'unseen', 'invalid')) +def test_known_camera_rejection_skips_preparation_and_preserves_updates(draw_core, change): + core = draw_core + if change in ('ahead', 'behind'): + core.camera.client.frame_id = 6 if change == 'ahead' else 4 + elif change == 'missing_frame': + core.camera.frame = None + elif change == 'timestamp': + core.camera.client.timestamp_sof += 1 + elif change == 'stale_state': + core.ui.sm.logMonoTime['narrowRoadEncodeIdx'] = 8_000_000_000 + else: + getattr(core.ui.sm, 'seen' if change == 'unseen' else 'valid')['narrowRoadEncodeIdx'] = False + assert core.draw(2, 0., 100) is False + core.context.begin_frame.assert_not_called() + core.adapter.build.assert_not_called() + core.gui._render_prepare[0].assert_not_called() + core.view.render.assert_not_called() + core.readback.start.assert_not_called() + assert core.last_route_frame is None and core.gui._frame == 0 + assert core.render_receipts == {'modelV2': 1} and core.ui.sm.updated['modelV2'] + + +@pytest.mark.parametrize('sample', (ClockSample(100, 100, 100, epoch=2), + ClockSample(100, 100, 100, epoch=3, valid=True))) +def test_invalid_or_other_epoch_clock_never_advances_draw(draw_core, sample): + core = draw_core + core.ui.replay_sample = sample + assert core.draw(2, 0., 100) is False + core.context.begin_frame.assert_not_called() + core.adapter.build.assert_not_called() + + +@pytest.mark.parametrize('replacement_id', (5, 6)) +def test_later_receive_keeps_final_identity_check(draw_core, replacement_id): + core = draw_core + def receive(**kwargs): + core.camera.client.frame_id = replacement_id + return SimpleNamespace(frame_id=replacement_id) + core.camera.client.recv.side_effect = receive + assert core.draw(2, 0., 100) is (replacement_id == 5) + core.camera.client.recv.assert_called_once_with(timeout_ms=0) + assert core.readback.start.call_count == int(replacement_id == 5) + + +def test_optional_placeholder_still_allows_later_camera_recovery(draw_core): + from openpilot.cereal.visionipc import VisionStreamType + core = draw_core + core.camera.stream_type = VisionStreamType.VISION_STREAM_WIDE_ROAD + core.camera.client.frame_id = 4 + def receive(**kwargs): + core.camera.client.frame_id = 5 + return SimpleNamespace(frame_id=5) + core.camera.client.recv.side_effect = receive + assert core.draw(2, 0., 100) + core.camera.client.recv.assert_called_once_with(timeout_ms=0) + core.readback.start.assert_called_once() + + +def test_eligible_retained_camera_draws_fresh_hud_and_route_time(draw_core): + core = draw_core + frame = core.camera.frame + core.camera.client.recv.return_value = None + assert core.draw(2, 0., 100) + core.ui.replay_sample = replace(core.ui.replay_sample, now_ns=1_050_000_000) + core.adapter.build.return_value.onroad = replace(core.adapter.build.return_value.onroad, speed_mps=20.) + assert core.draw(2, .05, 150) + core.ui.replay_sample = replace(core.ui.replay_sample, now_ns=1_100_000_000) + assert core.draw(2, .1, 200) # Animation advances even without another source update. + assert core.camera.frame is frame + assert core.view.render.call_args.args[0].speed_mps == 20. + assert [call.args[0] for call in core.context.begin_frame.call_args_list] == [1., 1.05, 1.1] + assert core.gui._frame == 3 + + +def test_rejected_camera_can_match_next_update_without_losing_model_receipt(draw_core): + core = draw_core + core.camera.client.frame_id = 6 + core.camera.frame = SimpleNamespace(frame_id=6) + assert not core.draw(2, 0., 100) + core.ui.sm['narrowRoadEncodeIdx'].frameId = 6 + def consume_model(*args, **kwargs): + assert core.ui.sm.updated['modelV2'] + core._receive_primary() + core.camera.client.recv.return_value = None + core.camera.render_camera_model_layer.side_effect = consume_model + assert core.draw(2, 0., 100) + assert core.render_receipts['modelV2'] == core.ui.sm.recv_frame['modelV2'] + assert core.context.begin_frame.call_count == 1 + + +def test_optional_missing_camera_still_publishes_existing_placeholder(draw_core, monkeypatch): + from openpilot.cereal.visionipc import VisionStreamType + core = draw_core + label = Mock() + monkeypatch.setattr('openpilot.system.ui.widgets.label.gui_label', label) + core.camera.stream_type = VisionStreamType.VISION_STREAM_WIDE_ROAD + core.camera.client.frame_id = 4 + core.camera.client.recv.return_value = None + assert core.draw(2, 0., 100) + label.assert_called_once() + core.readback.start.assert_called_once() + + +@pytest.mark.parametrize('camera_view', (1, 3, 4)) +@pytest.mark.parametrize('draw_core', (Profile.COMPACT,), indirect=True) +def test_compact_hidden_and_unrecorded_camera_keep_acquisition(draw_core, camera_view, monkeypatch): + core = draw_core + core.adapter.build.return_value.onroad = replace(core.adapter.build.return_value.onroad, appearance=replace( + core.adapter.build.return_value.onroad.appearance, camera_view=camera_view)) + monkeypatch.setattr('openpilot.system.ui.widgets.label.gui_label', Mock()) + assert core.draw(2, 0., 100) + core.camera.client.recv.assert_called_once_with(timeout_ms=0) + core.camera.render_camera_model_layer.assert_not_called() + + @pytest.mark.parametrize('profile', ('onroad', 'mici.onroad')) @pytest.mark.parametrize('replay', (False, True)) def test_camera_layers_allow_recorded_preview_and_preserve_live_start_gate(monkeypatch, profile, replay): diff --git a/tools/diagnostics/README.md b/tools/diagnostics/README.md index f5cf646ec8..a55ea66081 100644 --- a/tools/diagnostics/README.md +++ b/tools/diagnostics/README.md @@ -60,6 +60,52 @@ Focused tests (standard library compatible): python -m unittest tools.diagnostics.tests.test_performance ``` +## Replay compositor trace (20 FPS) + +`replay_render.py` is an opt-in wrapper for an existing owner-created replay worker. The normal worker never imports it. +Keep the existing owner, private session, decoder, encoder and control procedure. In a separately authorized diagnostic +owner's spawn callback, replace only `-m openpilot.starpilot.replay.render` with +`-m tools.diagnostics.replay_render --output /absolute/unused-trace.json --`, preserving all existing worker arguments +and environment. The wrapper refuses invalid private contexts and an existing output file; it does not connect to a +device, create an owner, spoof parked authority, or start a session on its own. Do not run a private producer beside live replay. + +Records stay in a 30,000-event ring until worker shutdown, when one mode-0600 JSON report is written. Overwritten records +are counted. Stop through the owning session and wait boundedly for owner idle; a successful stop response alone is insufficient. +Normal SIGTERM cleanup and worker errors flush the report; SIGKILL cannot. There are no pixel copies or per-frame file/console writes. + +The report includes source revision/hashes, worker arguments, attempt IDs, clock and camera metadata before/after acquisition, +early/late rejection reasons, inclusive and exclusive wall/thread-CPU timings, preparation, camera/model, HUD entry points, +alerts/PiP, conversion, readback, encoding, muxing, publication, sleep and readback-pending observations. Deadline lateness +mirrors the existing accepted-draw schedule; the initial draw has no previous deadline. Summary distributions cover only the +retained event window; outcome counters cover the worker lifetime. GC intervals overlap other spans and have no exclusive/CPU measurement. +The wrapper also accepts the unchanged worker before this patch: absent eligibility/preparation hooks are declared in +`capabilities`, and its rejected draws remain `unclassified_rejection` rather than invented early/late classifications. + +`connect_import` includes connection checks, possible FD exchange/import and texture setup. `primary_acquisition` includes +receive/synchronization and matching; the camera span also includes its later receive and submission. These are nested +boundaries, not isolated receive-only or pure GPU timings. `submit_to_observed_ready_ns` includes GPU queueing and polling +delay. HUD methods not individually wrapped remain in composition's exclusive remainder. Buffer metadata is an observation, +not an owned pixel snapshot. The wrapper retains no readback pixels beyond the original delivery context. + +Use a fresh unchanged baseline and candidate at the same source bundle, scene, rate and profile; record GPU/browser/encoder +conditions separately. Collect real CPU/load/temperature alongside this trace with the existing performance capture when +authorized. Compare instrumentation on/off and reverse promising A/B order. Keep picture/decoded-content comparisons separate. +This trace does not measure native upload/presentation, browser presentation, physical scanout or complete instrumentation overhead. + +Local harness checks: `./dev pytest -q tools/diagnostics/tests/test_replay_render.py`. + +Replay compatibility checks: + +```sh +./dev pytest -q openpilot/starpilot/replay/tests tools/diagnostics/tests/test_replay_render.py \ + openpilot/starpilot/ui/tests/test_replay_settings.py openpilot/starpilot/ui/tests/test_camera_display.py \ + openpilot/starpilot/ui/tests/test_camera_resources.py openpilot/starpilot/ui/tests/test_onroad_alerts.py \ + openpilot/starpilot/ui/tests/test_onroad_display_holds.py openpilot/selfdrive/ui/tests/test_camera_speed_gates.py \ + openpilot/starpilot/galaxy/tests/test_replay_http.py +node openpilot/starpilot/galaxy/tests/test_replay.mjs +node openpilot/starpilot/galaxy/tests/test_replay_media.mjs +``` + ## Offline longitudinal trace `longitudinal.py` reads one existing local rlog or qlog segment and prints one JSON object. It never opens a device, starts a publisher, or writes a log. Example: `./dev python -m tools.diagnostics.longitudinal /absolute/path/2-rlog.zst` (the `./dev` wrapper prints its own build progress before the tool's JSON). The report records input and source hashes, limits, message counts, mode transitions, rejected pairs and distributions. diff --git a/tools/diagnostics/replay_render.py b/tools/diagnostics/replay_render.py new file mode 100644 index 0000000000..36d7b561a7 --- /dev/null +++ b/tools/diagnostics/replay_render.py @@ -0,0 +1,330 @@ +"""Opt-in bounded timings for an owner-created replay worker; no decoder or rate changes.""" +from collections import Counter, deque +from contextlib import ExitStack, contextmanager +import argparse +import gc +import hashlib +import itertools +import json +import os +from pathlib import Path +import sys +import threading +import time +from types import SimpleNamespace + +from tools.diagnostics.performance import Distribution, metadata + +MAX_EVENTS = 30000 +STAMPS = ('narrowRoadEncodeIdx', 'wideRoadEncodeIdx', 'cabinEncodeIdx', + 'carState', 'controlsState', 'selfdriveState', 'modelV2', 'radarState') + + +class Trace: + def __init__(self): + self.events = deque(maxlen=MAX_EVENTS) + self.sequence = itertools.count(1) + self.local = threading.local() + self.count = self.attempt = 0 + self.outcomes = Counter() + self.core = None + self.pending_attempt = self.submitted_ns = None + self.last_draw = 0. + self.loop_now = None + self.await_loop_clock = False + self.connection_generation = 0 + self.capabilities = {} + self.configuration = {} + + def record(self, event): + self.events.append(event) + self.count += 1 + + @contextmanager + def span(self, stage): + stack = getattr(self.local, 'stack', []) + self.local.stack = stack + event = {'id': next(self.sequence), 'stage': stage, 'thread': threading.get_ident(), + 'attempt': getattr(self.local, 'attempt', None), 'parent': stack[-1]['id'] if stack else None, + 'start_ns': time.monotonic_ns()} + cpu = time.thread_time_ns() + entry = {'id': event['id'], 'child_wall': 0, 'child_cpu': 0} + stack.append(entry) + try: + yield event + except BaseException as error: + event['error_type'] = type(error).__name__ + raise + finally: + event['wall_ns'] = time.monotonic_ns() - event['start_ns'] + event['cpu_ns'] = time.thread_time_ns() - cpu + event['exclusive_wall_ns'] = event['wall_ns'] - entry['child_wall'] + event['exclusive_cpu_ns'] = event['cpu_ns'] - entry['child_cpu'] + stack.pop() + if stack: + stack[-1]['child_wall'] += event['wall_ns'] + stack[-1]['child_cpu'] += event['cpu_ns'] + self.record(event) + + def snapshot(self): + began = time.monotonic_ns() + core = self.core + sample, sm, camera = core.ui.replay_sample, core.ui.sm, core.camera + client = camera.client + connected = client is not None and client.is_connected() + frame = camera.frame if connected else None + return {'epoch': getattr(sample, 'epoch', None), 'route_ns': getattr(sample, 'now_ns', None), + 'boot_ns': getattr(sample, 'boot_ns', None), 'clock_valid': getattr(sample, 'valid', False), + 'paused': getattr(sample, 'paused', None), 'stream': int(camera.stream_type), + 'client': id(client) if client is not None else None, 'connected': connected, + 'switching': camera._switching, 'connection_generation': self.connection_generation, + 'pip_generation': core.pip_stream.generation, + 'client_frame_id': getattr(client, 'frame_id', None), + 'sof': getattr(client, 'timestamp_sof', None), 'eof': getattr(client, 'timestamp_eof', None), + 'buffer': {name: getattr(frame, name, None) for name in + ('frame_id', 'idx', 'fd', 'width', 'height', 'stride', 'uv_offset')}, + 'sources': {name: {'seen': sm.seen[name], 'valid': sm.valid[name], 'stamp': sm.logMonoTime[name], + **({'frame_id': sm[name].frameId, 'sof': sm[name].timestampSof, 'eof': sm[name].timestampEof} + if name.endswith('EncodeIdx') and sm.seen[name] else {})} + for name in STAMPS}, + 'observation_ns': time.monotonic_ns() - began} + + def report(self): + stages = {} + for event in self.events: + values = stages.setdefault(event['stage'], {key: Distribution() for key in + ('wall_ms', 'cpu_ms', 'exclusive_wall_ms', 'exclusive_cpu_ms')}) + for key, distribution in values.items(): + value = event.get(key.replace('_ms', '_ns')) + distribution.add(value / 1e6 if value is not None else None) + return {'schemaVersion': 1, 'kind': 'replay-render-trace', 'compositor_fps': 20, + 'capabilities': self.capabilities, 'configuration': self.configuration, + 'max_events': MAX_EVENTS, 'overwritten_events': self.count - len(self.events), + 'lifetime_draw_outcomes': dict(self.outcomes), + 'retained_window_stages': {stage: {key: value.summary() for key, value in values.items()} + for stage, values in stages.items()}, 'events': list(self.events)} + + +@contextmanager +def instrument(render, trace): + """Patch only this diagnostic worker; restore callbacks even when a frame fails.""" + patched = set() + trace.capabilities = {name: hasattr(render.Compositor, method) for name, method in + (('early_rejection_reason', '_draw_rejection'), ('preparation_hooks', '_prepare'))} + with ExitStack() as restore: + def replace(target, name, value): + previous = vars(target).get(name) + owned = name in vars(target) + setattr(target, name, value) + restore.callback(setattr, target, name, previous) if owned else restore.callback(delattr, target, name) + + def wrap(target, name, stage, *, camera=False, after=None): + key = id(target), name + if key in patched: + return + patched.add(key) + original = getattr(target, name) + def measured(*args, **kwargs): + before = trace.snapshot() if camera else None + with trace.span(stage) as event: + result = original(*args, **kwargs) + if after is not None: + after(event, result, args) + if camera: + event.update(before=before, after=trace.snapshot()) + if stage == 'stream_switch' or stage == 'connect_import' and not before['connected'] and event['after']['connected']: + trace.connection_generation += 1 + event['after']['connection_generation'] = trace.connection_generation + return result + replace(target, name, measured) + + def bind_view(event, result, args): + view = args[0].view + wrap(type(view), 'render', 'composition') + wrap(type(view.alert), 'render', 'alerts') + for name in ('current_speed', 'unified_speed', 'cem_gauge', 'steering_wheel', 'compact_hud', 'compact_sidebar', 'torque_bar'): + widget = getattr(view, name, None) + if widget is not None: + for method in vars(type(widget)): + if method == 'render' or method.startswith('render_'): + wrap(type(widget), method, 'hud') + + def bind_core(event, result, args): + core = trace.core = args[0] + trace.configuration = {'profile': getattr(getattr(core, 'profile', None), 'value', None), + 'width': getattr(core, 'width', None), 'height': getattr(core, 'height', None), + 'gui_target_fps': getattr(core.gui, 'target_fps', None)} + from openpilot.starpilot.system.android_auto import gpu_nv12 + camera_base = sys.modules[type(core.camera).__module__].CameraView + for method, stage in (('_render', 'camera'), ('_ensure_connection', 'connect_import'), + ('_handle_switch', 'target_acquisition'), ('_complete_switch', 'stream_switch')): + wrap(camera_base, method, stage, camera=True) + for target, method, stage in ( + (core.ui, 'update', 'replay_update'), (core.ui.sm, 'update', 'messages'), + (type(core.ui.replay_clock), 'sample', 'clock_sample'), (type(core.adapter), 'build', 'snapshot'), + (core.placement, 'maintain', 'placement'), (core.context, 'begin_frame', 'frame_clock'), + (gpu_nv12, 'compose_rgba', 'rgba_conversion'), (gpu_nv12.Nv12Converter, 'convert', 'nv12_conversion'), + (type(core.pip_stream), 'poll', 'pip_acquisition'), (type(core.pip), 'render', 'pip'), + (type(core.monitor), 'render', 'driver_monitor'), + (core.readback, 'finish', 'readback_map'), (core.readback, 'release', 'readback_release'), + (core.rl, 'begin_texture_mode', 'texture_begin'), (core.rl, 'end_texture_mode', 'texture_end'), + ): + wrap(target, method, stage) + def phase(name, callback, *args, **kwargs): + with trace.span('model' if name == 'model' else 'camera_phase'): + return original_phase(name, callback, *args, **kwargs) + original_phase = core.gui.measure_frame_phase + replace(core.gui, 'measure_frame_phase', phase) + def submitted(event, result, args): + trace.pending_attempt = trace.local.attempt + trace.submitted_ns = time.monotonic_ns() + def ready(event, result, args): + event['ready'] = result + if result and trace.submitted_ns is not None: + event['submit_to_observed_ready_ns'] = time.monotonic_ns() - trace.submitted_ns + trace.submitted_ns = None + def released(event, result, args): + trace.pending_attempt = trace.submitted_ns = None + wrap(core.readback, 'start', 'readback_submit', after=submitted) + wrap(core.readback, 'ready', 'readback_poll', after=ready) + # Record cancellation on seek too, without retaining the borrowed pixel memory. + original_release = core.readback.release + def release(): + try: + return original_release() + finally: + released(None, None, None) + replace(core.readback, 'release', release) + + original_draw, original_finish = render.Compositor.draw, render.Compositor.finish + def draw(core, *args): + trace.attempt += 1 + trace.local.attempt = trace.attempt + trace.local.rejection = None + before = trace.snapshot() + try: + with trace.span('draw') as event: + accepted = original_draw(core, *args) + reason = 'accepted' if accepted else trace.local.rejection or ( + 'late_camera_not_ready' if trace.capabilities['early_rejection_reason'] else 'unclassified_rejection') + trace.outcomes[reason] += 1 + event.update(accepted=accepted, reason=reason, captured_ns=args[2], position=args[1], before=before) + if accepted and trace.loop_now is not None: + trace.last_draw = max(trace.last_draw + render.FRAME_INTERVAL, trace.loop_now - render.FRAME_INTERVAL) + event['after'] = trace.snapshot() + return accepted + except BaseException: + trace.outcomes['error'] += 1 + raise + finally: + trace.local.attempt = None + + @contextmanager + def finish(core): + trace.local.attempt = trace.pending_attempt + try: + with trace.span('delivery_scope') as event: + with original_finish(core) as value: + event['delivered'] = value is not None + yield value + finally: + trace.local.attempt = None + + def eligibility(event, result, args): + trace.local.rejection = event['reason'] = result + + def updated(event, result, args): + trace.await_loop_clock = True + event['readback_pending'] = trace.core.readback.pending + + def monotonic(): + now = time.monotonic() + if trace.await_loop_clock: + trace.await_loop_clock, trace.loop_now = False, now + with trace.span('scheduler') as event: + event.update(readback_pending=trace.core.readback.pending, + deadline_lateness_ns=round(max(0., now - trace.last_draw - render.FRAME_INTERVAL) * 1e9) + if trace.last_draw else None) + return now + + def sleep(seconds): + with trace.span('scheduler_sleep') as event: + event['requested_ns'] = round(seconds * 1e9) + time.sleep(seconds) + + pending_gc = {} + def gc_event(phase, info): + generation = info['generation'] + if phase == 'start': + pending_gc[generation] = time.monotonic_ns() + elif generation in pending_gc: + start = pending_gc.pop(generation) + trace.record({'id': next(trace.sequence), 'stage': 'gc', 'start_ns': start, + 'wall_ns': time.monotonic_ns() - start, 'generation': generation, + 'collected': info['collected'], 'overlaps_other_stages': True}) + + wrap(render.Compositor, '_new_view', 'view_init', after=bind_view) + wrap(render.Compositor, '__init__', 'compositor_init', after=bind_core) + wrap(render.Compositor, 'update', 'update', after=updated) + wrap(render.Compositor, '_receive_primary', 'primary_acquisition', camera=True) + if trace.capabilities['early_rejection_reason']: + wrap(render.Compositor, '_draw_rejection', 'eligibility', after=eligibility) + if trace.capabilities['preparation_hooks']: + wrap(render.Compositor, '_prepare', 'preparation_hooks') + wrap(render.Compositor, 'reset_epoch', 'epoch_reset') + replace(render.Compositor, 'draw', draw) + replace(render.Compositor, 'finish', finish) + replace(render, 'time', SimpleNamespace(monotonic=monotonic, monotonic_ns=time.monotonic_ns, sleep=sleep)) + from openpilot.starpilot.system.android_auto.hw_encoder import HardwareH264Encoder + for target, method, stage in ((HardwareH264Encoder, 'encode_nv12', 'encode'), + (render.MediaMux, 'frame', 'mux'), (render.MediaStore, 'publish', 'web_publish'), + (render.ReplayFramePublisher, 'publish', 'native_publish')): + wrap(target, method, stage) + gc.callbacks.append(gc_event) + restore.callback(gc.callbacks.remove, gc_event) + yield + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--output', type=Path, required=True) + parser.add_argument('worker_args', nargs=argparse.REMAINDER) + args = parser.parse_args(argv) + if args.worker_args[:1] != ['--']: + parser.error('pass the existing owner-created worker arguments after --') + from openpilot.starpilot.replay import render + if render.FRAME_INTERVAL != .05: + parser.error('this diagnostic is restricted to the existing 20 FPS compositor') + # Reject a non-owner context before installing any instrumentation or starting the engine. + worker = argparse.ArgumentParser(add_help=False) + worker.add_argument('--session-dir', type=Path, required=True) + context, _ = worker.parse_known_args(args.worker_args[1:]) + render.validate_context(context.session_dir) + trace = Trace() + root = Path(render.__file__).resolve().parents[3] + source = metadata(root) + for name in ('openpilot/cereal/messaging/__init__.py', 'openpilot/selfdrive/ui/ui_state.py', + 'openpilot/selfdrive/ui/onroad/cameraview.py', 'openpilot/selfdrive/ui/mici/onroad/cameraview.py', + 'openpilot/tools/replay/display_clock.py', 'openpilot/starpilot/system/android_auto/headless_egl.py', + 'openpilot/starpilot/system/android_auto/hw_encoder.py'): + source['sourceSha256'][name] = hashlib.sha256((root / name).read_bytes()).hexdigest() + source['replayHashes'] = {name: hashlib.sha256(path.read_bytes()).hexdigest() for name, path in + (('render', Path(render.__file__)), ('owner', Path(render.__file__).with_name('owner.py')), + ('media', Path(render.__file__).with_name('media.py')), ('harness', Path(__file__)))} + fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600) + previous = sys.argv + try: + with os.fdopen(fd, 'w') as output: + try: + sys.argv = ['replay-render-diagnostic', *args.worker_args[1:]] + with instrument(render, trace): + render.main() + finally: + json.dump({**trace.report(), 'source': source, 'worker_args': args.worker_args[1:]}, output, allow_nan=False) + output.write('\n') + finally: + sys.argv = previous + + +if __name__ == '__main__': + main() diff --git a/tools/diagnostics/tests/test_replay_render.py b/tools/diagnostics/tests/test_replay_render.py new file mode 100644 index 0000000000..68c8350400 --- /dev/null +++ b/tools/diagnostics/tests/test_replay_render.py @@ -0,0 +1,236 @@ +from contextlib import contextmanager +from types import SimpleNamespace as NS +from unittest.mock import Mock +import gc +import json +import stat +import sys + +import pytest + +from tools.diagnostics import replay_render as diagnostic + + +def test_nested_timings_are_exclusive_and_failures_propagate(monkeypatch): + trace = diagnostic.Trace() + ticks = iter((100, 120, 150, 200)) + monkeypatch.setattr(diagnostic.time, 'monotonic_ns', lambda: next(ticks)) + monkeypatch.setattr(diagnostic.time, 'thread_time_ns', lambda: 0) + with pytest.raises(ValueError, match='original error'), trace.span('outer'): + with trace.span('inner'): + raise ValueError('original error') + inner, outer = trace.events + assert inner['wall_ns'] == inner['exclusive_wall_ns'] == 30 + assert outer['wall_ns'] == 100 and outer['exclusive_wall_ns'] == 70 + assert inner['parent'] == outer['id'] + assert inner['error_type'] == outer['error_type'] == 'ValueError' + + +def test_records_are_bounded_and_missing_cpu_is_not_zero(monkeypatch): + monkeypatch.setattr(diagnostic, 'MAX_EVENTS', 2) + trace = diagnostic.Trace() + for number in range(3): + trace.record({'stage': 'gc', 'wall_ns': number}) + report = trace.report() + assert len(report['events']) == 2 and report['overwritten_events'] == 1 + assert report['retained_window_stages']['gc']['cpu_ms']['p50'] is None + json.dumps(report, allow_nan=False) + + +class Hooks: + def update(self, *args): pass + def sample(self): return NS(valid=True, epoch=2) + def build(self, *args): pass + def render(self, *args): pass + def maintain(self, *args): pass + def begin_frame(self, *args): pass + def begin_texture_mode(self, *args): pass + def end_texture_mode(self, *args): pass + def measure_frame_phase(self, name, callback, *args, **kwargs): return callback(*args, **kwargs) + + +class CameraView: + def __init__(self): + self.client = NS(is_connected=lambda: True, frame_id=5, timestamp_sof=10, timestamp_eof=20) + self.frame = NS(frame_id=5, idx=0, fd=10, width=4, height=2, stride=4, uv_offset=8) + self.stream_type, self._switching = 0, False + def _render(self): self._ensure_connection() + def _ensure_connection(self): return True + def _handle_switch(self): pass + def _complete_switch(self): pass + + +class Pip(Hooks): + generation = 1 + def poll(self, *args): pass + + +class Readback: + def __init__(self): + self.pending = False + self.pixels = memoryview(b'pixels') + def start(self): self.pending = True + def ready(self): return self.pending + def finish(self): return self.pixels + def release(self): self.pending = False + + +class View(Hooks): + def __init__(self): self.alert = Hooks() + + +class Compositor: + def __init__(self): + self.ui = Hooks() + self.ui.replay_clock = Hooks() + self.ui.replay_sample = NS(epoch=2, now_ns=100, boot_ns=100, valid=True, paused=False) + self.ui.sm = Hooks() + self.ui.sm.seen = self.ui.sm.valid = dict.fromkeys(diagnostic.STAMPS, False) + self.ui.sm.logMonoTime = dict.fromkeys(diagnostic.STAMPS, 0) + self.camera, self.pip_stream, self.pip, self.monitor = CameraView(), Pip(), Pip(), Hooks() + self.adapter, self.placement, self.context, self.gui, self.rl = (Hooks() for _ in range(5)) + self.readback = Readback() + self._new_view() + self.rejection = None + def _new_view(self): self.view = View() + def update(self): return self.ui.replay_sample + def _receive_primary(self): self.camera._render() + def _draw_rejection(self, epoch): return self.rejection + def _prepare(self): pass + def reset_epoch(self, epoch): self.readback.release() + def draw(self, *args): + if self._draw_rejection(args[0]): + return False + self._receive_primary() + self._prepare() + self.gui.measure_frame_phase('model', lambda: None) + self.readback.start() + return True + @contextmanager + def finish(self): + if not self.readback.ready(): + yield None + return + try: + yield self.readback.finish(), (2, 0., 100) + finally: + self.readback.release() + + +class Output: + def frame(self): pass + def publish(self): pass + + +def test_worker_hooks_preserve_delivery_lifetime_and_restore_after_failure(): + trace = diagnostic.Trace() + module = NS(Compositor=Compositor, MediaMux=Output, MediaStore=Output, ReplayFramePublisher=Output, + FRAME_INTERVAL=.05, time=diagnostic.time) + original_draw, original_clock, callbacks = Compositor.draw, module.time, list(gc.callbacks) + with pytest.raises(RuntimeError, match='encoder failed'), diagnostic.instrument(module, trace): + core = Compositor() + core.update() + module.time.monotonic() + assert core.draw(2, 0., 100) + with core.finish() as value: + assert value[0].tobytes() == b'pixels' and core.readback.pending + raise RuntimeError('encoder failed') + assert not core.readback.pending + assert trace.pending_attempt is trace.submitted_ns is None + assert Compositor.draw is original_draw and module.time is original_clock + assert gc.callbacks == callbacks + delivery = next(event for event in trace.events if event['stage'] == 'delivery_scope') + assert delivery['attempt'] == 1 and delivery['error_type'] == 'RuntimeError' + ready = next(event for event in trace.events if event['stage'] == 'readback_poll') + assert ready['submit_to_observed_ready_ns'] >= 0 + assert trace.outcomes == {'accepted': 1} + json.dumps(trace.report(), allow_nan=False) + + +def test_early_reason_is_recorded_without_preparation(): + trace = diagnostic.Trace() + module = NS(Compositor=Compositor, MediaMux=Output, MediaStore=Output, ReplayFramePublisher=Output, + FRAME_INTERVAL=.05, time=diagnostic.time) + with diagnostic.instrument(module, trace): + core = Compositor() + core.rejection = 'camera_not_ready' + assert not core.draw(2, 0., 100) + assert trace.outcomes == {'camera_not_ready': 1} + assert not any(event['stage'] == 'preparation_hooks' for event in trace.events) + + +def test_unchanged_worker_does_not_invent_rejection_reasons(monkeypatch): + monkeypatch.delattr(Compositor, '_draw_rejection') + monkeypatch.delattr(Compositor, '_prepare') + monkeypatch.setattr(Compositor, 'draw', lambda *args: False) + trace = diagnostic.Trace() + module = NS(Compositor=Compositor, MediaMux=Output, MediaStore=Output, ReplayFramePublisher=Output, + FRAME_INTERVAL=.05, time=diagnostic.time) + with diagnostic.instrument(module, trace): + assert not Compositor().draw(2, 0., 100) + assert trace.capabilities == {'early_rejection_reason': False, 'preparation_hooks': False} + assert trace.outcomes == {'unclassified_rejection': 1} + + +def test_scheduler_reports_pending_inhibition_and_seek_cancels_readback(monkeypatch): + now = [10.] + monkeypatch.setattr(diagnostic.time, 'monotonic', lambda: now[0]) + trace = diagnostic.Trace() + module = NS(Compositor=Compositor, MediaMux=Output, MediaStore=Output, ReplayFramePublisher=Output, + FRAME_INTERVAL=.05, time=diagnostic.time) + with diagnostic.instrument(module, trace): + core = Compositor() + core.update() + module.time.monotonic() + assert core.draw(2, 0., 100) + now[0] += .02 + core.update() + module.time.monotonic() + scheduled = [event for event in trace.events if event['stage'] == 'scheduler'] + assert scheduled[0]['deadline_lateness_ns'] is None + assert scheduled[1]['deadline_lateness_ns'] == 20_000_000 and scheduled[1]['readback_pending'] + core.reset_epoch(3) + assert trace.pending_attempt is trace.submitted_ns is None + with core.finish() as value: + assert value is None + + +def test_snapshot_does_not_dereference_disconnected_buffer(): + class InvalidBuffer: + def __getattribute__(self, name): + raise AssertionError('disconnected buffer was read') + trace = diagnostic.Trace() + core = trace.core = Compositor() + core.camera.frame = InvalidBuffer() + core.camera.client.is_connected = lambda: False + assert trace.snapshot()['buffer']['frame_id'] is None + + +def test_cli_refuses_invalid_context_and_existing_output(tmp_path, monkeypatch): + from openpilot.starpilot.replay import render + output = tmp_path / 'trace.json' + launch = Mock() + monkeypatch.setattr(render, 'main', launch) + with pytest.raises((OSError, ValueError)): + diagnostic.main(['--output', str(output), '--', '--session-dir', str(tmp_path / 'missing')]) + assert not output.exists() + launch.assert_not_called() + monkeypatch.setattr(render, 'validate_context', lambda _: None) + output.write_text('keep') + with pytest.raises(FileExistsError): + diagnostic.main(['--output', str(output), '--', '--session-dir', str(tmp_path)]) + assert output.read_text() == 'keep' + launch.assert_not_called() + + +def test_cli_flushes_partial_report_on_worker_failure(tmp_path, monkeypatch): + from openpilot.starpilot.replay import render + output = tmp_path / 'trace.json' + monkeypatch.setattr(render, 'validate_context', lambda _: None) + monkeypatch.setattr(render, 'main', Mock(side_effect=SystemExit(1))) + previous = sys.argv + with pytest.raises(SystemExit): + diagnostic.main(['--output', str(output), '--', '--session-dir', str(tmp_path)]) + assert sys.argv is previous + assert stat.S_IMODE(output.stat().st_mode) == 0o600 + assert json.loads(output.read_text())['events'] == []