cabana: remove QtConcurrent (#37522)

tools/cabana/SConscript

@@ -20,7 +20,7 @@ SConscript(['#tools/replay/SConscript'])
 Import('replay_lib')
 
 qt_env = env.Clone()
-qt_modules = ["Widgets", "Gui", "Core", "Concurrent"]
+qt_modules = ["Widgets", "Gui", "Core"]
 
 qt_libs = []
 if arch == "Darwin":

tools/cabana/chart/chartswidget.cc

@@ -1,13 +1,13 @@
 #include "tools/cabana/chart/chartswidget.h"
 
 #include <algorithm>
+#include <future>
 
 #include <QApplication>
-#include <QFutureSynchronizer>
 #include <QMenu>
+#include <QMimeData>
 #include <QScrollBar>
 #include <QToolBar>
-#include <QtConcurrent>
 
 #include "tools/cabana/chart/chart.h"
 
@@ -171,10 +171,11 @@ void ChartsWidget::updateTabBar() {
 }
 
 void ChartsWidget::eventsMerged(const MessageEventsMap &new_events) {
-  QFutureSynchronizer<void> future_synchronizer;
+  std::vector<std::future<void>> futures;
   for (auto c : charts) {
-    future_synchronizer.addFuture(QtConcurrent::run(c, &ChartView::updateSeries, nullptr, &new_events));
+    futures.push_back(std::async(std::launch::async, &ChartView::updateSeries, c, nullptr, &new_events));
   }
+  for (auto &f : futures) f.get();
 }
 
 void ChartsWidget::timeRangeChanged(const std::optional<std::pair<double, double>> &time_range) {

tools/cabana/signalview.cc

@@ -1,6 +1,7 @@
 #include "tools/cabana/signalview.h"
 
 #include <algorithm>
+#include <future>
 
 #include <QCompleter>
 #include <QDialogButtonBox>
@@ -11,7 +12,6 @@
 #include <QPainterPath>
 #include <QPushButton>
 #include <QScrollBar>
-#include <QtConcurrent>
 #include <QVBoxLayout>
 
 #include "tools/cabana/commands.h"
@@ -641,13 +641,13 @@ void SignalView::updateState(const std::set<MessageId> *msgs) {
                delegate->button_size.height() - style()->pixelMetric(QStyle::PM_FocusFrameVMargin) * 2);
 
     auto [first, last] = can->eventsInRange(model->msg_id, std::make_pair(last_msg.ts -settings.sparkline_range, last_msg.ts));
-    QFutureSynchronizer<void> synchronizer;
+    std::vector<std::future<void>> futures;
     for (int i = first_visible.row(); i <= last_visible.row(); ++i) {
       auto item = model->getItem(model->index(i, 1));
-      synchronizer.addFuture(QtConcurrent::run(
-          &item->sparkline, &Sparkline::update, item->sig, first, last, settings.sparkline_range, size));
+      futures.push_back(std::async(std::launch::async,
+          &Sparkline::update, &item->sparkline, item->sig, first, last, settings.sparkline_range, size));
     }
-    synchronizer.waitForFinished();
+    for (auto &f : futures) f.get();
   }
 
   for (int i = 0; i < model->rowCount(); ++i) {

tools/cabana/streams/routes.cc

@@ -11,7 +11,7 @@
 #include <QMessageBox>
 #include <QPainter>
 #include <QPointer>
-#include <QtConcurrent>
+#include <thread>
 
 #include "tools/replay/py_downloader.h"
 
@@ -72,13 +72,13 @@ RoutesDialog::RoutesDialog(QWidget *parent) : QDialog(parent) {
 
   // Fetch devices
   QPointer<RoutesDialog> self = this;
-  QtConcurrent::run([self]() {
+  std::thread([self]() {
     std::string result = PyDownloader::getDevices();
     auto [success, error_code] = checkApiResponse(result);
     QMetaObject::invokeMethod(qApp, [self, r = QString::fromStdString(result), success, error_code]() {
       if (self) self->parseDeviceList(r, success, error_code);
     }, Qt::QueuedConnection);
-  });
+  }).detach();
 }
 
 void RoutesDialog::parseDeviceList(const QString &json, bool success, int error_code) {
@@ -114,14 +114,14 @@ void RoutesDialog::fetchRoutes() {
 
   int request_id = ++fetch_id_;
   QPointer<RoutesDialog> self = this;
-  QtConcurrent::run([self, did, start_ms, end_ms, preserved, request_id]() {
+  std::thread([self, did, start_ms, end_ms, preserved, request_id]() {
     std::string result = PyDownloader::getDeviceRoutes(did, start_ms, end_ms, preserved);
     if (!self || self->fetch_id_ != request_id) return;
     auto [success, error_code] = checkApiResponse(result);
     QMetaObject::invokeMethod(qApp, [self, r = QString::fromStdString(result), success, error_code, request_id]() {
       if (self && self->fetch_id_ == request_id) self->parseRouteList(r, success, error_code);
     }, Qt::QueuedConnection);
-  });
+  }).detach();
 }
 
 void RoutesDialog::parseRouteList(const QString &json, bool success, int error_code) {

tools/cabana/tools/findsignal.cc

@@ -1,10 +1,11 @@
 #include "tools/cabana/tools/findsignal.h"
 
+#include <thread>
+
 #include <QFormLayout>
 #include <QHBoxLayout>
 #include <QHeaderView>
 #include <QMenu>
-#include <QtConcurrent>
 #include <QTimer>
 #include <QVBoxLayout>
 
@@ -35,22 +36,35 @@ void FindSignalModel::search(std::function<bool(double)> cmp) {
   const auto prev_sigs = !histories.isEmpty() ? histories.back() : initial_signals;
   filtered_signals.clear();
   filtered_signals.reserve(prev_sigs.size());
-  QtConcurrent::blockingMap(prev_sigs, [&](auto &s) {
-    const auto &events = can->events(s.id);
-    auto first = std::upper_bound(events.cbegin(), events.cend(), s.mono_time, CompareCanEvent());
-    auto last = events.cend();
-    if (last_time < std::numeric_limits<uint64_t>::max()) {
-      last = std::upper_bound(events.cbegin(), events.cend(), last_time, CompareCanEvent());
-    }
 
-    auto it = std::find_if(first, last, [&](const CanEvent *e) { return cmp(get_raw_value(e->dat, e->size, s.sig)); });
-    if (it != last) {
-      auto values = s.values;
-      values += QString("(%1, %2)").arg(can->toSeconds((*it)->mono_time), 0, 'f', 3).arg(get_raw_value((*it)->dat, (*it)->size, s.sig));
-      std::lock_guard lk(lock);
-      filtered_signals.push_back({.id = s.id, .mono_time = (*it)->mono_time, .sig = s.sig, .values = values});
-    }
-  });
+  unsigned int num_threads = std::max(1u, std::thread::hardware_concurrency());
+  size_t chunk = (prev_sigs.size() + num_threads - 1) / num_threads;
+  std::vector<std::thread> threads;
+  for (unsigned int t = 0; t < num_threads && t * chunk < (size_t)prev_sigs.size(); ++t) {
+    size_t start = t * chunk;
+    size_t end = std::min(start + chunk, (size_t)prev_sigs.size());
+    threads.emplace_back([&, start, end]() {
+      for (size_t i = start; i < end; ++i) {
+        const auto &s = prev_sigs[i];
+        const auto &events = can->events(s.id);
+        auto first = std::upper_bound(events.cbegin(), events.cend(), s.mono_time, CompareCanEvent());
+        auto last = events.cend();
+        if (last_time < std::numeric_limits<uint64_t>::max()) {
+          last = std::upper_bound(events.cbegin(), events.cend(), last_time, CompareCanEvent());
+        }
+
+        auto it = std::find_if(first, last, [&](const CanEvent *e) { return cmp(get_raw_value(e->dat, e->size, s.sig)); });
+        if (it != last) {
+          auto values = s.values;
+          values += QString("(%1, %2)").arg(can->toSeconds((*it)->mono_time), 0, 'f', 3).arg(get_raw_value((*it)->dat, (*it)->size, s.sig));
+          std::lock_guard lk(lock);
+          filtered_signals.push_back({.id = s.id, .mono_time = (*it)->mono_time, .sig = s.sig, .values = values});
+        }
+      }
+    });
+  }
+  for (auto &th : threads) th.join();
+
   histories.push_back(filtered_signals);
 
   endResetModel();

tools/cabana/videowidget.cc

@@ -1,6 +1,7 @@
 #include "tools/cabana/videowidget.h"
 
 #include <algorithm>
+#include <thread>
 
 #include <QAction>
 #include <QActionGroup>
@@ -9,7 +10,6 @@
 #include <QPainter>
 #include <QStyleOptionSlider>
 #include <QVBoxLayout>
-#include <QtConcurrent>
 
 #include "tools/cabana/tools/routeinfo.h"
 
@@ -334,19 +334,31 @@ StreamCameraView::StreamCameraView(std::string stream_name, VisionStreamType str
 
 void StreamCameraView::parseQLog(std::shared_ptr<LogReader> qlog) {
   std::mutex mutex;
-  QtConcurrent::blockingMap(qlog->events.cbegin(), qlog->events.cend(), [this, &mutex](const Event &e) {
-    if (e.which == cereal::Event::Which::THUMBNAIL) {
-      capnp::FlatArrayMessageReader reader(e.data);
-      auto thumb_data = reader.getRoot<cereal::Event>().getThumbnail();
-      auto image_data = thumb_data.getThumbnail();
-      if (QPixmap thumb; thumb.loadFromData(image_data.begin(), image_data.size(), "jpeg")) {
-        QPixmap generated_thumb = generateThumbnail(thumb, can->toSeconds(thumb_data.getTimestampEof()));
-        std::lock_guard lock(mutex);
-        thumbnails[thumb_data.getTimestampEof()] = generated_thumb;
-        big_thumbnails[thumb_data.getTimestampEof()] = thumb;
+  const auto &events = qlog->events;
+  unsigned int num_threads = std::max(1u, std::thread::hardware_concurrency());
+  size_t chunk = (events.size() + num_threads - 1) / num_threads;
+  std::vector<std::thread> threads;
+  for (unsigned int t = 0; t < num_threads && t * chunk < events.size(); ++t) {
+    size_t start = t * chunk;
+    size_t end = std::min(start + chunk, events.size());
+    threads.emplace_back([this, &mutex, &events, start, end]() {
+      for (size_t i = start; i < end; ++i) {
+        const Event &e = events[i];
+        if (e.which == cereal::Event::Which::THUMBNAIL) {
+          capnp::FlatArrayMessageReader reader(e.data);
+          auto thumb_data = reader.getRoot<cereal::Event>().getThumbnail();
+          auto image_data = thumb_data.getThumbnail();
+          if (QPixmap thumb; thumb.loadFromData(image_data.begin(), image_data.size(), "jpeg")) {
+            QPixmap generated_thumb = generateThumbnail(thumb, can->toSeconds(thumb_data.getTimestampEof()));
+            std::lock_guard lock(mutex);
+            thumbnails[thumb_data.getTimestampEof()] = generated_thumb;
+            big_thumbnails[thumb_data.getTimestampEof()] = thumb;
+          }
+        }
       }
-    }
-  });
+    });
+  }
+  for (auto &th : threads) th.join();
   update();
 }