buffers.hpp

Go to the documentation of this file.

 
 
#pragma once
 
#include "async_task.hpp"
#include "async_task_promise.hpp"
#include "async_traits.hpp"
#include "epoch_context.hpp"
#include "shared_storage.hpp"
#include <uni20/common/trace.hpp>
 
#include <atomic>
#include <concepts>
#include <coroutine>
#include <cstddef>
#include <memory>
#include <mutex>
#include <new>
#include <optional>
#include <stdexcept>
#include <tuple>
#include <type_traits>
#include <utility>
 
namespace uni20::async
{
 
template <typename T> class Async;
 
template <typename T> class ReadMaybeAwaiter;
template <typename T> class ReadOrCancelAwaiter;
template <typename T> class OwningReadAccessProxy;
template <typename T> class OwningReadAwaiter;
 
template <typename T> class ReadBuffer { //}: public AsyncAwaiter {
  public:
    using value_type = T;
 
    ReadBuffer(EpochContextReader<T> reader) : reader_(std::move(reader)) {}
 
    ReadBuffer(ReadBuffer const& other) : reader_(other.reader_) { this->copy_exception_sink_from(other); }
 
    // No copy ctor here, although we could add one
    ReadBuffer& operator=(ReadBuffer const&) = delete;
 
    ReadBuffer(ReadBuffer&& other) noexcept : reader_(std::move(other.reader_))
    {
      this->move_exception_sink_from(other);
    }
 
    ReadBuffer& operator=(ReadBuffer&& other) noexcept
    {
      if (this != &other)
      {
        this->unregister_exception_sink(false);
        reader_ = std::move(other.reader_);
        this->move_exception_sink_from(other);
      }
      return *this;
    }
 
    ~ReadBuffer() noexcept { this->unregister_exception_sink(true); }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return reader_.node(); }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Reader; }
#endif
 
    ReadMaybeAwaiter<T const&> maybe() &;
 
    ReadMaybeAwaiter<T> maybe() &&;
 
    ReadOrCancelAwaiter<T const&> or_cancel() &;
 
    ReadOrCancelAwaiter<T> or_cancel() &&;
 
    [[nodiscard]] bool await_ready() const noexcept { return reader_.ready(); }
 
    void await_suspend(BasicTask&& t) noexcept
    {
      TRACE_MODULE(ASYNC, "ReadBuffer::await_suspend()", this, t.coroutine_handle());
      reader_.suspend(std::move(t));
    }
 
    [[nodiscard]] T const& await_resume() const& { return reader_.data(); }
 
    [[nodiscard]] OwningReadAccessProxy<T> await_resume() &&;
 
    void release() noexcept { reader_.release(); }
 
    ReadBuffer&& transfer() & noexcept { return std::move(*this); }
 
    ReadBuffer&& transfer() && noexcept { return std::move(*this); }
 
    void wait() const { this->reader_.wait(); }
 
    void wait(IAsyncScheduler& sched) const { this->reader_.wait(sched); }
 
    // T get_wait() && { return T(reader_.get_wait()); } // TODO: can this use move semantics?
    [[nodiscard]] T const& get_wait() const { return reader_.get_wait(); }
 
    [[nodiscard]] T const& get_wait(IAsyncScheduler& sched) const { return reader_.get_wait(sched); }
 
    auto operator co_await() & noexcept -> ReadBuffer& { return *this; }
    auto operator co_await() const& noexcept -> ReadBuffer const& { return *this; }
 
    auto operator co_await() && noexcept -> OwningReadAwaiter<T> { return OwningReadAwaiter<T>(std::move(reader_)); }
 
    std::shared_ptr<EpochContext> epoch_context_shared() const noexcept { return reader_.epoch_context_shared(); }
 
    void register_exception_sink(TaskPromiseBase& promise, bool explicit_sink) const
    {
      promise.register_exception_sink(exception_sink_, this->epoch_context_shared(), explicit_sink);
    }
 
  private:
    void copy_exception_sink_from(ReadBuffer const& other)
    {
      if (other.exception_sink_.owner)
      {
        other.exception_sink_.owner->register_exception_sink(exception_sink_, other.exception_sink_.epoch,
                                                             other.exception_sink_.explicit_sink);
      }
    }
 
    void move_exception_sink_from(ReadBuffer& other) noexcept
    {
      if (!other.exception_sink_.owner) return;
      auto* owner = other.exception_sink_.owner;
      auto epoch = std::move(other.exception_sink_.epoch);
      bool const explicit_sink = other.exception_sink_.explicit_sink;
      owner->unregister_exception_sink(other.exception_sink_, false);
      owner->register_exception_sink(exception_sink_, std::move(epoch), explicit_sink);
    }
 
    void unregister_exception_sink(bool from_destructor) noexcept
    {
      if (!exception_sink_.owner) return;
      exception_sink_.owner->unregister_exception_sink(exception_sink_, from_destructor);
    }
 
    EpochContextReader<T> reader_; 
    mutable TaskPromiseBase::ExceptionSinkNode exception_sink_{};
};
 
template <typename T> ReadBuffer<T>&& release(ReadBuffer<T>& in) { return std::move(in); }
 
template <typename T> ReadBuffer<T>&& release(ReadBuffer<T>&& in) { return std::move(in); }
 
