Fix race in MPSC algo (#1812)

* Fix race in MPSC algo

Relacy helped identify a race in the existing MPSC algo. I am having a
hard time exactly explaining what's going on, but in the newly added
unit test (the non Relacy one), I am able to observe three different
odd behaviors

 - a consumer consuming the same elemment in an finite loop, apparently
   due to the internal next pointers pointing in some sort of cycle
 - consumer returning &__nil_!
 - consumer never able to consume a produced value (node is lost)

With the non-relacy unit test, in the existing algo, if I insert a
random sleep of 0-10 microseconds in push_back after __back_ is
exchanged, I can observe one of the above behaviors nearly every
single time. The most common was the first behavior.

The existing algo claims it came from Dmitry Vyukov's implementation,
though one key difference is that the existing one uses an atomic
pointer to a Node for the "nil" object, whereas Dmitry's stores an
actual Node object embedded in the queue.

I re-implemented the version in stdexec exactly as it appears on
Dmitry's website (which I had to dig up on archive.org), and it passes
newly added Relacy (exploring many thread interleavings) and non-Relacy
unit tests.

I originally tracked down a bug in timed_thread_scheduler.cpp, where
sometimes `STDEXEC_ASSERT(op->command_ == command_type::command_type::stop);`
failed.

* Add Relacy tests to the build

* Default relacy only fofr GNU

* Remove old Relacy Makefile

* Simplify tests

* Improve implementation to be near to Dmitry's

* Final fixups

 - Use STDEXEC::__std::atomic
 - Update compiler requirements for Relacy
 - Compile Relacy with TBB off

---------

Co-authored-by: Maikel Nadolski <maikel.nadolski@gmail.com>
Co-authored-by: Eric Niebler <eniebler@nvidia.com>

Author: 3 people

CMakeLists.txt

@@ -150,6 +150,26 @@ else()
   set(stdexec_compiler_frontend ${CMAKE_CXX_COMPILER_ID})
 endif()
 
+# Build relacy tests by default only for GNU compiler and if tests are enabled
+# It will take work to get other platforms to work in the future.
+# Additionally, do not build the Relacy tests when a sanitizer is active. TSAN
+# doesn't do anything since Relacy simulates threads with fibers anyway. Relacy
+# has primtive (compared to ASAN) use-after-free detection which doesn't compose
+# with ASAN.
+# Additionally, we require GCC-12 or newer for Relacy, at __atomic.hpp has fallback
+# when std::atomic<std::shared_ptr<>> is not provided, and GCC-12 is the first
+# libstdc++ that provides it.
+if(
+  ${STDEXEC_BUILD_TESTS} AND
+  CMAKE_CXX_COMPILER_ID STREQUAL "GNU" AND
+  CMAKE_CXX_COMPILER_VERSION VERSION_GREATER_EQUAL 12 AND
+  NOT CMAKE_CXX_FLAGS MATCHES "-fsanitize")
+  set(STDEXEC_BUILD_RELACY_TESTS_DEFAULT ON)
+else()
+  set(STDEXEC_BUILD_RELACY_TESTS_DEFAULT OFF)
+endif()
+option(STDEXEC_BUILD_RELACY_TESTS "Build stdexec relacy tests" ${STDEXEC_BUILD_RELACY_TESTS_DEFAULT})
+
 set(stdexec_export_targets)
 
 # Define the main library

include/exec/timed_thread_scheduler.hpp

@@ -47,16 +47,20 @@ namespace experimental::execution
         stop
       };
 
+      // Default ctor for __intrusive_mpsc_queue's internal stub node
+      constexpr timed_thread_operation_base() = default;
+
       constexpr timed_thread_operation_base(
         void (*set_value)(timed_thread_operation_base*) noexcept,
         command_type command = command_type::schedule) noexcept
         : command_{command}
         , set_value_{set_value}
       {}
 
