epoch_context.hpp

Go to the documentation of this file.

 
 
#pragma once
 
#include "async_errors.hpp"
#include "async_node.hpp"
#include "async_task_promise.hpp"
#include "shared_storage.hpp"
#include "task_registry.hpp"
#include <atomic>
#include <coroutine>
#include <cstdio>
#include <cstdlib>
#include <exception>
#include <fmt/format.h>
#include <functional>
#include <memory>
#include <mutex>
#include <vector>
 
namespace uni20::async
{
 
template <typename T> class Async;
template <typename T> class EpochContextReader;
template <typename T> class EpochContextWriter;
 
 
// =============================================================================================
// EpochContext Invariants
// =============================================================================================
//
// Exactly one of the following must hold at all times:
//
//   * phase_ == Pending
//
//   * phase_ == Started
//      && total_writers_ == 0
//
//   * phase_ == Writing
//       && num_writers_ > 0
//
//   * phase_ == Reading
//       && num_writers_ == 0
//       && (num_readers_ > 0 || next_epoch_ == nullptr)
//
//   * phase_ == Finished
//
// Normally, num_writers_ <= 1. (Not sure if there is a use for > 1, as a transient?)
//
// ---------------------------------------------------------------------------------------------
//
// Destruction invariants:
//
//   * Destruction while (num_writers_ > 0 || num_readers_ > 0)
//       is erroneous (assertion failure).
//
// ---------------------------------------------------------------------------------------------
//
// Phase transition rules:
//
//   * Pending → Started
//       Triggered by start_epoch()
//
//   * Started → Writing
//       Triggered by total_writers_ > 0.
//
//   * Writing → Reading
//       Triggered when num_writers_ drops to to 0.
//       (i.e. writer has finished).
//       \pre num_writers_ = 0
//
//   * Reading → Finished
//       Triggered when num_readers_ drops to 0 && next_epoch_ != nullptr
//       or next_epoch_ is set and num_readers == 0
//       Invokes next_epoch_->start() and passes on cancelled_ and eptr_
//       \pre num_readers_ == 0 && next_epoch_ != nullptr
//
//   * Writing → Finished
//       Triggered when num_writers_ drops to to 0 && num_readers_ == 0 && next_eopch_ != nullptr
//       (i.e. writer has finished, and there are no readers).
//       \pre num_writers_ = 0 && num_readers_ == 0 && next_epoch_ != nullptr
//
// Note that immediate transitions are possible.
//
// ---------------------------------------------------------------------------------------------
//
// Registration rules:
//
//   * Writers can only be added when phase_ == Pending || phase_ == Started.
//   * Readers can only be added when phase_ != Finished.
//
// =============================================================================================
//
// Cancellation:
// An Epoch chain can be cancelled at any time. This indicates that the final result of this
// calculation is not going to be used and Reader tasks will either need to handle the cancel()
// or they will get an exception.
//
// Initialization status is handled separately.  The EpochContext simply handles writers and readers.
 
// WriteBuffer participates in epoch scheduling only; value construction state is tracked by shared_storage<T>.
// If a writer is released without emplacing/writing a value, the storage can remain unconstructed for this epoch.
//
// If a ReadBuffer's epoch has no value, ordinary reads throw buffer_read_uninitialized while
// .or_cancel() throws task_cancelled. Exceptions stored by failed writers always propagate unchanged.
//
// If there is a next_epoch_, then the cancellation and exception status is propogated forwards.
//
// co_await on a WriteBuffer on a cancelled or exception EpochContext passes the error into the coroutine.
//
 
class EpochQueue;
class ReverseEpochQueue;
 
class EpochContext {
  public:
    enum class Phase
    {
      Pending, 
      Started, 
      Writing, 
      Reading, 
      Finished 
    };
 
    struct TaskState
    {
        BasicTask task;
    };
 
    struct DebugSnapshot
    {
        int generation{0};
        Phase phase{Phase::Pending};
        EpochContext const* next_epoch{nullptr};
        int total_writers{0};
        int num_writers{0};
        int num_readers{0};
        bool writer_active{false};
        std::vector<std::coroutine_handle<>> reader_tasks{};
        std::vector<std::coroutine_handle<>> writer_tasks{};
#if UNI20_DEBUG_DAG
        NodeInfo const* node{nullptr};
#endif
    };
 