template <typename T> class OwningReadAccessProxy {
  public:
    using value_type = T;
 
    OwningReadAccessProxy() = delete;
    OwningReadAccessProxy(OwningReadAccessProxy const&) = delete;
    OwningReadAccessProxy& operator=(OwningReadAccessProxy const&) = delete;
    OwningReadAccessProxy(OwningReadAccessProxy&&) noexcept = default;
    OwningReadAccessProxy& operator=(OwningReadAccessProxy&&) noexcept = delete;
 
    [[nodiscard]] T const& get() const { return this->reader_.data(); }
 
    operator T const&() const { return this->get(); }
 
    T const* operator->() const { return std::addressof(this->get()); }
 
    T get_release()
    {
      T value = this->get();
      this->release();
      return value;
    }
 
    void release() noexcept { this->reader_.release(); }
 
  private:
    explicit OwningReadAccessProxy(EpochContextReader<T>&& reader) : reader_(std::move(reader)) {}
 
    friend class ReadBuffer<T>;
    friend class OwningReadAwaiter<T>;
    friend class ReadMaybeAwaiter<T>;
    friend class ReadOrCancelAwaiter<T>;
 
    EpochContextReader<T> reader_;
};
 
template <typename T> class OwningReadAwaiter {
  public:
    using value_type = T;
 
    explicit OwningReadAwaiter(EpochContextReader<T>&& reader) : reader_(std::move(reader)) {}
 
    [[nodiscard]] bool await_ready() const noexcept { return this->reader_.ready(); }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return reader_.node(); }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Reader; }
#endif
 
    void await_suspend(BasicTask&& t) noexcept
    {
      TRACE_MODULE(ASYNC, "OwningReadAwaiter::await_suspend()", this, t.coroutine_handle());
      this->reader_.suspend(std::move(t));
    }
 
    [[nodiscard]] OwningReadAccessProxy<T> await_resume() { return OwningReadAccessProxy<T>(std::move(this->reader_)); }
 
  private:
    EpochContextReader<T> reader_;
};
 
template <typename T> [[nodiscard]] OwningReadAccessProxy<T> ReadBuffer<T>::await_resume() &&
{
  return OwningReadAccessProxy<T>(std::move(reader_));
}
 
template <typename T> class ReadMaybeAwaiter {
  public:
    using value_type = std::optional<OwningReadAccessProxy<T>>;
 
    ReadMaybeAwaiter(ReadMaybeAwaiter&&) = default; // movable
 
    [[nodiscard]] bool await_ready() const noexcept { return reader_.ready(); }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return reader_.node(); }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Reader; }
#endif
 
    void await_suspend(BasicTask&& t) noexcept
    {
      TRACE_MODULE(ASYNC, "ReadBuffer::await_suspend()", this, t.coroutine_handle());
      this->reader_.suspend(std::move(t));
    }
 
    [[nodiscard]] value_type await_resume()
    {
      if (this->reader_.data_maybe() == nullptr)
      {
        this->reader_.release();
        return std::nullopt;
      }
      return value_type(OwningReadAccessProxy<T>(std::move(this->reader_)));
    }
 
  private:
    ReadMaybeAwaiter() = delete;
    ReadMaybeAwaiter(ReadMaybeAwaiter const&) = delete;
    ReadMaybeAwaiter& operator=(ReadMaybeAwaiter const&) = delete;
    ReadMaybeAwaiter& operator=(ReadMaybeAwaiter&&) = delete;
 
    ReadMaybeAwaiter(EpochContextReader<T>&& reader) : reader_(std::move(reader)) {}
 
    friend ReadBuffer<T>;
 
    EpochContextReader<T> reader_; 
};
 
template <typename T> class ReadMaybeAwaiter<T const&> {
  public:
    using value_type = T const*;
 
    ReadMaybeAwaiter(ReadMaybeAwaiter&&) = default; // movable
 
    [[nodiscard]] bool await_ready() const noexcept { return reader_.ready(); }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return reader_.node(); }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Reader; }
#endif
 
    void await_suspend(BasicTask&& t) noexcept
    {
      TRACE_MODULE(ASYNC, "ReadBuffer::await_suspend()", this, t.coroutine_handle());
      reader_.suspend(std::move(t));
    }
 
    [[nodiscard]] value_type await_resume() { return reader_.data_maybe(); }
 
  private:
    ReadMaybeAwaiter() = delete;
    ReadMaybeAwaiter(ReadMaybeAwaiter const&) = delete;
    ReadMaybeAwaiter& operator=(ReadMaybeAwaiter const&) = delete;
    ReadMaybeAwaiter& operator=(ReadMaybeAwaiter&&) = delete;
 
    ReadMaybeAwaiter(EpochContextReader<T>& reader) : reader_(reader) {}
 
    friend ReadBuffer<T>;
 
    EpochContextReader<T>& reader_; 
};
 
template <typename T> class ReadOrCancelAwaiter {
  public:
    using value_type = OwningReadAccessProxy<T>;
 
    ReadOrCancelAwaiter(ReadOrCancelAwaiter&&) = default; // movable
 
    [[nodiscard]] bool await_ready() const noexcept
    {
      // we must suspend here, because it is a possible cancellation point
      return false;
    }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return reader_.node(); }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Reader; }
