test, ci: Fix threadsanitizer errors in mptest

ci/configs/sanitize.bash

@@ -1,5 +1,6 @@
 CI_DESC="CI job running ThreadSanitizer"
 CI_DIR=build-sanitize
+NIX_ARGS=(--arg enableLibcxx true --argstr libcxxSanitizers "Thread" --argstr capnprotoSanitizers "thread")
 export CXX=clang++
 export CXXFLAGS="-ggdb -Werror -Wall -Wextra -Wpedantic -Wthread-safety -Wno-unused-parameter -fsanitize=thread"
 CMAKE_ARGS=()

include/mp/proxy-io.h

@@ -98,6 +98,7 @@ class LoggingErrorHandler : public kj::TaskSet::ErrorHandler
     EventLoop& m_loop;
 };
 
+//! Log flags. Update stringify function if changed!
 enum class Log {
     Trace = 0,
     Debug,
@@ -107,6 +108,8 @@ enum class Log {
     Raise,
 };
 
+kj::StringPtr KJ_STRINGIFY(Log flags);
+
 struct LogMessage {
 
     //! Message to be logged

shell.nix

@@ -3,12 +3,19 @@
 , enableLibcxx ? false # Whether to use libc++ toolchain and libraries instead of libstdc++
 , minimal ? false # Whether to create minimal shell without extra tools (faster when cross compiling)
 , capnprotoVersion ? null
+, capnprotoSanitizers ? null # Optional sanitizers to build cap'n proto with
 , cmakeVersion ? null
+, libcxxSanitizers ? null # Optional LLVM_USE_SANITIZER value to use for libc++, see https://llvm.org/docs/CMake.html
 }:
 
 let
   lib  = pkgs.lib;
-  llvm = crossPkgs.llvmPackages_20;
+  llvmBase = crossPkgs.llvmPackages_21;
+  llvm = llvmBase // lib.optionalAttrs (libcxxSanitizers != null) {
+    libcxx = llvmBase.libcxx.override {
+      devExtraCmakeFlags = [ "-DLLVM_USE_SANITIZER=${libcxxSanitizers}" ];
+    };
+  };
   capnprotoHashes = {
     "0.7.0" = "sha256-Y/7dUOQPDHjniuKNRw3j8dG1NI9f/aRWpf8V0WzV9k8=";
     "0.7.1" = "sha256-3cBpVmpvCXyqPUXDp12vCFCk32ZXWpkdOliNH37UwWE=";
@@ -35,7 +42,17 @@ let
   } // (lib.optionalAttrs (lib.versionOlder capnprotoVersion "0.10") {
     env = { }; # Drop -std=c++20 flag forced by nixpkgs
   }));
-  capnproto = capnprotoBase.override (lib.optionalAttrs enableLibcxx { clangStdenv = llvm.libcxxStdenv; });
+  capnproto = (capnprotoBase.overrideAttrs (old: lib.optionalAttrs (capnprotoSanitizers != null) {
+    env = (old.env or { }) // {
+      CXXFLAGS =
+        lib.concatStringsSep " " [
+          (old.env.CXXFLAGS or "")
+          "-fsanitize=${capnprotoSanitizers}"
+          "-fno-omit-frame-pointer"
+          "-g"
+        ];
+    };
+  })).override (lib.optionalAttrs enableLibcxx { clangStdenv = llvm.libcxxStdenv; });
   clang = if enableLibcxx then llvm.libcxxClang else llvm.clang;
   clang-tools = llvm.clang-tools.override { inherit enableLibcxx; };
   cmakeHashes = {

src/mp/proxy.cpp

@@ -24,6 +24,7 @@
 #include <kj/debug.h>
 #include <kj/function.h>
 #include <kj/memory.h>
+#include <kj/string.h>
 #include <map>
 #include <memory>
 #include <optional>
@@ -430,4 +431,16 @@ std::string LongThreadName(const char* exe_name)
     return g_thread_context.thread_name.empty() ? ThreadName(exe_name) : g_thread_context.thread_name;
 }
 
+kj::StringPtr KJ_STRINGIFY(Log v)
+{
+    switch (v) {
+        case Log::Trace:   return "Trace";
+        case Log::Debug:   return "Debug";
+        case Log::Info:    return "Info";
+        case Log::Warning: return "Warning";
+        case Log::Error:   return "Error";
+        case Log::Raise:   return "Raise";
+    }
+    return "<Log?>";
+}
 } // namespace mp

test/mp/test/test.cpp

@@ -10,10 +10,8 @@
 #include <capnp/rpc.h>
 #include <condition_variable>
 #include <cstring>
-#include <exception>
 #include <functional>
 #include <future>
-#include <iostream>
 #include <kj/async.h>
 #include <kj/async-io.h>
 #include <kj/common.h>
@@ -24,14 +22,15 @@
 #include <kj/test.h>
 #include <memory>
 #include <mp/proxy.h>
-#include "mp/proxy.capnp.h"
+#include <mp/proxy.capnp.h>
 #include <mp/proxy-io.h>
-#include "mp/util.h"
+#include <mp/util.h>
 #include <optional>
 #include <set>
 #include <stdexcept>
 #include <string>
 #include <string_view>
+#include <system_error>
 #include <thread>
 #include <utility>
 #include <vector>
@@ -67,9 +66,10 @@ class TestSetup
 
     TestSetup(bool client_owns_connection = true)
         : thread{[&] {
-              EventLoop loop("mptest", [](mp::LogMessage log_data) {
-                  std::cout << "LOG" << (int)log_data.level << ": " << log_data.message << "\n";
-                  if (log_data.level == mp::Log::Raise) throw std::runtime_error(log_data.message);
+              EventLoop loop("mptest", [](mp::LogMessage log) {
+                  // Info logs are not printed by default, but will be shown with `mptest --verbose`
+                  KJ_LOG(INFO, log.level, log.message);
+                  if (log.level == mp::Log::Raise) throw std::runtime_error(log.message);
               });
               auto pipe = loop.m_io_context.provider->newTwoWayPipe();
 
@@ -321,21 +321,21 @@ KJ_TEST("Make simultaneous IPC callbacks with same request_thread and callback_t
     setup.server->m_impl->m_fn = [&] {};
     foo->callFnAsync();
     ThreadContext& tc{g_thread_context};
-    std::optional<Thread::Client> callback_thread, request_thread;
-    {
+    Thread::Client *callback_thread, *request_thread;
+    foo->m_context.loop->sync([&] {
         Lock lock(tc.waiter->m_mutex);
-        callback_thread = tc.callback_threads.at(foo->m_context.connection)->m_client;
-        request_thread = tc.request_threads.at(foo->m_context.connection)->m_client;
-    }
+        callback_thread = &tc.callback_threads.at(foo->m_context.connection)->m_client;
+        request_thread = &tc.request_threads.at(foo->m_context.connection)->m_client;
+    });
 
     setup.server->m_impl->m_fn = [&] {
         try
         {
             signal.get_future().get();
         }
-        catch(const std::exception& e)
+        catch (const std::future_error& e)
         {
-            KJ_EXPECT(e.what() == std::string("Future already retrieved"));
+            KJ_EXPECT(e.code() == std::make_error_code(std::future_errc::future_already_retrieved));
         }
     };