cabana: fix stream and replay lifetimes (#38796)

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<MessageId, CanData> &lastMessages() const { return last_msgs; }
   bool isMessageActive(const MessageId &id) const;

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

openpilot/tools/cabana/tests/test_cabana.cc

@@ -1,15 +1,20 @@
 
+#include <atomic>
+#include <chrono>
 #include <cmath>
 #include <cstdlib>
 #include <ctime>
 #include <filesystem>
 #include <sstream>
+#include <stdexcept>
+#include <thread>
 
 #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<int> finished = 0;
+    const size_t chunks = std::clamp<size_t>(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();
 }
 

openpilot/tools/cabana/ui/threadpool.h

@@ -2,6 +2,7 @@
 
 #include <algorithm>
 #include <condition_variable>
+#include <exception>
 #include <functional>
 #include <future>
 #include <memory>
@@ -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<void(size_t begin, size_t
   for (; begin < n; begin += chunk) {
     futures.push_back(ThreadPool::instance().run([&fn, begin, end = std::min(begin + chunk, n)]() { fn(begin, end); }));
   }
-  fn(0, std::min(chunk, n));
-  for (auto &f : futures) f.get();
+  // All tasks must finish before captures can go out of scope, even if one fails.
+  std::exception_ptr error;
+  try { fn(0, std::min(chunk, n)); } catch (...) { error = std::current_exception(); }
+  for (auto &f : futures) {
+    try { f.get(); } catch (...) { if (!error) error = std::current_exception(); }
+  }
+  if (error) std::rethrow_exception(error);
 }

openpilot/tools/replay/replay.cc

@@ -40,8 +40,7 @@ void Replay::setupServices(const std::vector<std::string> &allow, const std::vec
   auto event_schema = capnp::Schema::from<cereal::Event>().asStruct();
   sockets_.resize(event_schema.getUnionFields().size(), nullptr);
 
-  std::vector<const char *> 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<std::string> &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<PubMaster>(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<PubMaster>(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) {

openpilot/tools/replay/replay.h

@@ -37,6 +37,7 @@ public:
   Replay(const std::string &route, std::vector<std::string> allow, std::vector<std::string> 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<const char *> active_services_;
   std::unique_ptr<PubMaster> pm_;
   std::vector<const char*> sockets_;
   std::unique_ptr<CameraServer> camera_server_;

openpilot/tools/replay/seg_mgr.cc

@@ -3,6 +3,10 @@
 #include <algorithm>
 
 SegmentManager::~SegmentManager() {
+  stop();
+}
+
+void SegmentManager::stop() {
   {
     std::unique_lock lock(mutex_);
     exit_ = true;

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<EventData>()) {}
   ~SegmentManager();
+  void stop();
 
   bool load();
   void setCurrentSegment(int seg_num);