#endif
 
    void await_suspend(BasicTask&& t) noexcept { this->reader_.suspend(std::move(t)); }
 
    [[nodiscard]] value_type await_resume()
    {
      T const* ptr = this->reader_.data_maybe();
      if (ptr == nullptr)
      {
        this->reader_.release();
        throw task_cancelled();
      }
      return value_type(std::move(this->reader_));
    }
 
  private:
    ReadOrCancelAwaiter() = delete;
    ReadOrCancelAwaiter(ReadOrCancelAwaiter const&) = delete;
    ReadOrCancelAwaiter& operator=(ReadOrCancelAwaiter const&) = delete;
    ReadOrCancelAwaiter& operator=(ReadOrCancelAwaiter&&) = delete;
 
    ReadOrCancelAwaiter(EpochContextReader<T>&& reader) : reader_(std::move(reader)) {}
 
    friend ReadBuffer<T>;
 
    EpochContextReader<T> reader_; 
};
 
template <typename T> class ReadOrCancelAwaiter<T const&> {
  public:
    using value_type = T;
 
    ReadOrCancelAwaiter(ReadOrCancelAwaiter&&) = default; // movable
 
    [[nodiscard]] bool await_ready() const noexcept
    {
      // we must suspend here, because it is a possible cancellation point
      return false;
    }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return reader_.node(); }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Reader; }
#endif
 
    void await_suspend(BasicTask&& t) noexcept { reader_.suspend(std::move(t)); }
 
    [[nodiscard]] T const& await_resume()
    {
      T const* ptr = reader_.data_maybe();
      if (!ptr) throw task_cancelled();
      return *ptr;
    }
 
  private:
    ReadOrCancelAwaiter() = delete;
    ReadOrCancelAwaiter(ReadOrCancelAwaiter const&) = delete;
    ReadOrCancelAwaiter& operator=(ReadOrCancelAwaiter const&) = delete;
    ReadOrCancelAwaiter& operator=(ReadOrCancelAwaiter&&) = delete;
 
    ReadOrCancelAwaiter(EpochContextReader<T>& reader) : reader_(reader) {}
 
    friend ReadBuffer<T>;
 
    EpochContextReader<T>& reader_; 
};
 
template <typename T> ReadMaybeAwaiter<T const&> ReadBuffer<T>::maybe() &
{
  return ReadMaybeAwaiter<T const&>(reader_);
}
 
template <typename T> ReadMaybeAwaiter<T> ReadBuffer<T>::maybe() && { return ReadMaybeAwaiter<T>(std::move(reader_)); }
 
template <typename T> ReadOrCancelAwaiter<T const&> ReadBuffer<T>::or_cancel() &
{
  return ReadOrCancelAwaiter<T const&>(reader_);
}
 
template <typename T> ReadOrCancelAwaiter<T> ReadBuffer<T>::or_cancel() &&
{
  return ReadOrCancelAwaiter<T>(std::move(reader_));
}
 
// Forward declaration of the proxy used for deferred writes
template <typename T> class WriteAccessProxy;
template <typename T> class OwningWriteAccessProxy;
template <typename T> class OwningWriteAwaiter;
template <typename T> class OwningStorageAccessProxy;
template <typename T> class OwningStorageAwaiter;
template <typename T> class OwningTakeAwaiter;
template <typename T> class WriteAssignProxy;
 
#if UNI20_DEBUG_ASYNC_TASKS
struct WriteProxyLifetimeState
{
    std::atomic<bool> alive{true};
};
#endif
 
template <typename T> class StorageAwaiter {
  public:
    StorageAwaiter(EpochContextWriter<T>* writer) : writer_(writer) {}
 
    [[nodiscard]] bool await_ready() const noexcept { return writer_->ready(); }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return writer_ ? writer_->node() : nullptr; }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Writer; }
#endif
 
    void await_suspend(BasicTask&& t) noexcept
    {
      TRACE_MODULE(ASYNC, "StorageAwaiter::await_suspend()", this, t.coroutine_handle());
      writer_->suspend(std::move(t));
    }
 
    [[nodiscard]] shared_storage<T>& await_resume()
    {
      writer_->resume();
      return writer_->storage();
    }
 
    [[nodiscard]] std::shared_ptr<EpochContext> epoch_context_shared() const noexcept
    {
      return writer_->epoch_context_shared();
    }
 
  private:
    EpochContextWriter<T>* writer_; // by pointer, since we don't want to take ownership
};
 
template <typename T> class TakeAwaiter {
  public:
    TakeAwaiter(EpochContextWriter<T>* writer) : writer_(writer) {}
 
    [[nodiscard]] bool await_ready() const noexcept { return writer_->ready(); }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return writer_ ? writer_->node() : nullptr; }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Writer; }
#endif
 
    void await_suspend(BasicTask&& t) noexcept
    {
      TRACE_MODULE(ASYNC, "TakeAwaiter::await_suspend()", this, t.coroutine_handle());
      writer_->suspend(std::move(t));
    }
 
    [[nodiscard]] T await_resume()
    {
      writer_->resume();
      return writer_->storage().take();
    }
 
    [[nodiscard]] std::shared_ptr<EpochContext> epoch_context_shared() const noexcept
    {
      return writer_->epoch_context_shared();
    }
 
  private:
    EpochContextWriter<T>* writer_; // by pointer, since we don't want to take ownership
};
 
