Implement HPX Future-Sender Bridge (P2300 interoperability)

libs/core/execution/CMakeLists.txt

@@ -15,9 +15,11 @@ set(execution_headers
     hpx/execution/algorithms/detail/predicates.hpp
     hpx/execution/algorithms/detail/single_result.hpp
     hpx/execution/algorithms/detail/sync_wait_domain.hpp
+    hpx/execution/algorithms/future_sender.hpp
     hpx/execution/algorithms/keep_future.hpp
     hpx/execution/algorithms/make_future.hpp
     hpx/execution/algorithms/run_loop.hpp
+    hpx/execution/algorithms/sender_future.hpp
     hpx/execution/algorithms/sync_wait.hpp
     hpx/execution/algorithms/when_all.hpp
     hpx/execution/algorithms/when_all_vector.hpp

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

@@ -9,6 +9,7 @@
 
 #include <hpx/assert.hpp>
 #include <hpx/execution/algorithms/detail/partial_algorithm.hpp>
+#include <hpx/execution/algorithms/future_sender.hpp>
 #include <hpx/modules/concepts.hpp>
 #include <hpx/modules/errors.hpp>
 #include <hpx/modules/execution_base.hpp>
@@ -18,312 +19,6 @@
 #include <type_traits>
 #include <utility>
 
-namespace hpx::execution::experimental { namespace detail {
-
-    ///////////////////////////////////////////////////////////////////////////
-    // Operation state for sender compatibility
-    HPX_CXX_CORE_EXPORT template <typename Receiver, typename Future>
-    class as_sender_operation_state
-    {
-    private:
-        using receiver_type = std::decay_t<Receiver>;
-        using future_type = std::decay_t<Future>;
-        using result_type = typename future_type::result_type;
-
-    public:
-        template <typename Receiver_>
-        as_sender_operation_state(Receiver_&& r, future_type f)
-          : receiver_(HPX_FORWARD(Receiver_, r))
-          , future_(HPX_MOVE(f))
-        {
-        }
-
-        as_sender_operation_state(as_sender_operation_state&&) = delete;
-        as_sender_operation_state& operator=(
-            as_sender_operation_state&&) = delete;
-        as_sender_operation_state(as_sender_operation_state const&) = delete;
-        as_sender_operation_state& operator=(
-            as_sender_operation_state const&) = delete;
-
-        void start() & noexcept
-        {
-            start_helper();
-        }
-
-    private:
-        void start_helper() & noexcept
-        {
-            hpx::detail::try_catch_exception_ptr(
-                [&]() {
-                    auto state = traits::detail::get_shared_state(future_);
-
-                    if (!state)
-                    {
-                        HPX_THROW_EXCEPTION(hpx::error::no_state,
-                            "as_sender_operation_state::start",
-                            "the future has no valid shared state");
-                    }
-
-                    auto on_completed = [this]() mutable {
-                        if (future_.has_value())
-                        {
-                            if constexpr (std::is_void_v<result_type>)
-                            {
-                                hpx::execution::experimental::set_value(
-                                    HPX_MOVE(receiver_));
-                            }
-                            else
-                            {
-                                hpx::execution::experimental::set_value(
-                                    HPX_MOVE(receiver_), future_.get());
-                            }
-                        }
-                        else if (future_.has_exception())
-                        {
-                            hpx::execution::experimental::set_error(
-                                HPX_MOVE(receiver_),
-                                future_.get_exception_ptr());
-                        }
-                    };
-
-                    if (!state->is_ready(std::memory_order_relaxed))
-                    {
-                        state->execute_deferred();
-
-                        // execute_deferred might have made the future ready
-                        if (!state->is_ready(std::memory_order_relaxed))
-                        {
-                            // The operation state has to be kept alive until
-                            // set_value is called, which means that we don't
-                            // need to move receiver and future into the
-                            // on_completed callback.
-                            state->set_on_completed(HPX_MOVE(on_completed));
-                        }
-                        else
-                        {
-                            on_completed();
-                        }
-                    }
-                    else
-                    {
-                        on_completed();
-                    }
-                },
-                [&](std::exception_ptr ep) {
-                    hpx::execution::experimental::set_error(
-                        HPX_MOVE(receiver_), HPX_MOVE(ep));
-                });
-        }
-
-        HPX_NO_UNIQUE_ADDRESS std::decay_t<Receiver> receiver_;
-        future_type future_;
-    };
-
-    HPX_CXX_CORE_EXPORT template <typename Future>
-    struct as_sender_sender_base
-    {
-        using result_type = typename std::decay_t<Future>::result_type;
-
-        std::decay_t<Future> future_;
-
-        template <bool IsVoid, typename _result_type>
-        struct set_value_void_checked
-        {
-            using type = hpx::execution::experimental::set_value_t(
-                _result_type);
-        };
-
-        template <typename _result_type>
-        struct set_value_void_checked<true, _result_type>
-        {
-            using type = hpx::execution::experimental::set_value_t();
-        };
-
-        using completion_signatures =
-            hpx::execution::experimental::completion_signatures<
-                typename set_value_void_checked<std::is_void_v<result_type>,
-                    result_type>::type,
-                hpx::execution::experimental::set_error_t(std::exception_ptr)>;
-    };
-
-    HPX_CXX_CORE_EXPORT template <typename Future>
-    struct as_sender_sender;
-
-    template <typename T>
-    struct as_sender_sender<hpx::future<T>>
-      : public as_sender_sender_base<hpx::future<T>>
-    {
-        using sender_concept = hpx::execution::experimental::sender_t;
-        using future_type = hpx::future<T>;
-        using base_type = as_sender_sender_base<hpx::future<T>>;
-        using base_type::future_;
-
-        template <typename Future>
-            requires(!std::is_same_v<std::decay_t<Future>, as_sender_sender>)
-        explicit as_sender_sender(Future&& future)
-          : base_type{HPX_FORWARD(Future, future)}
-        {
-        }
-
-        as_sender_sender(as_sender_sender&&) = default;
-        as_sender_sender& operator=(as_sender_sender&&) = default;
-        as_sender_sender(as_sender_sender const&) = delete;
-        as_sender_sender& operator=(as_sender_sender const&) = delete;
-
-        template <typename Self, typename... Env>
-        static consteval auto get_completion_signatures() noexcept ->
-            typename base_type::completion_signatures
-        {
-            return {};
-        }
-
-        template <typename Receiver>
-        auto connect(Receiver&& receiver) &&
-        {
-            return as_sender_operation_state<Receiver, future_type>{
-                HPX_FORWARD(Receiver, receiver), HPX_MOVE(future_)};
-        }
-    };
-}}    // namespace hpx::execution::experimental::detail
-
-namespace hpx::execution::experimental { namespace detail {
-    template <typename T>
-    struct as_sender_sender<hpx::shared_future<T>>
-      : as_sender_sender_base<hpx::shared_future<T>>
-    {
-        using sender_concept = hpx::execution::experimental::sender_t;
-        using future_type = hpx::shared_future<T>;
-        using base_type = as_sender_sender_base<hpx::shared_future<T>>;
-        using base_type::future_;
-
-        template <typename Future>
-            requires(!std::is_same_v<std::decay_t<Future>, as_sender_sender>)
-        explicit as_sender_sender(Future&& future)
-          : base_type{HPX_FORWARD(Future, future)}
-        {
-        }
-
-        as_sender_sender(as_sender_sender&&) = default;
-        as_sender_sender& operator=(as_sender_sender&&) = default;
-        as_sender_sender(as_sender_sender const&) = default;
-        as_sender_sender& operator=(as_sender_sender const&) = default;
-
-        template <typename Self, typename... Env>
-        static consteval auto get_completion_signatures() noexcept ->
-            typename base_type::completion_signatures
-        {
-            return {};
-        }
-
-        template <typename Receiver>
-        auto connect(Receiver&& receiver) &&
-        {
-            return as_sender_operation_state<Receiver, future_type>{
-                HPX_FORWARD(Receiver, receiver), HPX_MOVE(future_)};
-        }
-
-        template <typename Receiver>
-        auto connect(Receiver&& receiver) &
-        {
-            return as_sender_operation_state<Receiver, future_type>{
-                HPX_FORWARD(Receiver, receiver), future_};
-        }
-    };
-}}    // namespace hpx::execution::experimental::detail
-
-namespace hpx::execution::experimental { namespace detail {
-
-    ///////////////////////////////////////////////////////////////////////
-    // Scheduler-aware sender wrapper.
-    //
-    // Exposes the stored scheduler through its environment so that
-    // downstream sender algorithms (bulk, sync_wait, etc.) can query
-    // get_completion_scheduler<set_value_t> and obtain the scheduler
-    // that originated the work.
-    HPX_CXX_CORE_EXPORT template <typename Future, typename Scheduler>
-        requires(hpx::traits::is_future_v<std::decay_t<Future>>)
-    struct as_sender_sender_with_scheduler
-      : public as_sender_sender_base<std::decay_t<Future>>
-    {
-        using sender_concept = hpx::execution::experimental::sender_t;
-        using future_type = std::decay_t<Future>;
-        using scheduler_type = std::decay_t<Scheduler>;
-        using base_type = as_sender_sender_base<std::decay_t<Future>>;
-        using base_type::future_;
-
-        HPX_NO_UNIQUE_ADDRESS scheduler_type scheduler_;
-
-        // Environment that answers get_completion_scheduler queries.
-        struct env
-        {
-            scheduler_type sched;
-
-            auto query(
-                hpx::execution::experimental::get_domain_t) const noexcept
-            {
-                return hpx::execution::experimental::get_domain(sched);
-            }
-
-            template <typename CPO>
-                requires(std::is_same_v<CPO,
-                             hpx::execution::experimental::set_value_t> ||
-                    std::is_same_v<CPO,
-                        hpx::execution::experimental::set_stopped_t>)
-            auto query(
-                hpx::execution::experimental::get_completion_scheduler_t<CPO>)
-                const noexcept
-            {
-                return sched;
-            }
-        };
-
-        template <typename Future_, typename Scheduler_>
-            requires(!std::is_same_v<std::decay_t<Future_>,
-                        as_sender_sender_with_scheduler>)
-        explicit as_sender_sender_with_scheduler(
-            Future_&& future, Scheduler_&& scheduler)
-          : base_type{HPX_FORWARD(Future_, future)}
-          , scheduler_(HPX_FORWARD(Scheduler_, scheduler))
-        {
-        }
-
-        as_sender_sender_with_scheduler(
-            as_sender_sender_with_scheduler&&) = default;
-        as_sender_sender_with_scheduler& operator=(
-            as_sender_sender_with_scheduler&&) = default;
-        as_sender_sender_with_scheduler(
-            as_sender_sender_with_scheduler const&) = default;
-        as_sender_sender_with_scheduler& operator=(
-            as_sender_sender_with_scheduler const&) = default;
-
-        template <typename Self, typename... Env>
-        static consteval auto get_completion_signatures() noexcept ->
-            typename base_type::completion_signatures
-        {
-            return {};
-        }
-
-        template <typename Receiver>
-        auto connect(Receiver&& receiver) &&
-        {
-            return as_sender_operation_state<Receiver, future_type>{
-                HPX_FORWARD(Receiver, receiver), HPX_MOVE(future_)};
-        }
-
-        template <typename Receiver>
-        auto connect(Receiver&& receiver) &
-        {
-            return as_sender_operation_state<Receiver, future_type>{
-                HPX_FORWARD(Receiver, receiver), future_};
-        }
-
-        constexpr auto get_env() const noexcept
-        {
-            return env{scheduler_};
-        }
-    };
-}}    // namespace hpx::execution::experimental::detail
-
 namespace hpx::execution::experimental {
     // The as_sender CPO can be used to adapt any HPX future as a sender. The
     // value provided by the future will be used to call set_value on the
@@ -343,8 +38,8 @@ namespace hpx::execution::experimental {
         // clang-format on
         constexpr HPX_FORCEINLINE auto operator()(Future&& future) const
         {
-            return detail::as_sender_sender<std::decay_t<Future>>(
-                HPX_FORWARD(Future, future));
+            return detail::future_sender<std::decay_t<Future>>{
+                HPX_FORWARD(Future, future)};
         }
 
         // Scheduler-aware overload: wraps the future into a sender whose
@@ -356,9 +51,9 @@ namespace hpx::execution::experimental {
         constexpr HPX_FORCEINLINE auto operator()(
             Future&& future, Scheduler&& scheduler) const
         {
-            return detail::as_sender_sender_with_scheduler<std::decay_t<Future>,
-                std::decay_t<Scheduler>>(
-                HPX_FORWARD(Future, future), HPX_FORWARD(Scheduler, scheduler));
+            return detail::future_sender_with_scheduler<std::decay_t<Future>,
+                std::decay_t<Scheduler>>{
+                HPX_FORWARD(Future, future), HPX_FORWARD(Scheduler, scheduler)};
         }
 
         constexpr HPX_FORCEINLINE auto operator()() const