Check parallel scheduler macro value instead of definition

include/beman/execution/detail/common.hpp

@@ -17,6 +17,10 @@
 #define BEMAN_SPECIALIZE_EXPORT template <>
 #endif
 
+#ifndef BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND
+#define BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND 1
+#endif
+
 #define BEMAN_EXECUTION_TRY_EVAL(rcvr, expr)                                                    \
     do {                                                                                        \
         try {                                                                                   \

include/beman/execution/detail/default_parallel_scheduler_backend.hpp

@@ -3,7 +3,7 @@
 
 #ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_DEFAULT_PARALLEL_SCHEDULER_BACKEND
 #define INCLUDED_BEMAN_EXECUTION_DETAIL_DEFAULT_PARALLEL_SCHEDULER_BACKEND
-#ifdef BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND
+#if BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND
 
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_IMPORT_STD

include/beman/execution/detail/psched_bulk_sender.hpp

@@ -4,6 +4,7 @@
 #ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_PSCHED_BULK_SENDER
 #define INCLUDED_BEMAN_EXECUTION_DETAIL_PSCHED_BULK_SENDER
 
+#include <cassert>
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_IMPORT_STD
 import std;
@@ -75,7 +76,7 @@ import beman.execution.detail.value_types_of_t;
 // ----------------------------------------------------------------------------
 
 namespace beman::execution::detail {
-inline constexpr ::std::size_t psched_storage_alignment = alignof(void*);
+inline constexpr ::std::size_t psched_storage_alignment = alignof(::std::max_align_t);
 inline constexpr ::std::size_t psched_storage_size      = 6uz * sizeof(void*);
 template <bool IsChunked, typename Policy, typename Shape, typename Fn, typename Child>
 struct psched_bulk_sender {
@@ -101,12 +102,16 @@ struct psched_bulk_sender {
         auto get_env() const noexcept { return ::beman::execution::get_env(rcvr); }
 
         auto execute(::std::size_t begin, ::std::size_t end) noexcept -> void final {
+            assert(begin < end);
+            assert(IsChunked || !is_parallel_policy || end - begin == 1uz);
             const Shape first   = is_parallel_policy ? static_cast<Shape>(begin) : Shape(0);
             const Shape last    = is_parallel_policy ? static_cast<Shape>(end) : shape;
             const auto  call_fn = [=, this]<typename... Args>(const Args&... args) {
                 if constexpr (IsChunked) {
+                    static_assert(::std::invocable<Fn, Shape, Shape, const Args&...>);
                     fn(first, last, args...);
                 } else {
+                    static_assert(::std::invocable<Fn, Shape, const Args&...>);
                     for (Shape i = first; i < last; ++i) {
                         fn(i, args...);
                     }

include/beman/execution/detail/task.hpp

@@ -4,11 +4,11 @@
 #ifndef INCLUDED_BEMAN_EXECUTION_DETAIL_TASK
 #define INCLUDED_BEMAN_EXECUTION_DETAIL_TASK
 
+#include <cassert>
 #include <beman/execution/detail/common.hpp>
 #ifdef BEMAN_HAS_IMPORT_STD
 import std;
 #else
-#include <cassert>
 #include <concepts>
 #include <coroutine>
 #include <exception>

include/beman/execution/detail/task_scheduler.hpp

@@ -9,7 +9,6 @@
 import std;
 #else
 #include <algorithm>
-#include <cassert>
 #include <concepts>
 #include <exception>
 #include <memory>

include/beman/execution/detail/thread_pool_backend.hpp

@@ -92,11 +92,12 @@ class thread_pool_backend_base
 
     struct batched_bulk_task : task_base {
         struct cookie_type {
-            cookie_type(batched_bulk_task* head, ::std::size_t chunk_count) noexcept
-                : head(head), chunk_count(chunk_count), ref_count(chunk_count) {}
-            batched_bulk_task*           head;
-            ::std::size_t                chunk_count;
-            ::std::atomic<::std::size_t> ref_count;
+            cookie_type(batched_bulk_task* head, ::std::size_t chunk_count, bool one_by_one) noexcept
+                : head(head), chunk_count(chunk_count), one_by_one(one_by_one), ref_count(chunk_count) {}
+            batched_bulk_task* head;
+            ::std::size_t      chunk_count;
+            bool               one_by_one;
+            ::std::size_t      ref_count;
         };
 
         batched_bulk_task(cookie_type*                                                                  cookie,
@@ -106,8 +107,16 @@ class thread_pool_backend_base
             : cookie(cookie), proxy(proxy), i(i), j(j) {}
 
         auto exec() noexcept -> void override {
-            proxy.execute(i, j);
-            if (cookie->ref_count.fetch_sub(1uz, ::std::memory_order_acq_rel) == 1uz) {
+            if (cookie->one_by_one) {
+                for (::std::size_t k = i; k < j; ++k) {
+                    proxy.execute(k, k + 1uz);
+                }
+            } else {
+                proxy.execute(i, j);
+            }
+
+            ::std::atomic_ref<::std::size_t> ref_count{cookie->ref_count};
+            if (ref_count.fetch_sub(1uz, ::std::memory_order_acq_rel) == 1uz) {
                 auto       head        = cookie->head;
                 const auto chunk_count = cookie->chunk_count;
                 auto&      proxy_ref   = proxy;
@@ -158,14 +167,13 @@ class thread_pool_backend_base
     auto schedule_bulk_chunked(::std::size_t                                                                 shape,
                                ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy,
                                ::std::span<::std::byte> storage) noexcept -> void override {
-        const ::std::size_t chunk_length = (shape + num_threads() - 1uz) / num_threads();
-        schedule_bulk(shape, chunk_length, proxy, storage);
+        schedule_bulk(shape, false, proxy, storage);
     }
 
     auto schedule_bulk_unchunked(::std::size_t                                                                 shape,
                                  ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy,
                                  ::std::span<::std::byte> storage) noexcept -> void override {
-        schedule_bulk_chunked(shape, proxy, storage);
+        schedule_bulk(shape, true, proxy, storage);
     }
 
   protected:
@@ -175,15 +183,16 @@ class thread_pool_backend_base
     }
 
     auto schedule_bulk(::std::size_t                                                                 shape,
-                       ::std::size_t                                                                 chunk_length,
+                       bool                                                                          one_by_one,
                        ::beman::execution::parallel_scheduler_replacement::bulk_item_receiver_proxy& proxy,
                        ::std::span<::std::byte> storage) noexcept -> void {
         if (shape == 0uz) {
             schedule(proxy, storage);
             return;
         }
 
-        const ::std::size_t chunk_count = (shape + chunk_length - 1uz) / chunk_length;
+        const ::std::size_t chunk_length = (shape + num_threads() - 1uz) / num_threads();
+        const ::std::size_t chunk_count  = (shape + chunk_length - 1uz) / chunk_length;
         try {
             if (chunk_count == 1uz) {
                 push_back(::std::construct_at(reinterpret_cast<single_bulk_task*>(storage.data()), proxy, shape));
@@ -192,7 +201,7 @@ class thread_pool_backend_base
                     chunk_count * sizeof(batched_bulk_task), ::std::align_val_t{alignof(batched_bulk_task)}));
                 // NOLINTBEGIN(*-reinterpret-cast, *-pointer-arithmetic-on-polymorphic-object, *-ctr56-cpp)
                 auto cookie = ::std::construct_at(
-                    reinterpret_cast<batched_bulk_task::cookie_type*>(storage.data()), head, chunk_count);
+                    reinterpret_cast<batched_bulk_task::cookie_type*>(storage.data()), head, chunk_count, one_by_one);
 
                 batched_bulk_task* prev = nullptr;
                 for (::std::size_t i = 0; i < chunk_count; ++i) {

src/beman/execution/CMakeLists.txt

@@ -246,12 +246,11 @@ target_sources(
                 ${PROJECT_SOURCE_DIR}/include/beman/execution/detail/write_env.hpp
 )
 
-if(BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND)
-    target_compile_definitions(
-        ${BEMAN_EXECUTION_TARGET_NAME}
-        INTERFACE BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND
-    )
-endif()
+target_compile_definitions(
+    ${BEMAN_EXECUTION_TARGET_NAME}
+    INTERFACE
+        BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND=$<BOOL:${BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND}>
+)
 
 if(BEMAN_USE_MODULES)
     target_sources(

tests/beman/execution/exec-parallel-scheduler.test.cpp

@@ -100,9 +100,23 @@ auto test_parallel_scheduler_schedule() -> void {
         ASSERT(i == 114514);
     }
     {
-        test_std::sync_wait(test_std::schedule(sch) | test_std::bulk(test_std::par, 0uz, [](std::size_t) noexcept {}));
+        bool invoked = false;
         test_std::sync_wait(test_std::schedule(sch) |
-                            test_std::bulk(test_std::unseq, 0uz, [](std::size_t) noexcept {}));
+                            test_std::bulk(test_std::par, 0uz, [&invoked](std::size_t) noexcept { invoked = true; }));
+        ASSERT(not invoked);
+        test_std::sync_wait(
+            test_std::schedule(sch) |
+            test_std::bulk(test_std::unseq, 0uz, [&invoked](std::size_t) noexcept { invoked = true; }));
+        ASSERT(not invoked);
+        test_std::sync_wait(test_std::schedule(sch) |
+                            test_std::bulk_chunked(test_std::par, 0uz, [&invoked](std::size_t, std::size_t) noexcept {
+                                invoked = true;
+                            }));
+        ASSERT(not invoked);
+        test_std::sync_wait(
+            test_std::schedule(sch) |
+            test_std::bulk_unchunked(test_std::par, 0uz, [&invoked](std::size_t) noexcept { invoked = true; }));
+        ASSERT(not invoked);
     }
     {
         for (auto size : {1uz, 4uz, 8uz, 16uz, 32uz}) {
@@ -123,11 +137,31 @@ auto test_parallel_scheduler_schedule() -> void {
                 ASSERT(vec[i] == 2 * static_cast<int>(i) + 1);
             }
         }
+
+        for (auto size : {1uz, 4uz, 8uz, 16uz, 32uz}) {
+            std::vector<int> vec(size);
+            std::iota(vec.begin(), vec.end(), 0);
+
+            test_std::sync_wait(test_std::schedule(sch) |
+                                test_std::bulk_unchunked(test_std::par, vec.size(), [&vec](std::size_t i) noexcept {
+                                    vec[i] = 2 * vec[i];
+                                }));
+            for (std::size_t i = 0; i < vec.size(); ++i) {
+                ASSERT(vec[i] == 2 * static_cast<int>(i));
+            }
+
+            test_std::sync_wait(
+                test_std::schedule(sch) |
+                test_std::bulk_unchunked(test_std::seq, vec.size(), [&vec](std::size_t i) noexcept { ++vec[i]; }));
+            for (std::size_t i = 0; i < vec.size(); ++i) {
+                ASSERT(vec[i] == 2 * static_cast<int>(i) + 1);
+            }
+        }
     }
 }
 } // namespace
 
-#ifndef BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND
+#if !BEMAN_EXECUTION_WITH_DEFAULT_PARALLEL_SCHEDULER_BACKEND
 namespace beman::execution::parallel_scheduler_replacement {
 auto query_parallel_scheduler_backend() -> std::shared_ptr<parallel_scheduler_backend> {
     static auto backend = std::make_shared<::test_detail::thread_pool_backend>();