template <typename T> class TakeReleaseAwaiter {
  public:
    TakeReleaseAwaiter(EpochContextWriter<T>* writer) : writer_(writer) {}
 
    [[nodiscard]] bool await_ready() const noexcept { return writer_->ready(); }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return writer_ ? writer_->node() : nullptr; }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Writer; }
#endif
 
    void await_suspend(BasicTask&& t) noexcept
    {
      TRACE_MODULE(ASYNC, "TakeReleaseAwaiter::await_suspend()", this, t.coroutine_handle());
      writer_->suspend(std::move(t));
    }
 
    [[nodiscard]] T await_resume()
    {
      writer_->resume();
      T value = writer_->storage().take();
      writer_->release();
      return value;
    }
 
    [[nodiscard]] std::shared_ptr<EpochContext> epoch_context_shared() const noexcept
    {
      return writer_->epoch_context_shared();
    }
 
  private:
    EpochContextWriter<T>* writer_; // by pointer, since we don't want to take ownership
};
 
template <typename T> class OwningStorageAccessProxy {
  public:
    using value_type = shared_storage<T>;
 
    OwningStorageAccessProxy() = delete;
    OwningStorageAccessProxy(OwningStorageAccessProxy const&) = delete;
    OwningStorageAccessProxy& operator=(OwningStorageAccessProxy const&) = delete;
    OwningStorageAccessProxy(OwningStorageAccessProxy&&) noexcept = default;
    OwningStorageAccessProxy& operator=(OwningStorageAccessProxy&&) noexcept = delete;
 
    shared_storage<T>& get() { return writer_.storage(); }
    shared_storage<T> const& get() const { return writer_.storage(); }
 
    operator shared_storage<T>&() { return this->get(); }
    operator shared_storage<T> const&() const { return this->get(); }
 
    shared_storage<T>* operator->() { return std::addressof(this->get()); }
    shared_storage<T> const* operator->() const { return std::addressof(this->get()); }
 
    T take() { return writer_.storage().take(); }
 
    T take_release()
    {
      T value = this->take();
      this->release();
      return value;
    }
 
    void release() noexcept { writer_.release(); }
 
  private:
    explicit OwningStorageAccessProxy(EpochContextWriter<T>&& writer) : writer_(std::move(writer)) {}
 
    friend class OwningStorageAwaiter<T>;
 
    EpochContextWriter<T> writer_;
};
 
template <typename T> class OwningStorageAwaiter {
  public:
    using value_type = shared_storage<T>;
 
    explicit OwningStorageAwaiter(EpochContextWriter<T>&& writer) : writer_(std::move(writer)) {}
 
    [[nodiscard]] bool await_ready() const noexcept { return writer_.ready(); }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return writer_.node(); }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Writer; }
#endif
 
    void await_suspend(BasicTask&& t) noexcept
    {
      TRACE_MODULE(ASYNC, "OwningStorageAwaiter::await_suspend()", this, t.coroutine_handle());
      writer_.suspend(std::move(t));
    }
 
    [[nodiscard]] OwningStorageAccessProxy<T> await_resume()
    {
      writer_.resume();
      return OwningStorageAccessProxy<T>(std::move(writer_));
    }
 
    [[nodiscard]] std::shared_ptr<EpochContext> epoch_context_shared() const noexcept
    {
      return writer_.epoch_context_shared();
    }
 
  private:
    EpochContextWriter<T> writer_;
};
 
template <typename T> class OwningTakeAwaiter {
  public:
    explicit OwningTakeAwaiter(EpochContextWriter<T>&& writer) : writer_(std::move(writer)) {}
 
    [[nodiscard]] bool await_ready() const noexcept { return writer_.ready(); }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return writer_.node(); }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Writer; }
#endif
 
    void await_suspend(BasicTask&& t) noexcept
    {
      TRACE_MODULE(ASYNC, "OwningTakeAwaiter::await_suspend()", this, t.coroutine_handle());
      writer_.suspend(std::move(t));
    }
 
    [[nodiscard]] T await_resume()
    {
      writer_.resume();
      T value = writer_.storage().take();
      writer_.release();
      return value;
    }
 
    [[nodiscard]] std::shared_ptr<EpochContext> epoch_context_shared() const noexcept
    {
      return writer_.epoch_context_shared();
    }
 
  private:
    EpochContextWriter<T> writer_;
};
 
template <typename T> class WriteBuffer {
  public:
    using value_type = T;
    using element_type = std::conditional_t<is_async_alias_v<T>, T const&, T&>;
 
    explicit WriteBuffer(EpochContextWriter<T> writer) : writer_(std::move(writer)) {}
 
    ~WriteBuffer() noexcept
    {
      this->invalidate_proxy_state();
      this->unregister_exception_sink(true);
    }
 
    WriteBuffer(WriteBuffer const&) = delete;
    WriteBuffer& operator=(WriteBuffer const&) = delete;
 
    WriteBuffer(WriteBuffer&& other) noexcept : writer_(std::move(other.writer_))
    {
      this->move_exception_sink_from(other);
      other.invalidate_proxy_state();
    }
 
    WriteBuffer& operator=(WriteBuffer&& other) noexcept
    {
      if (this != &other)
      {
        this->invalidate_proxy_state();
        this->unregister_exception_sink(false);
        writer_ = std::move(other.writer_);
        this->move_exception_sink_from(other);
        this->reset_proxy_state();
        other.invalidate_proxy_state();
      }
      return *this;
    }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return writer_.node(); }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Writer; }
