From 5c56198dd85cbf2ccdcdb304025f8bcaa9769b25 Mon Sep 17 00:00:00 2001 From: Ian McFarlane Date: Wed, 28 Oct 2026 19:12:44 -0400 Subject: [PATCH 1/2] Fix messages being sent as two separate zmq messages Signed-off-by: Ian McFarlane --- src/client/include/Client.hpp | 4 +- src/client/test/ClientPublisher_tests.cpp | 53 +++++++++++++++++++++++ 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/src/client/include/Client.hpp b/src/client/include/Client.hpp index 69b829eb..bde7aed8 100644 --- a/src/client/include/Client.hpp +++ b/src/client/include/Client.hpp @@ -410,9 +410,9 @@ class MDClientCtx : public ClientBase { zmq::MessageFrame reqId{ std::to_string(req_id) }; reqId.send(_control_socket_send, ZMQ_SNDMORE).assertSuccess(); zmq::MessageFrame endpoint{ std::string(uri.str()) }; - endpoint.send(_control_socket_send, isSet ? ZMQ_SNDMORE : 0).assertSuccess(); + endpoint.send(_control_socket_send, ZMQ_SNDMORE).assertSuccess(); zmq::MessageFrame rbacframe{ std::move(rbac) }; - rbacframe.send(_control_socket_send, 0).assertSuccess(); + rbacframe.send(_control_socket_send, isSet ? ZMQ_SNDMORE : 0).assertSuccess(); if (isSet) { zmq::MessageFrame dataframe{ std::move(data) }; dataframe.send(_control_socket_send, 0).assertSuccess(); diff --git a/src/client/test/ClientPublisher_tests.cpp b/src/client/test/ClientPublisher_tests.cpp index f9ab9089..ae068ade 100644 --- a/src/client/test/ClientPublisher_tests.cpp +++ b/src/client/test/ClientPublisher_tests.cpp @@ -3,6 +3,9 @@ #include #include #include +#include + +#include namespace opencmw_client_publisher_test { @@ -65,6 +68,56 @@ TEST_CASE("Basic get/set test", "[ClientContext]") { REQUIRE(received == 2); } +struct PinThreadToSingleCpuGuard { + std::vector previousAffinity = opencmw::thread::getThreadAffinity(); + + PinThreadToSingleCpuGuard() { + std::vector singleCpu(previousAffinity.size(), false); + singleCpu[static_cast(std::ranges::find(previousAffinity, true) - previousAffinity.begin())] = true; + opencmw::thread::setThreadAffinity(singleCpu); + } + + ~PinThreadToSingleCpuGuard() { opencmw::thread::setThreadAffinity(previousAffinity); } +}; + +TEST_CASE("Regression test for messages not coming together as a single multipart message", "[MDClientCtx]") { + // this bug was that a message poller might see a message while it is only partially sent. With multiple cores, the issue + // does not reproduce reliably because the sender is able to get in the second half of the message through before the + // poller can read it + const PinThreadToSingleCpuGuard pinGuard; + constexpr int kNumRequests = 500; + const zmq::Context zctx{}; + MockServer server(zctx); + MDClientCtx client(zctx, 20ms, "burstClient"); + const auto endpoint = URI::factory(URI(server.address())).scheme("mdp").path("/a.service").build(); + + for (int i = 0; i < kNumRequests; ++i) { + opencmw::client::Command set; + set.command = Command::Set; + set.topic = endpoint; + set.data = IoBuffer("abc"); + client.request(std::move(set)); + } + + int setsReceived = 0; + int setsWithData = 0; + while (setsReceived < kNumRequests) { + server.processRequest([&](auto &&req, auto &reply) { + if (req.command == Command::Set) { + setsReceived++; + if (req.data.asString() == "abc") { + setsWithData++; + } + } + reply.topic = Message::URI::factory(endpoint).build(); + }); + } + + REQUIRE(setsReceived == kNumRequests); + REQUIRE(setsWithData == kNumRequests); + client.stop(); +} + TEST_CASE("Basic subscription test", "[ClientContext]") { constexpr std::string_view payload = "payload"; const zmq::Context zctx{}; From 1d2c066404cdb2fdae24de99c4462da4bbf29d37 Mon Sep 17 00:00:00 2001 From: Ian McFarlane Date: Wed, 28 Oct 2026 19:24:09 -0400 Subject: [PATCH 2/2] Bump CI compiler versions to equal the latest gnuradio docker image Fix fallout with emscripten version bumping, and clang. Also increase the wait time for the load test because the gcc coverage build is much slower now, for some reason. Signed-off-by: Ian McFarlane --- .github/workflows/build_cmake.yml | 18 +++++++++--------- cmake/Dependencies.cmake | 2 ++ concepts/client/dns_example.cpp | 10 +++++----- concepts/core/BasicThreadPool_example.cpp | 2 +- patches/catch2-include-new.diff | 13 +++++++++++++ src/client/test/EmscriptenClientTests.cpp | 4 ++-- .../EmscriptenRestClientTest.cpp | 2 +- src/core/include/Debug.hpp | 6 +++--- src/core/include/ThreadPool.hpp | 11 ++++++----- src/majordomo/test/majordomo_load_tests.cpp | 2 +- src/serialiser/include/IoSerialiser.hpp | 4 ++-- 11 files changed, 45 insertions(+), 29 deletions(-) create mode 100644 patches/catch2-include-new.diff diff --git a/.github/workflows/build_cmake.yml b/.github/workflows/build_cmake.yml index bac2542e..ad19c164 100644 --- a/.github/workflows/build_cmake.yml +++ b/.github/workflows/build_cmake.yml @@ -8,7 +8,7 @@ on: types: [ created ] env: - REFERENCE_CONFIG: 'gcc14' # configuration used for coverage etc + REFERENCE_CONFIG: 'gcc15' # configuration used for coverage etc jobs: buildKeycloakDockerImage: @@ -27,14 +27,14 @@ jobs: fail-fast: false matrix: configurations: - - name: gcc14 - compiler: gcc14 - cc: gcc-14 - cxx: g++-14 - - name: clang20 - compiler: clang20 - cc: clang-20 - cxx: clang++-20 + - name: gcc15 + compiler: gcc15 + cc: gcc-15 + cxx: g++-15 + - name: clang21 + compiler: clang21 + cc: clang-21 + cxx: clang++-21 - name: emscripten compiler: emscripten cmake-build-type: [ Release, Debug ] diff --git a/cmake/Dependencies.cmake b/cmake/Dependencies.cmake index d9cd9a88..4868ad9d 100644 --- a/cmake/Dependencies.cmake +++ b/cmake/Dependencies.cmake @@ -8,10 +8,12 @@ FetchContent_Declare( ) # fetch content support +set(catch2_patch_command git apply --ignore-space-change --ignore-whitespace "${CMAKE_CURRENT_SOURCE_DIR}/patches/catch2-include-new.diff") # either patch catch2 to work with recent clang, otherwise upgrade to catch 3 FetchContent_Declare( Catch2 GIT_REPOSITORY https://github.com/catchorg/Catch2.git GIT_TAG v2.13.9 # latest: v2.13.9 or v3.1.0 + PATCH_COMMAND ${catch2_patch_command} || true ) # dependency of mp-units, building examples, tests, etc is off by default diff --git a/concepts/client/dns_example.cpp b/concepts/client/dns_example.cpp index 00ddda64..5b46c258 100644 --- a/concepts/client/dns_example.cpp +++ b/concepts/client/dns_example.cpp @@ -2,7 +2,7 @@ #include #include #include -#ifndef EMSCRIPTEN +#ifndef __EMSCRIPTEN__ #include #include #include @@ -10,7 +10,7 @@ #include #else #include -#endif // EMSCRIPTEN +#endif // __EMSCRIPTEN__ #include @@ -29,7 +29,7 @@ using namespace opencmw::service::dns; * The server part is not available on emscripten */ -#ifndef EMSCRIPTEN +#ifndef __EMSCRIPTEN__ void run_dns_server(std::string_view httpAddress, std::string_view mdpAddress) { majordomo::Broker<> broker{ "Broker", {} }; std::string rootPath{ "./" }; @@ -75,7 +75,7 @@ int main(int argc, char *argv[]) { const std::vector args(argv + 1, argv + argc); std::string_view command = args[0]; if (command == "server" && args.size() == 3) { -#if defined(EMSCRIPTEN) +#if defined(__EMSCRIPTEN__) std::print("unable to run server on emscripten\n"); #else std::print("running server on addresses: {}, {}\n", args[1], args[2]); @@ -85,7 +85,7 @@ int main(int argc, char *argv[]) { // get client std::print("getting client for server {}\n", args[1]); std::vector> clients; -#ifndef EMSCRIPTEN +#ifndef __EMSCRIPTEN__ zmq::Context context; clients.emplace_back(std::make_unique(context, 20ms, "dnsTestClient")); #endif diff --git a/concepts/core/BasicThreadPool_example.cpp b/concepts/core/BasicThreadPool_example.cpp index eaa584d5..a9764935 100644 --- a/concepts/core/BasicThreadPool_example.cpp +++ b/concepts/core/BasicThreadPool_example.cpp @@ -20,7 +20,7 @@ int main() { // poolWork.execute([](const auto &...args) { std::print(std::runtime_format("Hello World from thread '{}'!\n"), args...); }, getThreadName()); // constexpr auto func1 = [](T&&...args) { return std::format(std::format_string("thread '{1}' scheduled task '{0}'!\n"), getThreadName(), args...); }; - constexpr auto func1 = [](const auto &...args) { return std::format("thread '{1}' scheduled task '{0}'!\n", args..., getThreadName()); }; + constexpr auto func1 = [](const std::string &taskName) { return std::format("thread '{1}' scheduled task '{0}'!\n", taskName, getThreadName()); }; std::future result = poolIO.execute<"customTaskName">(func1, getThreadName()); // do something else ... get result, wait if necessary std::cout << result.get() << std::endl; diff --git a/patches/catch2-include-new.diff b/patches/catch2-include-new.diff new file mode 100644 index 00000000..4635f836 --- /dev/null +++ b/patches/catch2-include-new.diff @@ -0,0 +1,13 @@ +diff --git a/single_include/catch2/catch.hpp b/single_include/catch2/catch.hpp +index d2a1242..4d1e884 100644 +--- a/single_include/catch2/catch.hpp ++++ b/single_include/catch2/catch.hpp +@@ -10,6 +10,8 @@ + */ + #ifndef TWOBLUECUBES_SINGLE_INCLUDE_CATCH_HPP_INCLUDED + #define TWOBLUECUBES_SINGLE_INCLUDE_CATCH_HPP_INCLUDED ++ ++#include + // start catch.hpp + + diff --git a/src/client/test/EmscriptenClientTests.cpp b/src/client/test/EmscriptenClientTests.cpp index dff16958..ea83551b 100644 --- a/src/client/test/EmscriptenClientTests.cpp +++ b/src/client/test/EmscriptenClientTests.cpp @@ -1,4 +1,4 @@ -#ifdef EMSCRIPTEN +#ifdef __EMSCRIPTEN__ #include #include #include @@ -438,7 +438,7 @@ void wait_for_results() { } DEBUG_LOG_EVERY_SECOND("waiting for results"); - if (auto s = still_running_tests() != "") { + if (still_running_tests() != "") { DEBUG_LOG_EVERY_SECOND("waiting for " + still_running_tests()) } else { DEBUG_LOG("~wait_for_results") diff --git a/src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp b/src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp index 095b79d0..2958c48a 100644 --- a/src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp +++ b/src/client/test/emscripten_rest_client/EmscriptenRestClientTest.cpp @@ -93,7 +93,7 @@ void checkBatchSubscriptionOverlap(TestState &state) { // PThread worker counts are Emscripten internals used only for this leak check. int runningWorkerCount() noexcept { - return EM_ASM_INT({ return PThread.runningWorkers.length; }); + return EM_ASM_INT({ return Object.keys(PThread.pthreads).length; }); } int unusedWorkerCount() noexcept { diff --git a/src/core/include/Debug.hpp b/src/core/include/Debug.hpp index 284553de..05e6f279 100644 --- a/src/core/include/Debug.hpp +++ b/src/core/include/Debug.hpp @@ -8,7 +8,7 @@ #include -#ifdef EMSCRIPTEN +#ifdef __EMSCRIPTEN__ #include #endif @@ -98,14 +98,14 @@ class DebugBeforeAfter { #define _DEBUG_PREFIXES_1 "" #define _DEBUG_PREFIXES_2 "" -#ifdef EMSCRIPTEN +#ifdef __EMSCRIPTEN__ #define _DEBUG_EMSCRIPTEN_THREADS_ "(emscripten mainthread/mainruntimethread) (" << emscripten_is_main_runtime_thread() << "/" << emscripten_is_main_browser_thread() << ")" #ifndef NDEBUGEMSCRIPTEN #undef _DEBUG_PREFIXES_1 #define _DEBUG_PREFIXES_1 _DEBUG_EMSCRIPTEN_THREADS_ #endif -#endif // EMSCRIPTEN +#endif // __EMSCRIPTEN__ #define _DEBUG_PREFIXES "\t\t\t|" << __FILE__ << ":" << __LINE__ << " @ " << __PRETTY_FUNCTION__ << ": " << _DEBUG_PREFIXES_1 << _DEBUG_PREFIXES_2 diff --git a/src/core/include/ThreadPool.hpp b/src/core/include/ThreadPool.hpp index 49970028..a05cbbf0 100644 --- a/src/core/include/ThreadPool.hpp +++ b/src/core/include/ThreadPool.hpp @@ -315,8 +315,8 @@ class BasicThreadPool { } // TODO: Do we need support for cancellation? - template> - requires(std::is_same_v) + template> + requires(std::invocable && std::is_same_v) void execute(Callable &&func, Args &&...args) { static thread_local SpinWait spinWait; if constexpr (cpuID >= 0) { @@ -344,8 +344,8 @@ class BasicThreadPool { } } - template> - requires(!std::is_same_v) + template> + requires(std::invocable && !std::is_same_v) [[nodiscard]] std::future execute(Callable &&func, Args &&...funcArgs) { if constexpr (cpuID >= 0) { if (cpuID >= _affinityMask.size() || !_affinityMask[cpuID]) { @@ -420,7 +420,8 @@ class BasicThreadPool { updateThreadConstraints(nThreads + 1, thread); } - template + template + requires std::invocable auto createTask(Callable &&func, Args &&...funcArgs) { const auto getTask = [&recycledTasks = _recycledTasks](Callable &&f, Args &&...args) { auto extracted = recycledTasks.pop(); diff --git a/src/majordomo/test/majordomo_load_tests.cpp b/src/majordomo/test/majordomo_load_tests.cpp index 6bbcdfe8..634fd797 100644 --- a/src/majordomo/test/majordomo_load_tests.cpp +++ b/src/majordomo/test/majordomo_load_tests.cpp @@ -111,7 +111,7 @@ TEST_CASE("Load test", "[majordomo][majordomoworker][load_test][http2]") { } } - waitFor(responseCount, kNUpdates * kSubscriptions * kNClients, 30s); + waitFor(responseCount, kNUpdates * kSubscriptions * kNClients, 90s); std::println("Received {} responses in {}ms (Net production time: {}ms)", responseCount.load(), std::chrono::duration_cast(std::chrono::system_clock::now() - start).count(), kInitialDelayMs + (kNUpdates * kIntervalMs)); // TODO maybe print more detailed distribution diff --git a/src/serialiser/include/IoSerialiser.hpp b/src/serialiser/include/IoSerialiser.hpp index 570460f6..6638fd76 100644 --- a/src/serialiser/include/IoSerialiser.hpp +++ b/src/serialiser/include/IoSerialiser.hpp @@ -380,7 +380,7 @@ constexpr void deserialise(IoBuffer &buffer, ReflectableClass auto &value, Deser } IoSerialiser::deserialise(buffer, field, getAnnotatedMember(unwrapPointerCreateIfAbsent(member(value)))); if constexpr (check != ProtocolCheck::IGNORE) { - info.setFields[parent.fieldName][static_cast(fieldIndex)] = true; + info.setFields[parent.fieldName][static_cast(fieldIndex)] = true; } } }); @@ -400,7 +400,7 @@ constexpr void deserialise(IoBuffer &buffer, ReflectableClass auto &value, Deser field.hierarchyDepth--; field.subfields = previousSubFields - 1; if constexpr (check != ProtocolCheck::IGNORE) { - info.setFields[parent.fieldName][static_cast(fieldIndex)] = true; + info.setFields[parent.fieldName][static_cast(fieldIndex)] = true; } } });