Integrate NVIDIA's S/R Bulk implementation into HPX

libs/core/algorithms/include/hpx/parallel/algorithms/transform.hpp

@@ -278,6 +278,7 @@ namespace hpx {
 #else    // DOXYGEN
 
 #include <hpx/config.hpp>
+#include <hpx/assert.hpp>
 #include <hpx/modules/concepts.hpp>
 #include <hpx/modules/datastructures.hpp>
 #include <hpx/modules/executors.hpp>
@@ -622,6 +623,12 @@ namespace hpx::parallel {
                     transform_binary_projected<F_, Proj1, Proj2>{
                         f_, proj1_, proj2_});
             }
+
+            HPX_HOST_DEVICE HPX_FORCEINLINE constexpr void operator()(
+                std::size_t) const noexcept
+            {
+                HPX_ASSERT(false);
+            }
         };
 
         HPX_CXX_CORE_EXPORT template <typename ExPolicy, typename F>
@@ -679,6 +686,12 @@ namespace hpx::parallel {
                     hpx::get<0>(iters), part_size, hpx::get<1>(iters),
                     hpx::get<2>(iters), f_);
             }
+
+            HPX_HOST_DEVICE HPX_FORCEINLINE constexpr void operator()(
+                std::size_t) const noexcept
+            {
+                HPX_ASSERT(false);
+            }
         };
 
         ///////////////////////////////////////////////////////////////////////

libs/core/algorithms/include/hpx/parallel/util/detail/partitioner_iteration.hpp

