Class awaitable_spsc_queue

example/spsc.cpp:17–118

/// This is a simple class making a lockfree::spsc_queue awaitable. It's not movable, since the spsc_queue isn't.
template<typename T, typename ... Options>
struct awaitable_spsc_queue
{
  boost::lockfree::spsc_queue<T, Options...> queue;

  template<typename ... Args>
  awaitable_spsc_queue(Args && ... args) : queue(std::forward<Args>(args)...) {}

  // if the queue gets destroyed, destroying the awaiters is all we can do.
  ~awaitable_spsc_queue()
  {
    if (auto r = reader.load(); r != nullptr)
      std::coroutine_handle<void>::from_address(r).destroy();

    if (auto w = writer.load(); w != nullptr)
      std::coroutine_handle<void>::from_address(w).destroy();
  }
  // to avoid locks, put the coroutine handles into atomics.
  std::atomic<void*> reader{nullptr}, writer{nullptr};

  // capture the read & write executor
  cobalt::executor read_executor, write_executor;

  // the awaitable to read a value from the queue
  struct read_op
  {
    awaitable_spsc_queue * this_;

    // if reads are available we don't need to suspend
    bool await_ready() {return this_->queue.read_available();}

    // The suspend implementation. We expect the coroutine promise to have an associated executor.
    // If it doesn't resumption will end up on the system_executor
    template<typename Promise>
    void await_suspend(std::coroutine_handle<Promise> h)
    {
      // Capture the read_executor.
      if constexpr (requires {h.promise().get_executor();})
        this_->read_executor = h.promise().get_executor();
      else
        this_->read_executor = cobalt::this_thread::get_executor();

      // Make sure there's only one coroutine awaiting the read
      assert(this_->reader == nullptr);
      // Store the handle of the awaiter.
      this_->reader.store(h.address());
    }
    T await_resume()
    {
      T res;
      // Grab the value from the queue
      this_->queue.pop(res);
      // if a writer is waiting post it to complete on it's thread
      auto w = cobalt::unique_handle<void>::from_address(this_->writer.exchange(nullptr));
      if (w)
        boost::asio::post(this_->write_executor, std::move(w));

      return res;
    }