diff --git a/include/stdexec/__detail/__diagnostics.hpp b/include/stdexec/__detail/__diagnostics.hpp index 313bac7c3..b333ba2c8 100644 --- a/include/stdexec/__detail/__diagnostics.hpp +++ b/include/stdexec/__detail/__diagnostics.hpp @@ -16,7 +16,7 @@ #pragma once #include "__config.hpp" -#include "__diagnostic_macros.hpp" +#include "__diagnostic_macros.hpp" // IWYU pragma: export #if STDEXEC_USE_MODULES() && !defined(STDEXEC_IN_MODULE_PURVIEW) @@ -110,6 +110,9 @@ namespace STDEXEC struct _WITH_SCHEDULER_ {}; + struct _WITH_ALLOCATOR_ + {}; + STDEXEC_MODULE_EXPORT_AUTHORING struct _TO_FIX_THIS_ERROR_ {}; diff --git a/include/stdexec/__detail/__task.hpp b/include/stdexec/__detail/__task.hpp index 66e1b4305..7d741e0bf 100644 --- a/include/stdexec/__detail/__task.hpp +++ b/include/stdexec/__detail/__task.hpp @@ -65,6 +65,10 @@ namespace STDEXEC struct with_stopped {}; + struct _THE_CURRENT_ENVIRONMENT_IS_INCOMPATIBLE_WITH_THE_TASK_ENVIRONMENT_; + struct _THE_ALLOCATOR_IN_THE_ENVIRONMENT_IS_INCOMPATIBLE_WITH_THE_TASK_ALLOCATOR_; + struct _THE_START_SCHEDULER_IN_THE_ENVIRONMENT_IS_INCOMPATIBLE_WITH_THE_TASK_START_SCHEDULER_; + namespace __task { //////////////////////////////////////////////////////////////////////////////// @@ -230,14 +234,10 @@ namespace STDEXEC }; template - concept __has_compatible_scheduler = // - requires(_ParentEnv const & __parent_env, _Alloc const & __alloc) { // - _Scheduler(STDEXEC::get_start_scheduler(__parent_env), __alloc); // - } || // - requires(_ParentEnv const & __parent_env) { // - _Scheduler(STDEXEC::get_start_scheduler(__parent_env)); // - } || // - requires { _Scheduler{}; }; + concept __has_compatible_scheduler = + __has_scheduler_compatible_with<_ParentEnv, _Scheduler, _Alloc> + || __has_scheduler_compatible_with<_ParentEnv, _Scheduler> + || __std::default_initializable<_Scheduler>; template concept __has_compatible_environment_with = @@ -343,6 +343,36 @@ namespace STDEXEC return __attrs{}; } + template + static consteval auto get_completion_signatures() + { + if constexpr (__task::__has_compatible_environment_with<_Env, _TaskEnv>) + { + return __concat_completion_signatures_t< + completion_signatures<__single_value_sig_t<_Ty>, set_stopped_t()>, + error_types>{}; + } + else if constexpr (!__task::__has_compatible_allocator<_Env, allocator_type>) + { + return __throw_compile_time_error< + _WHAT_(_THE_CURRENT_ENVIRONMENT_IS_INCOMPATIBLE_WITH_THE_TASK_ENVIRONMENT_), + _WHY_(_THE_ALLOCATOR_IN_THE_ENVIRONMENT_IS_INCOMPATIBLE_WITH_THE_TASK_ALLOCATOR_), + _WITH_ALLOCATOR_(allocator_type), + _WITH_ENVIRONMENT_(_Env)>(); + } + else + { + static_assert( + !__task::__has_compatible_scheduler<_Env, start_scheduler_type, allocator_type>); + return __throw_compile_time_error< + _WHAT_(_THE_CURRENT_ENVIRONMENT_IS_INCOMPATIBLE_WITH_THE_TASK_ENVIRONMENT_), + _WHY_( + _THE_START_SCHEDULER_IN_THE_ENVIRONMENT_IS_INCOMPATIBLE_WITH_THE_TASK_START_SCHEDULER_), + _WITH_SCHEDULER_(start_scheduler_type), + _WITH_ENVIRONMENT_(_Env)>(); + } + } + // This transforms a task into an __awaiter that can perform symmetric transfer when // co_awaited. template @@ -353,6 +383,22 @@ namespace STDEXEC return __awaiter<_ParentPromise>(static_cast(*this), __parent); } + // Connecting a task to a receiver, like co_awaiting it, requires the receiver's + // environment to be compatible with the task's configuration (allocator, start + // scheduler, stop token, ...). Unlike co_awaiting a task — which reports errors + // by throwing them as exceptions at the await point — connecting a task to a + // receiver delivers the task's errors with their declared types directly to the + // receiver, rather than always delivering them as std::exception_ptr. (The + // completion signatures advertised by get_completion_signatures above describe + // exactly what the operation state returned from this member delivers.) + template + requires __task::__has_compatible_environment_with, _TaskEnv> + [[nodiscard]] + constexpr auto connect(_Receiver __rcvr) && noexcept + { + return __opstate<_Receiver>(static_cast(*this), static_cast<_Receiver&&>(__rcvr)); + } + private: using __on_stopped_t = __forward_stop_request; using __stop_variant_t = __variant; @@ -586,6 +632,126 @@ namespace STDEXEC _ParentPromise& __parent_; }; + // The operation state produced by connecting a task to a receiver. Like __awaiter, + // it drives the task's coroutine to completion; unlike __awaiter, it has no parent + // coroutine to symmetrically transfer control back to, so instead it completes the + // receiver directly. Because the task's errors are stored (typed) in the error + // variant below, they can be delivered to the receiver with their declared types + // instead of being converted to exceptions and caught as std::exception_ptr. + template + struct STDEXEC_ATTRIBUTE(empty_bases) __opstate final + : __own_env_box> + , __awaiter_base + , __stop_callback_box_t> + { + constexpr explicit __opstate(task&& __task, _Receiver&& __rcvr) + noexcept(__nothrow_move_constructible<_Receiver>) + : __opstate::__own_env_box{__mk_own_env(STDEXEC::get_env(__rcvr))} + , __awaiter_base(static_cast(__task), STDEXEC::get_env(__rcvr), this->__own_env_) + , __rcvr_(static_cast<_Receiver&&>(__rcvr)) + {} + + STDEXEC_IMMOVABLE(__opstate); + + void start() & noexcept + { + // Register a stop callback that forwards stop requests from the receiver's + // stop token to the task's stop source, then resume the task's coroutine. + auto& __task_promise = this->__handle().promise(); + __task_promise.__state_ = this; + STDEXEC_TRY + { + this->__register_callback(STDEXEC::get_env(__rcvr_), __task_promise.__stop_); + STDEXEC::__coroutine_resume_nothrow(this->__handle()); + } + STDEXEC_CATCH_ALL + { + if constexpr (__nothrow_callback_registration>) + { + __std::unreachable(); + } + else + { + // The stop callback is not known to construct without throwing, so it may + // throw. In that case the task's coroutine never starts: destroy it and + // report the failure to the receiver as an exception. + auto const __coro = std::exchange(this->__task_.__coro_, {}); + STDEXEC::__coroutine_destroy_nothrow(__coro); + STDEXEC::set_error(static_cast<_Receiver&&>(__rcvr_), std::current_exception()); + } + } + } + + [[nodiscard]] + auto __completed() noexcept -> __std::coroutine_handle<> final + { + // Destroy the stop callback before completing the receiver: + this->__reset_callback(); + if (this->__stopped_) + { + // The task completed with with_stopped: destroy the coroutine and report + // the stopped completion. + auto const __coro = std::exchange(this->__task_.__coro_, {}); + STDEXEC::__coroutine_destroy_nothrow(__coro); + STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_)); + } + else if (!this->__errors_.__is_valueless()) + { + // The task completed with an error. Destroy the coroutine and deliver the + // error with its declared type -- not as an std::exception_ptr: + auto const __coro = std::exchange(this->__task_.__coro_, {}); + STDEXEC::__coroutine_destroy_nothrow(__coro); + __visit(STDEXEC::set_error, + std::move(this->__errors_), + static_cast<_Receiver&&>(__rcvr_)); + } + else + { + // The task completed successfully. Move/copy the result out of the + // coroutine before destroying it: + auto const __coro = std::exchange(this->__task_.__coro_, {}); + auto& __promise = __coro.promise(); + if constexpr (std::is_void_v<_Ty>) + { + __promise.__result(); + STDEXEC::__coroutine_destroy_nothrow(__coro); + STDEXEC::set_value(static_cast<_Receiver&&>(__rcvr_)); + } + else if constexpr (std::is_reference_v<_Ty>) + { + // A reference-valued task does not own its result; the referent is + // required to outlive the task (just as for the value returned from + // await_resume). Copy the reference out of the coroutine before + // destroying it, then deliver it to the receiver: + _Ty& __value = __promise.__result(); + STDEXEC::__coroutine_destroy_nothrow(__coro); + STDEXEC::set_value(static_cast<_Receiver&&>(__rcvr_), static_cast<_Ty>(__value)); + } + else + { + auto __value = static_cast<_Ty&&>(__promise.__result()); + STDEXEC::__coroutine_destroy_nothrow(__coro); + STDEXEC::set_value(static_cast<_Receiver&&>(__rcvr_), std::move(__value)); + } + } + return __std::noop_coroutine(); + } + + [[nodiscard]] + auto __canceled() noexcept -> __std::coroutine_handle<> final + { + // The task was stopped while awaiting a child operation. Destroy the + // coroutine and report the stopped completion to the receiver. + this->__reset_callback(); + auto const __coro = std::exchange(this->__task_.__coro_, {}); + STDEXEC::__coroutine_destroy_nothrow(__coro); + STDEXEC::set_stopped(static_cast<_Receiver&&>(__rcvr_)); + return __std::noop_coroutine(); + } + + _Receiver __rcvr_; + }; + struct __attrs { template @@ -786,8 +952,17 @@ namespace STDEXEC private: template friend struct __awaiter; + template + friend struct __opstate; friend struct __awaiter_base; + // On MSVC prior to 14.50, the compiler stores the coroutine handle returned + // from await_suspend in the suspended coroutine's frame, so when a connected + // task's __opstate::__completed destroys that frame before await_suspend + // returns, symmetric transfer would resume a use-after-free. See + // https://developercommunity.visualstudio.com/t/Incorrect-code-generation-for-symmetric-/1659260 + // Resume the continuation directly instead: a plain nested resume rather than + // a tail call, at the cost of stack growth in deeply chained tasks. struct __completed_awaiter { static constexpr bool await_ready() noexcept @@ -795,10 +970,14 @@ namespace STDEXEC return false; } - static constexpr auto await_suspend(__std::coroutine_handle<__promise> __coro) noexcept // - -> __std::coroutine_handle<> + static constexpr auto await_suspend(__std::coroutine_handle<__promise> __coro) noexcept { - return __coro.promise().__state_->__completed(); + __std::coroutine_handle<> const __continuation = __coro.promise().__state_->__completed(); +# ifdef STDEXEC_MSVC_CORO_DESTROY_BUG_WORKAROUND + __continuation.resume(); +# else + return __continuation; +# endif } static constexpr void await_resume() noexcept {} diff --git a/test/stdexec/types/test_task.cpp b/test/stdexec/types/test_task.cpp index 71a9b2558..4df68c748 100644 --- a/test/stdexec/types/test_task.cpp +++ b/test/stdexec/types/test_task.cpp @@ -33,7 +33,10 @@ import std; # else # include +# include +# include # include +# include # endif namespace ex = STDEXEC; @@ -365,6 +368,145 @@ namespace # endif # endif + // Regression test for NVIDIA/stdexec#2222: when a task is given an environment + // that declares custom error_types, the task's completion signatures -- and the + // errors it delivers when connected to a receiver -- must use those types + // rather than always reporting/delivering std::exception_ptr. + struct error_code_env + { + using error_types = ex::completion_signatures; + }; + + auto test_task_yields_error_code() noexcept -> ex::task + { + co_yield ex::with_error{std::make_error_code(std::errc::invalid_argument)}; + co_return 1; + } + +# if !STDEXEC_NO_STDCPP_EXCEPTIONS() + TEST_CASE("task's completion signatures and errors honor custom error_types", "[types][task]") + { + // This is the repro from issue 2222: it only compiles (and produces the + // error_code value at runtime) if the task's error is reported and delivered + // as std::error_code, not as std::exception_ptr: + auto s = test_task_yields_error_code() | ex::upon_error([](auto err) noexcept { return err; }) + | ex::into_variant(); + auto [r] = ex::sync_wait(std::move(s)).value(); + CHECK(std::holds_alternative>(r)); + CHECK(std::get>(r) + == std::make_tuple(std::make_error_code(std::errc::invalid_argument))); + } +# endif + + // A receiver that can be connected to a task whose environment declares + // set_error_t(std::error_code): + struct task_connect_env + { + ex::run_loop *__loop_; + + template < + ex::__one_of + _Query> + [[nodiscard]] + constexpr auto query(_Query) const noexcept -> ex::run_loop::scheduler + { + return __loop_->get_scheduler(); + } + }; + + struct error_code_task_receiver + { + using receiver_concept = ex::receiver_t; + + struct completion + { + enum class kind + { + none, + value, + error_code, + exception_ptr, + stopped + }; + + kind __kind_ = kind::none; + int __value_ = 0; + std::error_code __error_ = {}; + }; + + void set_value(int __value) noexcept + { + __completion_->__kind_ = completion::kind::value; + __completion_->__value_ = __value; + } + + void set_error(std::error_code __error) noexcept + { + __completion_->__kind_ = completion::kind::error_code; + __completion_->__error_ = __error; + } + + void set_error(std::exception_ptr) noexcept + { + __completion_->__kind_ = completion::kind::exception_ptr; + } + + void set_stopped() noexcept + { + __completion_->__kind_ = completion::kind::stopped; + } + + [[nodiscard]] + constexpr auto get_env() const noexcept -> task_connect_env + { + return {__loop_}; + } + + completion *__completion_; + ex::run_loop *__loop_; + }; + + auto test_task_connect_value() noexcept -> ex::task + { + co_return 42; + } + + auto test_task_connect_stopped() noexcept -> ex::task + { + co_yield ex::with_stopped(); + co_return 1; + } + + TEST_CASE("connecting a task delivers its declared error types to the receiver", "[types][task]") + { + ex::run_loop loop; + + { + error_code_task_receiver::completion completion; + auto op = ex::connect(test_task_connect_value(), + error_code_task_receiver{&completion, &loop}); + ex::start(op); + CHECK(completion.__kind_ == error_code_task_receiver::completion::kind::value); + CHECK(completion.__value_ == 42); + } + { + // The error must be delivered as std::error_code, not std::exception_ptr: + error_code_task_receiver::completion completion; + auto op = ex::connect(test_task_yields_error_code(), + error_code_task_receiver{&completion, &loop}); + ex::start(op); + CHECK(completion.__kind_ == error_code_task_receiver::completion::kind::error_code); + CHECK(completion.__error_ == std::make_error_code(std::errc::invalid_argument)); + } + { + error_code_task_receiver::completion completion; + auto op = ex::connect(test_task_connect_stopped(), + error_code_task_receiver{&completion, &loop}); + ex::start(op); + CHECK(completion.__kind_ == error_code_task_receiver::completion::kind::stopped); + } + } + struct error_as_value { constexpr error_as_value(ex::with_error error) noexcept @@ -473,9 +615,13 @@ namespace // In debug GCC builds, this test can cause a stack overflow due to // https://gcc.gnu.org/bugzilla/show_bug.cgi?id=94794, results in a symmetric - // transfer failing to be a tail call. + // transfer failing to be a tail call. Likewise, when + // STDEXEC_MSVC_CORO_DESTROY_BUG_WORKAROUND is defined (MSVC prior to 14.50), + // task's final suspend resumes its continuation directly instead of performing + // a symmetric transfer, which grows the stack with each nested task completion. # if !STDEXEC_GCC() \ || (defined(__OPTIMIZE__) && !defined(__SANITIZE_ADDRESS__) && !defined(__SANITIZE_THREAD__)) +# if !defined(STDEXEC_MSVC_CORO_DESTROY_BUG_WORKAROUND) auto sync() -> ex::task { co_return 42; @@ -509,6 +655,7 @@ namespace auto [i] = ex::sync_wait(std::move(t)).value(); CHECK(i == 84'000'042); } +# endif // !defined(STDEXEC_MSVC_CORO_DESTROY_BUG_WORKAROUND) # endif struct my_env