#endif
 
    [[nodiscard]] bool await_ready() const noexcept { return writer_.ready(); }
 
    void await_suspend(BasicTask&& t) noexcept
    {
      TRACE_MODULE(ASYNC, "WriteBuffer::await_suspend()", this, t.coroutine_handle());
      writer_.suspend(std::move(t));
    }
 
    [[nodiscard]] WriteAccessProxy<T> await_resume() &
    {
      writer_.resume();
      return WriteAccessProxy<T>(&writer_
#if UNI20_DEBUG_ASYNC_TASKS
                                 ,
                                 proxy_state_
#endif
      );
    }
 
    [[nodiscard]] WriteAccessProxy<T> await_resume() const&
    {
      auto* writer = const_cast<EpochContextWriter<T>*>(&writer_);
      writer->resume();
      return WriteAccessProxy<T>(writer
#if UNI20_DEBUG_ASYNC_TASKS
                                 ,
                                 proxy_state_
#endif
      );
    }
 
    [[nodiscard]] OwningWriteAccessProxy<T> await_resume() &&
    {
      writer_.resume();
      return OwningWriteAccessProxy<T>(std::move(writer_));
    }
 
    void release() noexcept
    {
      this->invalidate_proxy_state();
      writer_.release();
    }
 
    WriteBuffer&& transfer() & noexcept
    {
      this->invalidate_proxy_state();
      return std::move(*this);
    }
 
    WriteBuffer&& transfer() && noexcept
    {
      this->invalidate_proxy_state();
      return std::move(*this);
    }
 
    template <typename... Args>
      requires(!is_async_alias_v<T> && std::constructible_from<T, Args...>)
    T& emplace_assert(Args&&... args)
    {
      DEBUG_CHECK(writer_.ready(), "WriteBuffer must be immediately writable");
      writer_.emplace(std::forward<Args>(args)...);
      return writer_.data();
    }
 
    auto take() &
      requires(!is_async_alias_v<T>)
    {
      return TakeAwaiter<T>(&writer_);
    }
 
    auto take() && -> OwningTakeAwaiter<T>
      requires(!is_async_alias_v<T>)
    {
      this->invalidate_proxy_state();
      return OwningTakeAwaiter<T>(std::move(writer_));
    }
 
    auto take_release() &
      requires(!is_async_alias_v<T>)
    {
      this->invalidate_proxy_state();
      return TakeReleaseAwaiter<T>(&writer_);
    }
 
    auto take_release() && -> OwningTakeAwaiter<T>
      requires(!is_async_alias_v<T>)
    {
      this->invalidate_proxy_state();
      return OwningTakeAwaiter<T>(std::move(writer_));
    }
 
    auto storage() &
      requires(!is_async_alias_v<T>)
    {
      return StorageAwaiter<T>(&writer_);
    }
 
    auto storage() && -> OwningStorageAwaiter<T>
      requires(!is_async_alias_v<T>)
    {
      this->invalidate_proxy_state();
      return OwningStorageAwaiter<T>(std::move(writer_));
    }
 
    std::shared_ptr<EpochContext> epoch_context_shared() const noexcept { return writer_.epoch_context_shared(); }
 
    void register_exception_sink(TaskPromiseBase& promise, bool explicit_sink) const
    {
      promise.register_exception_sink(exception_sink_, this->epoch_context_shared(), explicit_sink);
    }
 
    T move_from_wait()
      requires(!is_async_alias_v<T>)
    {
      return writer_.move_from_wait();
    }
 
    template <typename U>
      requires(!is_async_alias_v<T> || detail::async_alias_assignment_source<T, U>)
    void write(U&& val)
    {
      async_assign(std::move(*this), std::forward<U>(val));
    }
 
    template <typename U>
    void write_assert(U&& val)
      requires((!is_async_alias_v<T> && std::assignable_from<T&, U &&>) || detail::async_alias_assignable_from<T, U &&>)
    {
      DEBUG_CHECK(writer_.ready(), "WriteBuffer must be immediately writable");
      if constexpr (is_async_alias_v<T>)
        detail::assign_async_alias(writer_.data(), std::forward<U>(val));
      else
        writer_.data() = std::forward<U>(val);
    }
 
    template <typename U>
    void write_move_assert(U&& val)
      requires(!is_async_alias_v<T> && std::assignable_from<T&, U &&>)
    {
      DEBUG_CHECK(writer_.ready(), "WriteBuffer must be immediately writable");
      writer_.data() = std::move(val);
    }
 
    [[nodiscard]] WriteAssignProxy<T> write();
 
    template <typename U>
      requires(!is_async_alias_v<T>)
    void write_move(U&& val)
    {
      async_move(std::move(*this), std::move(val));
    }
 
    auto operator co_await() & noexcept -> WriteBuffer& { return *this; }
    auto operator co_await() const& noexcept -> WriteBuffer const& { return *this; }
    auto operator co_await() && noexcept -> OwningWriteAwaiter<T>
    {
      this->invalidate_proxy_state();
      return OwningWriteAwaiter<T>(std::move(writer_));
    }
 
  private:
    void move_exception_sink_from(WriteBuffer& other) noexcept
    {
      if (!other.exception_sink_.owner) return;
      auto* owner = other.exception_sink_.owner;
      auto epoch = std::move(other.exception_sink_.epoch);
      bool const explicit_sink = other.exception_sink_.explicit_sink;
      owner->unregister_exception_sink(other.exception_sink_, false);
      owner->register_exception_sink(exception_sink_, std::move(epoch), explicit_sink);
    }
 
    void unregister_exception_sink(bool from_destructor) noexcept
    {
      if (!exception_sink_.owner) return;
      exception_sink_.owner->unregister_exception_sink(exception_sink_, from_destructor);
    }
 
    void invalidate_proxy_state() noexcept
    {
#if UNI20_DEBUG_ASYNC_TASKS
      if (proxy_state_) proxy_state_->alive.store(false, std::memory_order_release);
#endif
    }
 
    void reset_proxy_state() noexcept
    {
#if UNI20_DEBUG_ASYNC_TASKS
      proxy_state_ = std::make_shared<WriteProxyLifetimeState>();
#endif
    }
 
    EpochContextWriter<T> writer_;
    mutable TaskPromiseBase::ExceptionSinkNode exception_sink_{};
