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
0078       void operator()()
0079       {
0080         auto ex = r->exception;
0081         T rr{};
0082         if (r->result)
0083           rr = std::move(*r->result);
0084         r.reset();
0085         std::move(handler)(ex, std::move(rr));
0086       }
0087     };
0088
0089     p->awaited_from.reset(detail::post_coroutine(
0090         completion_handler{
0091             alloc, asio::get_associated_executor(h, exec), std::move(recs), std::move(h)
0092         }).address());
0093
0094     asio::dispatch(exec, std::coroutine_handle<detail::task_promise<T>>::from_promise(*p->promise));
0095   }
0096
0097   template<typename Handler>
0098   void operator()(Handler && h, task<void> a)
0099   {
0100     if (a.receiver_.done)
0101       return asio::dispatch(
0102           asio::get_associated_immediate_executor(h, exec),
0103           asio::append(std::forward<Handler>(h), a.receiver_.exception));
0104
0105
0106 #if !defined(BOOST_COBALT_NO_PMR)
0107     auto alloc = asio::get_associated_allocator(h, pmr::polymorphic_allocator<void>{boost::cobalt::this_thread::get_default_resource()});
0108 #else
0109     auto alloc = asio::get_associated_allocator(h);
0110 #endif
0111     auto recs = std::allocate_shared<detail::task_receiver<void>>(alloc, std::move(a.receiver_));
0112
0113     if (recs->done)
0114       return asio::dispatch(asio::get_associated_immediate_executor(h, exec),
0115                             asio::append(std::forward<Handler>(h), recs->exception));
0116
0117     auto sl = asio::get_associated_cancellation_slot(h);
0118     if (sl.is_connected())
0119       sl.assign(
0120           [ex = exec, recs](asio::cancellation_type ct)
0121           {
0122             asio::dispatch(ex, [recs, ct] {recs->cancel(ct);});
0123           });
0124
0125     auto p = recs.get();
0126
0127     p->promise->exec.emplace(exec);
0128     p->promise->exec_ = exec;
0129
0130     struct completion_handler
0131     {
0132       using allocator_type = std::decay_t<decltype(alloc)>;
0133
0134       const allocator_type &get_allocator() const { return alloc_; }
0135
0136       allocator_type alloc_;
0137
0138       using executor_type = std::decay_t<decltype(asio::get_associated_executor(h, exec))>;
0139       const executor_type &get_executor() const { return exec_; }
0140
0141       executor_type exec_;
0142       decltype(recs) r;
0143       Handler handler;
0144
0145       void operator()()
0146       {
0147         auto ex = r->exception;
0148         r.reset();
0149         std::move(handler)(ex);
0150       }
0151     };
0152
0153     p->awaited_from.reset(detail::post_coroutine(completion_handler{
0154         alloc, asio::get_associated_executor(h, exec), std::move(recs), std::forward<Handler>(h)
0155       }).address());
0156
0157     asio::dispatch(exec, std::coroutine_handle<detail::task_promise<void>>::from_promise(*p->promise));
0158   }
0159 };
0160
0161 }
0162
0163 #endif //BOOST_COBALT_DETAIL_SPAWN_HPP