    EpochContext() { TaskRegistry::register_epoch_context(this); }
 
    explicit EpochContext(std::shared_ptr<EpochContext> next) : counter_(next->counter_ - 1), next_epoch_(next)
    {
#if UNI20_DEBUG_DAG
      node_ = next ? next->node() : nullptr;
#endif
      TaskRegistry::register_epoch_context(this);
    }
 
    static std::shared_ptr<EpochContext> make_previous(std::shared_ptr<EpochContext> next)
    {
      return std::make_shared<EpochContext>(std::move(next));
    }
 
    ~EpochContext()
    {
      {
        std::lock_guard lock(mtx_);
        CHECK_EQUAL(num_writers_, 0);
        CHECK_EQUAL(num_readers_, 0);
      }
      TaskRegistry::destroy_epoch_context(this);
    }
 
    void set_next_epoch(std::shared_ptr<EpochContext> next)
    {
      std::unique_lock lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::set_next_epoch", this, next.get(), counter_);
      DEBUG_CHECK(!next_epoch_); // make sure that we haven't already set the next epoch pointer
      DEBUG_CHECK(phase_ <= Phase::Reading);
 
      next->counter_ = counter_ + 1; // increment the epoch counter
#if UNI20_DEBUG_DAG
      next->node_ = node_;
#endif
      next_epoch_ = std::move(next);
      if (phase_ == Phase::Reading && num_readers_ == 0) this->advance_finished_locked(std::move(lock));
    }
 
    // start executing the epoch, if it has not already started
    void start()
    {
      std::unique_lock lock(mtx_);
      //      DEBUG_CHECK(phase_ == Phase::Pending);
      if (phase_ == Phase::Pending) this->advance_start_locked(std::move(lock));
    }
 
    // start executing the epoch, assuming that we've already written
    // precondition: no existing writers
    void start_reading()
    {
      std::unique_lock lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::start_reading", this, counter_);
      DEBUG_CHECK(phase_ == Phase::Pending);
      DEBUG_CHECK(total_writers_ == 0);
      ++total_writers_;
      this->advance_start_locked(std::move(lock));
    }
 
    // start executing the epoch, propogating the exception and cancellation state from a previous epoch
    void start(std::exception_ptr eptr)
    {
      std::unique_lock lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::start", this, counter_);
      DEBUG_CHECK(phase_ == Phase::Pending, this);
      eptr_ = eptr;
      this->advance_start_locked(std::move(lock));
    }
 
    DebugSnapshot debug_snapshot() const
    {
      std::lock_guard lock(mtx_);
      DebugSnapshot snapshot;
      snapshot.generation = counter_;
      snapshot.phase = phase_;
      snapshot.next_epoch = next_epoch_.get();
      snapshot.total_writers = total_writers_;
      snapshot.num_writers = num_writers_;
      snapshot.num_readers = num_readers_;
      snapshot.writer_active = writer_active_;
#if UNI20_DEBUG_DAG
      snapshot.node = node_;
#endif
 
      snapshot.reader_tasks.reserve(reader_tasks_.size());
      for (auto const& task_state : reader_tasks_)
      {
        if (auto h = task_state.task.coroutine_handle()) snapshot.reader_tasks.push_back(h);
      }
 
      snapshot.writer_tasks.reserve(writer_tasks_.size());
      for (auto const& task_state : writer_tasks_)
      {
        if (auto h = task_state.task.coroutine_handle()) snapshot.writer_tasks.push_back(h);
      }
 
      return snapshot;
    }
 
    bool try_debug_snapshot(DebugSnapshot& snapshot) const
    {
      if (!mtx_.try_lock()) return false;
      std::lock_guard lock(mtx_, std::adopt_lock);
      snapshot = DebugSnapshot{};
      snapshot.generation = counter_;
      snapshot.phase = phase_;
      snapshot.next_epoch = next_epoch_.get();
      snapshot.total_writers = total_writers_;
      snapshot.num_writers = num_writers_;
      snapshot.num_readers = num_readers_;
      snapshot.writer_active = writer_active_;
#if UNI20_DEBUG_DAG
      snapshot.node = node_;
#endif
 
      snapshot.reader_tasks.reserve(reader_tasks_.size());
      for (auto const& task_state : reader_tasks_)
      {
        if (auto h = task_state.task.coroutine_handle()) snapshot.reader_tasks.push_back(h);
      }
 
      snapshot.writer_tasks.reserve(writer_tasks_.size());
      for (auto const& task_state : writer_tasks_)
      {
        if (auto h = task_state.task.coroutine_handle()) snapshot.writer_tasks.push_back(h);
      }
 
      return true;
    }
 
