Fix nvexec upon_stopped move-only callable forwarding (#2193)

* Fix nvexec upon_stopped callable forwarding

* clang-format

---------

Co-authored-by: Eric Niebler <eniebler@nvidia.com>

Author: fallintoplace

include/nvexec/stream/upon_stopped.cuh

@@ -160,7 +160,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

@@ -3,6 +3,8 @@
 #include <test_common/senders.hpp>
 #include <test_common/type_helpers.hpp>
 
+#include <type_traits>
+
 #include "common.cuh"
 #include "nvexec/stream_context.cuh"
 
@@ -12,6 +14,23 @@ 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)
@@ -20,7 +39,7 @@ namespace
     {}
 
     STDEXEC_ATTRIBUTE(host, device)
-    move_only_result(move_only_result&& other) noexcept
+    move_only_result(move_only_result &&other) noexcept
       : value_(other.value_)
     {
       other.value_ = 0;
@@ -83,6 +102,18 @@ 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{};