Integrate NVIDIA's S/R Bulk implementation into HPX

libs/core/execution/include/hpx/execution/algorithms/bulk.hpp

@@ -11,300 +11,6 @@
 
 #if defined(HPX_HAVE_STDEXEC)
 #include <hpx/execution_base/stdexec_forward.hpp>
-#endif
-
-#include <hpx/concepts/concepts.hpp>
-#include <hpx/datastructures/tuple.hpp>
-#include <hpx/datastructures/variant.hpp>
-#include <hpx/errors/try_catch_exception_ptr.hpp>
-#include <hpx/execution/algorithms/detail/partial_algorithm.hpp>
-#include <hpx/execution/algorithms/then.hpp>
-#include <hpx/execution_base/completion_scheduler.hpp>
-#include <hpx/execution_base/completion_signatures.hpp>
-#include <hpx/execution_base/receiver.hpp>
-#include <hpx/execution_base/sender.hpp>
-#include <hpx/functional/detail/tag_priority_invoke.hpp>
-#include <hpx/functional/invoke_result.hpp>
-#include <hpx/iterator_support/counting_shape.hpp>
-#include <hpx/type_support/pack.hpp>
-
-#include <exception>
-#include <iterator>
-#include <type_traits>
-#include <utility>
-
-namespace hpx::execution::experimental {
-
-    ///////////////////////////////////////////////////////////////////////////
-    namespace detail {
-
-        template <typename Sender, typename Shape, typename F>
-        struct bulk_sender
-        {
-            HPX_NO_UNIQUE_ADDRESS std::decay_t<Sender> sender;
-            HPX_NO_UNIQUE_ADDRESS std::decay_t<Shape> shape;
-            HPX_NO_UNIQUE_ADDRESS std::decay_t<F> f;
-
-#if defined(HPX_HAVE_STDEXEC)
-            using sender_concept = hpx::execution::experimental::sender_t;
-
-            template <typename... Args>
-            using default_set_value =
-                hpx::execution::experimental::completion_signatures<
-                    hpx::execution::experimental::set_value_t(Args...)>;
-
-            template <typename Arg>
-            using default_set_error =
-                hpx::execution::experimental::completion_signatures<
-                    hpx::execution::experimental::set_error_t(Arg)>;
-
-            using disable_set_stopped =
-                hpx::execution::experimental::completion_signatures<>;
-
-            // clang-format off
-            template <typename Env>
-            friend auto tag_invoke(get_completion_signatures_t,
-                bulk_sender const&, Env) noexcept -> hpx::execution::
-                experimental::transform_completion_signatures_of<Sender, Env,
-                    hpx::execution::experimental::completion_signatures<
-                        hpx::execution::experimental::set_error_t(
-                            std::exception_ptr)>,
-                    default_set_value, default_set_error, disable_set_stopped>;
-            // clang-format on
-
-            friend constexpr auto tag_invoke(
-                hpx::execution::experimental::get_env_t,
-                bulk_sender const& s) noexcept
-            {
-                return hpx::execution::experimental::get_env(s.sender);
-            }
 #else
-            using is_sender = void;
-
-            template <typename Env>
-            struct generate_completion_signatures
-            {
-                template <template <typename...> typename Tuple,
-                    template <typename...> typename Variant>
-                using value_types =
-                    value_types_of_t<Sender, Env, Tuple, Variant>;
-
-                template <template <typename...> typename Variant>
-                using error_types = hpx::util::detail::unique_concat_t<
-                    error_types_of_t<Sender, Env, Variant>,
-                    Variant<std::exception_ptr>>;
-
-                static constexpr bool sends_stopped = false;
-            };
-
-            // clang-format off
-            template <typename Env>
-            friend auto tag_invoke(
-                get_completion_signatures_t, bulk_sender const&, Env) noexcept
-                -> generate_completion_signatures<Env>;
-            // clang-format on
-
-            // clang-format off
-            template <typename CPO,
-                HPX_CONCEPT_REQUIRES_(
-                    hpx::execution::experimental::detail::is_receiver_cpo_v<CPO> &&
-                    hpx::execution::experimental::detail::has_completion_scheduler_v<
-                        CPO, std::decay_t<Sender>>
-                )>
-            // clang-format on
-            friend constexpr auto tag_invoke(
-                hpx::execution::experimental::get_completion_scheduler_t<CPO>
-                    tag,
-                bulk_sender const& s)
-            {
-                return tag(s.sender);
-            }
-#endif
-            template <typename Receiver>
-            struct bulk_receiver
-            {
-#if defined(HPX_HAVE_STDEXEC)
-                using receiver_concept =
-                    hpx::execution::experimental::receiver_t;
-#endif
-                HPX_NO_UNIQUE_ADDRESS std::decay_t<Receiver> receiver;
-                HPX_NO_UNIQUE_ADDRESS std::decay_t<Shape> shape;
-                HPX_NO_UNIQUE_ADDRESS std::decay_t<F> f;
-
-                template <typename Receiver_, typename Shape_, typename F_>
-                bulk_receiver(Receiver_&& receiver, Shape_&& shape, F_&& f)
-                  : receiver(HPX_FORWARD(Receiver_, receiver))
-                  , shape(HPX_FORWARD(Shape_, shape))
-                  , f(HPX_FORWARD(F_, f))
-                {
-                }
-
-                template <typename Error>
-                friend void tag_invoke(
-                    set_error_t, bulk_receiver&& r, Error&& error) noexcept
-                {
-                    hpx::execution::experimental::set_error(
-                        HPX_MOVE(r.receiver), HPX_FORWARD(Error, error));
-                }
-
-                friend void tag_invoke(
-                    set_stopped_t, bulk_receiver&& r) noexcept
-                {
-                    hpx::execution::experimental::set_stopped(
-                        HPX_MOVE(r.receiver));
-                }
-
-                template <typename... Ts>
-#if defined(HPX_HAVE_STDEXEC)
-                void set_value(Ts&&... ts) noexcept
-#else
-                void set_value(Ts&&... ts)
-#endif
-                {
-                    hpx::detail::try_catch_exception_ptr(
-                        [&]() {
-                            for (auto const& s : shape)
-                            {
-                                HPX_INVOKE(f, s, ts...);
-                            }
-                            hpx::execution::experimental::set_value(
-                                HPX_MOVE(receiver), HPX_FORWARD(Ts, ts)...);
-                        },
-                        [&](std::exception_ptr ep) {
-                            hpx::execution::experimental::set_error(
-                                HPX_MOVE(receiver), HPX_MOVE(ep));
-                        });
-                }
-
-                template <typename... Ts>
-                friend auto tag_invoke(
-                    set_value_t, bulk_receiver&& r, Ts&&... ts) noexcept
-                    -> decltype(hpx::execution::experimental::set_value(
-                                    std::declval<std::decay_t<Receiver>&&>(),
-                                    HPX_FORWARD(Ts, ts)...),
-                        void())
-                {
-                    // set_value is in a member function only because of a
-                    // compiler bug in GCC 7. When the body of set_value is
-                    // inlined here compilation fails with an internal compiler
-                    // error.
-                    r.set_value(HPX_FORWARD(Ts, ts)...);
-                }
-            };
-
-            template <typename Receiver>
-            friend auto tag_invoke(
-                connect_t, bulk_sender&& s, Receiver&& receiver)
-            {
-                return hpx::execution::experimental::connect(HPX_MOVE(s.sender),
-                    bulk_receiver<Receiver>(HPX_FORWARD(Receiver, receiver),
-                        HPX_MOVE(s.shape), HPX_MOVE(s.f)));
-            }
-
-            template <typename Receiver>
-            friend auto tag_invoke(
-                connect_t, bulk_sender& s, Receiver&& receiver)
-            {
-                return hpx::execution::experimental::connect(s.sender,
-                    bulk_receiver<Receiver>(
-                        HPX_FORWARD(Receiver, receiver), s.shape, s.f));
-            }
-        };
-    }    // namespace detail
-
-    ///////////////////////////////////////////////////////////////////////////
-    //
-    // execution::bulk is used to run a task repeatedly for every index in an
-    // index space.
-    //
-    // Returns a sender describing the task of invoking the provided function
-    // with every index in the provided shape along with the values sent by the
-    // input sender. The returned sender completes once all invocations have
-    // completed, or an error has occurred. If it completes by sending values,
-    // they are equivalent to those sent by the input sender.
-    //
-    // No instance of function will begin executing until the returned sender is
-    // started. Each invocation of function runs in an execution agent whose
-    // forward progress guarantees are determined by the scheduler on which they
-    // are run. All agents created by a single use of bulk execute with the same
-    // guarantee. This allows, for instance, a scheduler to execute all
-    // invocations of the function in parallel.
-    //
-    // The bulk operation is intended to be used at the point where the number
-    // of agents to be created is known and provided to bulk via its shape
-    // parameter. For some parallel computations, the number of agents to be
-    // created may be a function of the input data or dynamic conditions of the
-    // execution environment. In such cases, bulk can be combined with
-    // additional operations such as let_value to deliver dynamic shape
-    // information to the bulk operation.
-    //
-    inline constexpr struct bulk_t final
-      : hpx::functional::detail::tag_priority<bulk_t>
-    {
-    private:
-        // clang-format off
-        template <typename Sender, typename Shape, typename F,
-            HPX_CONCEPT_REQUIRES_(
-                is_sender_v<Sender> &&
-                experimental::detail::is_completion_scheduler_tag_invocable_v<
-                    hpx::execution::experimental::set_value_t, Sender,
-                    bulk_t, Shape, F
-                >
-            )>
-        // clang-format on
-        friend constexpr HPX_FORCEINLINE auto tag_override_invoke(
-            bulk_t, Sender&& sender, Shape const& shape, F&& f)
-        {
-            auto scheduler =
-                hpx::execution::experimental::get_completion_scheduler<
-                    hpx::execution::experimental::set_value_t>(
-#if defined(HPX_HAVE_STDEXEC)
-                    hpx::execution::experimental::get_env(sender)
-#else
-                    sender
-#endif
-                );
-
-            return hpx::functional::tag_invoke(bulk_t{}, HPX_MOVE(scheduler),
-                HPX_FORWARD(Sender, sender), shape, HPX_FORWARD(F, f));
-        }
-
-        // clang-format off
-        template <typename Sender, typename Shape, typename F,
-            HPX_CONCEPT_REQUIRES_(
-                is_sender_v<Sender> &&
-                std::is_integral_v<Shape>
-            )>
-        // clang-format on
-        friend constexpr HPX_FORCEINLINE auto tag_fallback_invoke(
-            bulk_t, Sender&& sender, Shape const& shape, F&& f)
-        {
-            return detail::bulk_sender<Sender, hpx::util::counting_shape<Shape>,
-                F>{HPX_FORWARD(Sender, sender),
-                hpx::util::counting_shape(shape), HPX_FORWARD(F, f)};
-        }
-
-        // clang-format off
-        template <typename Sender, typename Shape, typename F,
-            HPX_CONCEPT_REQUIRES_(
-                is_sender_v<Sender> &&
-                !std::is_integral_v<std::decay_t<Shape>>
-            )>
-        // clang-format on
-        friend constexpr HPX_FORCEINLINE auto tag_fallback_invoke(
-            bulk_t, Sender&& sender, Shape&& shape, F&& f)
-        {
-            return detail::bulk_sender<Sender, Shape, F>{
-                HPX_FORWARD(Sender, sender), HPX_FORWARD(Shape, shape),
-                HPX_FORWARD(F, f)};
-        }
 
-        template <typename Shape, typename F>
-        friend constexpr HPX_FORCEINLINE auto tag_fallback_invoke(
-            bulk_t, Shape&& shape, F&& f)
-        {
-            return detail::partial_algorithm<bulk_t, Shape, F>{
-                HPX_FORWARD(Shape, shape), HPX_FORWARD(F, f)};
-        }
-    } bulk{};
-}    // namespace hpx::execution::experimental
+#endif

