From 624d5b999517090325999a67466327cd4a20d055 Mon Sep 17 00:00:00 2001 From: Trey Moen <50057480+greatgitsby@users.noreply.github.com> Date: Mon, 7 Sep 2026 12:33:14 -0700 Subject: [PATCH] cabana: fix stream and replay lifetimes (#38796) --- .../tools/cabana/streams/abstractstream.h | 2 +- .../tools/cabana/streams/replaystream.cc | 1 + openpilot/tools/cabana/tests/test_cabana.cc | 25 +++++++++++++++++++ openpilot/tools/cabana/ui/threadpool.h | 12 ++++++--- openpilot/tools/replay/replay.cc | 18 +++++++------ openpilot/tools/replay/replay.h | 2 ++ openpilot/tools/replay/seg_mgr.cc | 4 +++ openpilot/tools/replay/seg_mgr.h | 1 + 8 files changed, 54 insertions(+), 11 deletions(-) diff --git a/openpilot/tools/cabana/streams/abstractstream.h b/openpilot/tools/cabana/streams/abstractstream.h index dedf9a2da6..45a9af92f4 100644 --- a/openpilot/tools/cabana/streams/abstractstream.h +++ b/openpilot/tools/cabana/streams/abstractstream.h @@ -42,7 +42,7 @@ public: inline double currentSec() const { return current_sec_; } inline uint64_t toMonoTime(double sec) const { return beginMonoTime() + std::max(sec, 0.0) * 1e9; } - inline double toSeconds(uint64_t mono_time) const { return std::max(0.0, (mono_time - beginMonoTime()) / 1e9); } + inline double toSeconds(uint64_t mono_time) const { return mono_time > beginMonoTime() ? (mono_time - beginMonoTime()) / 1e9 : 0.0; } inline const std::unordered_map &lastMessages() const { return last_msgs; } bool isMessageActive(const MessageId &id) const; diff --git a/openpilot/tools/cabana/streams/replaystream.cc b/openpilot/tools/cabana/streams/replaystream.cc index fe8048c215..c0ab46abc9 100644 --- a/openpilot/tools/cabana/streams/replaystream.cc +++ b/openpilot/tools/cabana/streams/replaystream.cc @@ -19,6 +19,7 @@ ReplayStream::ReplayStream() { ReplayStream::~ReplayStream() { cancelWaits(); + if (replay) replay->stop(); } // runs on replay's merge thread: a segment of CAN data takes ~30 ms to parse and group, which dropped diff --git a/openpilot/tools/cabana/tests/test_cabana.cc b/openpilot/tools/cabana/tests/test_cabana.cc index 983a80a459..268aa9a86d 100644 --- a/openpilot/tools/cabana/tests/test_cabana.cc +++ b/openpilot/tools/cabana/tests/test_cabana.cc @@ -1,15 +1,20 @@ +#include +#include #include #include #include #include #include +#include +#include #include "common/tests/native_test.h" #include "tools/cabana/dbc/dbcfile.h" #include "tools/cabana/dbc/dbcmanager.h" #include "tools/cabana/routes.h" #include "tools/cabana/ui/qtstate.h" +#include "tools/cabana/ui/threadpool.h" #include "tools/cabana/ui/chart/downsample.h" #include "tools/cabana/utils/strings.h" @@ -366,6 +371,25 @@ void test_qt_state_blobs() { REQUIRE(!qtstate::parseQtHeaderState(fromHex("000000fe00000000000000010000000000000000010000000000000000")).has_value()); } +void test_parallel_failure_joins_workers() { + for (bool fail_on_caller : {true, false}) { + std::atomic finished = 0; + const size_t chunks = std::clamp(std::thread::hardware_concurrency(), 2, 4) + 1; + bool caught = false; + try { + parallelFor(chunks, [&](size_t begin, size_t end) { + if (begin == (fail_on_caller ? 0u : 1u)) throw std::runtime_error("task failed"); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + ++finished; + }); + } catch (const std::runtime_error &) { + caught = true; + } + REQUIRE(caught); + REQUIRE(finished == chunks - 1); + } +} + void test_pixel_envelope() { struct Point { double x, y; @@ -405,6 +429,7 @@ void test_cabana_core() { test_route_timestamps(); test_route_api_response(); test_route_json(); + test_parallel_failure_joins_workers(); test_qt_state_blobs(); } diff --git a/openpilot/tools/cabana/ui/threadpool.h b/openpilot/tools/cabana/ui/threadpool.h index 5a45ec80f2..3580f55c78 100644 --- a/openpilot/tools/cabana/ui/threadpool.h +++ b/openpilot/tools/cabana/ui/threadpool.h @@ -2,6 +2,7 @@ #include #include +#include #include #include #include @@ -36,7 +37,7 @@ public: stop_ = true; } cv_.notify_all(); - for (auto &t : threads_) t.join(); + for (auto &thread : threads_) thread.join(); } private: @@ -76,6 +77,11 @@ inline void parallelFor(size_t n, const std::function &allow, const std::vec auto event_schema = capnp::Schema::from().asStruct(); sockets_.resize(event_schema.getUnionFields().size(), nullptr); - std::vector active_services; - active_services.reserve(services.size()); + auto &active_services = active_services_; for (const auto &[name, _] : services) { bool is_blocked = std::find(block.begin(), block.end(), name) != block.end(); @@ -55,9 +54,6 @@ void Replay::setupServices(const std::vector &allow, const std::vec std::string services_str = join(active_services, ", "); rInfo("active services: %s", services_str.c_str()); - if (!sm_) { - pm_ = std::make_unique(active_services); - } } void Replay::setupSegmentManager(bool has_filters) { @@ -73,6 +69,15 @@ void Replay::setupSegmentManager(bool has_filters) { } Replay::~Replay() { + stop(); + camera_server_.reset(); + seg_mgr_.reset(); +} + +void Replay::stop() { + // Merge callbacks access both Replay and its owner. Join them while all objects + // and their owning pointers are still alive, before stopping the playback thread. + seg_mgr_->stop(); if (stream_thread_.joinable()) { rInfo("shutdown: in progress..."); interruptStream([this]() { @@ -82,8 +87,6 @@ Replay::~Replay() { stream_thread_.join(); rInfo("shutdown: done"); } - camera_server_.reset(); - seg_mgr_.reset(); } bool Replay::load() { @@ -235,6 +238,7 @@ void Replay::publishMessage(const Event *e) { if (event_filter_ && event_filter_(e)) return; if (!sm_) { + if (!pm_) pm_ = std::make_unique(active_services_); // consumers with an event filter never need one auto bytes = e->data.asBytes(); int ret = pm_->send(sockets_[e->which], (capnp::byte *)bytes.begin(), bytes.size()); if (ret == -1) { diff --git a/openpilot/tools/replay/replay.h b/openpilot/tools/replay/replay.h index 7cd5fbecd0..59e1d67d48 100644 --- a/openpilot/tools/replay/replay.h +++ b/openpilot/tools/replay/replay.h @@ -37,6 +37,7 @@ public: Replay(const std::string &route, std::vector allow, std::vector block, SubMaster *sm = nullptr, uint32_t flags = REPLAY_FLAG_NONE, const std::string &data_dir = "", bool auto_source = false); ~Replay(); + void stop(); bool load(); RouteLoadError lastRouteError() const { return route().lastError(); } void start(int seconds = 0) { seekTo(min_seconds_ + seconds, false); } @@ -107,6 +108,7 @@ private: double min_seconds_ = 0; double max_seconds_ = 0; SubMaster *sm_ = nullptr; + std::vector active_services_; std::unique_ptr pm_; std::vector sockets_; std::unique_ptr camera_server_; diff --git a/openpilot/tools/replay/seg_mgr.cc b/openpilot/tools/replay/seg_mgr.cc index 0778cacbc1..074e7e53db 100644 --- a/openpilot/tools/replay/seg_mgr.cc +++ b/openpilot/tools/replay/seg_mgr.cc @@ -3,6 +3,10 @@ #include SegmentManager::~SegmentManager() { + stop(); +} + +void SegmentManager::stop() { { std::unique_lock lock(mutex_); exit_ = true; diff --git a/openpilot/tools/replay/seg_mgr.h b/openpilot/tools/replay/seg_mgr.h index 54e156fb60..55b412b25a 100644 --- a/openpilot/tools/replay/seg_mgr.h +++ b/openpilot/tools/replay/seg_mgr.h @@ -23,6 +23,7 @@ public: SegmentManager(const std::string &route_name, uint32_t flags, const std::string &data_dir = "", bool auto_source = false) : flags_(flags), route_(route_name, data_dir, auto_source), event_data_(std::make_shared()) {} ~SegmentManager(); + void stop(); bool load(); void setCurrentSegment(int seg_num);