1//
2// Copyright (c) 2022 Klemens Morgenstern (klemens.morgenstern@gmx.net)
3//
4// Distributed under the Boost Software License, Version 1.0. (See accompanying
5// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6//
7
8#ifndef BOOST_COBALT_DETAIL_TASK_HPP
9#define BOOST_COBALT_DETAIL_TASK_HPP
10
11#include <boost/cobalt/detail/exception.hpp>
12#include <boost/cobalt/detail/forward_cancellation.hpp>
13#include <boost/cobalt/detail/wrapper.hpp>
14#include <boost/cobalt/detail/this_thread.hpp>
15
16#include <boost/asio/bind_allocator.hpp>
17#include <boost/asio/cancellation_signal.hpp>
18
19
20#include <coroutine>
21#include <optional>
22#include <utility>
23
24namespace boost::cobalt
25{
26
27struct as_tuple_tag;
28struct as_result_tag;
29
30template<typename Return>
31struct task;
32
33namespace detail
34{
35
36template<typename T>
37struct task_receiver;
38
39template<typename T>
40struct task_value_holder
41{
42 std::optional<T> result;
43 bool result_taken = false;
44
45 system::result<T, std::exception_ptr> get_result_value()
46 {
47 result_taken = true;
48 BOOST_ASSERT(result);
49 return {system::in_place_value, std::move(*result)};
50 }
51
52 void return_value(T && ret)
53 {
54 result.emplace(std::move(ret));
55 static_cast<task_receiver<T>*>(this)->set_done();
56 }
57 void return_value(const T & ret)
58 {
59 result.emplace(ret);
60 static_cast<task_receiver<T>*>(this)->set_done();
61 }
62};
63
64template<>
65struct task_value_holder<void>
66{
67 bool result_taken = false;
68 system::result<void, std::exception_ptr> get_result_value()
69 {
70 result_taken = true;
71 return {system::in_place_value};
72 }
73
74 inline void return_void();
75};
76
77
78template<typename T>
79struct task_promise;
80
81template<typename T>
82struct task_receiver : task_value_holder<T>
83{
84 std::exception_ptr exception;
85 system::result<T, std::exception_ptr> get_result()
86 {
87 if (exception && !done) // detached error
88 return {system::in_place_error, std::exchange(obj&: exception, new_val: nullptr)};
89 else if (exception)
90 {
91 this->result_taken = true;
92 return {system::in_place_error, exception};
93 }
94 return this->get_result_value();
95 }
96
97 void unhandled_exception()
98 {
99 exception = std::current_exception();
100 set_done();
101 }
102
103 bool done = false;
104 unique_handle<void> awaited_from{nullptr};
105
106 void set_done()
107 {
108 done = true;
109 }
110
111 void cancel(asio::cancellation_type ct) const
112 {
113 if (!done)
114 promise->signal.emit(ct);
115 }
116
117 task_receiver() = default;
118 task_receiver(task_receiver && lhs)
119 : task_value_holder<T>(std::move(lhs)),
120 exception(std::move(lhs.exception)), done(lhs.done), awaited_from(std::move(lhs.awaited_from)),
121 promise(lhs.promise)
122 {
123 if (!done && !exception)
124 {
125 promise->receiver = this;
126 lhs.exception = moved_from_exception();
127 }
128
129 lhs.done = true;
130 }
131
132 ~task_receiver()
133 {
134 if (!done && promise && promise->receiver == this)
135 {
136 promise->receiver = nullptr;
137 if (!promise->started)
138 std::coroutine_handle<task_promise<T>>::from_promise(*promise).destroy();
139 }
140 }
141
142 task_receiver(task_promise<T> * promise)
143 : promise(promise)
144 {
145 promise->receiver = this;
146 }
147
148 struct awaitable
149 {
150 task_receiver * self;
151 asio::cancellation_slot cl;
152 awaitable(task_receiver * self) : self(self)
153 {
154 }
155
156 awaitable(awaitable && aw) : self(aw.self)
157 {
158 }
159
160 ~awaitable ()
161 {
162 }
163
164 bool await_ready() const { return self->done; }
165
166 template<typename Promise>
167 BOOST_NOINLINE std::coroutine_handle<void> await_suspend(std::coroutine_handle<Promise> h)
168 {
169 if (self->done) // ok, so we're actually done already, so noop
170 return std::coroutine_handle<void>::from_address(a: h.address());
171
172 if constexpr (requires (Promise p) {p.get_cancellation_slot();})
173 if ((cl = h.promise().get_cancellation_slot()).is_connected())
174 cl.emplace<forward_cancellation>(self->promise->signal);
175
176
177 if constexpr (requires (Promise p) {p.get_executor();})
178 self->promise->exec.emplace(h.promise().get_executor());
179 else
180 self->promise->exec.emplace(this_thread::get_executor());
181 self->promise->exec_ = self->promise->exec->get_executor();
182 self->awaited_from.reset(handle: h.address());
183
184 return std::coroutine_handle<task_promise<T>>::from_promise(*self->promise);
185 }
186
187 T await_resume(const boost::source_location & loc = BOOST_CURRENT_LOCATION)
188 {
189 if (cl.is_connected())
190 cl.clear();
191
192 return self->get_result().value(loc);
193 }
194
195 system::result<T, std::exception_ptr> await_resume(const as_result_tag &)
196 {
197 if (cl.is_connected())
198 cl.clear();
199 return self->get_result();
200 }
201
202 auto await_resume(const as_tuple_tag &)
203 {
204 if (cl.is_connected())
205 cl.clear();
206 auto res = self->get_result();
207 if constexpr (std::is_void_v<T>)
208 return res.error();
209 else
210 {
211 if (res.has_error())
212 return std::make_tuple(res.error(), T{});
213 else
214 return std::make_tuple(std::exception_ptr(), std::move(*res));
215 }
216
217 }
218
219 void interrupt_await() &
220 {
221 if (!self)
222 return ;
223 self->exception = detached_exception();
224 if (self->awaited_from)
225 self->awaited_from.release().resume();
226 }
227 };
228
229 task_promise<T> * promise;
230
231 awaitable get_awaitable() {return awaitable{this};}
232
233
234 void interrupt_await() &
235 {
236 exception = detached_exception();
237 awaited_from.release().resume();
238 }
239};
240
241inline void task_value_holder<void>::return_void()
242{
243 static_cast<task_receiver<void>*>(this)->set_done();
244}
245
246template<typename Return>
247struct task_promise_result
248{
249 task_receiver<Return>* receiver{nullptr};
250 void return_value(Return && ret)
251 {
252 if(receiver)
253 receiver->return_value(std::move(ret));
254 }
255 void return_value(const Return & ret)
256 {
257 if(receiver)
258 receiver->return_value(ret);
259 }
260};
261
262template<>
263struct task_promise_result<void>
264{
265 task_receiver<void>* receiver{nullptr};
266 void return_void()
267 {
268 if(receiver)
269 receiver->return_void();
270 }
271};
272
273struct async_initiate_spawn;
274
275template<typename Return>
276struct task_promise
277 : promise_memory_resource_base,
278 promise_cancellation_base<asio::cancellation_slot, asio::enable_total_cancellation>,
279 promise_throw_if_cancelled_base,
280 enable_awaitables<task_promise<Return>>,
281 enable_await_allocator<task_promise<Return>>,
282 enable_await_executor<task_promise<Return>>,
283 task_promise_result<Return>
284{
285 using promise_cancellation_base<asio::cancellation_slot, asio::enable_total_cancellation>::await_transform;
286 using promise_throw_if_cancelled_base::await_transform;
287 using enable_awaitables<task_promise<Return>>::await_transform;
288 using enable_await_allocator<task_promise<Return>>::await_transform;
289 using enable_await_executor<task_promise<Return>>::await_transform;
290
291 [[nodiscard]] task<Return> get_return_object()
292 {
293 return task<Return>{this};
294 }
295
296 mutable asio::cancellation_signal signal;
297
298 using executor_type = executor;
299 std::optional<asio::executor_work_guard<executor_type>> exec;
300 std::optional<executor_type> exec_;
301 const executor_type & get_executor() const
302 {
303 if (!exec)
304 throw_exception(e: asio::bad_executor());
305 BOOST_ASSERT(exec_);
306 return *exec_;
307 }
308
309 template<typename ... Args>
310 task_promise(Args & ...args)
311#if !defined(BOOST_COBALT_NO_PMR)
312 : promise_memory_resource_base(detail::get_memory_resource_from_args_global(args...))
313#endif
314 {
315 this->reset_cancellation_source(signal.slot());
316 }
317
318 struct initial_awaitable
319 {
320 task_promise * promise;
321
322 bool await_ready() const noexcept {return false;}
323 void await_suspend(std::coroutine_handle<>) {}
324
325 void await_resume()
326 {
327 promise->started = true;
328 }
329 };
330
331 auto initial_suspend()
332 {
333
334 return initial_awaitable{this};
335 }
336
337 struct final_awaitable
338 {
339 task_promise * promise;
340 bool await_ready() const noexcept
341 {
342 return promise->receiver && promise->receiver->awaited_from.get() == nullptr;
343 }
344
345 BOOST_NOINLINE
346 auto await_suspend(std::coroutine_handle<task_promise> h) noexcept
347 {
348 std::coroutine_handle<void> res = std::noop_coroutine();
349 if (promise->receiver && promise->receiver->awaited_from.get() != nullptr)
350 res = promise->receiver->awaited_from.release();
351
352
353 if (auto & rec = h.promise().receiver; rec != nullptr)
354 {
355 if (!rec->done && !rec->exception)
356 rec->exception = completed_unexpected();
357 rec->set_done();
358 rec->awaited_from.reset(nullptr);
359 rec = nullptr;
360 }
361 detail::self_destroy(h);
362 return res;
363 }
364
365 void await_resume() noexcept
366 {
367 }
368 };
369
370 auto final_suspend() noexcept
371 {
372 return final_awaitable{this};
373 }
374
375 void unhandled_exception()
376 {
377 if (this->receiver)
378 this->receiver->unhandled_exception();
379 else
380 throw ;
381 }
382
383 ~task_promise()
384 {
385 if (this->receiver)
386 {
387 if (!this->receiver->done && !this->receiver->exception)
388 this->receiver->exception = completed_unexpected();
389 this->receiver->set_done();
390 this->receiver->awaited_from.reset(nullptr);
391 }
392 }
393 bool started = false;
394 friend struct async_initiate;
395};
396
397}
398
399}
400
401#endif //BOOST_COBALT_DETAIL_TASK_HPP
402

source code of boost/libs/cobalt/include/boost/cobalt/detail/task.hpp