Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 12 additions & 7 deletions include/nvexec/stream/upon_stopped.cuh
Original file line number Diff line number Diff line change
Expand Up @@ -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
{
Expand All @@ -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>>;
Expand Down
53 changes: 53 additions & 0 deletions test/nvexec/upon_stopped.cpp
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
#include <stdexec/execution.hpp>
#include <test_common/catch2.hpp>
#include <test_common/senders.hpp>
#include <test_common/type_helpers.hpp>

#include "common.cuh"
#include "nvexec/stream_context.cuh"
Expand All @@ -10,6 +12,45 @@ using nvexec::is_on_gpu;

namespace
{
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]")
{
Expand Down Expand Up @@ -41,4 +82,16 @@ namespace

REQUIRE(flags_storage.all_set_once());
}

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
Loading