    std::vector<std::coroutine_handle<>> reader_task_handles() const
    {
      auto snapshot = this->debug_snapshot();
      return std::move(snapshot.reader_tasks);
    }
 
    std::vector<std::coroutine_handle<>> writer_task_handles() const
    {
      auto snapshot = this->debug_snapshot();
      return std::move(snapshot.writer_tasks);
    }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const
    {
      std::lock_guard lock(mtx_);
      return node_;
    }
 
    void set_node(NodeInfo const* node)
    {
      std::lock_guard lock(mtx_);
      node_ = node;
    }
#endif
 
  private:
    // Writer interface
    // internal private used only by EpochContextWriter<T>
    template <typename T> friend class EpochContextWriter;
    friend void propagate_unhandled_writer_exception(EpochContext* epoch, std::exception_ptr eptr) noexcept;
 
    void writer_acquire() noexcept
    {
      std::unique_lock lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::writer_acquire", this, counter_, phase_);
      // TODO: if we have reenterant writers, then we might be able to acquire writers in Writing phase
      DEBUG_CHECK(phase_ <= Phase::Started);
 
      ++num_writers_;
      ++total_writers_;
      if (phase_ == Phase::Started) this->advance_writing_locked(std::move(lock));
    }
 
    bool has_writer() const noexcept
    {
      std::lock_guard lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::has_writer", this, counter_, phase_, total_writers_);
      return total_writers_ > 0 || phase_ >= Phase::Writing;
    }
 
    void writer_clear_errors() noexcept
    {
      std::lock_guard lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::writer_clear_errors", this, counter_);
      DEBUG_CHECK(phase_ <= Phase::Writing);
      eptr_ = nullptr;
      // cancelled_ = false;
      // inherit_error_state_ = false;
    }
 
    void writer_bind(BasicTask&& task) noexcept
    {
      std::unique_lock lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::writer_bind", this, counter_, task.coroutine_handle(), phase_,
                         num_writers_, writer_active_);
      DEBUG_CHECK(phase_ <= Phase::Writing);
      DEBUG_CHECK(phase_ != Phase::Started); // if this Epoch was started, it should have transitioned to Writing by now
 
      // If the Epoch is running, then execute the task immediately
      if (phase_ == Phase::Writing && !writer_active_)
      {
        this->execute_writer_locked(std::move(lock), TaskState{std::move(task)});
        return;
      }
 
      // Otherwise, store the task for later
      writer_tasks_.emplace_back(std::move(task));
    }
 
    bool writer_try_make_active() noexcept
    {
      std::unique_lock lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::writer_try_make_active", this, counter_, phase_, writer_active_);
      DEBUG_CHECK(phase_ <= Phase::Writing);
      if (phase_ < Phase::Writing || writer_active_) return false;
      writer_active_ = true;
      return true;
    }
 
    // Notify about availability, without reserving the gate. The waiting writer
    // still acquires through writer_try_make_active() and may need to wait again.
    void writer_notify_when_ready(std::function<void()> notify)
    {
      std::unique_lock lock(mtx_);
      DEBUG_CHECK(phase_ <= Phase::Writing);
      if (phase_ == Phase::Writing && !writer_active_)
      {
        lock.unlock();
        notify();
        return;
      }
      writer_ready_waiters_.push_back(std::move(notify));
    }
 
    void writer_set_exception(std::exception_ptr e) noexcept
    {
      std::lock_guard lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::writer_set_exception", this, counter_);
      DEBUG_CHECK(phase_ <= Phase::Reading);
      if (!eptr_) eptr_ = e;
    }
 
    // void writer_set_cancel() noexcept
    // {
    //   std::lock_guard lock(mtx_);
    //   DEBUG_TRACE_MODULE(ASYNC, "EpochContext::writer_set_cancel", this, counter_);
    //   DEBUG_CHECK(phase_ == Phase::Writing);
    //   cancelled_ = true;
    // }
 