libs/core/execution/tests/unit/algorithm_bulk.cpp

@@ -39,13 +39,15 @@ struct custom_bulk_operation
     }
 };
 
+#if !defined(HPX_HAVE_STDEXEC)
 template <typename S>
 auto tag_invoke(ex::bulk_t, S&& s, int num, custom_bulk_operation t)
 {
     t.tag_invoke_overload_called = true;
     return ex::bulk(
         std::forward<S>(s), num, [t = std::move(t)](int n) { t(n); });
 }
+#endif
 
 int main()
 {
@@ -284,7 +286,12 @@ int main()
         auto os = ex::connect(std::move(s), std::move(r));
         ex::start(os);
         HPX_TEST(receiver_set_value_called);
+#if defined(HPX_HAVE_STDEXEC)
+        // stdexec doesn't use tag_invoke for bulk customization
+        HPX_TEST(!tag_invoke_overload_called);
+#else
         HPX_TEST(tag_invoke_overload_called);
+#endif
         HPX_TEST(custom_bulk_call_operator_called);
         HPX_TEST_EQ(custom_bulk_call_count, 10);
     }
@@ -395,7 +402,12 @@ int main()
         auto os = ex::connect(std::move(s), std::move(r));
         ex::start(os);
         HPX_TEST(receiver_set_error_called);