#if UNI20_DEBUG_ASYNC_TASKS
    std::shared_ptr<WriteProxyLifetimeState> proxy_state_{std::make_shared<WriteProxyLifetimeState>()};
#endif
 
    // Duplicating a WriteBuffer is no longer possible with the refactored EpochContext.
    // To re-enable this function we would need to make the
    // friend WriteBuffer dup(WriteBuffer& wb) { return WriteBuffer(wb.writer_); }
};
 
// For a ReadBuffer, we add the node to the ReadDependencies
template <typename T> void ProcessCoroutineArgument(TaskPromiseBase* promise, ReadBuffer<T> const& x)
{
#if UNI20_DEBUG_DAG
  promise->ReadDependencies.push_back(x.node());
#endif
}
 
// For a WriteBuffer, we add the node to the WriteDependencies
template <typename T> void ProcessCoroutineArgument(TaskPromiseBase* promise, WriteBuffer<T> const& x)
{
#if UNI20_DEBUG_DAG
  promise->WriteDependencies.push_back(x.node());
#endif
  x.register_exception_sink(*promise, false);
}
 
template <typename T> class Defer;
 
template <typename B>
concept exception_sink_buffer =
    requires(B& buffer, TaskPromiseBase& promise) { buffer.register_exception_sink(promise, true); };
 
template <exception_sink_buffer... Buffers> class PropagateExceptionsAwaiter {
  public:
    explicit PropagateExceptionsAwaiter(Buffers&... buffers) : buffers_(std::addressof(buffers)...) {}
 
    [[nodiscard]] bool await_ready() const noexcept { return true; }
 
    BasicTask await_suspend(BasicTask&& task) noexcept { return std::move(task); }
 
    void await_resume() noexcept {}
 
    void register_exception_sinks(TaskPromiseBase& promise) const
    {
      std::apply([&](auto*... buffers) { (buffers->register_exception_sink(promise, true), ...); }, buffers_);
    }
 
  private:
    std::tuple<Buffers*...> buffers_;
};
 
template <exception_sink_buffer... Buffers> auto propagate_exceptions_to(Buffers&... buffers)
{
  return PropagateExceptionsAwaiter<Buffers...>(buffers...);
}
 
namespace detail
{
 
template <typename T, typename U>
concept write_proxy_assignable_source = std::constructible_from<T, U&&> && std::is_assignable_v<T&, U&&>;
 
template <typename T, typename U>
  requires write_proxy_assignable_source<T, U>
void assign_write_proxy_value(EpochContextWriter<T>& writer, U&& source)
{
  if (writer.storage().constructed())
    writer.data() = std::forward<U>(source);
  else
    writer.emplace(std::forward<U>(source));
}
 
} // namespace detail
 
