diff --git a/include/nvexec/stream/common.cuh b/include/nvexec/stream/common.cuh index 6d4332814..14ccab1d4 100644 --- a/include/nvexec/stream/common.cuh +++ b/include/nvexec/stream/common.cuh @@ -429,10 +429,11 @@ namespace nv::execution using decayed_tuple_t = ::cuda::std::tuple<__decay_t...>; template - using variant_storage_t = - __for_each_completion_signature_t<__completion_signatures_of_t, - decayed_tuple_t, - __munique<__q<_nullable_variant_t>>::__f>; + using variant_storage_t = __for_each_completion_signature_t< + __completion_signatures_of_t, + decayed_tuple_t, + __munique< + __mbind_back_q<_nullable_variant_t, ::cuda::std::tuple>>::__f>; inline constexpr get_stream_provider_t get_stream_provider{}; diff --git a/include/nvexec/stream/continues_on.cuh b/include/nvexec/stream/continues_on.cuh index 2836312db..394749c26 100644 --- a/include/nvexec/stream/continues_on.cuh +++ b/include/nvexec/stream/continues_on.cuh @@ -239,9 +239,10 @@ namespace nv::execution::_strm template <__decay_copyable Self, class... Env> static consteval auto get_completion_signatures() - -> __completion_signatures_of_t<__copy_cvref_t, Env...> { - return {}; + return exec::concat_completion_signatures( + STDEXEC::get_completion_signatures<__copy_cvref_t, Env...>(), + STDEXEC::completion_signatures()); } [[nodiscard]] diff --git a/include/stdexec/__detail/__completion_behavior.hpp b/include/stdexec/__detail/__completion_behavior.hpp index c29f2ac07..479dff4d1 100644 --- a/include/stdexec/__detail/__completion_behavior.hpp +++ b/include/stdexec/__detail/__completion_behavior.hpp @@ -134,11 +134,17 @@ namespace STDEXEC struct __common_t { template <__behavior... _CSs> - requires(sizeof...(_CSs) > 0) STDEXEC_ATTRIBUTE(nodiscard, host, device) constexpr auto operator()(__constant_t<_CSs>... __cbs) const noexcept { - return (__cbs | ...); + if constexpr (sizeof...(_CSs) == 0) + { + return __completion_behavior::__unknown; + } + else + { + return (__cbs | ...); + } } }; diff --git a/include/stdexec/__detail/__let.hpp b/include/stdexec/__detail/__let.hpp index 509fe3a8b..41587142c 100644 --- a/include/stdexec/__detail/__let.hpp +++ b/include/stdexec/__detail/__let.hpp @@ -165,9 +165,10 @@ namespace STDEXEC using __sender_t = __invoke_result_t<_Fun, __decay_t<_Args>&...>; using __submit_t = __submit_result<__sender_t, __rcvr_env<_Receiver, _Env2>>; - constexpr bool __nothrow_store = (__nothrow_decay_copyable<_Args> && ...); - constexpr bool __nothrow_invoke = __nothrow_invocable<_Fun, __decay_t<_Args>&...>; - constexpr bool __nothrow_submit = + STDEXEC_CONSTEXPR_LOCAL bool __nothrow_store = (__nothrow_decay_copyable<_Args> && ...); + STDEXEC_CONSTEXPR_LOCAL bool __nothrow_invoke = + __nothrow_invocable<_Fun, __decay_t<_Args>&...>; + STDEXEC_CONSTEXPR_LOCAL bool __nothrow_submit = __nothrow_constructible_from<__submit_t, __sender_t, __second_rcvr_t>; STDEXEC_TRY @@ -315,172 +316,185 @@ namespace STDEXEC __op_state_variant_t __storage_{__no_init}; }; - // // The set_value completions of: - // // - // // * a let_value sender are: - // // * the value completions of the secondary senders - // // - // // * a let_error sender are: - // // * the value completions of the predecessor sender - // // * the value completions of the secondary senders - // // - // // * a let_stopped sender are: - // // * the value completions of the predecessor sender - // // * the value completions of the secondary sender - // // - // // The set_error completions of: - // // - // // * a let_value sender are: - // // * the error completions of the predecessor sender - // // * the error completions of the secondary senders - // // * the value completions of the predecessor sender if decay copying the arguments can throw - // // - // // * a let_error sender are: - // // * the error completions of the secondary senders - // // * the error completions of the predecessor sender if decay copying the errors can throw - // // - // // * a let_stopped sender are: - // // * the error completions of the predecessor sender - // // * the error completions of the secondary senders - // // - // // The set_stopped completions of: - // // - // // * a let_value sender are: - // // * the stopped completions of the predecessor sender - // // * the stopped completions of the secondary senders - // // - // // * a let_error sender are: - // // * the stopped completions of the predecessor sender - // // * the stopped completions of the secondary senders - // // - // // * a let_stopped sender are: - // // * the stopped completions of the secondary sender - // // - // template - // struct __result_completion_behavior_fn - // { - // template - // [[nodiscard]] - // static constexpr auto __impl() noexcept - // { - // using __sndr_t = - // __minvoke_or_q<__invoke_result_t, __not_a_sender<>, _Fun, __decay_t<_Ts>&...>; - // return STDEXEC::__get_completion_behavior<_SetTag, __sndr_t, _JoinEnv2...>(); - // } - - // template - // using __f = decltype(__impl<_Ts...>()); - // }; - - // template - // struct __domain_transform_fn - // { - // template - // using __f = __completion_domain_of_t<_SetTag, - // __invoke_result_t<_Fun, __decay_t<_As>&...>, - // __result_env_t<_SetTag, _Sender, _Env>...>; - // }; - - // //! @tparam _LetTag The tag type for the let_ operation. - // //! @tparam _SetTag The completion signal of the let_ sender itself that is being - // //! queried. For example, you may be querying a let_value sender for its set_error - // //! completion domain. - // template - // [[nodiscard]] - // consteval auto __get_completion_domain() noexcept - // { - // if constexpr (sender_in<_Sndr, _Env...>) - // { - // using __domain_transform_fn = __let::__domain_transform_fn<_SetTag, _Fun, _Sndr, _Env...>; - // return __minvoke_or_q<__gather_completion_signatures_t, - // indeterminate_domain<>, - // __t<_LetTag>, - // __completion_signatures_of_t<_Sndr, _Env...>, - // __domain_transform_fn, - // __qq<__common_domain_t>>(); - // } - // else - // { - // return indeterminate_domain<>{}; - // } - // } - - // template - // using __let_completion_domain_t = __unless_one_of_t< - // __result_of<__let::__get_completion_domain<_LetTag, _SetTag, _Sndr, _Fun, _Env...>>, - // indeterminate_domain<>>; - - // template - // struct __attrs - // { - // using __set_tag_t = STDEXEC::__t<_LetTag>; - - // template - // constexpr auto query(get_completion_scheduler_t<_Tag>) const = delete; - - // template - // [[nodiscard]] - // constexpr auto query(get_completion_domain_t<__set_tag_t>, _Env const &...) const noexcept - // -> __ensure_valid_domain_t< - // __let_completion_domain_t<_LetTag, __set_tag_t, _Sndr, _Fun, _Env...>> - // { - // return {}; - // } - - // template <__one_of _Tag, class... _Env> - // requires(__has_nothrow_completions_t<__set_tag_t, _Sndr, _Fun, _Env>::value && ...) - // [[nodiscard]] - // constexpr auto query(get_completion_domain_t<_Tag>, _Env const &...) const noexcept - // -> __ensure_valid_domain_t< - // __common_domain_t<__completion_domain_of_t<_Tag, _Sndr, __fwd_env_t<_Env>...>, - // __let_completion_domain_t<_LetTag, _Tag, _Sndr, _Fun, _Env...>>> - // { - // return {}; - // } - - // template - // requires(!__has_nothrow_completions_t<__set_tag_t, _Sndr, _Fun, _Env>::value) - // [[nodiscard]] - // constexpr auto query(get_completion_domain_t, _Env const &) const noexcept - // -> __ensure_valid_domain_t< - // __common_domain_t<__completion_domain_of_t<__set_tag_t, _Sndr, __fwd_env_t<_Env>>, - // __completion_domain_of_t>, - // __let_completion_domain_t<_LetTag, set_error_t, _Sndr, _Fun, _Env>>> - // { - // return {}; - // } - - // template - // [[nodiscard]] - // constexpr auto query(__get_completion_behavior_t<__set_tag_t>, _Env const &...) const noexcept - // { - // if constexpr (sender_in<_Sndr, __fwd_env_t<_Env>...>) - // { - // // The completion behavior of let_value(sndr, fn) is the union of the completion - // // behavior of sndr and all the senders that fn can potentially produce. (MSVC - // // needs the constexpr computation broken up, hence the local variables.) - // using __transform_fn = - // __result_completion_behavior_fn<__set_tag_t, - // _Fun, - // __result_env_t<__set_tag_t, _Sndr, _Env>...>; - // using __completions_t = __completion_signatures_of_t<_Sndr, __fwd_env_t<_Env>...>; - - // constexpr auto __pred_behavior = - // STDEXEC::__get_completion_behavior<__set_tag_t, _Sndr, __fwd_env_t<_Env>...>(); - // constexpr auto __result_behaviors = __gather_completion_signatures_t< - // __set_tag_t, - // __completions_t, - // __transform_fn, - // __mbind_front_q<__call_result_t, __completion_behavior::__common_t>>(); - - // return __pred_behavior | __result_behaviors; - // } - // else - // { - // return __completion_behavior::__unknown; - // } - // } - // }; + // The set_value completions of: + // + // * a let_value sender are: + // * the value completions of the secondary senders + // + // * a let_error sender are: + // * the value completions of the predecessor sender + // * the value completions of the secondary senders + // + // * a let_stopped sender are: + // * the value completions of the predecessor sender + // * the value completions of the secondary sender + // + // The set_error completions of: + // + // * a let_value sender are: + // * the error completions of the predecessor sender + // * the error completions of the secondary senders + // * the value completions of the predecessor sender if decay copying the arguments can throw + // + // * a let_error sender are: + // * the error completions of the secondary senders + // * the error completions of the predecessor sender if decay copying the errors can throw + // + // * a let_stopped sender are: + // * the error completions of the predecessor sender + // * the error completions of the secondary senders + // + // The set_stopped completions of: + // + // * a let_value sender are: + // * the stopped completions of the predecessor sender + // * the stopped completions of the secondary senders + // + // * a let_error sender are: + // * the stopped completions of the predecessor sender + // * the stopped completions of the secondary senders + // + // * a let_stopped sender are: + // * the stopped completions of the secondary sender + // + template + struct __result_completion_behavior_fn + { + template + [[nodiscard]] + static constexpr auto __impl() noexcept + { + using __sndr_t = + __minvoke_or_q<__invoke_result_t, __not_a_sender<>, _Fun, __decay_t<_Ts>&...>; + return STDEXEC::__get_completion_behavior<_SetTag, __sndr_t, _JoinEnv2...>(); + } + + template + using __f = decltype(__impl<_Ts...>()); + }; + + template + struct __domain_transform_fn + { + template + using __f = __completion_domain_of_t<_SetTag, + __invoke_result_t<_Fun, __decay_t<_As>&...>, + __result_env_t<_SetTag, _Sender, _Env>...>; + }; + + //! @tparam _LetTag The tag type for the let_ operation. + //! @tparam _SetTag The completion signal of the let_ sender itself that is being + //! queried. For example, you may be querying a let_value sender for its set_error + //! completion domain. + template + [[nodiscard]] + consteval auto __get_completion_domain() noexcept + { + if constexpr (sender_in<_Sndr, _Env...>) + { + using __domain_transform_fn = __let::__domain_transform_fn<_SetTag, _Fun, _Sndr, _Env...>; + return __minvoke_or_q<__gather_completion_signatures_t, + indeterminate_domain<>, + __t<_LetTag>, + __completion_signatures_of_t<_Sndr, _Env...>, + __domain_transform_fn, + __qq<__common_domain_t>>(); + } + else + { + return indeterminate_domain<>{}; + } + } + + template + using __let_completion_domain_t = __unless_one_of_t< + decltype(__let::__get_completion_domain<_LetTag, _SetTag, _Sndr, _Fun, _Env...>()), + indeterminate_domain<>>; + + template + struct __attrs + { + using __set_tag_t = STDEXEC::__t<_LetTag>; + + template + constexpr auto query(get_completion_scheduler_t<_Tag>) const = delete; + + template + [[nodiscard]] + constexpr auto query(get_completion_domain_t<__set_tag_t>, _Env const &...) const noexcept + -> __ensure_valid_domain_t< + __let_completion_domain_t<_LetTag, __set_tag_t, _Sndr, _Fun, _Env...>> + { + return {}; + } + + template <__one_of _Tag, class... _Env> + requires(__has_nothrow_completions_t<__set_tag_t, _Sndr, _Fun, _Env>::value && ...) + [[nodiscard]] + constexpr auto query(get_completion_domain_t<_Tag>, _Env const &...) const noexcept + -> __ensure_valid_domain_t< + __common_domain_t<__completion_domain_of_t<_Tag, _Sndr, __fwd_env_t<_Env>...>, + __let_completion_domain_t<_LetTag, _Tag, _Sndr, _Fun, _Env...>>> + { + return {}; + } + + template + requires(!__has_nothrow_completions_t<__set_tag_t, _Sndr, _Fun, _Env>::value) + [[nodiscard]] + constexpr auto query(get_completion_domain_t, _Env const &) const noexcept + -> __ensure_valid_domain_t< + __common_domain_t<__completion_domain_of_t<__set_tag_t, _Sndr, __fwd_env_t<_Env>>, + __completion_domain_of_t>, + __let_completion_domain_t<_LetTag, set_error_t, _Sndr, _Fun, _Env>>> + { + return {}; + } + + // This overload is needed so that let_stopped will look for customizations based on + // the set_stopped completion domain of the predecessor sender, not the set_value + // completion domain of the predecessor sender. Likewise for let_error and the + // set_error completion domain of the predecessor sender. + template + requires __callable, env_of_t<_Sndr>, _Env const &...> + [[nodiscard]] + constexpr auto query(get_completion_domain_t<>, _Env const &...) const noexcept + -> __call_result_t, env_of_t<_Sndr>, _Env const &...> + { + return {}; + } + + template + [[nodiscard]] + constexpr auto query(__get_completion_behavior_t<__set_tag_t>, _Env const &...) const noexcept + { + if constexpr (sender_in<_Sndr, __fwd_env_t<_Env>...>) + { + // The completion behavior of let_value(sndr, fn) is the union of the completion + // behavior of sndr and all the senders that fn can potentially produce. (MSVC + // needs the constexpr computation broken up, hence the local variables.) + using __transform_fn = + __result_completion_behavior_fn<__set_tag_t, + _Fun, + __result_env_t<__set_tag_t, _Sndr, _Env>...>; + using __completions_t = __completion_signatures_of_t<_Sndr, __fwd_env_t<_Env>...>; + + STDEXEC_CONSTEXPR_LOCAL auto __pred_behavior = + STDEXEC::__get_completion_behavior<__set_tag_t, _Sndr, __fwd_env_t<_Env>...>(); + STDEXEC_CONSTEXPR_LOCAL auto __result_behaviors = __gather_completion_signatures_t< + __set_tag_t, + __completions_t, + __transform_fn, + __mbind_front_q<__call_result_t, __completion_behavior::__common_t>>(); + + return __pred_behavior | __result_behaviors; + } + else + { + return __completion_behavior::__unknown; + } + } + }; //! Implementation of the `let_*_t` types, where `_SetTag` is, e.g., //! `set_value_t` for `let_value`. @@ -590,11 +604,10 @@ namespace STDEXEC } public: - static constexpr auto __get_attrs = - [](__ignore, __ignore, _Child const & __child) noexcept -> decltype(auto) + static constexpr auto __get_attrs = // + [](__ignore, _Fun const &, _Child const &) noexcept // { - // TODO(ericniebler): this needs a proper implementation - return __fwd_env(STDEXEC::get_env(__child)); + return __let::__attrs<_LetTag, _Child, _Fun>(); }; template diff --git a/test/nvexec/bulk.cpp b/test/nvexec/bulk.cpp index 35f5992f5..e5db94607 100644 --- a/test/nvexec/bulk.cpp +++ b/test/nvexec/bulk.cpp @@ -1,8 +1,12 @@ #include + +#include + +#include + #include #include "common.cuh" -#include "nvexec/stream_context.cuh" #include @@ -12,6 +16,21 @@ using nvexec::is_on_gpu; namespace { + using _attrs_base_t = + ex::env, nvexec::stream_scheduler>, + ex::prop, nvexec::stream_domain>>; + + struct _attrs : _attrs_base_t + {}; + + template + auto _proclaim_completes_on(nvexec::stream_scheduler sch) + { + return exec::write_attrs(_attrs{ + ex::env{ex::prop{ex::get_completion_scheduler, sch}, + ex::prop{ex::get_completion_domain, nvexec::stream_domain{}}} + }); + } TEST_CASE("nvexec bulk returns a sender", "[cuda][stream][adaptors][bulk]") { @@ -227,6 +246,8 @@ namespace [](std::size_t i, cuda::std::span out) { out[i] = (double) i; }) | STDEXEC::let_value([](cuda::std::span out) { return STDEXEC::just(out); }) + // proclaim that value completions will happen on the specified scheduler: + | ::_proclaim_completes_on(ctx.get_scheduler()) | STDEXEC::bulk(ex::par, nelems, [](std::size_t i, cuda::std::span out) diff --git a/test/nvexec/continues_on.cpp b/test/nvexec/continues_on.cpp index 69fdca48c..5f3fa718d 100644 --- a/test/nvexec/continues_on.cpp +++ b/test/nvexec/continues_on.cpp @@ -312,10 +312,13 @@ namespace template void record_streams_across(FirstSched first, SecondSched second, cudaStream_t* streams) { - auto sndr = STDEXEC::schedule(first) | STDEXEC::let_value([] { return nvexec::get_stream(); }) + auto sndr = STDEXEC::starts_on(first, nvexec::get_stream()) | STDEXEC::then([streams](cudaStream_t stream) { streams[0] = stream; }) | STDEXEC::continues_on(second) | STDEXEC::let_value([] { return nvexec::get_stream(); }) + // proclaim that value completions will happen on the specified scheduler: + | exec::write_attrs( + STDEXEC::prop{STDEXEC::get_completion_scheduler, second}) | STDEXEC::then([streams](cudaStream_t stream) { streams[1] = stream; }); STDEXEC::sync_wait(std::move(sndr)); diff --git a/test/nvexec/let_error.cpp b/test/nvexec/let_error.cpp index 22a937685..708650630 100644 --- a/test/nvexec/let_error.cpp +++ b/test/nvexec/let_error.cpp @@ -1,9 +1,12 @@ -#include #include + +#include + +#include + #include #include "common.cuh" -#include "nvexec/stream_context.cuh" namespace ex = STDEXEC; @@ -11,6 +14,21 @@ using nvexec::is_on_gpu; namespace { + using _attrs_base_t = + ex::env, nvexec::stream_scheduler>, + ex::prop, nvexec::stream_domain>>; + + struct _attrs : _attrs_base_t + {}; + + template + auto _proclaim_completes_on(nvexec::stream_scheduler sch) + { + return exec::write_attrs(_attrs{ + ex::env{ex::prop{ex::get_completion_scheduler, sch}, + ex::prop{ex::get_completion_domain, nvexec::stream_domain{}}} + }); + } TEST_CASE("nvexec let_error returns a sender", "[cuda][stream][adaptors][let_error]") { @@ -38,10 +56,10 @@ namespace flags.set(); } - return ex::just() - | exec::write_attrs(ex::prop{ex::get_domain, nvexec::stream_domain()}); - }); - STDEXEC::sync_wait(std::move(snd)); + return ex::just(); + }) + | ::_proclaim_completes_on(stream_ctx.get_scheduler()); + ex::sync_wait(std::move(snd)); REQUIRE(flags_storage.all_set_once()); } @@ -56,16 +74,16 @@ namespace auto snd = ex::just_error(42) | ex::continues_on(stream_ctx.get_scheduler()) | ex::let_error( - [flags](int err) + [flags, sch = stream_ctx.get_scheduler()](int err) { if (is_on_gpu() && err == 42) { flags.set(0); } - return ex::just() - | exec::write_attrs(ex::prop{ex::get_domain, nvexec::stream_domain()}); + return ex::just(); }) + | ::_proclaim_completes_on(stream_ctx.get_scheduler()) | a_sender( [flags] { @@ -74,7 +92,7 @@ namespace flags.set(1); } }); - STDEXEC::sync_wait(std::move(snd)); + ex::sync_wait(std::move(snd)); REQUIRE(flags_storage.all_set_once()); } @@ -98,7 +116,7 @@ namespace return ex::schedule(sch); }); - STDEXEC::sync_wait(std::move(snd)); + ex::sync_wait(std::move(snd)); REQUIRE(flags_storage.all_set_once()); } diff --git a/test/nvexec/let_value.cpp b/test/nvexec/let_value.cpp index a9345f636..f377dd4d6 100644 --- a/test/nvexec/let_value.cpp +++ b/test/nvexec/let_value.cpp @@ -165,7 +165,7 @@ namespace flags_storage_t flags_storage{}; auto flags = flags_storage.get(); - auto snd = ex::schedule(sch) | ex::let_value([] { return nvexec::get_stream(); }) + auto snd = ex::starts_on(sch, nvexec::get_stream()) | ex::then( [flags](cudaStream_t stream) { diff --git a/test/stdexec/algos/adaptors/test_let_error.cpp b/test/stdexec/algos/adaptors/test_let_error.cpp index 494c28cb3..c7d6edf23 100644 --- a/test/stdexec/algos/adaptors/test_let_error.cpp +++ b/test/stdexec/algos/adaptors/test_let_error.cpp @@ -15,6 +15,7 @@ */ #include +#define STDEXEC_PARALLEL_SCHEDULER_HEADER_ONLY 1 #include #include @@ -403,11 +404,12 @@ namespace TEST_CASE("let_error can be customized", "[adaptors][let_error]") { - basic_inline_scheduler sched{}; + auto attrs = ex::prop{ex::get_completion_domain, let_error_test_domain{}}; - // The customization will return a different value - auto snd = ex::just(std::string{"hello"}) - | ex::let_error([](std::exception_ptr) { return ex::just(std::string{"err"}); }); - wait_for_value(ex::starts_on(sched, std::move(snd)), std::string{"what error?"}); + // The customization will return a different stopped + auto snd = ex::schedule(ex::get_parallel_scheduler()) // + | exec::write_attrs(attrs) // + | ex::let_error([](std::exception_ptr) { return ex::just(std::string{"stopped"}); }); + wait_for_value(std::move(snd), std::string{"what error?"}); } } // namespace diff --git a/test/stdexec/algos/adaptors/test_let_stopped.cpp b/test/stdexec/algos/adaptors/test_let_stopped.cpp index 33b3988ad..f1dbcc27e 100644 --- a/test/stdexec/algos/adaptors/test_let_stopped.cpp +++ b/test/stdexec/algos/adaptors/test_let_stopped.cpp @@ -15,6 +15,7 @@ */ #include +#define STDEXEC_PARALLEL_SCHEDULER_HEADER_ONLY 1 #include #include @@ -232,10 +233,11 @@ namespace TEST_CASE("let_stopped can be customized", "[adaptors][let_stopped]") { - basic_inline_scheduler sched; + auto attrs = ex::prop{ex::get_completion_domain, let_stopped_test_domain{}}; // The customization will return a different stopped - auto snd = ex::just(std::string{"hello"}) | ex::continues_on(sched) + auto snd = ex::schedule(ex::get_parallel_scheduler()) // + | exec::write_attrs(attrs) // | ex::let_stopped([] { return ex::just(std::string{"stopped"}); }); wait_for_value(std::move(snd), std::string{"Don't stop me now"}); } diff --git a/test/stdexec/queries/test_get_completion_behavior.cpp b/test/stdexec/queries/test_get_completion_behavior.cpp index bd0a2c360..2c34ce41f 100644 --- a/test/stdexec/queries/test_get_completion_behavior.cpp +++ b/test/stdexec/queries/test_get_completion_behavior.cpp @@ -95,7 +95,7 @@ inline constexpr auto __reschedule_on_current = ex::get_scheduler() | ex::let_va // naming it fails the modular and nvc++ configurations. static_assert(ex::__completes_inline, - ex::env<>>); + ex::prop>); static_assert(!ex::__completes_inline, - ex::env<>>); + ex::prop>); diff --git a/test/test_common/schedulers.hpp b/test/test_common/schedulers.hpp index 2f67e6651..0843e2c22 100644 --- a/test/test_common/schedulers.hpp +++ b/test/test_common/schedulers.hpp @@ -45,7 +45,7 @@ namespace template struct sched_attrs { - sched_attrs(Scheduler sched, Tags...) + explicit sched_attrs(Scheduler sched, Tags...) : scheduler_(std::move(sched)) {}