    bool writer_cancel() noexcept
    {
      std::unique_lock lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::writer_release", this, counter_, phase_);
      DEBUG_CHECK(phase_ <= Phase::Writing);
      DEBUG_CHECK(num_writers_ >= 1);
 
      --num_writers_;
 
      if (num_writers_ == 0 && phase_ == Phase::Writing)
      {
        DEBUG_CHECK(!writer_active_);
        this->advance_reading_locked(std::move(lock));
        return true;
      }
      return false;
    }
 
    bool writer_release_active() noexcept
    {
      std::unique_lock lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::writer_release", this, counter_, phase_);
      DEBUG_CHECK(phase_ == Phase::Writing);
      DEBUG_CHECK(num_writers_ >= 1);
      DEBUG_CHECK(writer_active_);
 
      writer_active_ = false;
      --num_writers_;
 
      if (num_writers_ == 0 && phase_ == Phase::Writing)
      {
        this->advance_reading_locked(std::move(lock));
        return true;
      }
 
      if (!writer_tasks_.empty())
      {
        auto NextTask = std::move(writer_tasks_.back());
        writer_tasks_.pop_back();
        this->execute_writer_locked(std::move(lock), std::move(NextTask));
      }
      else
      {
        this->notify_writers_locked(std::move(lock));
      }
 
      return false;
    }
 
  private:
    // reader interface
    // internal private used only by EpochContextReader<T>
    template <typename T> friend class EpochContextReader;
 
    void reader_acquire() noexcept
    {
      std::lock_guard lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::reader_acquire", this, counter_);
      DEBUG_CHECK(phase_ <= Phase::Reading);
 
      ++num_readers_;
    }
 
    bool reader_ready() const noexcept
    {
      std::lock_guard lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::reader_ready", this, counter_);
      DEBUG_CHECK(phase_ <= Phase::Reading);
 
      return phase_ == Phase::Reading;
    }
 
    void reader_notify_when_ready(std::function<void()> notify)
    {
      std::unique_lock lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::reader_notify_when_ready", this, counter_, phase_);
      DEBUG_CHECK(phase_ <= Phase::Reading);
 
      if (phase_ == Phase::Reading)
      {
        lock.unlock();
        notify();
        return;
      }
      reader_ready_waiters_.push_back(std::move(notify));
    }
 
    void reader_bind(BasicTask&& h)
    {
      std::unique_lock lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::reader_bind", this, counter_, phase_);
      DEBUG_CHECK(phase_ <= Phase::Reading);
 
      if (phase_ == Phase::Reading)
      {
        DEBUG_TRACE_MODULE(ASYNC, "EpochContext::reader_bind is scheduling the task immediately", this, counter_);
        std::exception_ptr my_eptr = eptr_;
        lock.unlock();
        if (my_eptr) h.exception_on_resume(my_eptr);
        BasicTask::reschedule(std::move(h));
      }
      else
      {
        DEBUG_TRACE_MODULE(ASYNC, "EpochContext::reader_bind is adding the reader task", h.coroutine_handle());
        reader_tasks_.emplace_back(std::move(h));
      }
    }
 
    std::exception_ptr reader_exception() const noexcept
    {
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::reader_exception", this, counter_);
      DEBUG_PRECONDITION(phase_ == Phase::Reading);
 
      return eptr_;
    }
 
    // bool reader_cancelled() const noexcept
    // {
    //   DEBUG_TRACE_MODULE(ASYNC, "EpochContext::reader_bind", this, counter_);
    //   DEBUG_PRECONDITION(phase_ == Phase::Reading);
    //
    //   return cancelled_;
    // }
    //
    bool reader_release() noexcept
    {
      std::lock_guard lock(mtx_);
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::reader_release", this, counter_);
      DEBUG_CHECK(phase_ <= Phase::Reading);
      DEBUG_CHECK(num_readers_ >= 1);
 
      --num_readers_;
 
      if (phase_ == Phase::Reading && num_readers_ == 0 && next_epoch_ != nullptr)
      {
        phase_ = Phase::Finished;
        next_epoch_->start(eptr_);
      }
 
      return num_readers_ == 0;
    }
 
