Fix messages being sent as two separate zmq messages

.github/workflows/build_cmake.yml

@@ -8,7 +8,7 @@
     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 @@
       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 ]
@@ -77,7 +77,7 @@
         if: matrix.configurations.compiler == 'emscripten'
         shell: bash
         run: |
           export SYSTEM_NODE=`which node` # use system node instead of old version distributed with emsdk for threading support
           source $EMSDK_HOME/emsdk_env.sh
           emcmake cmake -S . -B ../build -DCMAKE_POLICY_VERSION_MINIMUM=3.5 -DCMAKE_BUILD_TYPE=${{ matrix.cmake-build-type }} -DCMAKE_CROSSCOMPILING_EMULATOR=${SYSTEM_NODE}
 

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

concepts/client/dns_example.cpp

@@ -2,15 +2,15 @@
 #include <services/dns.hpp>
 #include <services/dns_client.hpp>
 #include <utility>
-#ifndef EMSCRIPTEN
+#ifndef __EMSCRIPTEN__
 #include <Client.hpp>
 #include <concepts/majordomo/helpers.hpp>
 #include <majordomo/base64pp.hpp>
 #include <majordomo/Broker.hpp>
 #include <majordomo/Worker.hpp>
 #else
 #include <emscripten/trace.h>
-#endif // EMSCRIPTEN
+#endif // __EMSCRIPTEN__
 
 #include <RestClient.hpp>
 
@@ -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<std::string_view> 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<std::unique_ptr<opencmw::client::ClientBase>> clients;
-#ifndef EMSCRIPTEN
+#ifndef __EMSCRIPTEN__
         zmq::Context context;
         clients.emplace_back(std::make_unique<client::MDClientCtx>(context, 20ms, "dnsTestClient"));
 #endif

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  = []<typename... T>(T&&...args) { return std::format(std::format_string<std::string, T...>("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<std::string> result = poolIO.execute<"customTaskName">(func1, getThreadName());
     // do something else ... get result, wait if necessary
     std::cout << result.get() << std::endl;

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 <new>
+ // start catch.hpp
+
+

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();

src/client/test/ClientPublisher_tests.cpp

@@ -3,6 +3,9 @@
 #include <Client.hpp>
 #include <IoBuffer.hpp>
 #include <MockServer.hpp>
+#include <ThreadAffinity.hpp>
+
+#include <algorithm>
 
 namespace opencmw_client_publisher_test {
 
@@ -65,6 +68,56 @@
     REQUIRE(received == 2);
 }
 
+struct PinThreadToSingleCpuGuard {
+    std::vector<bool> previousAffinity = opencmw::thread::getThreadAffinity();
+
+    PinThreadToSingleCpuGuard() {
+        std::vector<bool> singleCpu(previousAffinity.size(), false);
+        singleCpu[static_cast<std::size_t>(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<STRICT>::factory(URI<STRICT>(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{};

src/client/test/EmscriptenClientTests.cpp

@@ -1,4 +1,4 @@
-#ifdef EMSCRIPTEN
+#ifdef __EMSCRIPTEN__
 #include <emscripten/emscripten.h>
 #include <emscripten/threading.h>
 #include <emscripten/trace.h>
@@ -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")

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 {

src/core/include/Debug.hpp

@@ -8,7 +8,7 @@
 
 #include <format>
 
-#ifdef EMSCRIPTEN
+#ifdef __EMSCRIPTEN__
 #include <emscripten/threading.h>
 #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
 

src/core/include/ThreadPool.hpp

@@ -315,8 +315,8 @@ class BasicThreadPool {
     }
 
     // TODO: Do we need support for cancellation?
-    template<const basic_fixed_string taskName = "", uint32_t priority = 0, int32_t cpuID = -1, std::invocable Callable, typename... Args, typename R = std::invoke_result_t<Callable, Args...>>
-        requires(std::is_same_v<R, void>)
+    template<const basic_fixed_string taskName = "", uint32_t priority = 0, int32_t cpuID = -1, typename Callable, typename... Args, typename R = std::invoke_result_t<Callable, Args...>>
+        requires(std::invocable<Callable, Args...> && std::is_same_v<R, void>)
     void execute(Callable &&func, Args &&...args) {
         static thread_local SpinWait spinWait;
         if constexpr (cpuID >= 0) {
@@ -344,8 +344,8 @@ class BasicThreadPool {
         }
     }
 
-    template<const basic_fixed_string taskName = "", uint32_t priority = 0, int32_t cpuID = -1, std::invocable Callable, typename... Args, typename R = std::invoke_result_t<Callable, Args...>>
-        requires(!std::is_same_v<R, void>)
+    template<const basic_fixed_string taskName = "", uint32_t priority = 0, int32_t cpuID = -1, typename Callable, typename... Args, typename R = std::invoke_result_t<Callable, Args...>>
+        requires(std::invocable<Callable, Args...> && !std::is_same_v<R, void>)
     [[nodiscard]] std::future<R> 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<const basic_fixed_string taskName = "", uint32_t priority = 0, int32_t cpuID = -1, std::invocable Callable, typename... Args>
+    template<const basic_fixed_string taskName = "", uint32_t priority = 0, int32_t cpuID = -1, typename Callable, typename... Args>
+        requires std::invocable<Callable, Args...>
     auto createTask(Callable &&func, Args &&...funcArgs) {
         const auto getTask = [&recycledTasks = _recycledTasks](Callable &&f, Args &&...args) {
             auto extracted = recycledTasks.pop();

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::milliseconds>(std::chrono::system_clock::now() - start).count(), kInitialDelayMs + (kNUpdates * kIntervalMs));
     // TODO maybe print more detailed distribution

src/serialiser/include/IoSerialiser.hpp

@@ -380,7 +380,7 @@ constexpr void deserialise(IoBuffer &buffer, ReflectableClass auto &value, Deser
                 }
                 IoSerialiser<protocol, MemberType>::deserialise(buffer, field, getAnnotatedMember(unwrapPointerCreateIfAbsent(member(value))));
                 if constexpr (check != ProtocolCheck::IGNORE) {
-                    info.setFields[parent.fieldName][static_cast<uint64_t>(fieldIndex)] = true;
+                    info.setFields[parent.fieldName][static_cast<std::size_t>(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<uint64_t>(fieldIndex)] = true;
+                        info.setFields[parent.fieldName][static_cast<std::size_t>(fieldIndex)] = true;
                     }
                 }
             });