reverse_value.hpp

Go to the documentation of this file.

#pragma once
#include "async.hpp"
#include "buffers.hpp"
#include "epoch_queue.hpp"
#include <type_traits>
 
namespace uni20::async
{
 
template <typename T, typename U> AsyncTask async_accumulate(ReadBuffer<T> a_, ReadBuffer<U> b_, WriteBuffer<T> out_)
{
  auto a = co_await a_.transfer().maybe();
  if (a)
  {
    T value = std::move(*a).get_release();
    auto b = co_await b_.transfer().maybe();
    if (b) value += std::move(*b).get_release();
    co_await out_ = std::move(value);
  }
  else
  {
    // GCC 13 workaround for `(co_await ...).get()`: stage through a named proxy.
    // GCC 14+ supports the one-liner form:
    //   U value = (co_await b_.transfer().or_cancel()).get();
    auto b = co_await b_.transfer().or_cancel();
    U value = b.get();
    b.release();
    co_await out_ += std::move(value);
  }
  co_return;
}
 
template <typename T, typename U>
AsyncTask async_accumulate_minus(ReadBuffer<T> a_, ReadBuffer<U> b_, WriteBuffer<T> out_)
{
  auto a = co_await a_.transfer().maybe();
  if (a)
  {
    T value = std::move(*a).get_release();
    auto b = co_await b_.transfer().maybe();
    if (b) value -= std::move(*b).get_release();
    co_await out_ = std::move(value);
  }
  else
  {
    auto b = co_await b_.transfer().or_cancel();
    U value = b.get();
    b.release();
    co_await out_ -= std::move(value);
  }
  co_return;
}
 
template <typename T> class ReverseValue {
  public:
    using value_type = T;
 
    ReverseValue() : async_(async_do_not_start), rqueue_(async_.queue().latest()) {}
 
    ReverseValue(ReverseValue&&) noexcept = default;
 
    ReverseValue& operator=(ReverseValue&& other) noexcept
    {
      if (this != &other)
      {
        this->finalize();
        async_ = std::move(other.async_);
        rqueue_ = std::move(other.rqueue_);
        started_ = std::exchange(other.started_, false);
      }
      return *this;
    }
 
    ReverseValue(ReverseValue const&) = delete;
 
    // Async<T> const& async() const noexcept { return async_; }
 
    // ReadBuffer<T> final()
    // {
    //   this->finalize();
    //   return async_.read();
    // }
 
    ~ReverseValue() { this->finalize(); }
 
    Async<T> final()
    {
      this->finalize();
      return async_;
    }
 
    [[nodiscard]] T final_wait()
    {
      this->finalize();
      return async_.get_wait();
    }
 
    void wait()
    {
      this->finalize();
      async_.wait();
    }
 
    [[nodiscard]] T get_wait() { return this->final_wait(); }
 
    ReadBuffer<T> input() const { return this->read_buffer(); }
 
    ReadBuffer<T> read() const { return this->read_buffer(); }
 
    [[nodiscard]] WriteBuffer<T> output() { return this->write_buffer(); }
 
    void finalize() const
    {
      if (!started_)
      {
        if (!rqueue_.is_started())
        {
          rqueue_.start();
        }
        started_ = true;
      }
    }
 
    template <typename U>
    ReverseValue& operator=(U&& v)
      requires std::constructible_from<T, U&&>
    {
      WriteBuffer<T> w(this->write_buffer());
      rqueue_.start();
      w.emplace_assert(std::forward<U>(v));
      return *this;
    }
 
    template <typename U> ReverseValue& operator=(Async<U> const& v)
    {
      async_assign(this->write_buffer(), v.read());
      rqueue_.start();
      return *this;
    }
 
    template <typename U> ReverseValue& operator=(Async<U>&& v)
    {
      async_move(this->write_buffer(), std::move(v));
      rqueue_.start();
      return *this;
    }
 
    template <typename U> ReverseValue& operator+=(Async<U> const& v)
    {
      // It is important that we construct the buffer objects in the right order. Writer first,
      // then reader, so the reader is the earlier epoch in the ReverseEpochQueue
      WriteBuffer<T> w(this->write_buffer());
      schedule(async_accumulate(this->read_buffer(), v.read(), std::move(w)));
      return *this;
    }
 
    template <typename U> ReverseValue& operator+=(ReverseValue<U> const& v)
    {
      WriteBuffer<T> w(this->write_buffer());
      schedule(async_accumulate(this->read_buffer(), v.read(), std::move(w)));
      return *this;
    }
 
    template <typename U> ReverseValue& operator+=(ReadBuffer<U> v)
    {
      WriteBuffer<T> w(this->write_buffer());
      schedule(async_accumulate(this->read_buffer(), std::move(v), std::move(w)));
      return *this;
    }
 
    template <typename U> ReverseValue& operator-=(Async<U> const& v)
    {
      WriteBuffer<T> w(this->write_buffer());
      schedule(async_accumulate_minus(this->read_buffer(), v.read(), std::move(w)));
      return *this;
    }
 
    template <typename U> ReverseValue& operator-=(ReverseValue<U> const& v)
    {
      WriteBuffer<T> w(this->write_buffer());
      schedule(async_accumulate_minus(this->read_buffer(), v.read(), std::move(w)));
      return *this;
    }
 
    template <typename U> ReverseValue& operator-=(ReadBuffer<U> v)
    {
      WriteBuffer<T> w(this->write_buffer());
      schedule(async_accumulate_minus(this->read_buffer(), std::move(v), std::move(w)));
      return *this;
    }
 
    Async<T>& backprop() &
    {
      this->finalize();
      return async_;
    }
    Async<T> const& backprop() const&
    {
      this->finalize();
      return async_;
    }
    Async<T> backprop() &&
    {
      this->finalize();
      return std::move(async_);
    }
 
    Async<T>& last_value() & { return async_; }
    Async<T> const& last_value() const& { return async_; }
    Async<T> last_value() && { return std::move(async_); }
 
    void start() { rqueue_.start(); }
 
  private:
    ReadBuffer<T> read_buffer() const { return ReadBuffer<T>(rqueue_.create_read_context(async_.storage())); }
    WriteBuffer<T> write_buffer() { return WriteBuffer<T>(rqueue_.create_write_context(async_.storage())); }
 
    Async<T> async_;
    mutable ReverseEpochQueue rqueue_; // must be mutable if we want read access to be logically const
    mutable bool started_{false};
};
 
template <typename T> struct async_value_type<ReverseValue<T>>
{
    using type = T;
};
 
template <typename T> ReadBuffer<T> read(ReverseValue<T> const& x) { return x.input(); }
 
} // namespace uni20::async