cobalt/detail/spawn.hpp

0001 //
0002 // Copyright (c) 2022 Klemens Morgenstern (klemens.morgenstern@gmx.net)
0003 //
0004 // Distributed under the Boost Software License, Version 1.0. (See accompanying
0005 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
0006 //
0007
0008 #ifndef BOOST_COBALT_DETAIL_SPAWN_HPP
0009 #define BOOST_COBALT_DETAIL_SPAWN_HPP
0010
0011 #include <boost/cobalt/task.hpp>
0012 #include <boost/asio/dispatch.hpp>
0013
0014 #include <boost/smart_ptr/allocate_unique.hpp>
0015
0016 namespace boost::cobalt
0017 {
0018 template<typename T>
0019 struct task;
0020 }
0021
0022 namespace boost::cobalt::detail
0023 {
0024
0025 struct async_initiate_spawn
0026 {
0027
0028   async_initiate_spawn(executor exec) : exec(exec) {}
0029
0030   using executor_type = executor;
0031   const executor_type & get_executor() const {return exec;}
0032   executor exec;
0033
0034   template<typename Handler, typename T>
0035   void operator()(Handler && h, task<T> a)
0036   {
0037     auto & rec = a.receiver_;
0038     if (rec.done)
0039       return asio::dispatch(
0040           asio::get_associated_immediate_executor(h, exec),
0041           asio::append(std::forward<Handler>(h), rec.exception, rec.exception ? T() : *rec.get_result()));
0042
0043 #if !defined(BOOST_COBALT_NO_PMR)
0044     auto dalloc = pmr::polymorphic_allocator<void>{boost::cobalt::this_thread::get_default_resource()};
0045     auto alloc = asio::get_associated_allocator(h, dalloc);
0046 #else
0047     auto alloc = asio::get_associated_allocator(h);
0048 #endif
0049     auto recs = std::allocate_shared<detail::task_receiver<T>>(alloc, std::move(rec));
0050
0051     auto sl = asio::get_associated_cancellation_slot(h);
0052     if (sl.is_connected())
0053       sl.assign(
0054           [ex = exec, recs](asio::cancellation_type ct)
0055           {
0056             asio::dispatch(ex, [recs, ct] {recs->cancel(ct);});
0057           });
0058
0059     auto p = recs.get();
0060
0061     p->promise->exec.emplace(exec);
0062     p->promise->exec_ = exec;
0063
0064     struct completion_handler
0065     {
0066       using allocator_type = std::decay_t<decltype(alloc)>;
0067
0068       allocator_type get_allocator() const { return alloc_; }
0069       allocator_type alloc_;
0070
0071       using executor_type = std::decay_t<decltype(asio::get_associated_executor(h, exec))>;
0072       const executor_type &get_executor() const { return exec_; }
0073       executor_type exec_;
0074
0075       decltype(recs) r;
0076       Handler handler;
0077       asio::cancellation_slot sl;
0078
0079       void operator()()
0080       {
0081         if (sl.is_connected())
0082           sl.clear();
0083         auto ex = r->exception;
0084         T rr{};
0085         if (r->result)
0086           rr = std::move(*r->result);
0087         r.reset();
0088         std::move(handler)(ex, std::move(rr));
0089       }
0090     };
0091
0092     p->awaited_from.reset(detail::post_coroutine(
0093         completion_handler{
0094             alloc, asio::get_associated_executor(h, exec), std::move(recs), std::move(h), sl
0095         }).address());
0096
0097     asio::dispatch(exec, std::coroutine_handle<detail::task_promise<T>>::from_promise(*p->promise));
0098   }
0099
0100   template<typename Handler>
0101   void operator()(Handler && h, task<void> a)
0102   {
0103     if (a.receiver_.done)
0104       return asio::dispatch(
0105           asio::get_associated_immediate_executor(h, exec),
0106           asio::append(std::forward<Handler>(h), a.receiver_.exception));
0107
0108
0109 #if !defined(BOOST_COBALT_NO_PMR)
0110     auto alloc = asio::get_associated_allocator(h, pmr::polymorphic_allocator<void>{boost::cobalt::this_thread::get_default_resource()});
0111 #else
0112     auto alloc = asio::get_associated_allocator(h);
0113 #endif
0114     auto recs = std::allocate_shared<detail::task_receiver<void>>(alloc, std::move(a.receiver_));
0115
0116     if (recs->done)
0117       return asio::dispatch(asio::get_associated_immediate_executor(h, exec),
0118                             asio::append(std::forward<Handler>(h), recs->exception));
0119
0120     auto sl = asio::get_associated_cancellation_slot(h);
0121     if (sl.is_connected())
0122       sl.assign(
0123           [ex = exec, recs](asio::cancellation_type ct)
0124           {
0125             asio::dispatch(ex, [recs, ct] {recs->cancel(ct);});
0126           });
0127
0128     auto p = recs.get();
0129
0130     p->promise->exec.emplace(exec);
0131     p->promise->exec_ = exec;
0132
0133     struct completion_handler
0134     {
0135       using allocator_type = std::decay_t<decltype(alloc)>;
0136
0137       const allocator_type &get_allocator() const { return alloc_; }
0138
0139       allocator_type alloc_;
0140
0141       using executor_type = std::decay_t<decltype(asio::get_associated_executor(h, exec))>;
0142       const executor_type &get_executor() const { return exec_; }
0143
0144       executor_type exec_;
0145       decltype(recs) r;
0146       Handler handler;
0147       asio::cancellation_slot sl;
0148
0149       void operator()()
0150       {
0151         if (sl.is_connected())
0152           sl.clear();
0153         auto ex = r->exception;
0154         r.reset();
0155         std::move(handler)(ex);
0156       }
0157     };
0158
0159     p->awaited_from.reset(detail::post_coroutine(completion_handler{
0160         alloc, asio::get_associated_executor(h, exec), std::move(recs), std::forward<Handler>(h), sl
0161       }).address());
0162
0163     asio::dispatch(exec, std::coroutine_handle<detail::task_promise<void>>::from_promise(*p->promise));
0164   }
0165 };
0166
0167 }
0168
0169 #endif //BOOST_COBALT_DETAIL_SPAWN_HPP