    // Helper functions
    void advance_start_locked(std::unique_lock<std::mutex> lock)
    {
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::advance_start_locked", this, counter_);
      DEBUG_PRECONDITION(phase_ == Phase::Pending);
      phase_ = Phase::Started;
 
      if (total_writers_ > 0) this->advance_writing_locked(std::move(lock));
    }
 
    void advance_writing_locked(std::unique_lock<std::mutex> lock) noexcept
    {
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::advance_writing_locked", this, counter_);
      DEBUG_PRECONDITION(phase_ == Phase::Started);
      phase_ = Phase::Writing;
 
      if (!writer_tasks_.empty())
      {
        auto NextTask = std::move(writer_tasks_.back());
        writer_tasks_.pop_back();
        this->execute_writer_locked(std::move(lock), std::move(NextTask));
        return;
      }
 
      if (num_writers_ == 0)
        this->advance_reading_locked(std::move(lock));
      else
        this->notify_writers_locked(std::move(lock));
    }
 
    void advance_reading_locked(std::unique_lock<std::mutex> lock) noexcept
    {
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::advance_reading_locked", this, counter_);
      DEBUG_PRECONDITION(phase_ == Phase::Writing);
      phase_ = Phase::Reading;
 
      if (num_readers_ > 0)
      {
        this->execute_readers_locked(std::move(lock));
        return;
      }
 
      if (num_readers_ == 0 && next_epoch_ != nullptr) this->advance_finished_locked(std::move(lock));
    }
 
    void advance_finished_locked(std::unique_lock<std::mutex> lock) noexcept
    {
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::advance_finished_locked", this, counter_);
      DEBUG_PRECONDITION(phase_ == Phase::Reading && next_epoch_);
      phase_ = Phase::Finished;
 
      if (next_epoch_)
      {
        next_epoch_->start(eptr_);
      }
    }
 
    void execute_writer_locked(std::unique_lock<std::mutex> lock, TaskState task)
    {
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::execute_writer_locked", this, counter_);
      DEBUG_CHECK(phase_ == Phase::Writing);
      DEBUG_CHECK(!writer_active_);
 
      if (eptr_) task.task.exception_on_resume(eptr_);
      writer_active_ = true;
      lock.unlock();
      BasicTask::reschedule(std::move(task.task));
    }
 
    void notify_writers_locked(std::unique_lock<std::mutex> lock) noexcept
    {
      DEBUG_CHECK(phase_ == Phase::Writing && !writer_active_);
      // As with reader callbacks, notification can release the last epoch owner.
      // Move callbacks out and do not access this after unlocking.
      std::vector<std::function<void()>> waiters;
      waiters.swap(writer_ready_waiters_);
      lock.unlock();
      for (auto& notify : waiters)
        notify();
    }
 
    void execute_readers_locked(std::unique_lock<std::mutex> lock)
    {
      DEBUG_TRACE_MODULE(ASYNC, "EpochContext::execute_readers_locked", this, counter_);
 
      // When we reschedule (or cancel) a task, we need to assume that the coroutine frame may get
      // destroyed at any time, which might result in the last reference to this EpochContext itself
      // disappearing.  This means that we cannot reference *this at all once we start rescheduling
      // tasks.  Hence copy all of the data structures we need into local storage and release the lock.
      std::exception_ptr my_eptr = eptr_;
      std::vector<TaskState> my_reader_tasks;
      std::swap(reader_tasks_, my_reader_tasks);
      std::vector<std::function<void()>> my_ready_waiters;
      std::swap(reader_ready_waiters_, my_ready_waiters);
      lock.unlock();
 
      for (auto& notify : my_ready_waiters)
      {
        notify();
      }
 
      for (auto&& task : my_reader_tasks)
      {
        if (my_eptr) task.task.exception_on_resume(my_eptr);
        BasicTask::reschedule(std::move(task.task));
      }
    }
 
    friend class EpochQueue;
    friend class ReverseEpochQueue;
 
    // All of the data is protected by the mutex
    mutable std::mutex mtx_;
 
    // Exception pointer - if this is set, then any attempt to get a buffer throws the exception.
    // This is also propogated to future epochs
    std::exception_ptr eptr_{nullptr};
 
    // Epoch counter. This is mainly for debugging
    int counter_{0};
 
