1 /*
2 * Copyright (c) 2021-2024 NVIDIA Corporation
3 *
4 * Licensed under the Apache License Version 2.0 with LLVM Exceptions
5 * (the "License"); you may not use this file except in compliance with
6 * the License. You may obtain a copy of the License at
7 *
8 * https://llvm.org/LICENSE.txt
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
15 */
16 #pragma once
17
18 #include "__execution_fwd.hpp"
19
20 // include these after __execution_fwd.hpp
21 #include "__concepts.hpp"
22 #include "__domain.hpp"
23 #include "__env.hpp"
24 #include "__meta.hpp"
25 #include "__operation_states.hpp"
26 #include "__schedulers.hpp"
27 #include "__senders_core.hpp"
28 #include "__transform_completion_signatures.hpp"
29 #include "__tuple.hpp"
30 #include "__variant.hpp"
31
32 namespace stdexec
33 {
34 /////////////////////////////////////////////////////////////////////////////
35 // [execution.senders.adaptors.schedule_from]
36 namespace __schfr
37 {
38 template <class... _Ts>
39 using __tuple_t = __tuple_for<__decay_t<_Ts>...>;
40
41 template <class... _Ts>
42 using __variant_t = __variant_for<__monostate, _Ts...>;
43
44 // Compute a variant type that is capable of storing the results of the
45 // input sender when it completes. The variant has type:
46 // variant<
47 // monostate,
48 // tuple<set_stopped_t>,
49 // tuple<set_value_t, __decay_t<_Values1>...>,
50 // tuple<set_value_t, __decay_t<_Values2>...>,
51 // ...
52 // tuple<set_error_t, __decay_t<_Error1>>,
53 // tuple<set_error_t, __decay_t<_Error2>>,
54 // ...
55 // >
56 template <class _CvrefSender, class _Env>
57 using __variant_for = //
58 __for_each_completion_signature<
59 __completion_signatures_of_t<_CvrefSender, _Env>, __tuple_t,
60 __munique<__qq<__variant_for>>::__f>;
61
62 template <class... _Values>
63 using __decay_value_sig = set_value_t (*)(__decay_t<_Values>...);
64
65 template <class _Error>
66 using __decay_error_sig = set_error_t (*)(__decay_t<_Error>);
67
68 template <class... _Ts>
69 using __all_nothrow_decay_copyable =
70 __mbool<(__nothrow_decay_copyable<_Ts> && ...)>;
71
72 template <class _CvrefSender, class... _Env>
73 using __all_nothrow_decay_copyable_results = //
74 __for_each_completion_signature<
75 __completion_signatures_of_t<_CvrefSender, _Env...>,
76 __all_nothrow_decay_copyable, __mand_t>;
77
78 template <class _Scheduler, class _CvrefSender, class... _Env>
79 using __completions_t = //
80 __mtry_q<__concat_completion_signatures>::__f<
81 __transform_completion_signatures<
82 __completion_signatures_of_t<_CvrefSender, _Env...>,
83 __decay_value_sig, __decay_error_sig, set_stopped_t (*)(),
84 __completion_signature_ptrs>,
85 transform_completion_signatures<
86 __completion_signatures_of_t<schedule_result_t<_Scheduler>,
87 _Env...>,
88 __eptr_completion_if_t<
89 __all_nothrow_decay_copyable_results<_CvrefSender, _Env...>>,
90 __mconst<completion_signatures<>>::__f>>;
91
92 template <class _SchedulerId>
93 struct __environ
94 {
95 using _Scheduler = stdexec::__t<_SchedulerId>;
96
97 struct __t
98 {
99 using __id = __environ;
100
101 _Scheduler __sched_;
102
103 template <__one_of<set_value_t, set_stopped_t> _Tag>
querystdexec::__schfr::__environ::__t104 auto query(get_completion_scheduler_t<_Tag>) const noexcept
105 {
106 return __sched_;
107 }
108
querystdexec::__schfr::__environ::__t109 auto query(get_domain_t) const noexcept
110 {
111 return query_or(get_domain, __sched_, default_domain());
112 }
113 };
114 };
115
116 template <class _Scheduler, class _Sexpr, class _Receiver>
117 struct __state;
118
119 template <class _State>
120 STDEXEC_ATTRIBUTE((always_inline))
__make_visitor_fn(_State * __state)121 auto __make_visitor_fn(_State* __state) noexcept
122 {
123 return [__state]<class _Tup>(_Tup& __tupl) noexcept -> void {
124 if constexpr (__same_as<_Tup, __monostate>)
125 {
126 std::terminate(); // reaching this indicates a bug in schedule_from
127 }
128 else
129 {
130 __tupl.apply(
131 [&]<class... _Args>(auto __tag,
132 _Args&... __args) noexcept -> void {
133 __tag(std::move(__state->__receiver()),
134 static_cast<_Args&&>(__args)...);
135 },
136 __tupl);
137 }
138 };
139 }
140
141 // This receiver is to be completed on the execution context associated with the
142 // scheduler. When the source sender completes, the completion information is
143 // saved off in the operation state so that when this receiver completes, it can
144 // read the completion out of the operation state and forward it to the output
145 // receiver after transitioning to the scheduler's context.
146 template <class _Scheduler, class _Sexpr, class _Receiver>
147 struct __receiver2
148 {
149 using receiver_concept = receiver_t;
150
set_valuestdexec::__schfr::__receiver2151 void set_value() noexcept
152 {
153 __state_->__data_.visit(__schfr::__make_visitor_fn(__state_),
154 __state_->__data_);
155 }
156
157 template <class _Error>
set_errorstdexec::__schfr::__receiver2158 void set_error(_Error&& __err) noexcept
159 {
160 stdexec::set_error(static_cast<_Receiver&&>(__state_->__receiver()),
161 static_cast<_Error&&>(__err));
162 }
163
set_stoppedstdexec::__schfr::__receiver2164 void set_stopped() noexcept
165 {
166 stdexec::set_stopped(static_cast<_Receiver&&>(__state_->__receiver()));
167 }
168
get_envstdexec::__schfr::__receiver2169 auto get_env() const noexcept -> env_of_t<_Receiver>
170 {
171 return stdexec::get_env(__state_->__receiver());
172 }
173
174 __state<_Scheduler, _Sexpr, _Receiver>* __state_;
175 };
176
177 template <class _Scheduler, class _Sexpr, class _Receiver>
178 struct __state : __enable_receiver_from_this<_Sexpr, _Receiver>, __immovable
179 {
180 using __variant_t = __variant_for<__child_of<_Sexpr>, env_of_t<_Receiver>>;
181 using __receiver2_t = __receiver2<_Scheduler, _Sexpr, _Receiver>;
182
183 __variant_t __data_;
184 connect_result_t<schedule_result_t<_Scheduler>, __receiver2_t> __state2_;
185 STDEXEC_APPLE_CLANG(__state* __self_;)
186
__statestdexec::__schfr::__state187 explicit __state(_Scheduler __sched) :
188 __data_(), __state2_(connect(schedule(__sched), __receiver2_t{this}))
189 STDEXEC_APPLE_CLANG(, __self_(this))
190 {}
191 };
192
193 struct schedule_from_t
194 {
195 template <scheduler _Scheduler, sender _Sender>
operator ()stdexec::__schfr::schedule_from_t196 auto operator()(_Scheduler&& __sched,
197 _Sender&& __sndr) const -> __well_formed_sender auto
198 {
199 using _Env = __t<__environ<__id<__decay_t<_Scheduler>>>>;
200 auto __env = _Env{{static_cast<_Scheduler&&>(__sched)}};
201 auto __domain = query_or(get_domain, __sched, default_domain());
202 return stdexec::transform_sender(
203 __domain, __make_sexpr<schedule_from_t>(
204 std::move(__env), static_cast<_Sender&&>(__sndr)));
205 }
206
207 using _Sender = __1;
208 using _Env = __0;
209 using __legacy_customizations_t = __types<tag_invoke_t(
210 schedule_from_t, get_completion_scheduler_t<set_value_t>(_Env&),
211 _Sender)>;
212 };
213
214 struct __schedule_from_impl : __sexpr_defaults
215 {
216 template <class _Sender>
217 using __scheduler_t =
218 __decay_t<__call_result_t<get_completion_scheduler_t<set_value_t>,
219 env_of_t<_Sender>>>;
220
221 static constexpr auto get_attrs = //
222 []<class _Data, class _Child>(const _Data& __data,
223 const _Child& __child) noexcept {
224 return __env::__join(__data, stdexec::get_env(__child));
225 };
226
227 static constexpr auto get_completion_signatures = //
228 []<class _Sender, class... _Env>(_Sender&&, _Env&&...) noexcept
229 -> __completions_t<__scheduler_t<_Sender>, __child_of<_Sender>,
230 _Env...> {
231 static_assert(sender_expr_for<_Sender, schedule_from_t>);
232 return {};
233 };
234
235 static constexpr auto get_state =
236 []<class _Sender, class _Receiver>(_Sender&& __sndr, _Receiver&) {
237 static_assert(sender_expr_for<_Sender, schedule_from_t>);
238 auto __sched =
239 get_completion_scheduler<set_value_t>(stdexec::get_env(__sndr));
240 using _Scheduler = decltype(__sched);
241 return __state<_Scheduler, _Sender, _Receiver>{__sched};
242 };
243
244 static constexpr auto complete = //
245 []<class _State, class _Receiver, class _Tag, class... _Args>(
246 __ignore, _State& __state, _Receiver& __rcvr, _Tag __tag,
247 _Args&&... __args) noexcept -> void {
248 STDEXEC_APPLE_CLANG(
249 __state.__self_ == &__state ? void() : std::terminate());
250 // Write the tag and the args into the operation state so that we can
251 // forward the completion from within the scheduler's execution context.
252 if constexpr (__nothrow_callable<__tup::__mktuple_t, _Tag, _Args...>)
253 {
254 __state.__data_.emplace_from(__tup::__mktuple, __tag,
255 static_cast<_Args&&>(__args)...);
256 }
257 else
258 {
259 try
260 {
261 __state.__data_.emplace_from(__tup::__mktuple, __tag,
262 static_cast<_Args&&>(__args)...);
263 }
264 catch (...)
265 {
266 stdexec::set_error(static_cast<_Receiver&&>(__rcvr),
267 std::current_exception());
268 return;
269 }
270 }
271
272 // Enqueue the schedule operation so the completion happens on the
273 // scheduler's execution context.
274 stdexec::start(__state.__state2_);
275 };
276 };
277 } // namespace __schfr
278
279 using __schfr::schedule_from_t;
280 inline constexpr schedule_from_t schedule_from{};
281
282 template <>
283 struct __sexpr_impl<schedule_from_t> : __schfr::__schedule_from_impl
284 {};
285 } // namespace stdexec
286