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