Merge remote-tracking branch 'origin/main' into P3481R2_bulk_issues2

Author: ericniebler

CMakeLists.txt

@@ -416,7 +416,7 @@ set(STDEXEC_ASIO_IMPLEMENTATION "boost" CACHE STRING "boost")
 set_property(CACHE STDEXEC_ASIO_IMPLEMENTATION PROPERTY STRINGS boost standalone)
 
 if(STDEXEC_ENABLE_ASIO)
-    set(STDEXEC_ASIO_USES_ASIO FALSE)
+    set(STDEXEC_ASIO_USES_BOOST FALSE)
     set(STDEXEC_ASIO_USES_STANDALONE FALSE)
 
     include(rapids-find)

cmake/clangd_compile_info.cmake

@@ -21,7 +21,7 @@ set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
 # Symlink the compile command output to the source dir, where clangd will find it.
 set(compile_commands_file "${CMAKE_BINARY_DIR}/compile_commands.json")
 set(compile_commands_link "${CMAKE_SOURCE_DIR}/compile_commands.json")
-message(STATUS "Creating symlink from ${compile_commands_link} to ${compile_commands_file}...")
+message(STATUS "Creating symlink from \"${compile_commands_file}\" to \"${compile_commands_link}\"...")
 stdexec_execute_non_fatal_process(COMMAND
   "${CMAKE_COMMAND}" -E rm -f "${compile_commands_link}")
 stdexec_execute_non_fatal_process(COMMAND

cmake/import_standalone_asio.cmake

@@ -11,7 +11,7 @@ function(import_standalone_asio)
     set(options "")
     set(args TAG VERSION)
     set(multi_args "")
-    cmake_parse_arguments(IMPORT_STANDALONE_ASIO "${options}" "${args}" "${multi_args}")
+    cmake_parse_arguments(IMPORT_STANDALONE_ASIO "${options}" "${args}" "${multi_args}" ${ARGN})
 
     CPMAddPackage("gh:chriskohlhoff/asio#${IMPORT_STANDALONE_ASIO_TAG}@${IMPORT_STANDALONE_ASIO_VERSION}")
 

include/exec/__detail/__system_context_default_impl.hpp

@@ -16,10 +16,11 @@
 #pragma once
 
 #include "__system_context_replaceability_api.hpp"
-#include "stdexec/execution.hpp"
-#include "exec/static_thread_pool.hpp"
+
+#include "../../stdexec/execution.hpp"
+#include "../static_thread_pool.hpp"
 #if STDEXEC_ENABLE_LIBDISPATCH
-#  include "exec/libdispatch_queue.hpp"
+#  include "../libdispatch_queue.hpp" // IWYU pragma: keep
 #endif
 
 #include <atomic>

include/exec/__detail/__system_context_replaceability_api.hpp

@@ -17,7 +17,7 @@
 #ifndef STDEXEC_SYSTEM_CONTEXT_REPLACEABILITY_API_H
 #define STDEXEC_SYSTEM_CONTEXT_REPLACEABILITY_API_H
 
-#include "stdexec/__detail/__execution_fwd.hpp"
+#include "../../stdexec/__detail/__execution_fwd.hpp"
 
 #include <cstdint>
 #include <cstddef>

include/exec/libdispatch_queue.hpp

@@ -27,7 +27,7 @@
 #    define __has_extension(x) false
 #  endif
 
-#  include "stdexec/execution.hpp"
+#  include "../stdexec/execution.hpp"
 #  include <dispatch/dispatch.h>
 
 namespace exec {

include/exec/start_now.hpp

@@ -15,13 +15,13 @@
  */
 #pragma once
 
-#include "stdexec/__detail/__execution_fwd.hpp"
+#include "../stdexec/__detail/__execution_fwd.hpp"
 
-#include "stdexec/__detail/__concepts.hpp"
-#include "stdexec/__detail/__env.hpp"
-#include "stdexec/__detail/__receivers.hpp"
-#include "stdexec/__detail/__senders.hpp"
-#include "stdexec/__detail/__meta.hpp"
+#include "../stdexec/__detail/__concepts.hpp"
+#include "../stdexec/__detail/__env.hpp"
+#include "../stdexec/__detail/__receivers.hpp"
+#include "../stdexec/__detail/__senders.hpp"
+#include "../stdexec/__detail/__meta.hpp"
 
 #include "async_scope.hpp"
 

include/exec/static_thread_pool.hpp

@@ -617,6 +617,8 @@ namespace exec {
         void notify_one_sleeping();
         void set_stealing();
         void clear_stealing();
+        void set_sleeping();
+        void clear_sleeping();
 
         bwos::lifo_queue<task_base*, numa_allocator<task_base*>> local_queue_;
         __intrusive_queue<&task_base::next> pending_queue_{};
@@ -633,7 +635,7 @@ namespace exec {
       void run(std::uint32_t index) noexcept;
       void join() noexcept;
 
-      alignas(64) std::atomic<std::uint32_t> numThiefs_{};
+      alignas(64) std::atomic<std::uint32_t> numActive_{};
       alignas(64) remote_queue_list remotes_;
       std::uint32_t threadCount_;
       std::uint32_t maxSteals_{threadCount_ + 1};
@@ -703,6 +705,7 @@ namespace exec {
       threads_.reserve(threadCount);
 
       try {
+        numActive_.store(threadCount << 16u, std::memory_order_relaxed);
         for (std::uint32_t i = 0; i < threadCount; ++i) {
           threads_.emplace_back([this, i] { run(i); });
         }
@@ -725,9 +728,9 @@ namespace exec {
     }
 
     inline void static_thread_pool_::run(std::uint32_t threadIndex) noexcept {
+      STDEXEC_ASSERT(threadIndex < threadCount_);
       // NOLINTNEXTLINE(bugprone-unused-return-value)
       numa_.bind_to_node(threadStates_[threadIndex]->numa_node());
-      STDEXEC_ASSERT(threadIndex < threadCount_);
       while (true) {
         // Make a blocking call to de-queue a task if we don't already have one.
         auto [task, queueIndex] = threadStates_[threadIndex]->pop();
@@ -946,12 +949,31 @@ namespace exec {
       pending_queue_.prepend(std::move(tasks));
     }
 
+    inline void static_thread_pool_::thread_state::set_sleeping() {
+      pool_->numActive_.fetch_sub(1u << 16u, std::memory_order_relaxed);
+    }
+
+    // wakeup a worker thread and maintain the invariant that we always one active thief as long as a potential victim is awake
+    inline void static_thread_pool_::thread_state::clear_sleeping() {
+      const std::uint32_t numActive =
+        pool_->numActive_.fetch_add(1u << 16u, std::memory_order_relaxed);
+      if (numActive == 0) {
+        notify_one_sleeping();
+      }
+    }
+
     inline void static_thread_pool_::thread_state::set_stealing() {
-      pool_->numThiefs_.fetch_add(1, std::memory_order_relaxed);
+      const std::uint32_t diff = 1u - (1u << 16u);
+      pool_->numActive_.fetch_add(diff, std::memory_order_relaxed);
     }
 
+    // put a thief to sleep but maintain the invariant that we always have one active thief as long as a potential victim is awake
     inline void static_thread_pool_::thread_state::clear_stealing() {
-      if (pool_->numThiefs_.fetch_sub(1, std::memory_order_relaxed) == 1) {
+      constexpr std::uint32_t diff = 1 - (1u << 16u);
+      const std::uint32_t numActive = pool_->numActive_.fetch_sub(diff, std::memory_order_relaxed);
+      const std::uint32_t numVictims = numActive >> 16u;
+      const std::uint32_t numThiefs = numActive & 0xffffu;
+      if (numThiefs == 1 && numVictims != 0) {
         notify_one_sleeping();
       }
     }
@@ -1003,9 +1025,14 @@ namespace exec {
           if (result.task) {
             return result;
           }
+          set_sleeping();
           cv_.wait(lock);
+          lock.unlock();
+          clear_sleeping();
+        }
+        if (lock.owns_lock()) {
+          lock.unlock();
         }
-        lock.unlock();
         state_.store(state::running, std::memory_order_relaxed);
         result = try_pop();
       }

include/exec/system_context.hpp

@@ -17,7 +17,7 @@
 
 #include <utility>
 
-#include "stdexec/execution.hpp"
+#include "../stdexec/execution.hpp"
 #include "__detail/__system_context_replaceability_api.hpp"
 
 #ifndef STDEXEC_SYSTEM_CONTEXT_SCHEDULE_OP_SIZE

include/execpools/asio/asio_thread_pool.hpp

@@ -15,7 +15,8 @@ namespace execpools {
    public:
     asio_thread_pool()
       : pool_()
-      , executor_(pool_.executor()) = default;
+      , executor_(pool_.executor()) {
+    }
 
     explicit asio_thread_pool(uint32_t num_threads)
       : pool_(num_threads)