Integrate NVIDIA's S/R Bulk implementation into HPX

libs/core/algorithms/include/hpx/parallel/container_algorithms/for_each.hpp

@@ -542,53 +542,6 @@ namespace hpx::ranges {
                 HPX_FORWARD(ExPolicy, policy), hpx::util::begin(rng),
                 hpx::util::end(rng), HPX_MOVE(f), HPX_MOVE(proj));
         }
-
-#if defined(HPX_HAVE_STDEXEC)
-        // Sender algorithm support for stdexec integration
-        template <typename Sender, typename ExPolicy, typename F,
-            typename Proj = hpx::identity>
-        // clang-format off
-            requires (
-                hpx::execution::experimental::sender<Sender> &&
-                std::invocable<F,
-                    typename hpx::execution::experimental::value_types_of_t<
-                        Sender, hpx::execution::experimental::empty_env>::
-                        template apply<std::tuple>>
-            )
-        // clang-format on
-        friend auto tag_fallback_invoke(hpx::ranges::for_each_t,
-            Sender&& sender, ExPolicy&& policy, F&& f, Proj&& proj = Proj{})
-        {
-            return HPX_FORWARD(Sender, sender) |
-                hpx::execution::experimental::let_value(
-                    [policy = HPX_FORWARD(ExPolicy, policy),
-                        f = HPX_FORWARD(F, f),
-                        proj = HPX_FORWARD(Proj, proj)](auto&& rng) mutable {
-                        return hpx::execution::experimental::just(hpx::for_each(
-                            policy, HPX_FORWARD(decltype(rng), rng),
-                            HPX_MOVE(f), HPX_MOVE(proj)));
-                    });
-        }
-
-        // Partial algorithm support for stdexec senders
-        template <typename ExPolicy>
-        friend auto tag_fallback_invoke(
-            hpx::ranges::for_each_t, ExPolicy&& policy)
-        {
-            return [policy = HPX_FORWARD(ExPolicy, policy)](
-                       auto&& sender) mutable {
-                return HPX_FORWARD(decltype(sender), sender) |
-                    hpx::execution::experimental::let_value(
-                        [policy = HPX_MOVE(policy)](
-                            auto&& rng, auto&& f) mutable {
-                            return hpx::execution::experimental::just(
-                                hpx::for_each(policy,
-                                    HPX_FORWARD(decltype(rng), rng),
-                                    HPX_FORWARD(decltype(f), f)));
-                        });
-            };
-        }
-#endif
     } for_each{};
 
     ///////////////////////////////////////////////////////////////////////////

libs/core/execution_base/include/hpx/execution_base/completion_signatures.hpp

