Fixing use after free

- Properly synchronizing `finish` with `run` with task counts
- Adding sync_wait relacy test for verification
- adapting atomic wrappers to account for missing std::atomic_ref in relacy

Author: sithhell

include/stdexec/__detail/__atomic.hpp

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

include/stdexec/__detail/__run_loop.hpp

@@ -27,7 +27,6 @@
 
 #include "__atomic.hpp"
 #include "stdexec/__detail/__config.hpp"
-#include <atomic>
 #include <cstddef>
 
 namespace stdexec {
@@ -56,16 +55,21 @@ namespace stdexec {
     STDEXEC_ATTRIBUTE(host, device) void finish() noexcept {
       // Increment our task count to avoid lifetime issues. This is preventing
       // a use-after-free issue if finish is called from a different thread.
-      __task_count_.fetch_add(1,    __std::memory_order_release);
+      // We increment the task counter by two to avoid the run loop to exit before
+      // we scheduled the noop task
+      __task_count_.fetch_add(2, __std::memory_order_release);
       if (!__finishing_.exchange(true, __std::memory_order_acq_rel)) {
         // push an empty work item to the queue to wake up the consuming thread
         // and let it finish.
         // The count will be decremented once the tasks executes.
         __queue_.push(&__noop_task);
+        // If the task got pushed, simply subtract one again, the other increment
+        // happens when the noop task got executed.
+        __task_count_.fetch_sub(1, __std::memory_order_release);
         return;
       }
-      // We are done finishing. Decrement the count, which signals final completion.
-      __task_count_.fetch_sub(1,    __std::memory_order_release);
+      // We are done finishing. Decrement the count by two, which signals final completion.
+      __task_count_.fetch_sub(2, __std::memory_order_release);
     }
 
     struct __task : __immovable {

test/rrd/Makefile

@@ -1,13 +1,13 @@
 # User-customizable variables:
 CXX ?= c++
 CXX_STD ?= c++20
-CXXFLAGS ?= -I relacy -I relacy/relacy/fakestd -O1 -std=$(CXX_STD) -I ../../include -I ../../test -g
+CXXFLAGS ?= -DSTDEXEC_RELACY -I relacy -I relacy/relacy/fakestd -O1 -std=$(CXX_STD) -I ../../include -I ../../test -g
 DEPFLAGS ?= -MD -MF $(@).d -MP -MT $(@)
 build_dir = build
 
 .SECONDARY:
 
-test_programs = split async_scope
+test_programs = split async_scope sync_wait
 
 test_exe_files = $(foreach name,$(test_programs),$(build_dir)/$(name))
 

test/rrd/sync_wait.cpp

@@ -0,0 +1,27 @@
+#include "../../relacy/relacy_std.hpp"
+
+#include <stdexec/execution.hpp>
+#include <exec/static_thread_pool.hpp>
+
+namespace ex = stdexec;
+
+struct sync_wait_bg_thread : rl::test_suite<sync_wait_bg_thread, 1> {
+  static size_t const dynamic_thread_count = 1;
+
+  void thread(unsigned) {
+    exec::static_thread_pool pool{1};
+    auto sender = ex::schedule(pool.get_scheduler()) | ex::then([] { return 42; });
+
+    auto [val] = ex::sync_wait(sender).value();
+    RL_ASSERT(val == 42);
+  }
+};
+
+auto main() -> int {
+  rl::test_params p;
+  p.iteration_count = 50000;
+  p.execution_depth_limit = 10000;
+  p.search_type = rl::random_scheduler_type;
+  rl::simulate<sync_wait_bg_thread>(p);
+  return 0;
+}