    // Pointer to the following epoch
    std::shared_ptr<EpochContext> next_epoch_;
 
    Phase phase_{Phase::Pending};
 
    // Writer interface
    int total_writers_{0};
    int num_writers_{0}; 
    std::vector<TaskState> writer_tasks_;
    std::vector<std::function<void()>> writer_ready_waiters_;
    bool writer_active_{false};
 
    // Reader interface
    int num_readers_{0};
    std::vector<TaskState> reader_tasks_;
    std::vector<std::function<void()>> reader_ready_waiters_;
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node_{nullptr};
#endif
 
    // NOTE: we should be able to use atomic operations for most data.
    // Data that is only written once in some phase, and read only in a later phase,
    // do not need any protection at all because it is sequenced by the phase change.
    // std::atomic<Phase> phase_{Phase::Pending};
    // std::atomic<int> num_writers_{0};
    // std::atomic<int> num_readers_{0};
    // std::shared_ptr<EpochContext> next_epoch_;  // shared_ptr for lifetime management
    // std::atomic<EpochContext*> next_epoch_ptr_; // copy of next_epoch_ in an atomic pointer
    //
    // bool inherit_error_state{true};       // written only before phase_ = Reading
    // bool cancelled_{false};               // written only before phase_ = Reading
    // std::exception_ptr eptr_{nullptr};    // written only before phase_ = Reading
    // BasicTask writer_tasks_;               // written only before phase_ = Writing
    // std::vector<BasicTask> reader_tasks_; // protected by mutex
    // std::mutex reader_tasks_mtx_;         // protects reader_tasks_
    //
    // starting next_epoch_ is gated on phase_ transitioning to Finished.
    // This is a competition between set_next_epoch() and release_writer().
    //
    // void set_next_epoch(std::shared_ptr<EpochContext> next) {
    //     DEBUG_TRACE_MODULE(ASYNC, "EpochContext::set_next_epoch", this, next.get(), counter_);
    //
    //     DEBUG_CHECK(!next_epoch_); // not set before
    //     DEBUG_CHECK(phase_.load(std::memory_order_relaxed) <= Phase::Reading);
    //
    //     // Establish ownership first
    //     next->counter_ = counter_ + 1;
    //     next_epoch_ = std::move(next);
    //
    //     // Publish raw pointer with release semantics
    //     next_epoch_ptr_.store(next_epoch_.get(), std::memory_order_release);
    //
    //     // If readers are already done, start the next epoch now
    //     if (num_readers_ == 0) {
    //         Phase expected = Phase::Reading;
    //         if (phase_.compare_exchange_strong(expected, Phase::Finished, std::memory_order_acq_rel)) {
    //             auto* next_ptr = next_epoch_ptr_.load(std::memory_order_acquire);
    //             if (next_ptr)
    //                 next_ptr->start(eptr_, cancelled_);
    //         }
    //     }
    // }
    //
    // void reader_release() noexcept
    // {
    //   int readers = --num_readers_;
    //
    //   if (readers == 0)
    //   {
    //     auto* next_ptr = next_epoch_ptr_.load(std::memory_order_acquire);
    //     if (next_ptr && phase_.load(std::memory_order_relaxed) == Phase::Reading)
    //     {
    //       Phase expected = Phase::Reading;
    //       if (phase_.compare_exchange_strong(expected, Phase::Finished, std::memory_order_acq_rel))
    //       {
    //         next_ptr->start(eptr_, cancelled_);
    //       }
    //     }
    //   }
    // }
};
 
inline void propagate_unhandled_writer_exception(EpochContext* epoch, std::exception_ptr eptr) noexcept
{
  if (!epoch || !eptr) return;
  epoch->writer_set_exception(std::move(eptr));
}
 
