core/tests/multithreading.cpp

0001 // SPDX-License-Identifier: LGPL-3.0-or-later
0002 // Copyright (C) 2026 Wouter Deconinck
0003 //
0004 // Tests for concurrent use of algorithms, logging, properties, services and random numbers,
0005 // as in multithreaded frameworks. Run these with the thread sanitizer (USE_TSAN) to detect
0006 // data races that do not change the results.
0007 //
0008 #include <algorithms/algorithm.h>
0009 #include <algorithms/logger.h>
0010 #include <algorithms/random.h>
0011
0012 #include <catch2/catch_test_macros.hpp>
0013
0014 #include <algorithm>
0015 #include <cstddef>
0016 #include <cstdint>
0017 #include <functional>
0018 #include <latch>
0019 #include <limits>
0020 #include <mutex>
0021 #include <set>
0022 #include <string>
0023 #include <string_view>
0024 #include <thread>
0025 #include <vector>
0026
0027 using namespace algorithms;
0028
0029 namespace {
0030
0031 constexpr std::size_t kThreads    = 8;
0032 constexpr std::size_t kIterations = 1000;
0033
0034 // Run f(thread index) on kThreads threads that all start at the same time
0035 void runConcurrently(const std::function<void(std::size_t)>& f) {
0036   std::latch start{kThreads};
0037   std::vector<std::jthread> threads;
0038   for (std::size_t t = 0; t < kThreads; ++t) {
0039     threads.emplace_back([&f, &start, t] {
0040       start.arrive_and_wait();
0041       f(t);
0042     });
0043   }
0044 }
0045
0046 // Scales the input by a configurable factor
0047 class Scale : public Algorithm<Input<int, std::optional<int>, std::vector<int>>,
0048                                Output<int, std::vector<int>>> {
0049 public:
0050   Scale()
0051       : Algorithm{"Scale",
0052                   {"value", "offset", "values"},
0053                   {"scaled", "scaledValues"},
0054                   "Scales values by a factor, with an optional offset"} {}
0055
0056   void process(const Input& input, const Output& output) const final {
0057     const auto& [value, offset, values] = input;
0058     const auto& [scaled, scaled_values] = output;
0059
0060     const int add = (offset != nullptr) ? *offset : 0;
0061     *scaled       = *value * m_factor + add;
0062     for (std::size_t i = 0; i < std::min(values.size(), scaled_values.size()); ++i) {
0063       *scaled_values[i] = *values[i] * m_factor + add;
0064     }
0065   }
0066
0067 private:
0068   Property<int> m_factor{this, "factor", 3, "Scale factor"};
0069 };
0070
0071 class TestLogger : public LoggerMixin {
0072 public:
0073   TestLogger() : LoggerMixin("TestLogger") {}
0074   using LoggerMixin::endmsg;
0075   using LoggerMixin::info;
0076 };
0077
0078 // Records all log messages, from any thread
0079 class Recorder {
0080 public:
0081   // installed once, as the LogSvc keeps the action for the rest of the process
0082   static Recorder& instance() {
0083     static Recorder& recorder = []() -> Recorder& {
0084       static Recorder r;
0085       LogSvc::instance().init([](LogLevel, std::string_view, std::string_view msg) {
0086         std::lock_guard<std::mutex> lock{r.m_mutex};
0087         r.m_messages.emplace_back(msg);
0088       });
0089       return r;
0090     }();
0091     return recorder;
0092   }
0093   void clear() {
0094     std::lock_guard<std::mutex> lock{m_mutex};
0095     m_messages.clear();
0096   }
0097   std::vector<std::string> messages() {
0098     std::lock_guard<std::mutex> lock{m_mutex};
0099     return m_messages;
0100   }
0101
0102 private:
0103   std::mutex m_mutex;
0104   std::vector<std::string> m_messages;
0105 };
0106
0107 // Distinct service types, to be accessed for the first time concurrently
0108 constexpr const char* kIndexedSvcNames[] = {"IndexedSvc0", "IndexedSvc1", "IndexedSvc2",
0109                                             "IndexedSvc3", "IndexedSvc4", "IndexedSvc5",
0110                                             "IndexedSvc6", "IndexedSvc7"};
0111 template <std::size_t N> class IndexedSvc : public Service<IndexedSvc<N>> {
0112 public:
0113   void init() {}
0114   constexpr static const char* kName = kIndexedSvcNames[N];
0115
0116 protected:
0117   IndexedSvc() : Service<IndexedSvc<N>>(kName) {}
0118   friend class Service<IndexedSvc<N>>;
0119 };
0120
0121 } // namespace
0122
0123 TEST_CASE("Algorithms can process concurrently on one instance", "[multithreading]") {
0124   Scale algo;
0125   algo.setProperty("factor", 2);
0126   const Scale& shared = algo;
0127
0128   std::vector<std::size_t> failures(kThreads, 0);
0129   runConcurrently([&](std::size_t t) {
0130     for (std::size_t i = 0; i < kIterations; ++i) {
0131       const int value  = static_cast<int>(t * kIterations + i);
0132       const int offset = static_cast<int>(t);
0133       const std::vector<int> values{value, value + 1};
0134       int scaled = 0;
0135       std::vector<int> scaled_values(2, 0);
0136
0137       shared.process({&value, (i % 2 == 0) ? &offset : nullptr, {&values[0], &values[1]}},
0138                      {&scaled, {&scaled_values[0], &scaled_values[1]}});
0139
0140       const int add = (i % 2 == 0) ? offset : 0;
0141       if (scaled != 2 * value + add || scaled_values[0] != 2 * value + add ||
0142           scaled_values[1] != 2 * (value + 1) + add) {
0143         ++failures[t];
0144       }
0145     }
0146   });
0147   CHECK(failures == std::vector<std::size_t>(kThreads, 0));
0148 }
0149
0150 TEST_CASE("Algorithms can be constructed and configured concurrently", "[multithreading]") {
0151   std::vector<int> results(kThreads, 0);
0152   runConcurrently([&](std::size_t t) {
0153     Scale algo;
0154     algo.setProperty("factor", static_cast<int>(t));
0155     algo.validate();
0156     const int value = 10;
0157     int scaled = 0;
0158     algo.process({&value, nullptr, {}}, {&scaled, {}});
0159     results[t] = scaled;
0160   });
0161   for (std::size_t t = 0; t < kThreads; ++t) {
0162     CHECK(results[t] == static_cast<int>(10 * t));
0163   }
0164 }
0165
0166 TEST_CASE("Properties can be read concurrently", "[multithreading]") {
0167   const Scale algo;
0168   std::vector<std::size_t> failures(kThreads, 0);
0169   runConcurrently([&](std::size_t t) {
0170     for (std::size_t i = 0; i < kIterations; ++i) {
0171       if (algo.getProperty<int64_t>("factor") != 3 || !algo.hasProperty("factor") ||
0172           !algo.missingProperties().empty()) {
0173         ++failures[t];
0174       }
0175     }
0176   });
0177   CHECK(failures == std::vector<std::size_t>(kThreads, 0));
0178 }
0179
0180 TEST_CASE("Loggers report complete messages from concurrent threads", "[multithreading]") {
0181   auto& recorder = Recorder::instance();
0182   recorder.clear();
0183   const TestLogger logger;
0184
0185   runConcurrently([&](std::size_t t) {
0186     for (std::size_t i = 0; i < kIterations / 10; ++i) {
0187       logger.info(fmt::format("plain {} {}", t, i));
0188       logger.info("formatted {} {}", t, i);
0189     }
0190   });
0191
0192   std::set<std::string> expected;
0193   for (std::size_t t = 0; t < kThreads; ++t) {
0194     for (std::size_t i = 0; i < kIterations / 10; ++i) {
0195       expected.insert(fmt::format("plain {} {}", t, i));
0196       expected.insert(fmt::format("formatted {} {}", t, i));
0197     }
0198   }
0199   const auto messages = recorder.messages();
0200   CHECK(messages.size() == expected.size());
0201   CHECK(std::set<std::string>(messages.begin(), messages.end()) == expected);
0202 }
0203
0204 TEST_CASE("Logger streams report complete messages from concurrent threads",
0205           "[multithreading]") {
0206   auto& recorder = Recorder::instance();
0207   recorder.clear();
0208   const TestLogger logger;
0209
0210   runConcurrently([&](std::size_t t) {
0211     for (std::size_t i = 0; i < kIterations / 10; ++i) {
0212       logger.info() << "stream " << t << " " << i << TestLogger::endmsg;
0213     }
0214   });
0215
0216   std::set<std::string> expected;
0217   for (std::size_t t = 0; t < kThreads; ++t) {
0218     for (std::size_t i = 0; i < kIterations / 10; ++i) {
0219       expected.insert(fmt::format("stream {} {}", t, i));
0220     }
0221   }
0222   const auto messages = recorder.messages();
0223   CHECK(messages.size() == expected.size());
0224   CHECK(std::set<std::string>(messages.begin(), messages.end()) == expected);
0225 }
0226
0227 TEST_CASE("Different services can be accessed for the first time concurrently",
0228           "[multithreading]") {
0229   std::vector<const ServiceBase*> services(kThreads, nullptr);
0230   runConcurrently([&](std::size_t t) {
0231     switch (t) {
0232     case 0: services[t] = &IndexedSvc<0>::instance(); break;
0233     case 1: services[t] = &IndexedSvc<1>::instance(); break;
0234     case 2: services[t] = &IndexedSvc<2>::instance(); break;
0235     case 3: services[t] = &IndexedSvc<3>::instance(); break;
0236     case 4: services[t] = &IndexedSvc<4>::instance(); break;
0237     case 5: services[t] = &IndexedSvc<5>::instance(); break;
0238     case 6: services[t] = &IndexedSvc<6>::instance(); break;
0239     default: services[t] = &IndexedSvc<7>::instance(); break;
0240     }
0241   });
0242   const auto& registered = ServiceSvc::instance().services();
0243   for (std::size_t t = 0; t < kThreads; ++t) {
0244     const auto name = "IndexedSvc" + std::to_string(t);
0245     REQUIRE(registered.count(name) == 1);
0246     CHECK(registered.at(name) == services[t]);
0247   }
0248 }
0249
0250 TEST_CASE("A service can be accessed for the first time concurrently", "[multithreading]") {
0251   std::vector<const RandomSvc*> services(kThreads, nullptr);
0252   runConcurrently([&](std::size_t t) { services[t] = &RandomSvc::instance(); });
0253   CHECK(std::all_of(services.begin(), services.end(),
0254                     [&](const auto* svc) { return svc == services[0]; }));
0255 }
0256
0257 TEST_CASE("A generator can be shared between threads", "[multithreading]") {
0258   const Generator gen{RandomSvc::instance().createEngine(1), 16};
0259   std::vector<std::vector<uint64_t>> draws(kThreads);
0260   runConcurrently([&](std::size_t t) {
0261     for (std::size_t i = 0; i < kIterations; ++i) {
0262       draws[t].push_back(gen.uniform_int<uint64_t>(0, std::numeric_limits<uint64_t>::max()));
0263     }
0264   });
0265
0266   // full-range draws map one-to-one onto the engine output, so all draws are distinct
0267   std::set<uint64_t> unique;
0268   for (const auto& d : draws) {
0269     unique.insert(d.begin(), d.end());
0270   }
0271   CHECK(unique.size() == kThreads * kIterations);
0272 }
0273
0274 TEST_CASE("Generators of one engine can be used in parallel threads", "[multithreading]") {
0275   const auto engine = RandomSvc::instance().createEngine(2);
0276   std::vector<std::vector<uint64_t>> draws(kThreads);
0277   std::vector<double> gaussians(kThreads, 0.);
0278   runConcurrently([&](std::size_t t) {
0279     const Generator gen{engine, 16};
0280     for (std::size_t i = 0; i < kIterations; ++i) {
0281       draws[t].push_back(gen.uniform_int<uint64_t>(0, std::numeric_limits<uint64_t>::max()));
0282       gaussians[t] += gen.gaussian();
0283       gen.poisson(2.5);
0284     }
0285   });
0286
0287   // each generator refreshes its cache from the shared engine, so all draws are distinct
0288   std::set<uint64_t> unique;
0289   for (const auto& d : draws) {
0290     unique.insert(d.begin(), d.end());
0291   }
0292   CHECK(unique.size() == kThreads * kIterations);
0293 }