template <typename T> class WriteAccessProxy {
  public:
    using value_type = T;
 
    WriteAccessProxy() = delete;
    WriteAccessProxy(WriteAccessProxy const&) noexcept = default;
    WriteAccessProxy& operator=(WriteAccessProxy const&) = delete;
    WriteAccessProxy(WriteAccessProxy&&) noexcept = default;
    WriteAccessProxy& operator=(WriteAccessProxy&&) = delete;
 
    template <typename U>
      requires(!std::same_as<std::remove_cvref_t<U>, WriteAccessProxy>) &&
              ((!is_async_alias_v<T> && detail::write_proxy_assignable_source<T, U>) ||
               detail::async_alias_assignable_from<T, U &&>)
    WriteAccessProxy& operator=(U&& u)
    {
      if constexpr (is_async_alias_v<T>)
        detail::assign_async_alias(this->writer().data(), std::forward<U>(u));
      else
        detail::assign_write_proxy_value(this->writer(), std::forward<U>(u));
      return *this;
    }
 
    template <typename... Args>
      requires(!is_async_alias_v<T> && std::constructible_from<T, Args...>)
    T& emplace(Args&&... args)
    {
      return this->writer().emplace(std::forward<Args>(args)...);
    }
 
    template <typename U>
      requires(!is_async_alias_v<T>) && requires(T& value, U&& x) { value += std::forward<U>(x); }
    WriteAccessProxy& operator+=(U&& x)
    {
      auto& storage = this->writer().storage();
      if (storage.constructed())
      {
        this->get() += std::forward<U>(x);
      }
      else if constexpr (std::constructible_from<T, U&&>)
      {
        storage.emplace(std::forward<U>(x));
      }
      else
      {
        this->get() += std::forward<U>(x);
      }
      return *this;
    }
 
    template <typename U>
      requires(!is_async_alias_v<T>) && requires(T& value, U&& x) { value -= std::forward<U>(x); }
    WriteAccessProxy& operator-=(U&& x)
    {
      auto& storage = this->writer().storage();
      if (storage.constructed())
      {
        this->get() -= std::forward<U>(x);
      }
      else if constexpr (requires { -std::declval<U&&>(); } &&
                         std::constructible_from<T, decltype(-std::declval<U&&>())>)
      {
        auto neg = -std::forward<U>(x);
        storage.emplace(std::move(neg));
      }
      else
      {
        this->get() -= std::forward<U>(x);
      }
      return *this;
    }
 
    template <typename U>
      requires(!is_async_alias_v<T>) && requires(T& value, U&& x) { value *= std::forward<U>(x); }
    WriteAccessProxy& operator*=(U&& x)
    {
      this->get() *= std::forward<U>(x);
      return *this;
    }
 
    template <typename U>
      requires(!is_async_alias_v<T>) && requires(T& value, U&& x) { value /= std::forward<U>(x); }
    WriteAccessProxy& operator/=(U&& x)
    {
      this->get() /= std::forward<U>(x);
      return *this;
    }
 
    T take()
      requires(!is_async_alias_v<T>)
    {
      return this->writer().storage().take();
    }
 
    T take_release()
      requires(!is_async_alias_v<T>)
    {
      T value = this->take();
      this->release();
      return value;
    }
 
    void release() noexcept
    {
#if UNI20_DEBUG_ASYNC_TASKS
      if (auto state = proxy_state_.lock(); state)
      {
        state->alive.store(false, std::memory_order_release);
      }
#endif
      if (writer_)
      {
        writer_->release();
        writer_ = nullptr;
      }
    }
 
    [[nodiscard]] T& get() const
      requires(!is_async_alias_v<T>)
    {
      return this->writer().data();
    }
 
    [[nodiscard]] T const& get() const
      requires(is_async_alias_v<T>)
    {
      return this->writer().data();
    }
 
    operator T&() const
      requires(!is_async_alias_v<T>)
    {
      return this->get();
    }
 
    operator T const&() const
      requires(is_async_alias_v<T>)
    {
      return this->get();
    }
 
    T* operator->() const
      requires(!is_async_alias_v<T>)
    {
      return std::addressof(this->get());
    }
 
    T const* operator->() const
      requires(is_async_alias_v<T>)
    {
      return std::addressof(this->get());
    }
 
  private:
#if UNI20_DEBUG_ASYNC_TASKS
#endif
    explicit WriteAccessProxy(EpochContextWriter<T>* writer
#if UNI20_DEBUG_ASYNC_TASKS
                              ,
                              std::shared_ptr<WriteProxyLifetimeState> proxy_state
#endif
                              ) noexcept
        : writer_(writer)
#if UNI20_DEBUG_ASYNC_TASKS
          ,
          proxy_state_(std::move(proxy_state))
#endif
    {}
 
    [[nodiscard]] EpochContextWriter<T>& writer() const
    {
#if UNI20_DEBUG_ASYNC_TASKS
      auto state = proxy_state_.lock();
      DEBUG_CHECK(state && state->alive.load(std::memory_order_acquire),
                  "WriteAccessProxy used after WriteBuffer moved, released, or destroyed");
#endif
      DEBUG_CHECK(writer_, "WriteAccessProxy has no writer");
      return *writer_;
    }
 
    friend class WriteBuffer<T>;
 
    EpochContextWriter<T>* writer_;
#if UNI20_DEBUG_ASYNC_TASKS
    std::weak_ptr<WriteProxyLifetimeState> proxy_state_;
#endif
};
 