-      STDEXEC::__std::atomic<void*> next_{nullptr};
-      command_type                  command_;
-      void (*set_value_)(timed_thread_operation_base*) noexcept;
+      using set_value_fn_t = void (*)(timed_thread_operation_base*) noexcept;
+      STDEXEC::__std::atomic<timed_thread_operation_base*> next_{nullptr};
+      command_type                                         command_;
+      set_value_fn_t                                       set_value_;
     };
 
     template <class Tp>

include/stdexec/__detail/__atomic.hpp

@@ -64,7 +64,7 @@ namespace STDEXEC::__std
   using std::atomic_thread_fence;
   using std::atomic_signal_fence;
 
-#  if __cpp_lib_atomic_ref >= 2018'06L && !defined(STDEXEC_RELACY)
+#  if __cpp_lib_atomic_ref >= 2018'06L
   using std::atomic_ref;
 #  else
   inline constexpr int __atomic_flag_map[] = {

include/stdexec/__detail/__intrusive_mpsc_queue.hpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) Dmitiy V'jukov
+ * Copyright (c) Dmitry V'jukov
  * Copyright (c) 2024 Maikel Nadolski
  * Copyright (c) 2024 NVIDIA Corporation
  *
@@ -23,72 +23,84 @@
 
 #include "__atomic.hpp"
 
-#include "./__spin_loop_pause.hpp"
+#include "stdexec/__detail/__config.hpp"
 
 namespace STDEXEC
 {
   template <auto _Ptr>
   class __intrusive_mpsc_queue;
 
-  template <class _Node, __std::atomic<void*> _Node::* _Next>
+  // _Node must be default_initializable only for the queue to construct an
+  // internal "stub" node - only the _Next data element is accessed internally.
+  template <class _Node, __std::atomic<_Node*> _Node::* _Next>
+    requires __std::default_initializable<_Node>
   class __intrusive_mpsc_queue<_Next>
   {
-    __std::atomic<void*>  __back_{&__nil_};
-    void*                 __front_{&__nil_};
-    __std::atomic<_Node*> __nil_ = nullptr;
+    __std::atomic<_Node*> __head_{&__stub_};
+    _Node*                __tail_{&__stub_};
+    _Node                 __stub_{};
 
-    constexpr void push_back_nil()
+   public:
+    __intrusive_mpsc_queue()
     {
-      __nil_.store(nullptr, __std::memory_order_relaxed);
-      auto* __prev = static_cast<_Node*>(__back_.exchange(&__nil_, __std::memory_order_acq_rel));
-      (__prev->*_Next).store(&__nil_, __std::memory_order_release);
+      (__stub_.*_Next).store(nullptr, __std::memory_order_release);
     }
 
-   public:
     constexpr auto push_back(_Node* __new_node) noexcept -> bool
     {
       (__new_node->*_Next).store(nullptr, __std::memory_order_relaxed);
-      void* __prev_back = __back_.exchange(__new_node, __std::memory_order_acq_rel);
-      bool  __is_nil    = __prev_back == static_cast<void*>(&__nil_);
-      if (__is_nil)
-      {
-        __nil_.store(__new_node, __std::memory_order_release);
-      }
-      else
-      {
-        (static_cast<_Node*>(__prev_back)->*_Next).store(__new_node, __std::memory_order_release);
-      }
-      return __is_nil;
+      _Node* __prev = __head_.exchange(__new_node, __std::memory_order_acq_rel);
+      (__prev->*_Next).store(__new_node, __std::memory_order_release);
+      return __prev == &__stub_;
     }
 
     constexpr auto pop_front() noexcept -> _Node*
     {
-      if (__front_ == static_cast<void*>(&__nil_))
+      _Node* __tail = this->__tail_;
+      STDEXEC_ASSERT(__tail != nullptr);
+      _Node* __next = (__tail->*_Next).load(__std::memory_order_acquire);
+      // If tail is pointing to the stub node we need to advance it once more
+      if (&__stub_ == __tail)
       {
-        _Node* __next = __nil_.load(__std::memory_order_acquire);
-        if (!__next)
+        if (nullptr == __next)
         {
           return nullptr;
         }
-        __front_ = __next;
+        this->__tail_ = __next;
+        __tail        = __next;
+        __next        = (__next->*_Next).load(__std::memory_order_acquire);
+      }
+      // Normal case: there is a next node and we can just advance the tail
+      if (nullptr != __next)
+      {
+        this->__tail_ = __next;
+        return __tail;
       }
-      auto* __front = static_cast<_Node*>(__front_);
-      void* __next  = (__front->*_Next).load(__std::memory_order_acquire);
-      if (__next)
+      // Next is nullptr here means that either:
+      // 1) There are no more nodes in the queue
+      // 2) A producer is in the middle of adding a new node
+      _Node const * __head = this->__head_.load(__std::memory_order_acquire);
+      // A producer is in the middle of adding a new node
+      // we cannot return tail as we cannot link the next node yet
+      if (__tail != __head)
       {
-        __front_ = __next;
-        return __front;
+        return nullptr;
       }
-      STDEXEC_ASSERT(!__next);
-      push_back_nil();
-      do
+      // No more nodes in the queue - we need to insert a stub node
+      // to be able to link to an eventual empty state (or new nodes)
+      push_back(&__stub_);
+      // Now re-attempt to load next
+      __next = (__tail->*_Next).load(__std::memory_order_acquire);
+      if (nullptr != __next)
       {
-        __spin_loop_pause();
-        __next = (__front->*_Next).load(__std::memory_order_acquire);
+        // Successfully linked either a new node or the stub node
+        this->__tail_ = __next;
+        return __tail;
       }
-      while (!__next);
-      __front_ = __next;
-      return __front;
+      // A producer is in the middle of adding a new node since next is still nullptr
+      // and not our stub node, thus we cannot link the next node yet
+      return nullptr;
     }
   };
-}  // namespace STDEXEC
+
+}  // namespace STDEXEC