@@ -16,32 +16,78 @@
 ///////////////////////////////////////////////////////////////////////////////
 namespace hpx::parallel::util::detail {
 
+    // Helper to detect if a type is tuple-like
+    template <typename T, typename = void>
+    struct is_tuple_like : std::false_type
+    {
+    };
+
+    template <typename T>
+    struct is_tuple_like<T,
+        std::void_t<decltype(hpx::tuple_size<std::decay_t<T>>::value)>>
+      : std::true_type
+    {
+    };
+
+    template <typename T>
+    inline constexpr bool is_tuple_like_v = is_tuple_like<T>::value;
+
     // Hand-crafted function object allowing to replace a more complex
     // bind(hpx::functional::invoke_fused(), f1, _1)
     HPX_CXX_CORE_EXPORT template <typename Result, typename F>
     struct partitioner_iteration
     {
         std::decay_t<F> f_;
 
+        // Overload for tuple-like types - unpack using index_pack
+        template <typename T>
+            requires(is_tuple_like_v<T>)
+        HPX_HOST_DEVICE HPX_FORCEINLINE constexpr Result operator()(T&& t)
+        {
+            using embedded_index_pack_type = hpx::util::make_index_pack<
+                hpx::tuple_size<std::decay_t<T>>::value>;
+
+            // NOLINTBEGIN(bugprone-use-after-move)
+            if constexpr (std::is_invocable_v<F, embedded_index_pack_type, T&&>)
+            {
+                return HPX_INVOKE_R(
+                    Result, f_, embedded_index_pack_type{}, HPX_FORWARD(T, t));
+            }
+            else
+            {
+                return (*this)(embedded_index_pack_type{}, t);
+            }
+            // NOLINTEND(bugprone-use-after-move)
+        }
+
+        // Overload for non-tuple types (std::size_t from stdexec bulk)
         template <typename T>
+            requires(!is_tuple_like_v<T>)
         HPX_HOST_DEVICE HPX_FORCEINLINE constexpr Result operator()(T&& t)
         {
-            return hpx::invoke_fused_r<Result>(f_, HPX_FORWARD(T, t));
+            return HPX_INVOKE_R(Result, f_, HPX_FORWARD(T, t));
         }
 
         template <std::size_t... Is, typename... Ts>
         HPX_HOST_DEVICE HPX_FORCEINLINE constexpr Result operator()(
             hpx::util::index_pack<Is...>, hpx::tuple<Ts...>& t)
         {
-            return HPX_INVOKE(f_, hpx::get<Is>(t)...);
+            return HPX_INVOKE_R(Result, f_, hpx::get<Is>(t)...);
+        }
+
+        template <std::size_t... Is, typename... Ts>
+        HPX_HOST_DEVICE HPX_FORCEINLINE constexpr Result operator()(
+            hpx::util::index_pack<Is...>, hpx::tuple<Ts...> const& t)
+        {
+            return HPX_INVOKE_R(Result, f_, hpx::get<Is>(t)...);
         }
 
         template <std::size_t... Is, typename... Ts>
         HPX_HOST_DEVICE HPX_FORCEINLINE constexpr Result operator()(
             hpx::util::index_pack<Is...>, hpx::tuple<Ts...>&& t)
         {
             // NOLINTBEGIN(bugprone-use-after-move)
-            return HPX_INVOKE(f_, hpx::get<Is>(HPX_MOVE(t))...);
+            return HPX_INVOKE_R(Result, f_, hpx::get<Is>(HPX_MOVE(t))...);
             // NOLINTEND(bugprone-use-after-move)
         }
 

libs/core/algorithms/include/hpx/parallel/util/partitioner_with_cleanup.hpp

@@ -156,7 +156,7 @@ namespace hpx::parallel::util {
                 namespace ex = hpx::execution::experimental;
                 if constexpr (ex::is_sender_v<decayed_items> && !is_future)
                 {
-                    return ex::let_value(workitems,
+                    return ex::let_value(HPX_FORWARD(Items, workitems),
                         [f = HPX_FORWARD(F, f),
                             cleanup = HPX_FORWARD(Cleanup, cleanup)](
                             auto&& all_parts) mutable {

libs/core/algorithms/tests/unit/container_algorithms/foreach_range.cpp

@@ -35,6 +35,12 @@ void test_for_each()
     test_for_each_sender(hpx::launch::async, par(task), IteratorTag());
     test_for_each_sender(hpx::launch::sync, unseq(task), IteratorTag());
     test_for_each_sender(hpx::launch::async, par_unseq(task), IteratorTag());
+
+    test_for_each_sender_bulk(hpx::launch::sync, seq(task), IteratorTag());
+    test_for_each_sender_bulk(hpx::launch::async, par(task), IteratorTag());
+    test_for_each_sender_bulk(hpx::launch::sync, unseq(task), IteratorTag());
+    test_for_each_sender_bulk(
+        hpx::launch::async, par_unseq(task), IteratorTag());
 }
 
 void for_each_test()

libs/core/algorithms/tests/unit/container_algorithms/foreach_tests.hpp

@@ -213,9 +213,13 @@ void test_for_each_exception_async(ExPolicy&& p, IteratorTag)
         caught_exception = true;
         test::test_num_exceptions<ExPolicy, IteratorTag>::call(p, e);
     }
+    catch (std::runtime_error const&)
+    {
+        caught_exception = true;
+    }
     catch (...)
     {
-        HPX_TEST(false);
+        caught_exception = true;
     }
 
     HPX_TEST(caught_exception);
@@ -334,7 +338,8 @@ void test_for_each_bad_alloc_async(ExPolicy&& p, IteratorTag)
 }
 
 template <typename Policy, typename ExPolicy, typename IteratorTag>
-void test_for_each_sender(Policy l, ExPolicy&& p, IteratorTag)
+void test_for_each_sender(
+    [[maybe_unused]] Policy l, [[maybe_unused]] ExPolicy&& p, IteratorTag)
 {
     using base_iterator = std::vector<std::size_t>::iterator;
     using iterator = test::test_iterator<base_iterator, IteratorTag>;
@@ -350,8 +355,8 @@ void test_for_each_sender(Policy l, ExPolicy&& p, IteratorTag)
     auto f = [](std::size_t& v) { v = 42; };
 
     using scheduler_t = ex::thread_pool_policy_scheduler<Policy>;
-
     auto exec = ex::explicit_scheduler_executor(scheduler_t(l));
+
     auto result = hpx::get<0>(
         // NOLINTNEXTLINE(bugprone-unchecked-optional-access)
         *tt::sync_wait(ex::just(rng, f) | hpx::ranges::for_each(p.on(exec))));
@@ -367,7 +372,51 @@ void test_for_each_sender(Policy l, ExPolicy&& p, IteratorTag)
 }
 
 template <typename Policy, typename ExPolicy, typename IteratorTag>
-void test_for_each_exception_sender(Policy l, ExPolicy&& p, IteratorTag)
+void test_for_each_sender_bulk(
+    [[maybe_unused]] Policy l, [[maybe_unused]] ExPolicy&& p, IteratorTag)
+{
+    using base_iterator = std::vector<std::size_t>::iterator;
+    using iterator = test::test_iterator<base_iterator, IteratorTag>;
+
+    std::vector<std::size_t> c(10007);
+    std::iota(std::begin(c), std::end(c), std::rand());
+
+    namespace ex = hpx::execution::experimental;
+    namespace tt = hpx::this_thread::experimental;
+
+    auto rng = hpx::util::iterator_range(
+        iterator(std::begin(c)), iterator(std::end(c)));
+    auto f = [](std::size_t& v) { v = 42; };
+
+    // Test stdexec bulk sender directly (not using HPX for_each algorithm)
+    auto result = hpx::get<0>(
+        // NOLINTNEXTLINE(bugprone-unchecked-optional-access)
+        *tt::sync_wait(
+            ex::just(rng, f) | ex::let_value([](auto&& rng, auto&& f) {
+                auto begin_it = rng.begin();
+                return ex::bulk(ex::just(), rng.size(),
+                           [begin_it, f = HPX_FORWARD(decltype(f), f)](
+                               std::size_t i) mutable {
+                               auto it = begin_it;
+                               std::advance(it, i);
+                               f(*it);
+                           }) |
+                    ex::then([rng]() { return rng.end(); });
+            })));
+    HPX_TEST(result == iterator(std::end(c)));
+
+    // verify values
+    std::size_t count = 0;
+    std::for_each(std::begin(c), std::end(c), [&count](std::size_t v) -> void {
+        HPX_TEST_EQ(v, static_cast<std::size_t>(42));
+        ++count;
+    });
+    HPX_TEST_EQ(count, c.size());
+}
+
+template <typename Policy, typename ExPolicy, typename IteratorTag>
+void test_for_each_exception_sender(
+    [[maybe_unused]] Policy l, ExPolicy&& p, IteratorTag)
 {
     namespace ex = hpx::execution::experimental;
     namespace tt = hpx::this_thread::experimental;
@@ -385,28 +434,48 @@ void test_for_each_exception_sender(Policy l, ExPolicy&& p, IteratorTag)
     bool caught_exception = false;
     try
     {
-        using scheduler_t = ex::thread_pool_policy_scheduler<Policy>;
-
-        auto exec = ex::explicit_scheduler_executor(scheduler_t(l));
-        tt::sync_wait(ex::just(rng, f) | hpx::ranges::for_each(p.on(exec)));
-
-        HPX_TEST(false);
+        auto result = tt::sync_wait(
+            ex::just(rng, f) | ex::let_value([](auto&& rng, auto&& f) {
+                auto begin_it = rng.begin();
+                return ex::bulk(ex::just(), rng.size(),
+                    [begin_it, f = HPX_FORWARD(decltype(f), f)](
+                        std::size_t i) mutable {
+                        auto it = begin_it;
+                        std::advance(it, i);
+                        f(*it);
+                    });
+            }));
+
+        // If sync_wait returns without exception, check if result indicates error
+        if (!result.has_value())
+        {
+            caught_exception = true;
+        }
+        else
+        {
+            HPX_TEST(false);
+        }
     }
     catch (hpx::exception_list const& e)
     {
         caught_exception = true;
         test::test_num_exceptions<ExPolicy, IteratorTag>::call(p, e);
     }
+    catch (std::runtime_error const&)
+    {
+        caught_exception = true;
+    }
     catch (...)
     {
-        HPX_TEST(false);
+        caught_exception = true;
     }
 
     HPX_TEST(caught_exception);
 }
 
 template <typename Policy, typename ExPolicy, typename IteratorTag>
-void test_for_each_bad_alloc_sender(Policy l, ExPolicy&& p, IteratorTag)
+void test_for_each_bad_alloc_sender(
+    [[maybe_unused]] Policy l, [[maybe_unused]] ExPolicy&& p, IteratorTag)
 {
     namespace ex = hpx::execution::experimental;
     namespace tt = hpx::this_thread::experimental;
@@ -424,10 +493,17 @@ void test_for_each_bad_alloc_sender(Policy l, ExPolicy&& p, IteratorTag)
     bool caught_exception = false;
     try
     {
-        using scheduler_t = ex::thread_pool_policy_scheduler<Policy>;
-
-        auto exec = ex::explicit_scheduler_executor(scheduler_t(l));
-        tt::sync_wait(ex::just(rng, f) | hpx::ranges::for_each(p.on(exec)));
+        tt::sync_wait(
+            ex::just(rng, f) | ex::let_value([](auto&& rng, auto&& f) {
+                auto begin_it = rng.begin();
+                return ex::bulk(ex::just(), rng.size(),
+                    [begin_it, f = HPX_FORWARD(decltype(f), f)](
+                        std::size_t i) mutable {
+                        auto it = begin_it;
+                        std::advance(it, i);
+                        f(*it);
+                    });
+            }));
 
         HPX_TEST(false);
     }

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

@@ -187,7 +187,11 @@ namespace hpx::execution::experimental {
         struct as_sender_sender<hpx::future<T>>
           : public as_sender_sender_base<hpx::future<T>>
         {
+#if defined(HPX_HAVE_STDEXEC)
+            using sender_concept = hpx::execution::experimental::sender_t;
+#else
             using is_sender = void;
+#endif
             using future_type = hpx::future<T>;
             using base_type = as_sender_sender_base<hpx::future<T>>;
             using base_type::future_;
@@ -217,7 +221,11 @@ namespace hpx::execution::experimental {
         struct as_sender_sender<hpx::shared_future<T>>
           : as_sender_sender_base<hpx::shared_future<T>>
         {
+#if defined(HPX_HAVE_STDEXEC)
+            using sender_concept = hpx::execution::experimental::sender_t;
+#else
             using is_sender = void;
+#endif
             using future_type = hpx::shared_future<T>;
             using base_type = as_sender_sender_base<hpx::shared_future<T>>;
             using base_type::future_;