epoch_queue.hpp

Go to the documentation of this file.

 
 
#pragma once
 
#include "epoch_context.hpp"
#include <atomic>
#include <deque>
#include <memory>
#include <mutex>
#include <uni20/common/trace.hpp>
#include <utility>
 
namespace uni20::async
{
 
class EpochQueue {
  public:
    EpochQueue() : current_(std::make_shared<EpochContext>()) { TRACE_MODULE(ASYNC, "EpochQueue Constructor", this); }
 
    EpochQueue(EpochQueue const&) = delete;
    EpochQueue& operator=(EpochQueue const&) = delete;
    EpochQueue(EpochQueue&&) = delete;
    EpochQueue& operator=(EpochQueue&&) = delete;
 
    ~EpochQueue() { TRACE_MODULE(ASYNC, "EpochQueue Destructor", this); }
 
    template <typename T> [[nodiscard]] EpochContextReader<T> create_read_context(shared_storage<T> storage) const
    {
      TRACE_MODULE(ASYNC, "EpochQueue::create_read_context", this);
      // if (!current_->has_writer() && storage.constructed()) current_->start();
      return EpochContextReader<T>(storage, current_);
    }
 
    template <typename T> EpochContextWriter<T> create_write_context(shared_storage<T> storage)
    {
      TRACE_MODULE(ASYNC, "EpochQueue::create_write_context", this);
      if (current_->has_writer())
      {
        std::shared_ptr<EpochContext> next = std::make_shared<EpochContext>();
        current_->set_next_epoch(next);
        current_ = next;
      }
      return EpochContextWriter<T>(storage, current_);
    }
 
    [[nodiscard]] std::shared_ptr<EpochContext> latest() const noexcept { return current_; }
 
    [[nodiscard]] bool has_pending_writers() const noexcept { return current_ && current_->has_writer(); }
 
#if UNI20_DEBUG_DAG
    template <typename T> void initialize_node(shared_storage<T> const& storage)
    {
      if (current_) current_->set_node(NodeInfo::create(storage));
    }
 
    [[nodiscard]] NodeInfo const* debug_node() const { return current_ ? current_->node() : nullptr; }
#endif
 
  private:
    std::shared_ptr<EpochContext> current_;
};
 
// FIXME: we can probably fix the oddities with the ordering of getting the buffers by
// pre-constructing the WriteBuffer.
// If the first epoch goes away then we cancel() it
class ReverseEpochQueue {
  public:
    ReverseEpochQueue() : first_(std::make_shared<EpochContext>())
    {
      TRACE("Constructing ReverseEpochQueue EpochContext", first_.get());
    }
 
    ReverseEpochQueue(ReverseEpochQueue&) = delete;
    ReverseEpochQueue& operator=(ReverseEpochQueue&) = delete;
 
    ReverseEpochQueue(ReverseEpochQueue&&) = default;
    ReverseEpochQueue& operator=(ReverseEpochQueue&&) = default;
 
    ~ReverseEpochQueue() = default;
 
    explicit ReverseEpochQueue(std::shared_ptr<EpochContext> first) : first_(std::move(first))
    {
      TRACE_MODULE(ASYNC, "ReverseEpochQueue Constructor", this);
      DEBUG_CHECK(!first_->has_writer());
    }
 
    template <typename T> [[nodiscard]] EpochContextReader<T> create_read_context(shared_storage<T> storage)
    {
      TRACE_MODULE(ASYNC, "ReverseEpochQueue::create_read_context", this);
      DEBUG_CHECK(first_);
      if (first_->has_writer()) first_ = EpochContext::make_previous(first_);
      return EpochContextReader<T>(storage, first_);
    }
 
    template <typename T> EpochContextWriter<T> create_write_context(shared_storage<T> storage)
    {
      TRACE_MODULE(ASYNC, "ReverseEpochQueue::create_write_context", this);
      DEBUG_CHECK(first_);
      if (first_->has_writer()) first_ = EpochContext::make_previous(first_);
      // CHECK(!first_->has_writer());
      return EpochContextWriter<T>(storage, first_);
    }
 
    void start()
    {
      DEBUG_CHECK(first_);
      // CHECK(first_->has_writer());
 
      // FIXME: we need to start() here in such a way that if we don't have an initial writer then
      // we set an error state (which will generally be absorbed by the backprop accumulation knowing
      // how to handle missing data)
      first_->writer_acquire();
      first_->writer_cancel();
      first_->start();
      first_.reset();
    }
 
    [[nodiscard]] bool is_started() const { return !first_; }
 
  private:
    std::shared_ptr<EpochContext> first_;
};
 
} // namespace uni20::async