template <typename T> class OwningWriteAccessProxy {
  public:
    using value_type = T;
 
    OwningWriteAccessProxy() = delete;
    OwningWriteAccessProxy(OwningWriteAccessProxy const&) = delete;
    OwningWriteAccessProxy& operator=(OwningWriteAccessProxy const&) = delete;
    OwningWriteAccessProxy(OwningWriteAccessProxy&&) noexcept = default;
    OwningWriteAccessProxy& operator=(OwningWriteAccessProxy&&) noexcept = delete;
 
    template <typename U>
      requires(!std::same_as<std::remove_cvref_t<U>, OwningWriteAccessProxy>) &&
              ((!is_async_alias_v<T> && detail::write_proxy_assignable_source<T, U>) ||
               detail::async_alias_assignable_from<T, U &&>)
    OwningWriteAccessProxy& operator=(U&& u)
    {
      if constexpr (is_async_alias_v<T>)
        detail::assign_async_alias(writer_.data(), std::forward<U>(u));
      else
        detail::assign_write_proxy_value(writer_, std::forward<U>(u));
      return *this;
    }
 
    template <typename... Args>
      requires(!is_async_alias_v<T> && std::constructible_from<T, Args...>)
    T& emplace(Args&&... args)
    {
      return writer_.emplace(std::forward<Args>(args)...);
    }
 
    template <typename U>
      requires(!is_async_alias_v<T>) && requires(T& value, U&& x) { value += std::forward<U>(x); }
    OwningWriteAccessProxy& operator+=(U&& x)
    {
      auto& storage = writer_.storage();
      if (storage.constructed())
      {
        this->get() += std::forward<U>(x);
      }
      else if constexpr (std::constructible_from<T, U&&>)
      {
        storage.emplace(std::forward<U>(x));
      }
      else
      {
        this->get() += std::forward<U>(x);
      }
      return *this;
    }
 
    template <typename U>
      requires(!is_async_alias_v<T>) && requires(T& value, U&& x) { value -= std::forward<U>(x); }
    OwningWriteAccessProxy& operator-=(U&& x)
    {
      auto& storage = writer_.storage();
      if (storage.constructed())
      {
        this->get() -= std::forward<U>(x);
      }
      else if constexpr (requires { -std::declval<U&&>(); } &&
                         std::constructible_from<T, decltype(-std::declval<U&&>())>)
      {
        auto neg = -std::forward<U>(x);
        storage.emplace(std::move(neg));
      }
      else
      {
        this->get() -= std::forward<U>(x);
      }
      return *this;
    }
 
    template <typename U>
      requires(!is_async_alias_v<T>) && requires(T& value, U&& x) { value *= std::forward<U>(x); }
    OwningWriteAccessProxy& operator*=(U&& x)
    {
      this->get() *= std::forward<U>(x);
      return *this;
    }
 
    template <typename U>
      requires(!is_async_alias_v<T>) && requires(T& value, U&& x) { value /= std::forward<U>(x); }
    OwningWriteAccessProxy& operator/=(U&& x)
    {
      this->get() /= std::forward<U>(x);
      return *this;
    }
 
    T& get()
      requires(!is_async_alias_v<T>)
    {
      return writer_.data();
    }
    T const& get() const { return writer_.data(); }
 
    T take()
      requires(!is_async_alias_v<T>)
    {
      return writer_.storage().take();
    }
 
    T take_release()
      requires(!is_async_alias_v<T>)
    {
      T value = this->take();
      this->release();
      return value;
    }
 
    void release() noexcept { writer_.release(); }
 
    operator T&()
      requires(!is_async_alias_v<T>)
    {
      return this->get();
    }
    operator T const&() const { return this->get(); }
 
    T* operator->()
      requires(!is_async_alias_v<T>)
    {
      return std::addressof(this->get());
    }
    T const* operator->() const { return std::addressof(this->get()); }
 
  private:
    explicit OwningWriteAccessProxy(EpochContextWriter<T>&& writer) : writer_(std::move(writer)) {}
 
    friend class WriteBuffer<T>;
    friend class OwningWriteAwaiter<T>;
 
    EpochContextWriter<T> writer_;
};
 
template <typename T> class OwningWriteAwaiter {
  public:
    explicit OwningWriteAwaiter(EpochContextWriter<T>&& writer) : writer_(std::move(writer)) {}
 
    [[nodiscard]] bool await_ready() const noexcept { return writer_.ready(); }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return writer_.node(); }
    static constexpr auto debug_task_role() noexcept { return uni20::TaskRegistry::EpochTaskRole::Writer; }
#endif
 
    void await_suspend(BasicTask&& task) noexcept
    {
      TRACE_MODULE(ASYNC, "OwningWriteAwaiter::await_suspend()", this, task.coroutine_handle());
      writer_.suspend(std::move(task));
    }
 
    [[nodiscard]] OwningWriteAccessProxy<T> await_resume()
    {
      writer_.resume();
      return OwningWriteAccessProxy<T>(std::move(writer_));
    }
 
    [[nodiscard]] std::shared_ptr<EpochContext> epoch_context_shared() const noexcept
    {
      return writer_.epoch_context_shared();
    }
 
  private:
    EpochContextWriter<T> writer_;
};
 
template <typename T> using WriteAwaitProxy = WriteAccessProxy<T>;
template <typename T> using OwningWriteProxy = OwningWriteAccessProxy<T>;
 
template <typename T> class WriteAssignProxy {
  public:
    using value_type = T;
 
    WriteAssignProxy() = delete;
    WriteAssignProxy(WriteAssignProxy const&) = delete;
    WriteAssignProxy& operator=(WriteAssignProxy const&) = delete;
    WriteAssignProxy(WriteAssignProxy&&) noexcept = default;
    WriteAssignProxy& operator=(WriteAssignProxy&&) noexcept = delete;
 
    template <typename U>
      requires(!std::same_as<std::remove_cvref_t<U>, WriteAssignProxy>) &&
              (!is_async_alias_v<T> || detail::async_alias_assignment_source<T, U>)
    void operator=(U&& u)
    {
      async_assign(WriteBuffer<T>(std::move(writer_)), std::forward<U>(u));
    }
 
  private:
    explicit WriteAssignProxy(EpochContextWriter<T>&& writer) : writer_(std::move(writer)) {}
 
    friend class WriteBuffer<T>;
 
    EpochContextWriter<T> writer_;
};
 
template <typename T> using WriteProxy = WriteAssignProxy<T>;
 
template <typename T> WriteAssignProxy<T> WriteBuffer<T>::write() { return WriteAssignProxy<T>(std::move(writer_)); }
 
} // namespace uni20::async