Fix nvexec upon_stopped move-only callable forwarding

include/nvexec/stream/upon_stopped.cuh

@@ -106,7 +106,7 @@ namespace nv::execution::_strm
               status == cudaSuccess)
           {
             opstate_.defer_temp_storage_destruction(d_result);
-            opstate_.propagate_completion_signal(STDEXEC::set_value, *d_result);
+            opstate_.propagate_completion_signal(STDEXEC::set_value, std::move(*d_result));
           }
           else
           {
@@ -131,19 +131,24 @@ namespace nv::execution::_strm
   struct upon_stopped_sender : stream_sender_base
   {
     using sender_concept = STDEXEC::sender_tag;
-    using _set_error_t   = completion_signatures<set_error_t(std::exception_ptr)>;
 
     template <class Receiver>
     using receiver_t = _upon_stopped::receiver<Receiver, Fun>;
 
+    template <class Self, class... Env>
+    using __error_completions_t =
+      __minvoke_q<__concat_completion_signatures_t,
+                  __with_error_invoke_t<__mbind_front_q<__callable_error_t, upon_stopped_t>,
+                                        set_stopped_t,
+                                        Fun,
+                                        __copy_cvref_t<Self, Sender>,
+                                        Env...>,
+                  completion_signatures<set_error_t(cudaError_t)>>;
+
     template <class Self, class... Env>
     using completion_signatures = __transform_completion_signatures_t<
       __completion_signatures_of_t<__copy_cvref_t<Self, Sender>, Env...>,
-      __with_error_invoke_t<__mbind_front_q<__callable_error_t, upon_stopped_t>,
-                            set_stopped_t,
-                            Fun,
-                            __copy_cvref_t<Self, Sender>,
-                            Env...>,
+      __error_completions_t<Self, Env...>,
       __cmplsigs::__default_set_value,
       __cmplsigs::__default_set_error,
       __set_value_from_t<Fun>>;
@@ -163,7 +168,7 @@ namespace nv::execution::_strm
         static_cast<Self&&>(self).sndr_,
         static_cast<Receiver&&>(rcvr),
         [&](_strm::opstate_base<Receiver>& stream_provider) -> receiver_t<Receiver>
-        { return receiver_t<Receiver>(self.fun_, stream_provider); });
+        { return receiver_t<Receiver>(static_cast<Self&&>(self).fun_, stream_provider); });
     }
     STDEXEC_EXPLICIT_THIS_END(connect)
 

test/nvexec/upon_stopped.cpp

@@ -1,5 +1,9 @@
 #include <stdexec/execution.hpp>
 #include <test_common/catch2.hpp>
+#include <test_common/senders.hpp>
+#include <test_common/type_helpers.hpp>
+
+#include <type_traits>
 
 #include "common.cuh"
 #include "nvexec/stream_context.cuh"
@@ -10,6 +14,62 @@ using nvexec::is_on_gpu;
 
 namespace
 {
+  struct move_only_stopped_handler
+  {
+    move_only_stopped_handler()                                  = default;
+    move_only_stopped_handler(move_only_stopped_handler const &) = delete;
+
+    STDEXEC_ATTRIBUTE(host, device)
+    move_only_stopped_handler(move_only_stopped_handler &&) = default;
+
+    STDEXEC_ATTRIBUTE(host, device) auto operator()() const -> int
+    {
+      return 42;
+    }
+  };
+
+  static_assert(std::is_trivially_copyable_v<move_only_stopped_handler>);
+  static_assert(!std::is_copy_constructible_v<move_only_stopped_handler>);
+
+  struct move_only_result
+  {
+    STDEXEC_ATTRIBUTE(host, device)
+    explicit move_only_result(int value) noexcept
+      : value_(value)
+    {}
+
+    STDEXEC_ATTRIBUTE(host, device)
+    move_only_result(move_only_result&& other) noexcept
+      : value_(other.value_)
+    {
+      other.value_ = 0;
+    }
+
+    move_only_result(move_only_result const &) = delete;
+
+    STDEXEC_ATTRIBUTE(host, device)
+    ~move_only_result() = default;
+
+    STDEXEC_ATTRIBUTE(host, device)
+    auto value() const noexcept -> int
+    {
+      return value_;
+    }
+
+   private:
+    int value_;
+  };
+
+  TEST_CASE("nvexec upon_stopped advertises CUDA launch errors",
+            "[cuda][stream][adaptors][upon_stopped]")
+  {
+    auto fun = []() noexcept {};
+    using sender_t = nvexec::_strm::upon_stopped_sender<a_sender_of<ex::set_stopped_t()>,
+                                                        decltype(fun)>;
+    sender_t snd{a_sender_of<ex::set_stopped_t()>{}, std::move(fun)};
+
+    check_err_types<ex::__mset<cudaError_t>>(snd);
+  }
 
   TEST_CASE("nvexec upon_stopped returns a sender", "[cuda][stream][adaptors][upon_stopped]")
   {
@@ -41,4 +101,28 @@ namespace
 
     REQUIRE(flags_storage.all_set_once());
   }
+
+  TEST_CASE("nvexec upon_stopped supports move-only function objects",
+            "[cuda][stream][adaptors][upon_stopped]")
+  {
+    nvexec::stream_context stream_ctx{};
+
+    auto snd = ex::just_stopped() | ex::continues_on(stream_ctx.get_scheduler())
+             | ex::upon_stopped(move_only_stopped_handler{});
+    auto const [result] = STDEXEC::sync_wait(std::move(snd)).value();
+
+    REQUIRE(result == 42);
+  }
+
+  TEST_CASE("nvexec upon_stopped moves its result", "[cuda][stream][adaptors][upon_stopped]")
+  {
+    nvexec::stream_context stream_ctx{};
+
+    auto snd = ex::just_stopped() | ex::continues_on(stream_ctx.get_scheduler())
+             | ex::upon_stopped([] { return move_only_result{42}; });
+
+    auto [result] = STDEXEC::sync_wait(std::move(snd)).value();
+
+    REQUIRE(result.value() == 42);
+  }
 }  // namespace