constexpr std::string_view format_as(uni20::async::EpochContext::Phase p)
{
  using Phase = uni20::async::EpochContext::Phase;
  switch (p)
  {
    case Phase::Pending:
      return "Pending";
    case Phase::Started:
      return "Started";
    case Phase::Writing:
      return "Writing";
    case Phase::Reading:
      return "Reading";
    case Phase::Finished:
      return "Finished";
  }
  return "Unknown";
}
 
 
template <typename T> class EpochContextReader {
  public:
    EpochContextReader() = delete;
 
    EpochContextReader(EpochContextReader const& other) : storage_(other.storage_), epoch_(other.epoch_)
    {
      epoch_->reader_acquire();
    }
 
    EpochContextReader& operator=(EpochContextReader const& other)
    {
      if (this != &other)
      {
        if (other.epoch_) other.epoch_->reader_acquire();
        this->release();
        storage_ = other.storage_;
        epoch_ = other.epoch_;
      }
      return *this;
    }
 
    EpochContextReader(shared_storage<T> storage, std::shared_ptr<EpochContext> epoch) noexcept
        : storage_(std::move(storage)), epoch_(std::move(epoch))
    {
      epoch_->reader_acquire();
    }
 
    EpochContextReader(EpochContextReader&& other) noexcept
        : storage_(std::move(other.storage_)), epoch_(std::move(other.epoch_))
    {}
 
    EpochContextReader& operator=(EpochContextReader&& other) noexcept
    {
      if (this != &other)
      {
        this->release();
        storage_ = std::move(other.storage_);
        epoch_ = std::move(other.epoch_);
      }
      return *this;
    }
 
    ~EpochContextReader() { this->release(); }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return epoch_ ? epoch_->node() : nullptr; }
#endif
 
    void suspend(BasicTask&& t)
    {
      TRACE_MODULE(ASYNC, "EpochContextReader::suspend", this, t.coroutine_handle(), epoch_.get(), epoch_->counter_);
      epoch_->reader_bind(std::move(t));
    }
 
    [[nodiscard]] bool ready() const noexcept { return epoch_->reader_ready(); }
 
    void notify_when_ready(std::function<void()> notify) const { epoch_->reader_notify_when_ready(std::move(notify)); }
 
    [[nodiscard]] T const& data() const
    {
      DEBUG_PRECONDITION(epoch_->reader_ready());
      DEBUG_TRACE_MODULE(ASYNC, "EpochContextReader::data", epoch_.get(), epoch_->counter_);
      if (auto e = epoch_->reader_exception(); e) std::rethrow_exception(e);
      auto* ptr = storage_.get();
      if (!ptr)
      {
#if UNI20_DEBUG_ASYNC_TASKS
        constexpr char const* reason = "EpochContextReader::data access on uninitialized storage";
        auto const mode = TaskRegistry::dump_mode();
        if (mode != TaskRegistry::DumpMode::None)
        {
          auto const epoch_counter = epoch_ ? epoch_->counter_ : -1;
          TRACE_STACK(reason, this, epoch_.get(), epoch_counter);
          if (mode == TaskRegistry::DumpMode::Full)
            TaskRegistry::dump();
          else
            TaskRegistry::dump_epoch_context(epoch_.get(), reason);
        }
#endif
        throw buffer_read_uninitialized{};
      }
      return *ptr;
    }
 
    // \brief Optionally retrieves the value stored in this buffer, if available.
    [[nodiscard]] T const* data_maybe() const
    {
      DEBUG_PRECONDITION(epoch_->reader_ready());
      if (auto e = epoch_->reader_exception(); e) std::rethrow_exception(e);
      return storage_.get();
    }
 
    [[nodiscard]] std::optional<T> data_option() const
    {
      DEBUG_PRECONDITION(epoch_->reader_ready());
      if (auto e = epoch_->reader_exception(); e) std::rethrow_exception(e);
      auto* ptr = storage_.get();
      if (!ptr) return std::nullopt;
      return *ptr;
    }
 
    void release() noexcept
    {
      if (epoch_)
      {
        epoch_->reader_release();
        epoch_.reset();
      }
    }
 
    void wait() const;
 
    void wait(IAsyncScheduler& sched) const;
 
    [[nodiscard]] T const& get_wait() const;
 
    [[nodiscard]] T const& get_wait(IAsyncScheduler& sched) const;
 
    [[nodiscard]] std::shared_ptr<EpochContext> epoch_context_shared() const noexcept { return epoch_; }
 
  private:
    shared_storage<T> storage_;
    std::shared_ptr<EpochContext> epoch_{}; 
};
 
