awaitable.hpp 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760
  1. //
  2. // impl/awaitable.hpp
  3. // ~~~~~~~~~~~~~~~~~~
  4. //
  5. // Copyright (c) 2003-2022 Christopher M. Kohlhoff (chris at kohlhoff dot com)
  6. //
  7. // Distributed under the Boost Software License, Version 1.0. (See accompanying
  8. // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
  9. //
  10. #ifndef ASIO_IMPL_AWAITABLE_HPP
  11. #define ASIO_IMPL_AWAITABLE_HPP
  12. #if defined(_MSC_VER) && (_MSC_VER >= 1200)
  13. # pragma once
  14. #endif // defined(_MSC_VER) && (_MSC_VER >= 1200)
  15. #include "asio/detail/config.hpp"
  16. #include <exception>
  17. #include <new>
  18. #include <tuple>
  19. #include "asio/cancellation_signal.hpp"
  20. #include "asio/cancellation_state.hpp"
  21. #include "asio/detail/thread_context.hpp"
  22. #include "asio/detail/thread_info_base.hpp"
  23. #include "asio/detail/throw_error.hpp"
  24. #include "asio/detail/type_traits.hpp"
  25. #include "asio/error.hpp"
  26. #include "asio/post.hpp"
  27. #include "asio/system_error.hpp"
  28. #include "asio/this_coro.hpp"
  29. #include "asio/detail/push_options.hpp"
  30. namespace asio {
  31. namespace detail {
  32. struct awaitable_thread_has_context_switched {};
  33. // An awaitable_thread represents a thread-of-execution that is composed of one
  34. // or more "stack frames", with each frame represented by an awaitable_frame.
  35. // All execution occurs in the context of the awaitable_thread's executor. An
  36. // awaitable_thread continues to "pump" the stack frames by repeatedly resuming
  37. // the top stack frame until the stack is empty, or until ownership of the
  38. // stack is transferred to another awaitable_thread object.
  39. //
  40. // +------------------------------------+
  41. // | top_of_stack_ |
  42. // | V
  43. // +--------------+---+ +-----------------+
  44. // | | | |
  45. // | awaitable_thread |<---------------------------+ awaitable_frame |
  46. // | | attached_thread_ | |
  47. // +--------------+---+ (Set only when +---+-------------+
  48. // | frames are being |
  49. // | actively pumped | caller_
  50. // | by a thread, and |
  51. // | then only for V
  52. // | the top frame.) +-----------------+
  53. // | | |
  54. // | | awaitable_frame |
  55. // | | |
  56. // | +---+-------------+
  57. // | |
  58. // | | caller_
  59. // | :
  60. // | :
  61. // | |
  62. // | V
  63. // | +-----------------+
  64. // | bottom_of_stack_ | |
  65. // +------------------------------->| awaitable_frame |
  66. // | |
  67. // +-----------------+
  68. template <typename Executor>
  69. class awaitable_frame_base
  70. {
  71. public:
  72. #if !defined(ASIO_DISABLE_AWAITABLE_FRAME_RECYCLING)
  73. void* operator new(std::size_t size)
  74. {
  75. return asio::detail::thread_info_base::allocate(
  76. asio::detail::thread_info_base::awaitable_frame_tag(),
  77. asio::detail::thread_context::top_of_thread_call_stack(),
  78. size);
  79. }
  80. void operator delete(void* pointer, std::size_t size)
  81. {
  82. asio::detail::thread_info_base::deallocate(
  83. asio::detail::thread_info_base::awaitable_frame_tag(),
  84. asio::detail::thread_context::top_of_thread_call_stack(),
  85. pointer, size);
  86. }
  87. #endif // !defined(ASIO_DISABLE_AWAITABLE_FRAME_RECYCLING)
  88. // The frame starts in a suspended state until the awaitable_thread object
  89. // pumps the stack.
  90. auto initial_suspend() noexcept
  91. {
  92. return suspend_always();
  93. }
  94. // On final suspension the frame is popped from the top of the stack.
  95. auto final_suspend() noexcept
  96. {
  97. struct result
  98. {
  99. awaitable_frame_base* this_;
  100. bool await_ready() const noexcept
  101. {
  102. return false;
  103. }
  104. void await_suspend(coroutine_handle<void>) noexcept
  105. {
  106. this->this_->pop_frame();
  107. }
  108. void await_resume() const noexcept
  109. {
  110. }
  111. };
  112. return result{this};
  113. }
  114. void set_except(std::exception_ptr e) noexcept
  115. {
  116. pending_exception_ = e;
  117. }
  118. void set_error(const asio::error_code& ec)
  119. {
  120. this->set_except(std::make_exception_ptr(asio::system_error(ec)));
  121. }
  122. void unhandled_exception()
  123. {
  124. set_except(std::current_exception());
  125. }
  126. void rethrow_exception()
  127. {
  128. if (pending_exception_)
  129. {
  130. std::exception_ptr ex = std::exchange(pending_exception_, nullptr);
  131. std::rethrow_exception(ex);
  132. }
  133. }
  134. void clear_cancellation_slot()
  135. {
  136. this->attached_thread_->entry_point()->cancellation_state_.slot().clear();
  137. }
  138. template <typename T>
  139. auto await_transform(awaitable<T, Executor> a) const
  140. {
  141. if (attached_thread_->entry_point()->throw_if_cancelled_)
  142. if (!!attached_thread_->get_cancellation_state().cancelled())
  143. do_throw_error(asio::error::operation_aborted, "co_await");
  144. return a;
  145. }
  146. // This await transformation obtains the associated executor of the thread of
  147. // execution.
  148. auto await_transform(this_coro::executor_t) noexcept
  149. {
  150. struct result
  151. {
  152. awaitable_frame_base* this_;
  153. bool await_ready() const noexcept
  154. {
  155. return true;
  156. }
  157. void await_suspend(coroutine_handle<void>) noexcept
  158. {
  159. }
  160. auto await_resume() const noexcept
  161. {
  162. return this_->attached_thread_->get_executor();
  163. }
  164. };
  165. return result{this};
  166. }
  167. // This await transformation obtains the associated cancellation state of the
  168. // thread of execution.
  169. auto await_transform(this_coro::cancellation_state_t) noexcept
  170. {
  171. struct result
  172. {
  173. awaitable_frame_base* this_;
  174. bool await_ready() const noexcept
  175. {
  176. return true;
  177. }
  178. void await_suspend(coroutine_handle<void>) noexcept
  179. {
  180. }
  181. auto await_resume() const noexcept
  182. {
  183. return this_->attached_thread_->get_cancellation_state();
  184. }
  185. };
  186. return result{this};
  187. }
  188. // This await transformation resets the associated cancellation state.
  189. auto await_transform(this_coro::reset_cancellation_state_0_t) noexcept
  190. {
  191. struct result
  192. {
  193. awaitable_frame_base* this_;
  194. bool await_ready() const noexcept
  195. {
  196. return true;
  197. }
  198. void await_suspend(coroutine_handle<void>) noexcept
  199. {
  200. }
  201. auto await_resume() const
  202. {
  203. return this_->attached_thread_->reset_cancellation_state();
  204. }
  205. };
  206. return result{this};
  207. }
  208. // This await transformation resets the associated cancellation state.
  209. template <typename Filter>
  210. auto await_transform(
  211. this_coro::reset_cancellation_state_1_t<Filter> reset) noexcept
  212. {
  213. struct result
  214. {
  215. awaitable_frame_base* this_;
  216. Filter filter_;
  217. bool await_ready() const noexcept
  218. {
  219. return true;
  220. }
  221. void await_suspend(coroutine_handle<void>) noexcept
  222. {
  223. }
  224. auto await_resume()
  225. {
  226. return this_->attached_thread_->reset_cancellation_state(
  227. ASIO_MOVE_CAST(Filter)(filter_));
  228. }
  229. };
  230. return result{this, ASIO_MOVE_CAST(Filter)(reset.filter)};
  231. }
  232. // This await transformation resets the associated cancellation state.
  233. template <typename InFilter, typename OutFilter>
  234. auto await_transform(
  235. this_coro::reset_cancellation_state_2_t<InFilter, OutFilter> reset)
  236. noexcept
  237. {
  238. struct result
  239. {
  240. awaitable_frame_base* this_;
  241. InFilter in_filter_;
  242. OutFilter out_filter_;
  243. bool await_ready() const noexcept
  244. {
  245. return true;
  246. }
  247. void await_suspend(coroutine_handle<void>) noexcept
  248. {
  249. }
  250. auto await_resume()
  251. {
  252. return this_->attached_thread_->reset_cancellation_state(
  253. ASIO_MOVE_CAST(InFilter)(in_filter_),
  254. ASIO_MOVE_CAST(OutFilter)(out_filter_));
  255. }
  256. };
  257. return result{this,
  258. ASIO_MOVE_CAST(InFilter)(reset.in_filter),
  259. ASIO_MOVE_CAST(OutFilter)(reset.out_filter)};
  260. }
  261. // This await transformation determines whether cancellation is propagated as
  262. // an exception.
  263. auto await_transform(this_coro::throw_if_cancelled_0_t)
  264. noexcept
  265. {
  266. struct result
  267. {
  268. awaitable_frame_base* this_;
  269. bool await_ready() const noexcept
  270. {
  271. return true;
  272. }
  273. void await_suspend(coroutine_handle<void>) noexcept
  274. {
  275. }
  276. auto await_resume()
  277. {
  278. return this_->attached_thread_->throw_if_cancelled();
  279. }
  280. };
  281. return result{this};
  282. }
  283. // This await transformation sets whether cancellation is propagated as an
  284. // exception.
  285. auto await_transform(this_coro::throw_if_cancelled_1_t throw_if_cancelled)
  286. noexcept
  287. {
  288. struct result
  289. {
  290. awaitable_frame_base* this_;
  291. bool value_;
  292. bool await_ready() const noexcept
  293. {
  294. return true;
  295. }
  296. void await_suspend(coroutine_handle<void>) noexcept
  297. {
  298. }
  299. auto await_resume()
  300. {
  301. this_->attached_thread_->throw_if_cancelled(value_);
  302. }
  303. };
  304. return result{this, throw_if_cancelled.value};
  305. }
  306. // This await transformation is used to run an async operation's initiation
  307. // function object after the coroutine has been suspended. This ensures that
  308. // immediate resumption of the coroutine in another thread does not cause a
  309. // race condition.
  310. template <typename Function>
  311. auto await_transform(Function f,
  312. typename enable_if<
  313. is_convertible<
  314. typename result_of<Function(awaitable_frame_base*)>::type,
  315. awaitable_thread<Executor>*
  316. >::value
  317. >::type* = nullptr)
  318. {
  319. struct result
  320. {
  321. Function function_;
  322. awaitable_frame_base* this_;
  323. bool await_ready() const noexcept
  324. {
  325. return false;
  326. }
  327. void await_suspend(coroutine_handle<void>) noexcept
  328. {
  329. function_(this_);
  330. }
  331. void await_resume() const noexcept
  332. {
  333. }
  334. };
  335. return result{std::move(f), this};
  336. }
  337. // Access the awaitable thread's has_context_switched_ flag.
  338. auto await_transform(detail::awaitable_thread_has_context_switched) noexcept
  339. {
  340. struct result
  341. {
  342. awaitable_frame_base* this_;
  343. bool await_ready() const noexcept
  344. {
  345. return true;
  346. }
  347. void await_suspend(coroutine_handle<void>) noexcept
  348. {
  349. }
  350. bool& await_resume() const noexcept
  351. {
  352. return this_->attached_thread_->entry_point()->has_context_switched_;
  353. }
  354. };
  355. return result{this};
  356. }
  357. void attach_thread(awaitable_thread<Executor>* handler) noexcept
  358. {
  359. attached_thread_ = handler;
  360. }
  361. awaitable_thread<Executor>* detach_thread() noexcept
  362. {
  363. attached_thread_->entry_point()->has_context_switched_ = true;
  364. return std::exchange(attached_thread_, nullptr);
  365. }
  366. void push_frame(awaitable_frame_base<Executor>* caller) noexcept
  367. {
  368. caller_ = caller;
  369. attached_thread_ = caller_->attached_thread_;
  370. attached_thread_->entry_point()->top_of_stack_ = this;
  371. caller_->attached_thread_ = nullptr;
  372. }
  373. void pop_frame() noexcept
  374. {
  375. if (caller_)
  376. caller_->attached_thread_ = attached_thread_;
  377. attached_thread_->entry_point()->top_of_stack_ = caller_;
  378. attached_thread_ = nullptr;
  379. caller_ = nullptr;
  380. }
  381. void resume()
  382. {
  383. coro_.resume();
  384. }
  385. void destroy()
  386. {
  387. coro_.destroy();
  388. }
  389. protected:
  390. coroutine_handle<void> coro_ = nullptr;
  391. awaitable_thread<Executor>* attached_thread_ = nullptr;
  392. awaitable_frame_base<Executor>* caller_ = nullptr;
  393. std::exception_ptr pending_exception_ = nullptr;
  394. };
  395. template <typename T, typename Executor>
  396. class awaitable_frame
  397. : public awaitable_frame_base<Executor>
  398. {
  399. public:
  400. awaitable_frame() noexcept
  401. {
  402. }
  403. awaitable_frame(awaitable_frame&& other) noexcept
  404. : awaitable_frame_base<Executor>(std::move(other))
  405. {
  406. }
  407. ~awaitable_frame()
  408. {
  409. if (has_result_)
  410. static_cast<T*>(static_cast<void*>(result_))->~T();
  411. }
  412. awaitable<T, Executor> get_return_object() noexcept
  413. {
  414. this->coro_ = coroutine_handle<awaitable_frame>::from_promise(*this);
  415. return awaitable<T, Executor>(this);
  416. };
  417. template <typename U>
  418. void return_value(U&& u)
  419. {
  420. new (&result_) T(std::forward<U>(u));
  421. has_result_ = true;
  422. }
  423. template <typename... Us>
  424. void return_values(Us&&... us)
  425. {
  426. this->return_value(std::forward_as_tuple(std::forward<Us>(us)...));
  427. }
  428. T get()
  429. {
  430. this->caller_ = nullptr;
  431. this->rethrow_exception();
  432. return std::move(*static_cast<T*>(static_cast<void*>(result_)));
  433. }
  434. private:
  435. alignas(T) unsigned char result_[sizeof(T)];
  436. bool has_result_ = false;
  437. };
  438. template <typename Executor>
  439. class awaitable_frame<void, Executor>
  440. : public awaitable_frame_base<Executor>
  441. {
  442. public:
  443. awaitable<void, Executor> get_return_object()
  444. {
  445. this->coro_ = coroutine_handle<awaitable_frame>::from_promise(*this);
  446. return awaitable<void, Executor>(this);
  447. };
  448. void return_void()
  449. {
  450. }
  451. void get()
  452. {
  453. this->caller_ = nullptr;
  454. this->rethrow_exception();
  455. }
  456. };
  457. struct awaitable_thread_entry_point {};
  458. template <typename Executor>
  459. class awaitable_frame<awaitable_thread_entry_point, Executor>
  460. : public awaitable_frame_base<Executor>
  461. {
  462. public:
  463. awaitable_frame()
  464. : top_of_stack_(0),
  465. has_executor_(false),
  466. has_context_switched_(false),
  467. throw_if_cancelled_(true)
  468. {
  469. }
  470. ~awaitable_frame()
  471. {
  472. if (has_executor_)
  473. u_.executor_.~Executor();
  474. }
  475. awaitable<awaitable_thread_entry_point, Executor> get_return_object()
  476. {
  477. this->coro_ = coroutine_handle<awaitable_frame>::from_promise(*this);
  478. return awaitable<awaitable_thread_entry_point, Executor>(this);
  479. };
  480. void return_void()
  481. {
  482. }
  483. void get()
  484. {
  485. this->caller_ = nullptr;
  486. this->rethrow_exception();
  487. }
  488. private:
  489. template <typename> friend class awaitable_frame_base;
  490. template <typename, typename> friend class awaitable_handler_base;
  491. template <typename> friend class awaitable_thread;
  492. union u
  493. {
  494. u() {}
  495. ~u() {}
  496. char c_;
  497. Executor executor_;
  498. } u_;
  499. awaitable_frame_base<Executor>* top_of_stack_;
  500. asio::cancellation_slot parent_cancellation_slot_;
  501. asio::cancellation_state cancellation_state_;
  502. bool has_executor_;
  503. bool has_context_switched_;
  504. bool throw_if_cancelled_;
  505. };
  506. template <typename Executor>
  507. class awaitable_thread
  508. {
  509. public:
  510. typedef Executor executor_type;
  511. typedef cancellation_slot cancellation_slot_type;
  512. // Construct from the entry point of a new thread of execution.
  513. awaitable_thread(awaitable<awaitable_thread_entry_point, Executor> p,
  514. const Executor& ex, cancellation_slot parent_cancel_slot,
  515. cancellation_state cancel_state)
  516. : bottom_of_stack_(std::move(p))
  517. {
  518. bottom_of_stack_.frame_->top_of_stack_ = bottom_of_stack_.frame_;
  519. new (&bottom_of_stack_.frame_->u_.executor_) Executor(ex);
  520. bottom_of_stack_.frame_->has_executor_ = true;
  521. bottom_of_stack_.frame_->parent_cancellation_slot_ = parent_cancel_slot;
  522. bottom_of_stack_.frame_->cancellation_state_ = cancel_state;
  523. }
  524. // Transfer ownership from another awaitable_thread.
  525. awaitable_thread(awaitable_thread&& other) noexcept
  526. : bottom_of_stack_(std::move(other.bottom_of_stack_))
  527. {
  528. }
  529. // Clean up with a last ditch effort to ensure the thread is unwound within
  530. // the context of the executor.
  531. ~awaitable_thread()
  532. {
  533. if (bottom_of_stack_.valid())
  534. {
  535. // Coroutine "stack unwinding" must be performed through the executor.
  536. auto* bottom_frame = bottom_of_stack_.frame_;
  537. (post)(bottom_frame->u_.executor_,
  538. [a = std::move(bottom_of_stack_)]() mutable
  539. {
  540. (void)awaitable<awaitable_thread_entry_point, Executor>(
  541. std::move(a));
  542. });
  543. }
  544. }
  545. awaitable_frame<awaitable_thread_entry_point, Executor>* entry_point()
  546. {
  547. return bottom_of_stack_.frame_;
  548. }
  549. executor_type get_executor() const noexcept
  550. {
  551. return bottom_of_stack_.frame_->u_.executor_;
  552. }
  553. cancellation_state get_cancellation_state() const noexcept
  554. {
  555. return bottom_of_stack_.frame_->cancellation_state_;
  556. }
  557. void reset_cancellation_state()
  558. {
  559. bottom_of_stack_.frame_->cancellation_state_ =
  560. cancellation_state(bottom_of_stack_.frame_->parent_cancellation_slot_);
  561. }
  562. template <typename Filter>
  563. void reset_cancellation_state(ASIO_MOVE_ARG(Filter) filter)
  564. {
  565. bottom_of_stack_.frame_->cancellation_state_ =
  566. cancellation_state(bottom_of_stack_.frame_->parent_cancellation_slot_,
  567. ASIO_MOVE_CAST(Filter)(filter));
  568. }
  569. template <typename InFilter, typename OutFilter>
  570. void reset_cancellation_state(ASIO_MOVE_ARG(InFilter) in_filter,
  571. ASIO_MOVE_ARG(OutFilter) out_filter)
  572. {
  573. bottom_of_stack_.frame_->cancellation_state_ =
  574. cancellation_state(bottom_of_stack_.frame_->parent_cancellation_slot_,
  575. ASIO_MOVE_CAST(InFilter)(in_filter),
  576. ASIO_MOVE_CAST(OutFilter)(out_filter));
  577. }
  578. bool throw_if_cancelled() const
  579. {
  580. return bottom_of_stack_.frame_->throw_if_cancelled_;
  581. }
  582. void throw_if_cancelled(bool value)
  583. {
  584. bottom_of_stack_.frame_->throw_if_cancelled_ = value;
  585. }
  586. cancellation_slot_type get_cancellation_slot() const noexcept
  587. {
  588. return bottom_of_stack_.frame_->cancellation_state_.slot();
  589. }
  590. // Launch a new thread of execution.
  591. void launch()
  592. {
  593. bottom_of_stack_.frame_->top_of_stack_->attach_thread(this);
  594. pump();
  595. }
  596. protected:
  597. template <typename> friend class awaitable_frame_base;
  598. // Repeatedly resume the top stack frame until the stack is empty or until it
  599. // has been transferred to another resumable_thread object.
  600. void pump()
  601. {
  602. do
  603. bottom_of_stack_.frame_->top_of_stack_->resume();
  604. while (bottom_of_stack_.frame_ && bottom_of_stack_.frame_->top_of_stack_);
  605. if (bottom_of_stack_.frame_)
  606. {
  607. awaitable<awaitable_thread_entry_point, Executor> a(
  608. std::move(bottom_of_stack_));
  609. a.frame_->rethrow_exception();
  610. }
  611. }
  612. awaitable<awaitable_thread_entry_point, Executor> bottom_of_stack_;
  613. };
  614. } // namespace detail
  615. } // namespace asio
  616. #if !defined(GENERATING_DOCUMENTATION)
  617. # if defined(ASIO_HAS_STD_COROUTINE)
  618. namespace std {
  619. template <typename T, typename Executor, typename... Args>
  620. struct coroutine_traits<asio::awaitable<T, Executor>, Args...>
  621. {
  622. typedef asio::detail::awaitable_frame<T, Executor> promise_type;
  623. };
  624. } // namespace std
  625. # else // defined(ASIO_HAS_STD_COROUTINE)
  626. namespace std { namespace experimental {
  627. template <typename T, typename Executor, typename... Args>
  628. struct coroutine_traits<asio::awaitable<T, Executor>, Args...>
  629. {
  630. typedef asio::detail::awaitable_frame<T, Executor> promise_type;
  631. };
  632. }} // namespace std::experimental
  633. # endif // defined(ASIO_HAS_STD_COROUTINE)
  634. #endif // !defined(GENERATING_DOCUMENTATION)
  635. #include "asio/detail/pop_options.hpp"
  636. #endif // ASIO_IMPL_AWAITABLE_HPP