Line data    Source code

       1              : #include "async_event.hpp"
       2              : #include <boost/asio/append.hpp>
       3              : #include <boost/asio/execution_context.hpp>
       4              : #include <boost/asio/post.hpp>
       5              : #include <mutex>
       6              : #include <utility>
       7              : #include <vector>
       8              : 
       9              : namespace thinger::asio {
      10              : 
      11              : // Ends a wait whose cancellation slot was emitted. It stays in the slot until the waiter's next
      12              : // operation replaces it, so it acts once: a second emission finds nothing to cancel
      13              : struct async_event::canceller {
      14              :     async_event* event;
      15              :     std::list<waiter>::iterator it;
      16              : 
      17         2376 :     void operator()(boost::asio::cancellation_type type) {
      18         2376 :         if (type == boost::asio::cancellation_type::none || event == nullptr) return;
      19         1775 :         auto* cancelled = std::exchange(event, nullptr);
      20         1775 :         cancelled->complete(it, boost::asio::error::operation_aborted);
      21              :     }
      22              : };
      23              : 
      24              : // The events of one io_context. Its shutdown (the io_context destroyed) destroys the waits still
      25              : // pending on them, unresumed, as asio destroys the pending operations of a timer: a waiter parked
      26              : // here is no operation of the io_context, and a coroutine waiting on an event of an object it
      27              : // keeps alive (a connection) would otherwise never be freed
      28              : class async_event::registry : public boost::asio::execution_context::service {
      29              : public:
      30              :     static inline boost::asio::execution_context::id id;
      31              : 
      32         2932 :     explicit registry(boost::asio::execution_context& context) : service(context) {}
      33              : 
      34        34832 :     void add(async_event& event) {
      35        34832 :         std::lock_guard lock(mutex_);
      36        34832 :         event.next_ = head_;
      37        34832 :         if (head_ != nullptr) head_->prev_ = &event;
      38        34832 :         head_ = &event;
      39        34832 :         event.linked_ = true;
      40        34832 :     }
      41              : 
      42        34130 :     void remove(async_event& event) {
      43        34130 :         std::lock_guard lock(mutex_);
      44        34130 :         if (event.prev_ != nullptr) event.prev_->next_ = event.next_;
      45        25410 :         else head_ = event.next_;
      46        34130 :         if (event.next_ != nullptr) event.next_->prev_ = event.prev_;
      47        34130 :         event.prev_ = event.next_ = nullptr;
      48        34130 :         event.linked_ = false;
      49        34130 :     }
      50              : 
      51         5910 :     void shutdown() override {
      52         5910 :         std::vector<waiter> pending;
      53              :         {
      54         5910 :             std::lock_guard lock(mutex_);
      55         6612 :             for (auto* event = std::exchange(head_, nullptr); event != nullptr;) {
      56         1404 :                 for (auto& waiter : event->waiters_) {
      57          702 :                     if (waiter.slot.is_connected()) waiter.slot.clear();
      58          702 :                     pending.push_back(std::move(waiter));
      59              :                 }
      60          702 :                 event->waiters_.clear();
      61              :                 // Without waits, it leaves the registry
      62          702 :                 event->linked_ = false;
      63          702 :                 event->prev_ = nullptr;
      64          702 :                 event = std::exchange(event->next_, nullptr);
      65              :             }
      66         5910 :         }
      67              :         // Destroyed here, out of the lock: a coroutine destroyed may own events of this
      68              :         // io_context, which are no longer linked
      69         5910 :     }
      70              : 
      71              : private:
      72              :     std::mutex mutex_;
      73              :     async_event* head_ = nullptr;
      74              : };
      75              : 
      76        45888 : async_event::async_event(boost::asio::any_io_executor executor)
      77        45888 :     : executor_(std::move(executor)) {}
      78              : 
      79        45888 : async_event::~async_event() {
      80              :     // The last wait completed leaves the registry
      81        45888 :     complete_all(boost::asio::error::operation_aborted);
      82        45888 : }
      83              : 
      84        43584 : void async_event::set() {
      85        43584 :     set_ = true;
      86        43584 :     complete_all({});
      87        43584 : }
      88              : 
      89         3570 : void async_event::reset() {
      90         3570 :     set_ = false;
      91         3570 : }
      92              : 
      93           30 : bool async_event::is_set() const {
      94           30 :     return set_;
      95              : }
      96              : 
      97        34916 : awaitable<bool> async_event::wait() {
      98              :     if (set_) co_return true;
      99              :     auto [ec] = co_await boost::asio::async_initiate<decltype(use_nothrow_awaitable), void(boost::system::error_code)>(
     100        34850 :         [this](auto handler) { add_waiter(std::move(handler)); }, use_nothrow_awaitable);
     101              :     // The event may be gone (destroyed with this wait pending): only the result is used
     102              :     co_return !ec;
     103        69832 : }
     104              : 
     105        34850 : void async_event::add_waiter(boost::asio::any_completion_handler<void(boost::system::error_code)> handler) {
     106        34850 :     auto slot = boost::asio::get_associated_cancellation_slot(handler);
     107        34850 :     auto work = boost::asio::prefer(boost::asio::get_associated_executor(handler, executor_),
     108        34850 :                                     boost::asio::execution::outstanding_work.tracked);
     109        34850 :     auto it = waiters_.insert(waiters_.end(), waiter{std::move(handler), std::move(work), slot});
     110        34850 :     if (slot.is_connected()) slot.emplace<canceller>(this, it);
     111        34850 :     if (!linked_) {
     112        34832 :         if (registry_ == nullptr) {
     113        33028 :             registry_ = &boost::asio::use_service<registry>(boost::asio::query(executor_, boost::asio::execution::context));
     114              :         }
     115        34832 :         registry_->add(*this);
     116              :     }
     117        34850 : }
     118              : 
     119        34148 : void async_event::complete(std::list<waiter>::iterator it, boost::system::error_code ec) {
     120        34148 :     auto completed = std::move(*it);
     121        34148 :     waiters_.erase(it);
     122        34148 :     if (waiters_.empty() && linked_) registry_->remove(*this);
     123              :     // Posted, then run on the waiter's executor: it never resumes inside set() or the destructor
     124        34148 :     boost::asio::post(executor_, boost::asio::append(std::move(completed.handler), ec));
     125        34148 : }
     126              : 
     127        89472 : void async_event::complete_all(boost::system::error_code ec) {
     128       121845 :     while (!waiters_.empty()) {
     129              :         // Its canceller would act on an event that no longer knows the wait
     130        32373 :         if (waiters_.front().slot.is_connected()) waiters_.front().slot.clear();
     131        32373 :         complete(waiters_.begin(), ec);
     132              :     }
     133        89472 : }
     134              : 
     135              : }