From 1c51095fa3672573b906aadcd1acfcdcf913ba24 Mon Sep 17 00:00:00 2001
From: Mikael Simberg <mikael.simberg@iki.fi>
Date: Mon, 10 May 2021 11:59:12 +0200
Subject: [PATCH] Add execution::keep_future to pass the future instead of its
 value to continuations when futures are used as senders

keep_future passes the future instead of the value to help manage future
lifetimes. Especially shared_future | on | ... is not possible without
this helper if the wrapped value in the future is noncopyable (since
shared_future::get returns a const& and on cannot store a const&, nor
does it need to keep the operation state alive long enough).
---
 libs/parallelism/execution/CMakeLists.txt     |   1 +
 .../hpx/execution/algorithms/keep_future.hpp  | 159 ++++++++++++
 .../executors/tests/unit/p0443_executor.cpp   | 239 ++++++++++++++++++
 3 files changed, 399 insertions(+)
 create mode 100644 libs/parallelism/execution/include/hpx/execution/algorithms/keep_future.hpp

diff --git a/libs/parallelism/execution/CMakeLists.txt b/libs/parallelism/execution/CMakeLists.txt
index b8f65c5712cf..a898c25ff0eb 100644
--- a/libs/parallelism/execution/CMakeLists.txt
+++ b/libs/parallelism/execution/CMakeLists.txt
@@ -15,6 +15,7 @@ set(execution_headers
     hpx/execution/algorithms/ensure_started.hpp
     hpx/execution/algorithms/just.hpp
     hpx/execution/algorithms/just_on.hpp
+    hpx/execution/algorithms/keep_future.hpp
     hpx/execution/algorithms/let_error.hpp
     hpx/execution/algorithms/let_value.hpp
     hpx/execution/algorithms/on.hpp
diff --git a/libs/parallelism/execution/include/hpx/execution/algorithms/keep_future.hpp b/libs/parallelism/execution/include/hpx/execution/algorithms/keep_future.hpp
new file mode 100644
index 000000000000..069c2da7d234
--- /dev/null
+++ b/libs/parallelism/execution/include/hpx/execution/algorithms/keep_future.hpp
@@ -0,0 +1,159 @@
+//  Copyright (c) 2021 ETH Zurich
+//
+//  SPDX-License-Identifier: BSL-1.0
+//  Distributed under the Boost Software License, Version 1.0. (See accompanying
+//  file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
+
+#pragma once
+
+#include <hpx/assert.hpp>
+#include <hpx/execution/algorithms/detail/partial_algorithm.hpp>
+#include <hpx/execution_base/operation_state.hpp>
+#include <hpx/execution_base/receiver.hpp>
+#include <hpx/execution_base/sender.hpp>
+#include <hpx/functional/tag_fallback_invoke.hpp>
+#include <hpx/futures/detail/future_data.hpp>
+#include <hpx/futures/future.hpp>
+#include <hpx/futures/traits/acquire_shared_state.hpp>
+
+#include <exception>
+#include <type_traits>
+#include <utility>
+
+namespace hpx { namespace execution { namespace experimental {
+    namespace detail {
+        template <typename Receiver, typename Future>
+        struct operation_state
+        {
+            std::decay_t<Receiver> receiver;
+            std::decay_t<Future> future;
+
+            void start() & noexcept
+            {
+                try
+                {
+                    auto state = hpx::traits::detail::get_shared_state(future);
+
+                    if (!state)
+                    {
+                        HPX_THROW_EXCEPTION(no_state, "operation_state::start",
+                            "the future has no valid shared state asdasd");
+                    }
+
+                    state->set_on_completed(
+                        [this, receiver = std::move(receiver)]() mutable {
+                            hpx::execution::experimental::set_value(
+                                std::move(receiver), std::move(future));
+                        });
+                }
+                catch (...)
+                {
+                    hpx::execution::experimental::set_error(
+                        std::move(receiver), std::current_exception());
+                }
+            }
+        };
+        template <typename Future>
+        struct keep_future_sender_base
+        {
+            std::decay_t<Future> future;
+
+            template <template <typename...> class Tuple,
+                template <typename...> class Variant>
+            using value_types = Variant<Tuple<std::decay_t<Future>>>;
+
+            template <template <typename...> class Variant>
+            using error_types = Variant<std::exception_ptr>;
+
+            static constexpr bool sends_done = false;
+        };
+
+        template <typename Future>
+        struct keep_future_sender;
+
+        template <typename T>
+        struct keep_future_sender<hpx::future<T>>
+          : public keep_future_sender_base<hpx::future<T>>
+        {
+            using future_type = hpx::future<T>;
+            using base_type = keep_future_sender_base<hpx::future<T>>;
+            using base_type::future;
+
+            template <typename Future,
+                typename = std::enable_if_t<!std::is_same<std::decay_t<Future>,
+                    keep_future_sender>::value>>
+            explicit keep_future_sender(Future&& future)
+              : base_type{std::forward<Future>(future)}
+            {
+            }
+
+            keep_future_sender(keep_future_sender&&) = default;
+            keep_future_sender& operator=(keep_future_sender&&) = default;
+            keep_future_sender(keep_future_sender const&) = delete;
+            keep_future_sender& operator=(keep_future_sender const&) = delete;
+
+            template <typename Receiver>
+            operation_state<Receiver, future_type> connect(
+                Receiver&& receiver) &&
+            {
+                return {std::forward<Receiver>(receiver), std::move(future)};
+            }
+        };
+
+        template <typename T>
+        struct keep_future_sender<hpx::shared_future<T>>
+          : keep_future_sender_base<hpx::shared_future<T>>
+        {
+            using future_type = hpx::shared_future<T>;
+            using base_type = keep_future_sender_base<hpx::shared_future<T>>;
+            using base_type::future;
+
+            template <typename Future,
+                typename = std::enable_if_t<!std::is_same<std::decay_t<Future>,
+                    keep_future_sender>::value>>
+            explicit keep_future_sender(Future&& future)
+              : base_type{std::forward<Future>(future)}
+            {
+            }
+
+            keep_future_sender(keep_future_sender&&) = default;
+            keep_future_sender& operator=(keep_future_sender&&) = default;
+            keep_future_sender(keep_future_sender const&) = default;
+            keep_future_sender& operator=(keep_future_sender const&) = default;
+
+            template <typename Receiver>
+            operation_state<Receiver, future_type> connect(
+                Receiver&& receiver) &&
+            {
+                return {std::forward<Receiver>(receiver), std::move(future)};
+            }
+
+            template <typename Receiver>
+            operation_state<Receiver, future_type> connect(
+                Receiver&& receiver) &
+            {
+                return {std::forward<Receiver>(receiver), future};
+            }
+        };
+    }    // namespace detail
+
+    HPX_INLINE_CONSTEXPR_VARIABLE struct keep_future_t final
+      : hpx::functional::tag_fallback<keep_future_t>
+    {
+    private:
+        template <typename Future,
+            typename = std::enable_if_t<
+                hpx::traits::is_future<std::decay_t<Future>>::value>>
+        friend constexpr HPX_FORCEINLINE auto tag_fallback_invoke(
+            keep_future_t, Future&& future)
+        {
+            return detail::keep_future_sender<std::decay_t<Future>>(
+                std::forward<Future>(future));
+        }
+
+        friend constexpr HPX_FORCEINLINE auto tag_fallback_invoke(keep_future_t)
+        {
+            return detail::partial_algorithm<keep_future_t>{};
+        }
+    } keep_future{};
+}}}    // namespace hpx::execution::experimental
diff --git a/libs/parallelism/executors/tests/unit/p0443_executor.cpp b/libs/parallelism/executors/tests/unit/p0443_executor.cpp
index f3e48c754184..6c254214f55d 100644
--- a/libs/parallelism/executors/tests/unit/p0443_executor.cpp
+++ b/libs/parallelism/executors/tests/unit/p0443_executor.cpp
@@ -8,6 +8,7 @@
 #include <hpx/execution.hpp>
 #include <hpx/functional.hpp>
 #include <hpx/init.hpp>
+#include <hpx/local/future.hpp>
 #include <hpx/modules/testing.hpp>
 #include <hpx/mutex.hpp>
 #include <hpx/thread.hpp>
@@ -23,6 +24,22 @@
 #include <type_traits>
 #include <utility>
 
+struct custom_type_non_default_constructible_non_copyable
+{
+    int x;
+    custom_type_non_default_constructible_non_copyable() = delete;
+    explicit custom_type_non_default_constructible_non_copyable(int x)
+      : x(x){};
+    custom_type_non_default_constructible_non_copyable(
+        custom_type_non_default_constructible_non_copyable&&) = default;
+    custom_type_non_default_constructible_non_copyable& operator=(
+        custom_type_non_default_constructible_non_copyable&&) = default;
+    custom_type_non_default_constructible_non_copyable(
+        custom_type_non_default_constructible_non_copyable const&) = delete;
+    custom_type_non_default_constructible_non_copyable& operator=(
+        custom_type_non_default_constructible_non_copyable const&) = delete;
+};
+
 namespace ex = hpx::execution::experimental;
 
 ///////////////////////////////////////////////////////////////////////////////
@@ -1185,6 +1202,227 @@ void test_detach()
     }
 }
 