include/stdexec/__detail/__stop_token.hpp

@@ -24,6 +24,14 @@
 #  include <stop_token>  // IWYU pragma: export
 #endif
 
+// This shouldn't be necessary, but some standard library implementations claim support
+// for jthread but don't actually provide std::stop_token.
+STDEXEC_NAMESPACE_STD_BEGIN
+  class stop_token;
+  template <class _Callback>
+  class stop_callback;
+STDEXEC_NAMESPACE_STD_END
+
 STDEXEC_P2300_NAMESPACE_BEGIN()
 
   template <template <class> class _Callback>
@@ -82,7 +90,7 @@ STDEXEC_P2300_NAMESPACE_BEGIN()
    private:
     struct __callback_type
     {
-      constexpr explicit __callback_type(never_stop_token, STDEXEC::__ignore) noexcept { }
+      constexpr explicit __callback_type(never_stop_token, STDEXEC::__ignore) noexcept {}
     };
    public:
     template <class>

test/CMakeLists.txt

@@ -65,6 +65,7 @@ set(stdexec_test_sources
     stdexec/detail/test_completion_signatures.cpp
     stdexec/detail/test_demangle.cpp
     stdexec/detail/test_utility.cpp
+    stdexec/detail/test_intrusive_mpsc_queue.cpp
     stdexec/schedulers/test_task_scheduler.cpp
     stdexec/schedulers/test_parallel_scheduler.cpp
     stdexec/queries/test_env.cpp
@@ -154,6 +155,10 @@ icm_add_build_failure_test(
     FOLDER test
 )
 
+if(STDEXEC_BUILD_RELACY_TESTS)
+    add_subdirectory(rrd)
+endif()
+
 # # Adding multiple tests with a glob
 # icm_glob_build_failure_tests(
 #     PATTERN *_fail*.cpp

test/rrd/CMakeLists.txt

@@ -0,0 +1,46 @@
+include(FetchContent)
+
+FetchContent_Declare(
+  relacy
+  GIT_REPOSITORY https://github.com/dvyukov/relacy
+  GIT_TAG master
+)
+
+FetchContent_Populate(relacy)
+
+add_library(relacy INTERFACE)
+target_include_directories(relacy INTERFACE
+  $<BUILD_INTERFACE:${relacy_SOURCE_DIR}/relacy/fakestd>
+  $<BUILD_INTERFACE:${relacy_SOURCE_DIR}>
+  $<INSTALL_INTERFACE:include>
+)
+
+function(add_relacy_test target_name)
+  add_executable(${target_name} ${target_name}.cpp)
+
+  target_link_libraries(${target_name} PRIVATE relacy)
+  # Disable TBB - Relacy simulates all threads with fibers anyway...
+  target_compile_definitions(${target_name} PRIVATE _GLIBCXX_USE_TBB_PAR_BACKEND=0)
+  target_include_directories(${target_name} PRIVATE ../../include)
+  target_include_directories(${target_name} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR})
+  set_target_properties(${target_name} PROPERTIES
+      CXX_STANDARD 20
+      CXX_STANDARD_REQUIRED ON
+      CXX_EXTENSIONS OFF)
+
+  add_test(NAME relacy-${target_name} COMMAND ${target_name})
+endfunction()
+
+set(relacy_tests
+  async_scope
+  intrusive_mpsc_queue
+  split
+  sync_wait
+)
+
+foreach(test ${relacy_tests})
+  add_relacy_test(${test})
+endforeach()
+
+# Target to build all relacy tests
+add_custom_target(relacy-tests DEPENDS ${relacy_tests})

