cobalt/detail/promise.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_PROMISE_HPP
0009 #define BOOST_COBALT_DETAIL_PROMISE_HPP
0010
0011 #include <boost/cobalt/detail/exception.hpp>
0012 #include <boost/cobalt/detail/forward_cancellation.hpp>
0013 #include <boost/cobalt/detail/wrapper.hpp>
0014 #include <boost/cobalt/detail/this_thread.hpp>
0015 #include <boost/cobalt/noop.hpp>
0016 #include <boost/cobalt/op.hpp>
0017 #include <boost/cobalt/unique_handle.hpp>
0018
0019 #include <boost/asio/cancellation_signal.hpp>
0020 #include <boost/asio/bind_allocator.hpp>
0021
0022 #include <boost/core/exchange.hpp>
0023 #include <coroutine>
0024 #include <optional>
0025 #include <utility>
0026
0027 namespace boost::cobalt
0028 {
0029
0030 struct as_tuple_tag;
0031 struct as_result_tag;
0032
0033 template<typename Return>
0034 struct promise;
0035
0036 namespace detail
0037 {
0038
0039 template<typename T>
0040 struct promise_receiver;
0041
0042 template<typename T>
0043 struct promise_value_holder
0044 {
0045   std::optional<T> result;
0046   bool result_taken = false;
0047
0048   system::result<T, std::exception_ptr> get_result_value()
0049   {
0050     result_taken = true;
0051     BOOST_ASSERT(result);
0052     return {system::in_place_value, std::move(*result)};
0053   }
0054
0055   void return_value(T && ret)
0056   {
0057     result.emplace(std::move(ret));
0058     static_cast<promise_receiver<T>*>(this)->set_done();
0059   }
0060
0061   void return_value(const T & ret)
0062   {
0063     result.emplace(ret);
0064     static_cast<promise_receiver<T>*>(this)->set_done();
0065   }
0066   constexpr promise_value_holder() = default;
0067   constexpr promise_value_holder(noop<T> value) noexcept(std::is_nothrow_move_constructible_v<T>) : result(std::move(value.value)) {}
0068
0069 };
0070
0071 template<>
0072 struct promise_value_holder<void>
0073 {
0074   bool result_taken = false;
0075   system::result<void, std::exception_ptr> get_result_value()
0076   {
0077     result_taken = true;
0078     return {system::in_place_value};
0079   }
0080
0081   inline void return_void();
0082
0083   constexpr promise_value_holder() = default;
0084   constexpr promise_value_holder(noop<void>) {}
0085 };
0086
0087
0088
0089 template<typename T>
0090 struct promise_receiver : promise_value_holder<T>
0091 {
0092   std::exception_ptr exception;
0093   system::result<T, std::exception_ptr> get_result()
0094   {
0095     if (exception && !done) // detached error
0096       return {system::in_place_error, std::exchange(exception, nullptr)};
0097     else if (exception)
0098     {
0099       this->result_taken = true;
0100       return {system::in_place_error, exception};
0101     }
0102     return this->get_result_value();
0103   }
0104   void unhandled_exception()
0105   {
0106     exception = std::current_exception();
0107     set_done();
0108   }
0109
0110   bool done = false;
0111   unique_handle<void>  awaited_from{nullptr};
0112
0113   void set_done()
0114   {
0115     done = true;
0116   }
0117
0118   promise_receiver() = default;
0119   promise_receiver(noop<T> value) : promise_value_holder<T>(std::move(value)), done(true) {}
0120   promise_receiver(promise_receiver && lhs) noexcept
0121       : promise_value_holder<T>(std::move(lhs)),
0122         exception(std::move(lhs.exception)), done(lhs.done), awaited_from(std::move(lhs.awaited_from)),
0123         reference(lhs.reference), cancel_signal(lhs.cancel_signal)
0124   {
0125     if (!done && !exception)
0126     {
0127       *reference = this;
0128       lhs.exception = moved_from_exception();
0129     }
0130
0131     lhs.done = true;
0132   }
0133
0134   promise_receiver& operator=(promise_receiver && lhs) noexcept
0135   {
0136     if (*reference == this)
0137     {
0138       *reference = nullptr;
0139     }
0140
0141     promise_value_holder<T>::operator=(std::move(lhs));
0142     exception = std::move(lhs.exception);
0143     done = std::move(lhs.done);
0144     awaited_from = std::move(lhs.awaited_from);
0145     reference = std::move(lhs.reference);
0146     cancel_signal = std::move(lhs.cancel_signal);
0147
0148     if (!done && !exception)
0149     {
0150       *reference = this;
0151       lhs.exception = moved_from_exception();
0152     }
0153
0154     return *this;
0155   }
0156
0157
0158   ~promise_receiver()
0159   {
0160     if (!done && *reference == this)
0161       *reference = nullptr;
0162   }
0163
0164   promise_receiver(promise_receiver * &reference, asio::cancellation_signal & cancel_signal)
0165       : reference(&reference), cancel_signal(&cancel_signal)
0166   {
0167     reference = this;
0168   }
0169
0170   struct awaitable
0171   {
0172     promise_receiver * self;
0173     std::exception_ptr ex;
0174     asio::cancellation_slot cl;
0175
0176     awaitable(promise_receiver * self) : self(self)
0177     {
0178     }
0179
0180     awaitable(awaitable && aw) : self(aw.self)
0181     {
0182     }
0183
0184     ~awaitable ()
0185     {
0186     }
0187     bool await_ready() const { return self->done; }
0188
0189     template<typename Promise>
0190     bool await_suspend(std::coroutine_handle<Promise> h)
0191     {
0192       if (self->done) // ok, so we're actually done already, so noop
0193         return false;
0194
0195       if (ex)
0196         return false;
0197
0198       if (self->awaited_from != nullptr) // we're already being awaited, that's an error!
0199       {
0200         ex = already_awaited();
0201         return false;
0202       }
0203
0204       if constexpr (requires (Promise p) {p.get_cancellation_slot();})
0205         if ((cl = h.promise().get_cancellation_slot()).is_connected())
0206           cl.emplace<forward_cancellation>(*self->cancel_signal);
0207
0208       self->awaited_from.reset(h.address());
0209       return true;
0210     }
0211
0212     T await_resume(const boost::source_location & loc = BOOST_CURRENT_LOCATION)
0213     {
0214       if (cl.is_connected())
0215         cl.clear();
0216       if (ex)
0217         std::rethrow_exception(ex);
0218       return self->get_result().value(loc);
0219     }
0220
0221     system::result<T, std::exception_ptr> await_resume(const as_result_tag &)
0222     {
0223       if (cl.is_connected())
0224         cl.clear();
0225       if (ex)
0226         return {system::in_place_error, std::move(ex)};
0227       return self->get_result();
0228     }
0229
0230     auto await_resume(const as_tuple_tag &)
0231     {
0232       if (cl.is_connected())
0233         cl.clear();
0234
0235       if constexpr (std::is_void_v<T>)
0236       {
0237         if (ex)
0238           return std::move(ex);
0239         return self->get_result().error();
0240       }
0241       else
0242       {
0243         if (ex)
0244           return std::make_tuple(std::move(ex), T{});
0245         auto res = self->get_result();
0246         if (res.has_error())
0247           return std::make_tuple(res.error(), T{});
0248         else
0249           return std::make_tuple(std::exception_ptr(), std::move(*res));
0250       }
0251     }
0252     void interrupt_await() &
0253     {
0254       if (!self)
0255         return ;
0256       ex = detached_exception();
0257       if (self->awaited_from)
0258         self->awaited_from.release().resume();
0259     }
0260   };
0261
0262   promise_receiver  **reference;
0263   asio::cancellation_signal * cancel_signal;
0264
0265   awaitable get_awaitable() {return awaitable{this};}
0266
0267
0268   void interrupt_await() &
0269   {
0270     exception = detached_exception();
0271     awaited_from.release().resume();
0272   }
0273 };
0274
0275 inline void promise_value_holder<void>::return_void()
0276 {
0277   static_cast<promise_receiver<void>*>(this)->set_done();
0278 }
0279
0280 template<typename Return>
0281 struct cobalt_promise_result
0282 {
0283   promise_receiver<Return>* receiver{nullptr};
0284   void return_value(Return && ret)
0285   {
0286     if(receiver)
0287       receiver->return_value(std::move(ret));
0288   }
0289
0290   void return_value(const Return & ret)
0291   {
0292     if(receiver)
0293       receiver->return_value(ret);
0294   }
0295
0296 };
0297
0298 template<>
0299 struct cobalt_promise_result<void>
0300 {
0301   promise_receiver<void>* receiver{nullptr};
0302   void return_void()
0303   {
0304     if(receiver)
0305       receiver->return_void();
0306   }
0307 };
0308
0309 template<typename Return>
0310 struct cobalt_promise
0311     : promise_memory_resource_base,
0312       promise_cancellation_base<asio::cancellation_slot, asio::enable_total_cancellation>,
0313       promise_throw_if_cancelled_base,
0314       enable_awaitables<cobalt_promise<Return>>,
0315       enable_await_allocator<cobalt_promise<Return>>,
0316       enable_await_executor<cobalt_promise<Return>>,
0317       enable_await_deferred,
0318       cobalt_promise_result<Return>
0319 {
0320   using promise_cancellation_base<asio::cancellation_slot, asio::enable_total_cancellation>::await_transform;
0321   using promise_throw_if_cancelled_base::await_transform;
0322   using enable_awaitables<cobalt_promise<Return>>::await_transform;
0323   using enable_await_allocator<cobalt_promise<Return>>::await_transform;
0324   using enable_await_executor<cobalt_promise<Return>>::await_transform;
0325   using enable_await_deferred::await_transform;
0326
0327   [[nodiscard]] promise<Return> get_return_object()
0328   {
0329     return promise<Return>{this};
0330   }
0331
0332   mutable asio::cancellation_signal signal;
0333
0334   using executor_type = executor;
0335   executor_type exec;
0336   const executor_type & get_executor() const {return exec;}
0337
0338   template<typename ... Args>
0339   cobalt_promise(Args & ...args)
0340       :
0341 #if !defined(BOOST_COBALT_NO_PMR)
0342   promise_memory_resource_base(detail::get_memory_resource_from_args(args...)),
0343 #endif
0344         exec{detail::get_executor_from_args(args...)}
0345   {
0346     this->reset_cancellation_source(signal.slot());
0347   }
0348
0349   std::suspend_never initial_suspend() noexcept {return {};}
0350   auto final_suspend() noexcept
0351   {
0352     return final_awaitable{this};
0353   }
0354
0355   void unhandled_exception()
0356   {
0357     if (this->receiver)
0358       this->receiver->unhandled_exception();
0359     else
0360       throw ;
0361   }
0362
0363   ~cobalt_promise()
0364   {
0365     if (this->receiver)
0366     {
0367       if (!this->receiver->done && !this->receiver->exception)
0368         this->receiver->exception = completed_unexpected();
0369       this->receiver->set_done();
0370       this->receiver->awaited_from.reset(nullptr);
0371     }
0372
0373   }
0374  private:
0375   struct final_awaitable
0376   {
0377     cobalt_promise * promise;
0378     bool await_ready() const noexcept
0379     {
0380       return promise->receiver && promise->receiver->awaited_from.get() == nullptr;
0381     }
0382
0383     std::coroutine_handle<void> await_suspend(std::coroutine_handle<cobalt_promise> h) noexcept
0384     {
0385       std::coroutine_handle<void> res = std::noop_coroutine();
0386       if (promise->receiver && promise->receiver->awaited_from.get() != nullptr)
0387         res = promise->receiver->awaited_from.release();
0388
0389
0390       if (auto &rec = h.promise().receiver; rec != nullptr)
0391       {
0392         if (!rec->done && !rec->exception)
0393           rec->exception = completed_unexpected();
0394         rec->set_done();
0395         rec->awaited_from.reset(nullptr);
0396         rec = nullptr;
0397       }
0398       detail::self_destroy(h);
0399       return res;
0400     }
0401
0402     void await_resume() noexcept
0403     {
0404     }
0405   };
0406
0407
0408 };
0409
0410 }
0411
0412 }
0413
0414 #endif //BOOST_COBALT_DETAIL_PROMISE_HPP