From cfc0eca07a9e036f1606973cbce5ec320ee69a7d Mon Sep 17 00:00:00 2001 From: firestarsdog <229254897+firestarsdog@users.noreply.github.com> Date: Thu, 15 Oct 2026 01:56:39 -0400 Subject: [PATCH] Replay diagnostic checkpoint --- tools/diagnostics/README.md | 10 ++- tools/diagnostics/replay_render.py | 32 ++++++++-- tools/diagnostics/tests/test_replay_render.py | 64 ++++++++++++++++++- 3 files changed, 95 insertions(+), 11 deletions(-) diff --git a/tools/diagnostics/README.md b/tools/diagnostics/README.md index a55ea66081..c187757d41 100644 --- a/tools/diagnostics/README.md +++ b/tools/diagnostics/README.md @@ -69,9 +69,13 @@ owner's spawn callback, replace only `-m openpilot.starpilot.replay.render` with 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. +Records stay in a 30,000-event ring until one mode-0600 JSON report is written. Overwritten records are counted. +After pausing through the owner and confirming readiness, send SIGUSR1 to this diagnostic worker PID to request a checkpoint. +It writes at the next valid paused loop boundary with no borrowed readback pending; wait for a complete JSON report before stopping. +The report declares `report_boundary=paused_checkpoint` and excludes subsequent teardown. Without a checkpoint it attempts a +shutdown report, but the owner's cleanup deadline can kill it before that completes. Stop through the owning session and wait +boundedly for owner idle; a successful stop response alone is insufficient. SIGKILL cannot flush. 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, diff --git a/tools/diagnostics/replay_render.py b/tools/diagnostics/replay_render.py index 36d7b561a7..4b153bcc28 100644 --- a/tools/diagnostics/replay_render.py +++ b/tools/diagnostics/replay_render.py @@ -8,6 +8,7 @@ import itertools import json import os from pathlib import Path +import signal import sys import threading import time @@ -35,6 +36,7 @@ class Trace: self.connection_generation = 0 self.capabilities = {} self.configuration = {} + self.flush_requested = False def record(self, event): self.events.append(event) @@ -90,8 +92,9 @@ class Trace: 'observation_ns': time.monotonic_ns() - began} def report(self): + events, count = list(self.events), self.count stages = {} - for event in self.events: + for event in 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(): @@ -99,14 +102,14 @@ class Trace: 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), + 'max_events': MAX_EVENTS, 'overwritten_events': count - len(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)} + for stage, values in stages.items()}, 'events': events} @contextmanager -def instrument(render, trace): +def instrument(render, trace, *, flush=None): """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 @@ -251,6 +254,11 @@ def instrument(render, trace): with trace.span('scheduler_sleep') as event: event['requested_ns'] = round(seconds * 1e9) time.sleep(seconds) + sample = trace.core.ui.replay_sample if trace.core is not None else None + if (flush is not None and trace.flush_requested and sample is not None and sample.valid and sample.paused and + not trace.core.readback.pending): + trace.flush_requested = False + flush() pending_gc = {} def gc_event(phase, info): @@ -315,13 +323,23 @@ def main(argv=None): previous = sys.argv try: with os.fdopen(fd, 'w') as output: + saved = False + def save(boundary): + nonlocal saved + if not saved: + output.write(json.dumps({**trace.report(), 'source': source, 'worker_args': args.worker_args[1:], + 'report_boundary': boundary}, separators=(',', ':'), allow_nan=False)) + output.write('\n') + output.flush() + saved = True + previous_signal = signal.signal(signal.SIGUSR1, lambda *_: setattr(trace, 'flush_requested', True)) try: sys.argv = ['replay-render-diagnostic', *args.worker_args[1:]] - with instrument(render, trace): + with instrument(render, trace, flush=lambda: save('paused_checkpoint')): render.main() finally: - json.dump({**trace.report(), 'source': source, 'worker_args': args.worker_args[1:]}, output, allow_nan=False) - output.write('\n') + signal.signal(signal.SIGUSR1, previous_signal) + save('worker_exit') finally: sys.argv = previous diff --git a/tools/diagnostics/tests/test_replay_render.py b/tools/diagnostics/tests/test_replay_render.py index 68c8350400..c5f0837217 100644 --- a/tools/diagnostics/tests/test_replay_render.py +++ b/tools/diagnostics/tests/test_replay_render.py @@ -3,6 +3,7 @@ from types import SimpleNamespace as NS from unittest.mock import Mock import gc import json +import signal import stat import sys @@ -206,6 +207,43 @@ def test_snapshot_does_not_dereference_disconnected_buffer(): assert trace.snapshot()['buffer']['frame_id'] is None +def test_requested_checkpoint_waits_for_a_valid_paused_frame_without_borrowed_pixels(): + trace = diagnostic.Trace() + reports = [] + module = NS(Compositor=Compositor, MediaMux=Output, MediaStore=Output, ReplayFramePublisher=Output, + FRAME_INTERVAL=.05, time=diagnostic.time) + with diagnostic.instrument(module, trace, flush=lambda: reports.append(trace.report())): + core = Compositor() + trace.flush_requested = True + module.time.sleep(0) + assert reports == [] + core.ui.replay_sample.paused = True + core.readback.pending = True + module.time.sleep(0) + assert reports == [] + core.readback.pending = False + core.ui.replay_sample.valid = False + module.time.sleep(0) + assert reports == [] + core.ui.replay_sample.valid = True + module.time.sleep(0) + module.time.sleep(0) + assert len(reports) == 1 and reports[0]['events'] + + +def test_report_keeps_a_stable_window_when_gc_records_during_summary(monkeypatch): + trace = diagnostic.Trace() + trace.record({'stage': 'draw', 'wall_ns': 100}) + original = diagnostic.Distribution.add + def add(distribution, *args): + trace.record({'stage': 'gc', 'wall_ns': 1}) + return original(distribution, *args) + monkeypatch.setattr(diagnostic.Distribution, 'add', add) + report = trace.report() + assert len(report['events']) == 1 and report['events'][0]['stage'] == 'draw' + assert report['overwritten_events'] == 0 + + def test_cli_refuses_invalid_context_and_existing_output(tmp_path, monkeypatch): from openpilot.starpilot.replay import render output = tmp_path / 'trace.json' @@ -233,4 +271,28 @@ def test_cli_flushes_partial_report_on_worker_failure(tmp_path, monkeypatch): 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'] == [] + report = json.loads(output.read_text()) + assert not any(event['stage'] == 'draw' for event in report['events']) + assert report['report_boundary'] == 'worker_exit' + + +def test_cli_checkpoint_is_complete_before_teardown_and_is_not_appended_on_exit(tmp_path, monkeypatch): + from openpilot.starpilot.replay import render + output = tmp_path / 'trace.json' + monkeypatch.setattr(render, 'validate_context', lambda _: None) + monkeypatch.setattr(render, 'Compositor', Compositor) + saved = [] + def worker(): + core = render.Compositor() + core.ui.replay_sample.paused = True + signal.raise_signal(signal.SIGUSR1) + render.time.sleep(0) + saved.append(output.read_text()) + assert json.loads(saved[0])['report_boundary'] == 'paused_checkpoint' + raise SystemExit(1) + monkeypatch.setattr(render, 'main', worker) + previous = signal.getsignal(signal.SIGUSR1) + with pytest.raises(SystemExit): + diagnostic.main(['--output', str(output), '--', '--session-dir', str(tmp_path)]) + assert signal.getsignal(signal.SIGUSR1) is previous + assert output.read_text() == saved[0]