spawn.hpp 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156
  1. //
  2. // Copyright (c) 2022 Klemens Morgenstern ([email protected])
  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. #ifndef BOOST_COBALT_DETAIL_SPAWN_HPP
  8. #define BOOST_COBALT_DETAIL_SPAWN_HPP
  9. #include <boost/cobalt/task.hpp>
  10. #include <boost/asio/dispatch.hpp>
  11. #include <boost/smart_ptr/allocate_unique.hpp>
  12. namespace boost::cobalt
  13. {
  14. template<typename T>
  15. struct task;
  16. }
  17. namespace boost::cobalt::detail
  18. {
  19. struct async_initiate_spawn
  20. {
  21. template<typename Handler, typename T>
  22. void operator()(Handler && h, task<T> a, executor exec)
  23. {
  24. auto & rec = a.receiver_;
  25. if (rec.done)
  26. return asio::dispatch(
  27. asio::get_associated_immediate_executor(h, exec),
  28. asio::append(std::forward<Handler>(h), rec.exception, rec.exception ? T() : *rec.get_result()));
  29. #if !defined(BOOST_COBALT_NO_PMR)
  30. auto dalloc = pmr::polymorphic_allocator<void>{boost::cobalt::this_thread::get_default_resource()};
  31. auto alloc = asio::get_associated_allocator(h, dalloc);
  32. #else
  33. auto alloc = asio::get_associated_allocator(h);
  34. #endif
  35. auto recs = std::allocate_shared<detail::task_receiver<T>>(alloc, std::move(rec));
  36. auto sl = asio::get_associated_cancellation_slot(h);
  37. if (sl.is_connected())
  38. sl.assign(
  39. [exec, recs](asio::cancellation_type ct)
  40. {
  41. asio::dispatch(exec, [recs, ct] {recs->cancel(ct);});
  42. });
  43. auto p = recs.get();
  44. p->promise->exec.emplace(exec);
  45. p->promise->exec_ = exec;
  46. struct completion_handler
  47. {
  48. using allocator_type = std::decay_t<decltype(alloc)>;
  49. allocator_type get_allocator() const { return alloc_; }
  50. allocator_type alloc_;
  51. using executor_type = std::decay_t<decltype(asio::get_associated_executor(h, exec))>;
  52. const executor_type &get_executor() const { return exec_; }
  53. executor_type exec_;
  54. decltype(recs) r;
  55. Handler handler;
  56. void operator()()
  57. {
  58. auto ex = r->exception;
  59. T rr{};
  60. if (r->result)
  61. rr = std::move(*r->result);
  62. r.reset();
  63. std::move(handler)(ex, std::move(rr));
  64. }
  65. };
  66. p->awaited_from.reset(detail::post_coroutine(
  67. completion_handler{
  68. alloc, asio::get_associated_executor(h, exec), std::move(recs), std::move(h)
  69. }).address());
  70. asio::dispatch(exec, std::coroutine_handle<detail::task_promise<T>>::from_promise(*p->promise));
  71. }
  72. template<typename Handler>
  73. void operator()(Handler && h, task<void> a, executor exec)
  74. {
  75. if (a.receiver_.done)
  76. return asio::dispatch(
  77. asio::get_associated_immediate_executor(h, exec),
  78. asio::append(std::forward<Handler>(h), a.receiver_.exception));
  79. #if !defined(BOOST_COBALT_NO_PMR)
  80. auto alloc = asio::get_associated_allocator(h, pmr::polymorphic_allocator<void>{boost::cobalt::this_thread::get_default_resource()});
  81. #else
  82. auto alloc = asio::get_associated_allocator(h);
  83. #endif
  84. auto recs = std::allocate_shared<detail::task_receiver<void>>(alloc, std::move(a.receiver_));
  85. if (recs->done)
  86. return asio::dispatch(asio::get_associated_immediate_executor(h, exec),
  87. asio::append(std::forward<Handler>(h), recs->exception));
  88. auto sl = asio::get_associated_cancellation_slot(h);
  89. if (sl.is_connected())
  90. sl.assign(
  91. [exec, recs](asio::cancellation_type ct)
  92. {
  93. asio::dispatch(exec, [recs, ct] {recs->cancel(ct);});
  94. });
  95. auto p = recs.get();
  96. p->promise->exec.emplace(exec);
  97. p->promise->exec_ = exec;
  98. struct completion_handler
  99. {
  100. using allocator_type = std::decay_t<decltype(alloc)>;
  101. const allocator_type &get_allocator() const { return alloc_; }
  102. allocator_type alloc_;
  103. using executor_type = std::decay_t<decltype(asio::get_associated_executor(h, exec))>;
  104. const executor_type &get_executor() const { return exec_; }
  105. executor_type exec_;
  106. decltype(recs) r;
  107. Handler handler;
  108. void operator()()
  109. {
  110. auto ex = r->exception;
  111. r.reset();
  112. std::move(handler)(ex);
  113. }
  114. };
  115. p->awaited_from.reset(detail::post_coroutine(completion_handler{
  116. alloc, asio::get_associated_executor(h, exec), std::move(recs), std::forward<Handler>(h)
  117. }).address());
  118. asio::dispatch(exec, std::coroutine_handle<detail::task_promise<void>>::from_promise(*p->promise));
  119. }
  120. };
  121. }
  122. #endif //BOOST_COBALT_DETAIL_SPAWN_HPP