RFC: run loop finish forward progress

.github/workflows/ci.cpu.yml

@@ -142,11 +142,46 @@ jobs:
           path: /tmp/sccache*.log
           compression-level: 9
 
+  valgrind-run-loop:
+    runs-on: ubuntu-latest
+    name: CPU (gcc 14, Valgrind run_loop)
+    container:
+      options: -u root
+      image: rapidsai/devcontainers:26.08-cpp-gcc14-cuda12.9
+    steps:
+      - name: Checkout stdexec
+        uses: actions/checkout@v4
+        with:
+          persist-credentials: false
+
+      - name: Install Valgrind
+        run: |
+          apt-get update
+          DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends valgrind
+
+      - name: Build run_loop reproducer
+        run: |
+          cmake -S . -B build -GNinja \
+            -DCMAKE_BUILD_TYPE=Debug \
+            -DCMAKE_CXX_STANDARD=20 \
+            -DCMAKE_CXX_EXTENSIONS=OFF \
+            -DSTDEXEC_BUILD_TESTS=ON
+          cmake --build build --target test.run_loop_finish_repro -v
+
+      - name: Run run_loop reproducer under Valgrind
+        run: |
+          timeout 60s valgrind \
+            --fair-sched=no \
+            --error-exitcode=1 \
+            --quiet \
+            ./build/test/test.run_loop_finish_repro
+
   ci-cpu:
     runs-on: ubuntu-latest
     name: CI (CPU)
     needs:
       - build-cpu
+      - valgrind-run-loop
     steps:
       - run: echo "CI (CPU) success"
 

include/stdexec/__detail/__run_loop.hpp

@@ -69,8 +69,21 @@ namespace STDEXEC
         }
         // drain the queue, taking care to execute any tasks that get added while
         // executing the remaining tasks (also wait for other tasks that might still be in flight):
-        while (__execute_all() || __task_count_.load(__std::memory_order_acquire) > 0)
-          ;
+        while (true)
+        {
+          if (__execute_all())
+          {
+            continue;
+          }
+
+          if (__task_count_.load(__std::memory_order_acquire) == 0)
+          {
+            break;
+          }
+
+          // Another thread still has work in flight. Let it make progress.
+          std::this_thread::yield();
+        }
       }
 
       STDEXEC_ATTRIBUTE(host, device)

test/CMakeLists.txt

@@ -120,6 +120,15 @@ target_compile_options(
     $<$<CXX_COMPILER_ID:Clang>:-Wno-gnu-line-marker>)
 target_link_libraries(common_test_settings
                       INTERFACE $<TARGET_NAME_IF_EXISTS:TBB::tbb>)
+
+add_executable(test.run_loop_finish_repro EXCLUDE_FROM_ALL
+               run_loop_finish_repro.cpp)
+add_compile_diagnostics(test.run_loop_finish_repro)
+target_link_libraries(
+  test.run_loop_finish_repro
+  PUBLIC STDEXEC::stdexec stdexec_executable_flags
+  PRIVATE common_test_settings)
+
 # target_compile_definitions( common_test_settings INTERFACE
 # $<$<NOT:$<AND:$<CXX_COMPILER_ID:NVHPC>,$<COMPILE_LANGUAGE:CXX>>>:STDEXEC_ENABLE_EXTRA_TYPE_CHECKING>)
 

test/run_loop_finish_repro.cpp

@@ -0,0 +1,31 @@
+#include <stdexec/execution.hpp>
+
+#include <thread>
+
+namespace ex = STDEXEC_NAMESPACE;
+
+void run_once()
+{
+  ex::run_loop loop;
+
+  std::thread worker([&loop] { loop.run(); });
+
+  // Ensure run() is actively servicing the loop before initiating
+  // shutdown from this thread.
+  if (!ex::sync_wait(ex::schedule(loop.get_scheduler()) | ex::then([]() {})))
+  {
+    loop.finish();
+    worker.join();
+  }
+
+  loop.finish();
+  worker.join();
+}
+
+int main()
+{
+  for (int i = 0; i < 1000; ++i)
+  {
+    run_once();
+  }
+}