@@ -508,15 +508,15 @@ namespace hpx::execution::experimental {
     // The sender_of concept defines the requirements for a sender type that on
     // successful completion sends the specified set of value types.
     template <typename Sender, typename Signal,
-        typename Env = hpx::execution::experimental::env<>>
+        typename Env = hpx::execution::experimental::empty_env>
     struct is_sender_of
       : std::bool_constant<
             hpx::execution::experimental::sender_of<Sender, Signal, Env>>
     {
     };
 
     template <typename Sender, typename Signal,
-        typename Env = hpx::execution::experimental::env<>>
+        typename Env = hpx::execution::experimental::empty_env>
     inline constexpr bool is_sender_of_v =
         is_sender_of<Sender, Signal, Env>::value;
 
@@ -533,7 +533,7 @@ namespace hpx::execution::experimental {
     // 3. Otherwise, single-sender-value-type<S, E> is ill-formed.
     //
     template <typename Sender,
-        typename Env = hpx::execution::experimental::env<>>
+        typename Env = hpx::execution::experimental::empty_env>
     using single_sender_value_t =
         hpx::execution::experimental::stdexec_internal::__single_sender_value_t<
             Sender, Env>;
@@ -1008,9 +1008,8 @@ namespace hpx::execution::experimental {
             value_or_void_t<Value>, std::exception_ptr>;
 
         template <typename Promise>
-        using coroutine_env_t =
-            hpx::util::detected_or<hpx::execution::experimental::empty_env,
-                hpx::functional::tag_invoke_result_t, get_env_t, Promise>;
+        using coroutine_env_t = hpx::util::detected_or<exec_envs::empty_env,
+            hpx::functional::tag_invoke_result_t, get_env_t, Promise>;
 
         template <typename Value>
         struct receiver_base

libs/core/execution_base/include/hpx/execution_base/stdexec_forward.hpp

@@ -133,9 +133,6 @@ namespace hpx::execution::experimental {
     // Continue on
     using stdexec::continue_on;
     using stdexec::continue_on_t;
-    // Backward compatibility alias for continues_on
-    inline constexpr stdexec::continue_on_t continues_on{};
-    using continues_on_t = stdexec::continue_on_t;
 
     // Transfer just
     using stdexec::transfer_just;
@@ -322,12 +319,12 @@ namespace hpx::execution::experimental {
         }
 
         using stdexec::__connect_awaitable_t;
-    }    // namespace stdexec_internal
 
-    // Additional stdexec concepts and utilities needed for domain customization
-    using stdexec::__completes_on;
-    using stdexec::__starts_on;
-    using stdexec::sender_expr_for;
+        // Additional stdexec concepts and utilities needed for domain customization
+        using stdexec::__completes_on;
+        using stdexec::__starts_on;
+        using stdexec::sender_expr_for;
+    }    // namespace stdexec_internal
 }    // namespace hpx::execution::experimental
 
 // Leaving this as a placeholder

libs/core/executors/include/hpx/executors/thread_pool_scheduler.hpp

@@ -78,10 +78,10 @@ namespace hpx::execution::experimental {
     // Concept to match bulk sender types
     // Note: We keep bulk_t handling as pragmatic workaround for stdexec template issues
     template <typename Sender>
-    concept any_bulk_sender =
-        hpx::execution::experimental::sender_expr_for<Sender,
+    concept bulk_chunked_or_unchunked_sender =
+        hpx::execution::experimental::stdexec_internal::sender_expr_for<Sender,
             hpx::execution::experimental::bulk_chunked_t> ||
-        hpx::execution::experimental::sender_expr_for<Sender,
+        hpx::execution::experimental::stdexec_internal::sender_expr_for<Sender,
             hpx::execution::experimental::bulk_unchunked_t>;
 
     // Domain customization for stdexec bulk operations
@@ -95,10 +95,10 @@ namespace hpx::execution::experimental {
     {
         // Unified transform_sender for all bulk operations without environment
         // (completes_on pattern)
-        template <any_bulk_sender Sender>
+        template <bulk_chunked_or_unchunked_sender Sender>
         auto transform_sender(Sender&& sndr) const noexcept
         {
-            static_assert(hpx::execution::experimental::__completes_on<Sender,
+            static_assert(hpx::execution::experimental::stdexec_internal::__completes_on<Sender,
                               thread_pool_policy_scheduler<Policy>>,
                 "No thread_pool_policy_scheduler instance can be found in the "
                 "sender's "
@@ -115,7 +115,7 @@ namespace hpx::execution::experimental {
 
             auto iota_shape = std::views::iota(decltype(shape){0}, shape);
 
-            if constexpr (hpx::execution::experimental::sender_expr_for<Sender,
+            if constexpr (hpx::execution::experimental::stdexec_internal::sender_expr_for<Sender,
                               hpx::execution::experimental::bulk_unchunked_t>)
             {
                 // This should be launching one hpx thread for each index
@@ -131,7 +131,7 @@ namespace hpx::execution::experimental {
                     };
             }
             else if constexpr (
-                hpx::execution::experimental::sender_expr_for<Sender,
+                hpx::execution::experimental::stdexec_internal::sender_expr_for<Sender,
                     hpx::execution::experimental::bulk_chunked_t>)
             {
                 // This should be launching one hpx thread for each chunk
@@ -150,10 +150,10 @@ namespace hpx::execution::experimental {
 
         // Unified transform_sender for all bulk operations with environment
         // (starts_on pattern)
-        template <any_bulk_sender Sender, typename Env>
+        template <bulk_chunked_or_unchunked_sender Sender, typename Env>
         auto transform_sender(Sender&& sndr, const Env& env) const noexcept
         {
-            static_assert(hpx::execution::experimental::__starts_on<Sender,
+            static_assert(hpx::execution::experimental::stdexec_internal::__starts_on<Sender,
                               thread_pool_policy_scheduler<Policy>, Env>,
                 "No thread_pool_policy_scheduler instance can be found in the "
                 "receiver's "
@@ -167,7 +167,7 @@ namespace hpx::execution::experimental {
 
             auto iota_shape = std::views::iota(decltype(shape){0}, shape);
 
-            if constexpr (hpx::execution::experimental::sender_expr_for<Sender,
+            if constexpr (hpx::execution::experimental::stdexec_internal::sender_expr_for<Sender,
                               hpx::execution::experimental::bulk_unchunked_t>)
             {
                 return hpx::execution::experimental::detail::
@@ -182,7 +182,7 @@ namespace hpx::execution::experimental {
                     };
             }
             else if constexpr (
-                hpx::execution::experimental::sender_expr_for<Sender,
+                hpx::execution::experimental::stdexec_internal::sender_expr_for<Sender,
                     hpx::execution::experimental::bulk_chunked_t>)
             {
                 return hpx::execution::experimental::detail::
@@ -451,7 +451,6 @@ namespace hpx::execution::experimental {
             {
                 return {s.scheduler, HPX_FORWARD(Receiver, receiver)};
             }
-#if defined(HPX_HAVE_STDEXEC)
             struct env
             {
                 std::decay_t<Scheduler> const& sched;
@@ -470,12 +469,14 @@ namespace hpx::execution::experimental {
                     return e.sched;
                 }
 
+#if defined(HPX_HAVE_STDEXEC)
                 // Add domain query to sender environment
                 friend constexpr auto tag_invoke(
                     stdexec::get_domain_t, env const& e) noexcept
                 {
                     return stdexec::get_domain(e.sched);
                 }
+#endif
             };
 
             friend constexpr auto tag_invoke(
@@ -498,35 +499,6 @@ namespace hpx::execution::experimental {
             {
                 return s.scheduler;
             }
-#else
-            struct env_no_stdexec
-            {
-                thread_pool_policy_scheduler<Policy> sched;
-
-                friend constexpr auto tag_invoke(
-                    hpx::execution::experimental::get_completion_scheduler_t<
-                        hpx::execution::experimental::set_value_t>,
-                    env_no_stdexec const& e) noexcept
-                {
-                    return e.sched;
-                }
-
-                friend constexpr auto tag_invoke(
-                    hpx::execution::experimental::get_completion_scheduler_t<
-                        hpx::execution::experimental::set_stopped_t>,
-                    env_no_stdexec const& e) noexcept
-                {
-                    return e.sched;
-                }
-            };
-
-            friend constexpr auto tag_invoke(
-                hpx::execution::experimental::get_env_t,
-                sender const& s) noexcept
-            {
-                return env_no_stdexec{s.scheduler};
-            };
-#endif
         };
 
 #if defined(HPX_HAVE_STDEXEC)

libs/core/executors/include/hpx/executors/thread_pool_scheduler_bulk.hpp

@@ -82,7 +82,7 @@ namespace hpx::execution::experimental::detail {
             chunk_size *= 2;
         }
         return static_cast<std::uint32_t>(
-            (std::max) (chunk_size, std::uint64_t(1)));
+            (std::max)(chunk_size, std::uint64_t(1)));
     }
 
     // For bulk_unchunked: f(index, ...)
@@ -99,18 +99,8 @@ namespace hpx::execution::experimental::detail {
     constexpr void bulk_scheduler_invoke_helper_chunked(
         hpx::util::index_pack<Is...>, F&& f, Start&& start, End&& end, Ts& ts)
     {
-        if constexpr (sizeof...(Is) == 0)
-        {
-            // No additional arguments from sender values
-            HPX_INVOKE(HPX_FORWARD(F, f), HPX_FORWARD(Start, start),
-                HPX_FORWARD(End, end));
-        }
-        else
-        {
-            // Additional arguments from sender values
-            HPX_INVOKE(HPX_FORWARD(F, f), HPX_FORWARD(Start, start),
-                HPX_FORWARD(End, end), hpx::get<Is>(ts)...);
-        }
+        HPX_INVOKE(HPX_FORWARD(F, f), HPX_FORWARD(Start, start),
+            HPX_FORWARD(End, end), hpx::get<Is>(ts)...);
     }
 
     inline hpx::threads::mask_type full_mask(
@@ -186,10 +176,8 @@ namespace hpx::execution::experimental::detail {
             auto const i_begin =
                 static_cast<std::size_t>(index) * task_f->chunk_size;
             auto const i_end =
-                (std::min) (i_begin + task_f->chunk_size, task_f->size);
+                (std::min)(i_begin + task_f->chunk_size, task_f->size);
 
-            auto it =
-                std::ranges::next(hpx::util::begin(op_state->shape), i_begin);
             if constexpr (OperationState::is_chunked)
             {
                 // bulk_chunked: f(start, end, values...)
@@ -199,6 +187,8 @@ namespace hpx::execution::experimental::detail {
             else
             {
                 // bulk_unchunked: f(index, values...) for each element
+                auto it = std::ranges::next(
+                    hpx::util::begin(op_state->shape), i_begin);
                 for (std::uint32_t i = i_begin; i != i_end; (void) ++i)
                 {
                     bulk_scheduler_invoke_helper(
@@ -428,8 +418,8 @@ namespace hpx::execution::experimental::detail {
             auto& queue = op_state->queues[worker_thread].data_;
             auto const num_steps = size / num_threads + 1;
             auto const part_begin = worker_thread;
-            auto part_end = (std::min) (size + num_threads - 1,
-                part_begin + num_steps * num_threads);
+            auto part_end = (std::min)(
+                size + num_threads - 1, part_begin + num_steps * num_threads);
             auto const remainder = (part_end - part_begin) % num_threads;
             if (remainder != 0)
             {
@@ -502,7 +492,6 @@ namespace hpx::execution::experimental::detail {
         {
             // Don't spawn tasks if there is no work to be done
             auto const size =
-                // static_cast<std::uint32_t>(op_state->shape);
                 static_cast<std::uint32_t>(hpx::util::size(op_state->shape));
             if (size == 0)
             {
@@ -644,7 +633,11 @@ namespace hpx::execution::experimental::detail {
         }
 
         // clang-format off
-        template <typename... Ts>
+        template <typename... Ts,
+            HPX_CONCEPT_REQUIRES_(
+                hpx::is_invocable_v<F, range_value_type,
+                    std::add_lvalue_reference_t<Ts>...>
+            )>
         // clang-format on
         friend void tag_invoke(hpx::execution::experimental::set_value_t,
             bulk_receiver&& r, Ts&&... ts) noexcept
@@ -906,6 +899,60 @@ namespace hpx::execution::experimental::detail {
     };
 }    // namespace hpx::execution::experimental::detail
 
+#if !defined(HPX_HAVE_STDEXEC)
+namespace hpx::execution::experimental {
+    // clang-format off
+    template <typename Policy, typename Sender, typename Shape, typename F,
+        HPX_CONCEPT_REQUIRES_(
+            !std::is_integral_v<Shape>
+        )>
+    // clang-format on
+    constexpr auto tag_invoke(bulk_t,
+        thread_pool_policy_scheduler<Policy> scheduler, Sender&& sender,
+        Shape const& shape, F&& f)
+    {
+        if constexpr (std::is_same_v<Policy, launch::sync_policy>)
+        {
+            // fall back to non-bulk scheduling if sync execution was requested
+            return detail::bulk_sender<Sender, Shape, F>{
+                HPX_FORWARD(Sender, sender), shape, HPX_FORWARD(F, f)};
+        }
+        else
+        {
+            return detail::thread_pool_bulk_sender<Policy, Sender, Shape, F>{
+                HPX_MOVE(scheduler), HPX_FORWARD(Sender, sender), shape,
+                HPX_FORWARD(F, f)};
+        }
+    }
+
+    // clang-format off
+    template <typename Policy, typename Sender, typename Count, typename F,
+        HPX_CONCEPT_REQUIRES_(
+            std::is_integral_v<Count>
+        )>
+    // clang-format on
+    constexpr decltype(auto) tag_invoke(bulk_t,
+        thread_pool_policy_scheduler<Policy> scheduler, Sender&& sender,
+        Count const& count, F&& f)
+    {
+        if constexpr (std::is_same_v<Policy, launch::sync_policy>)
+        {
+            // fall back to non-bulk scheduling if sync execution was requested
+            return detail::bulk_sender<Sender, hpx::util::counting_shape<Count>,
+                F>{HPX_FORWARD(Sender, sender),
+                hpx::util::counting_shape(count), HPX_FORWARD(F, f)};
+        }
+        else
+        {
+            return detail::thread_pool_bulk_sender<Policy, Sender,
+                hpx::util::counting_shape<Count>, F>{HPX_MOVE(scheduler),
+                HPX_FORWARD(Sender, sender), hpx::util::counting_shape(count),
+                HPX_FORWARD(F, f)};
+        }
+    }
+}    // namespace hpx::execution::experimental
+#endif
+
 // Note: With stdexec integration, bulk operations are now customized
 // through the domain system in thread_pool_scheduler.hpp rather than
 // direct tag_invoke customizations. The thread_pool_domain<Policy>