awaitable.hpp 19 KB

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