+void test_keep_future_sender()
+{
+    // the future should be passed to transform, not it's contained value
+    {
+        hpx::make_ready_future<void>() | ex::keep_future() |
+            ex::transform([](auto f) {
+                static_assert(std::is_same<std::decay_t<decltype(f)>,
+                                  hpx::future<void>>::value,
+                    "f should be a future<void>");
+                HPX_TEST(f.is_ready());
+            }) |
+            ex::sync_wait();
+    }
+
+    {
+        hpx::make_ready_future<void>().share() | ex::keep_future() |
+            ex::transform([](auto f) {
+                static_assert(std::is_same<std::decay_t<decltype(f)>,
+                                  hpx::shared_future<void>>::value,
+                    "f should be a shared_future<void>");
+                HPX_TEST(f.is_ready());
+            }) |
+            ex::sync_wait();
+    }
+
+    {
+        hpx::make_ready_future<int>(42) | ex::keep_future() |
+            ex::transform([](auto f) {
+                static_assert(std::is_same<std::decay_t<decltype(f)>,
+                                  hpx::future<int>>::value,
+                    "f should be a future<int>");
+                HPX_TEST(f.is_ready());
+                HPX_TEST_EQ(f.get(), 42);
+            }) |
+            ex::sync_wait();
+    }
+
+    {
+        hpx::make_ready_future<int>(42).share() | ex::keep_future() |
+            ex::transform([](auto f) {
+                static_assert(std::is_same<std::decay_t<decltype(f)>,
+                                  hpx::shared_future<int>>::value,
+                    "f should be a shared_future<int>");
+                HPX_TEST(f.is_ready());
+                HPX_TEST_EQ(f.get(), 42);
+            }) |
+            ex::sync_wait();
+    }
+
+    {
+        std::atomic<bool> called{false};
+        auto f = hpx::async([&]() { called = true; });
+
+        auto r = ex::sync_wait(std::move(f) | ex::keep_future());
+        static_assert(
+            std::is_same<std::decay_t<decltype(r)>, hpx::future<void>>::value,
+            "sync_wait should return future<void>");
+
+        HPX_TEST(called);
+        HPX_TEST(r.is_ready());
+
+        bool exception_thrown = false;
+        try
+        {
+            ex::sync_wait(std::move(f) | ex::keep_future());
+        }
+        catch (...)
+        {
+            exception_thrown = true;
+        };
+        HPX_TEST(exception_thrown);
+    }
+
+    {
+        std::atomic<bool> called{false};
+        auto f = hpx::async([&]() {
+            called = true;
+            return 42;
+        });
+
+        auto r = ex::sync_wait(std::move(f) | ex::keep_future());
+        static_assert(
+            std::is_same<std::decay_t<decltype(r)>, hpx::future<int>>::value,
+            "sync_wait should return future<int>");
+
+        HPX_TEST(called);
+        HPX_TEST(r.is_ready());
+        HPX_TEST_EQ(r.get(), 42);
+
+        bool exception_thrown = false;
+        try
+        {
+            ex::sync_wait(std::move(f) | ex::keep_future());
+        }
+        catch (...)
+        {
+            exception_thrown = true;
+        };
+        HPX_TEST(exception_thrown);
+    }
+
+    {
+        std::atomic<bool> called{false};
+        auto f = hpx::async([&]() {
+            called = true;
+            return 42;
+        });
+
+        HPX_TEST_EQ(
+            ex::sync_wait(ex::transform(std::move(f) | ex::keep_future(),
+                [](hpx::future<int>&& f) { return f.get() / 2; })),
+            21);
+        HPX_TEST(called);
+    }
+
+    {
+        std::atomic<std::size_t> calls{0};
+        auto sf = hpx::async([&]() { ++calls; }).share();
+
+        ex::sync_wait(sf | ex::keep_future());
+        ex::sync_wait(sf | ex::keep_future());
+        ex::sync_wait(std::move(sf) | ex::keep_future());
+        HPX_TEST_EQ(calls, std::size_t(1));
+
+        bool exception_thrown = false;
+        try
+        {
+            ex::sync_wait(sf | ex::keep_future());
+        }
+        catch (...)
+        {
+            exception_thrown = true;
+        };
+        HPX_TEST(exception_thrown);
+    }
+
+    {
+        std::atomic<std::size_t> calls{0};
+        auto sf = hpx::async([&]() {
+            ++calls;
+            return 42;
+        }).share();
+
+        HPX_TEST_EQ(ex::sync_wait(sf | ex::keep_future()).get(), 42);
+        HPX_TEST_EQ(ex::sync_wait(sf | ex::keep_future()).get(), 42);
+        HPX_TEST_EQ(ex::sync_wait(std::move(sf) | ex::keep_future()).get(), 42);
+        HPX_TEST_EQ(calls, std::size_t(1));
+
+        bool exception_thrown = false;
+        try
+        {
+            ex::sync_wait(sf | ex::keep_future());
+        }
+        catch (...)
+        {
+            exception_thrown = true;
+        };
+        HPX_TEST(exception_thrown);
+    }
+
+    // Keep future alive across on
+    {
+        auto f = hpx::async([&]() { return 42; });
+
+        auto r = std::move(f) | ex::keep_future() | ex::on(ex::executor{}) |
+            ex::sync_wait();
+        HPX_TEST(r.is_ready());
+        HPX_TEST_EQ(r.get(), 42);
+    }
+
+    {
+        auto sf = hpx::async([&]() { return 42; }).share();
+
+        auto r = std::move(sf) | ex::keep_future() | ex::on(ex::executor{}) |
+            ex::sync_wait();
+        HPX_TEST(r.is_ready());
+        HPX_TEST_EQ(r.get(), 42);
+    }
+
+    {
+        auto sf = hpx::async([&]() {
+            return custom_type_non_default_constructible_non_copyable{42};
+        }).share();
+
+        // NOTE: Without keep_future this should fail to compile, since
+        // sync_wait would receive a const& to the value which requires a copy
+        // or storing a const&. The copy is not possible because the type is
+        // noncopyable, and storing a reference is not acceptable since the
+        // reference may outlive the value.
+        auto r = std::move(sf) | ex::keep_future() | ex::on(ex::executor{}) |
+            ex::sync_wait();
+        HPX_TEST(r.is_ready());
+        HPX_TEST_EQ(r.get().x, 42);
+    }
+
+    // Use unwrapping with keep_future
+    {
+        auto f = hpx::async([]() { return 42; });
+        auto sf = hpx::async([]() { return 3.14; }).share();
+
+        auto fun = hpx::util::unwrapping(
+            [](int&& x, double const& y) { return x * 2 + (int(y) / 2); });
+        HPX_TEST_EQ(ex::when_all(std::move(f) | ex::keep_future(),
+                        sf | ex::keep_future()) |
+                ex::transform(fun) | ex::sync_wait(),
+            85);
+    }
+
+    {
+        auto f = hpx::async([]() { return 42; });
+        auto sf = hpx::async([]() { return 3.14; }).share();
+
+        auto fun = hpx::util::unwrapping(
+            [](int&& x, double const& y) { return x * 2 + (int(y) / 2); });
+        HPX_TEST_EQ(ex::when_all(std::move(f) | ex::keep_future(),
+                        sf | ex::keep_future()) |
+                ex::on(ex::executor{}) | ex::transform(fun) | ex::sync_wait(),
+            85);
+    }
+}
+
 ///////////////////////////////////////////////////////////////////////////////
 int hpx_main()
 {
@@ -1206,6 +1444,7 @@ int hpx_main()
     test_just_on_two_args();
     test_when_all();
     test_future_sender();
+    test_keep_future_sender();
     test_ensure_started();
     test_ensure_started_when_all();
     test_let_value();