test/rrd/Makefile

@@ -15,13 +15,20 @@ STDEXEC library could needs to use `x.fetch_add(1)` to be compatible with Relacy
 
 ## Instructions
 
-Run the following commands from within this directory (`./tests/rrd`).
+Configure and build stdexec following the build instructions in the top level
+[README.md](../../README.md). There are a couple relacy specific build and ctest
+targets, though they are part of the standard build and ctest and will be run
+automatically if cmake is configured with `-DSTDEXEC_BUILD_RELACY_TESTS=1`.
+`STDEXEC_BUILD_RELACY_TESTS` is set by default for GCC today.
+
+Run the following on a Linux machine with GCC as the toolchain.
 
 ```
-git clone -b STDEXEC https://github.com/dvyukov/relacy
-CXX=g++-11 make -j 4
-./build/split
-./build/async_scope
+mkdir build && cd build
+cmake ..
+make relacy-tests -j 4
+ctest -R relacy # Run all relacy tests
+./test/rrd/sync_wait # Run a specific relacy test directly
 ```
 
 ## Recommended use
@@ -35,8 +42,16 @@ out a more stable build on all environments/compilers, we should revisit this.
 ## Supported platforms
 
 The STDEXEC Relacy tests have been verified to build and run on
- * Linux based GCC+11 with libstdc++ (`x86_64`)
- * Mac with Apple Clang 15 with libc++ (`x86_64`)
+ * Linux based GCC+12-14 with libstdc++ (`x86_64`)
+ * Mac with Apple Clang 15 and 17 with libc++ (`x86_64`)
+
+## Caveat
+
+Relacy relies on a less than robust approach to implement its runtime: it replaces
+std:: names with its own versions, for example, std::atomic and std::mutex, as well
+as pthread_* APIs. As libstdc++/libc++ evolve, newer versions may not be compatible with
+Relacy. In these cases, changes to Relacy are needed to correctly intercept and replace
+std:: names.
 
-G++12 and newer are known to have issues that could be addressed with patches
-to Relacy.
+When the compilers and standard libraries release new versions, we will need to test the
+new versions can compile the stdexec Relacy tests before enabling the new compiler.