template <typename T> class EpochContextWriter {
  public:
    EpochContextWriter() = delete;
 
    EpochContextWriter(shared_storage<T> storage, std::shared_ptr<EpochContext> epoch) noexcept
        : storage_(std::move(storage)), epoch_(std::move(epoch))
    {
      TRACE_MODULE(ASYNC, "EpochContextWriter Constructor", this, epoch_.get());
      epoch_->writer_acquire();
    }
 
    // EpochContextWriter(EpochContextWriter const& other) : storage_(other.storage_), epoch_(other.epoch_)
    // {
    //   epoch_->writer_acquire_duplicate(static_cast<void const*>(this));
    // }
 
    explicit operator bool() const noexcept { return epoch_ != nullptr; }
 
    EpochContextWriter& operator=(EpochContextWriter const&) = delete;
 
    EpochContextWriter(EpochContextWriter&& other) noexcept
        : storage_(std::move(other.storage_)), epoch_(std::move(other.epoch_)), acquired_(std::move(other.acquired_))
    {
      other.epoch_.reset();
    }
 
    EpochContextWriter& operator=(EpochContextWriter&& other) noexcept
    {
      if (this != &other)
      {
        this->release();
        storage_ = std::move(other.storage_);
        epoch_ = std::move(other.epoch_);
        acquired_ = other.acquired_;
        other.epoch_.reset();
      }
      return *this;
    }
 
    ~EpochContextWriter() { this->release(); }
 
#if UNI20_DEBUG_DAG
    NodeInfo const* node() const { return epoch_ ? epoch_->node() : nullptr; }
#endif
 
    bool ready() const noexcept
    {
      if (acquired_) return true;
      acquired_ = epoch_->writer_try_make_active();
      return acquired_;
    }
 
    void suspend(BasicTask&& t)
    {
      DEBUG_CHECK(!acquired_);
      epoch_->writer_bind(std::move(t));
    }
 
    void resume() const noexcept { acquired_ = true; }
 
    T& data() const
    {
      auto* ptr = storage_.get();
      if (!ptr)
      {
#if UNI20_DEBUG_ASYNC_TASKS
        constexpr char const* reason = "EpochContextWriter::data access on uninitialized storage";
        auto const mode = TaskRegistry::dump_mode();
        if (mode != TaskRegistry::DumpMode::None)
        {
          auto const epoch_counter = epoch_ ? epoch_->counter_ : -1;
          TRACE_STACK(reason, this, epoch_.get(), epoch_counter);
          if (mode == TaskRegistry::DumpMode::Full)
            TaskRegistry::dump();
          else
            TaskRegistry::dump_epoch_context(epoch_.get(), reason);
        }
#endif
        throw buffer_write_uninitialized{};
      }
      return *ptr;
    }
 
    template <typename... Args>
      requires std::constructible_from<T, Args...>
    T& emplace(Args&&... args)
    {
      try
      {
        return storage_.emplace(std::forward<Args>(args)...);
      }
      catch (...)
      {
#if UNI20_DEBUG_ASYNC_TASKS
        constexpr char const* reason = "EpochContextWriter::emplace failed while constructing storage";
        auto const mode = TaskRegistry::dump_mode();
        if (mode != TaskRegistry::DumpMode::None)
        {
          auto const epoch_counter = epoch_ ? epoch_->counter_ : -1;
          TRACE_STACK(reason, this, epoch_.get(), epoch_counter);
          if (mode == TaskRegistry::DumpMode::Full)
            TaskRegistry::dump();
          else
            TaskRegistry::dump_epoch_context(epoch_.get(), reason);
        }
#endif
        throw;
      }
    }
 
    shared_storage<T>& storage() { return storage_; }
 
    void set_exception(std::exception_ptr e) const noexcept { epoch_->writer_set_exception(e); }
 
    std::shared_ptr<EpochContext> epoch_context_shared() const noexcept { return epoch_; }
 
    bool value_is_initialized() const noexcept { return storage_.constructed(); }
 
    void release() noexcept
    {
      if (epoch_)
      {
        if (acquired_)
          epoch_->writer_release_active();
        else
          epoch_->writer_cancel(); // releasing without ever
        epoch_.reset();
        storage_.reset();
      }
    }
 
    T&& move_from_wait();
 
  private:
    mutable shared_storage<T> storage_;
    std::shared_ptr<EpochContext> epoch_;
    mutable bool acquired_{false};
};
 
} // namespace uni20::async