+#if defined(HPX_HAVE_STDEXEC)
+        // stdexec doesn't use tag_invoke for bulk customization
+        HPX_TEST(!tag_invoke_overload_called);
+#else
         HPX_TEST(tag_invoke_overload_called);
+#endif
         HPX_TEST(custom_bulk_call_operator_called);
         HPX_TEST_EQ(custom_bulk_call_count, 3);
     }

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

@@ -138,9 +138,23 @@ namespace hpx::execution::experimental {
     using stdexec::transfer_just;
     using stdexec::transfer_just_t;
 
-    // Bulk (NOT FORWARDED)
-    //    using stdexec::bulk_t;
-    //    using stdexec::bulk;
+    // Bulk
+    using stdexec::bulk;
+    using stdexec::bulk_chunked;
+    using stdexec::bulk_chunked_t;
+    using stdexec::bulk_t;
+    using stdexec::bulk_unchunked;
+    using stdexec::bulk_unchunked_t;
+
+    // Execution policies (required for stdexec bulk)
+    using stdexec::par;
+    using stdexec::par_unseq;
+    using stdexec::parallel_policy;
+    using stdexec::parallel_unsequenced_policy;
+    using stdexec::seq;
+    using stdexec::sequenced_policy;
+    using stdexec::unseq;
+    using stdexec::unsequenced_policy;
 
     // Split
     using stdexec::split;