async/async.hpp

Go to the documentation of this file.

 
 
// NOTE: Immediately-invoked coroutine lambdas must not capture variables.
// Captures (by reference or value) are stored in the lambda frame, which is destroyed
// after the lambda returns. If the coroutine suspends, any captured state becomes dangling.
// Instead, pass all needed data via parameters, which are safely moved into the coroutine frame.
 
#pragma once
 
#include "async_errors.hpp"
#include "async_node.hpp"
#include "async_traits.hpp"
#include "buffers.hpp"
#include "epoch_queue.hpp"
#include <concepts>
#include <initializer_list>
#include <memory>
#include <string>
#include <type_traits>
#include <uni20/common/demangle.hpp>
#include <uni20/config.hpp>
#include <utility>
 
namespace uni20::async
{
 
class DebugScheduler;
 
struct async_do_not_start_t
{};
 
inline constexpr async_do_not_start_t async_do_not_start{};
 
template <typename T> class ReverseValue; // forward declaration so we can add it as a friend of Async<T>
 
namespace detail
{
 
template <typename Source>
concept async_source = std::same_as<std::remove_cvref_t<Source>, Async<async_value_t<Source>>>;
 
template <typename Target, typename Source>
concept value_rebind_source = (async_source<Source> && std::constructible_from<Target, async_value_t<Source> const&>) ||
                              (!async_source<Source> && std::constructible_from<Target, Source&&>);
 
template <typename Target, typename Source>
concept async_assignment_source =
    (!std::same_as<std::remove_cvref_t<Source>, Async<Target>>) &&
    ((!is_async_alias_v<Target> && value_rebind_source<Target, Source>) ||
     (is_async_alias_v<Target> && async_alias_assignable_from<Target, async_value_t<Source> const&>));
 
} // namespace detail
 
template <typename T, typename U>
  requires(!is_async_alias_v<T> && detail::async_source<U> && std::constructible_from<T, async_value_t<U> const&>)
void async_initialize(Async<T>& destination, U&& source);
 
template <typename Alias, typename Parent, typename... Args>
[[nodiscard]] Async<Alias> make_async_alias(Async<Parent> const& parent, Args&&... args);
 
template <typename Alias, typename Parent, typename... Args>
[[nodiscard]] Async<Alias> make_async_alias_from_parent(Async<Parent>& parent, Args&&... args);
 
template <typename Alias, typename Parent, typename... Args>
[[nodiscard]] Async<Alias> make_async_alias_from_parent(Async<Parent> const& parent, Args&&... args);
 
template <typename T> class Async {
  public:
    using value_type = T;
 
    static_assert(std::is_object_v<T> && !std::is_array_v<T>, "Async<T> requires a non-array object type");
    static_assert(std::destructible<T>, "Async<T> requires a non-throwing destructor");
 
    Async()
      requires(!is_async_alias_v<T>)
        : storage_(make_unconstructed_shared_storage<T>()), queue_(std::make_shared<EpochQueue>())
    {
      queue_->latest()->start();
#if UNI20_DEBUG_DAG
      queue_->initialize_node(storage_);
#endif
    }
 
    Async(async_do_not_start_t tag)
      requires(!is_async_alias_v<T>)
        : storage_(make_unconstructed_shared_storage<T>()), queue_(std::make_shared<EpochQueue>())
    {
      (void)tag;
#if UNI20_DEBUG_DAG
      queue_->initialize_node(storage_);
#endif
    }
 
    template <typename U>
      requires(std::convertible_to<U, T> && !is_async_alias_v<T>)
    Async(U&& val) : storage_(make_shared_storage<T>(std::forward<U>(val))), queue_(std::make_shared<EpochQueue>())
    {
      queue_->latest()->start_reading();
#if UNI20_DEBUG_DAG
      queue_->initialize_node(storage_);
#endif
    }
 
    template <typename U>
      requires(std::constructible_from<T, U> && (!std::convertible_to<U, T>) && !is_async_alias_v<T>)
    explicit Async(U&& u) : storage_(make_shared_storage<T>(std::forward<U>(u))), queue_(std::make_shared<EpochQueue>())
    {
      queue_->latest()->start_reading();
#if UNI20_DEBUG_DAG
      queue_->initialize_node(storage_);
#endif
    }
 
    template <typename... Args>
      requires(std::constructible_from<T, Args...> && !is_async_alias_v<T>)
    Async(Args&&... args)
        : storage_(make_shared_storage<T>(std::forward<Args>(args)...)), queue_(std::make_shared<EpochQueue>())
    {
      queue_->latest()->start_reading();
#if UNI20_DEBUG_DAG
      queue_->initialize_node(storage_);
#endif
    }
 
    template <typename U, typename... Args>
      requires(std::constructible_from<T, std::initializer_list<U>&, Args...> && !is_async_alias_v<T>)
    Async(std::initializer_list<U> init, Args&&... args)
        : storage_(make_shared_storage<T>(init, std::forward<Args>(args)...)), queue_(std::make_shared<EpochQueue>())
    {
      queue_->latest()->start_reading();
#if UNI20_DEBUG_DAG
      queue_->initialize_node(storage_);
#endif
    }
 
    Async(Async const& rhs)
      requires(!is_async_alias_v<T> && std::constructible_from<T, T const&>)
        : Async()
    {
      async_initialize(*this, rhs);
    }
 
    Async(Async const&)
      requires(!is_async_alias_v<T> && !std::constructible_from<T, T const&>)
    = delete;
 
    Async(Async const& rhs)
      requires(is_async_alias_v<T>)
        : storage_(rhs.storage_), queue_(rhs.queue_)
    {}
 
    Async& debug_name(std::string const& label)
    {
#if UNI20_DEBUG_DAG
      TaskRegistry::name_async_value(queue_->debug_node(), label);
#else
      static_cast<void>(label);
#endif
      return *this;
    }
 
    Async const& debug_name(std::string const& label) const
    {
#if UNI20_DEBUG_DAG
      TaskRegistry::name_async_value(queue_->debug_node(), label);
#else
      static_cast<void>(label);
#endif
      return *this;
    }
 
    template <typename U>
      requires detail::async_assignment_source<T, U>
    Async& operator=(U&& rhs);
 
    Async& operator=(Async const& rhs)
      requires(!is_async_alias_v<T> && std::constructible_from<T, T const&>)
    {
      if (this != &rhs)
      {
        Async<T> replacement;
        async_initialize(replacement, rhs);
        *this = std::move(replacement);
      }
      return *this;
    }
 
    Async& operator=(Async const&)
      requires(!is_async_alias_v<T> && !std::constructible_from<T, T const&>)
    = delete;
 
    Async& operator=(Async const& rhs)
      requires(is_async_alias_v<T> && detail::async_alias_assignable_from<T, T const&>)
    {
      if (this != &rhs) async_assign(*this, rhs);
      return *this;
    }
 
    Async& operator=(Async const&)
      requires(is_async_alias_v<T> && !detail::async_alias_assignable_from<T, T const&>)
    = delete;
 
    Async(Async&& other) noexcept = default;
    Async& operator=(Async&& other) noexcept
      requires(!is_async_alias_v<T>)
    {
      if (this != &other)
      {
        storage_ = std::move(other.storage_);
        queue_ = std::move(other.queue_);
      }
      return *this;
    }
 
    Async& operator=(Async&& other)
      requires(is_async_alias_v<T> && detail::async_alias_assignable_from<T, T const&>)
    {
      if (this != &other) async_assign(*this, other);
      return *this;
    }
 
    Async& operator=(Async&&)
      requires(is_async_alias_v<T> && !detail::async_alias_assignable_from<T, T const&>)
    = delete;
 
    ~Async() = default;
 
    ReadBuffer<T> read() const { return ReadBuffer<T>(queue_->create_read_context(storage_)); }
 
    WriteBuffer<T> write() { return WriteBuffer<T>(queue_->create_write_context(storage_)); }
 
    // template <typename Sched> T& get_wait(Sched& sched)
    // {
    //   while (queue_.has_pending_writers())
    //   {
    //     TRACE_MODULE(ASYNC, "Has pending writers");
    //     sched.run();
    //   }
    //   return *value_;
    // }
 
    void wait() const;
 
    void wait(IAsyncScheduler& sched) const;
 
    [[nodiscard]] T const& get_wait() const;
 
    [[nodiscard]] T const& get_wait(IAsyncScheduler& sched) const;
 
    // TODO: we could have a version that returns a ref-counted proxy, which enables reference rather than copy
 
    T move_from_wait()
      requires(!is_async_alias_v<T> && std::constructible_from<T, T &&>);
 
    void unsafe_set(T const& x)
      requires(!is_async_alias_v<T> && std::assignable_from<T&, T const&>)
    {
      auto* ptr = require_value();
      *ptr = x;
    }
 
    T unsafe_value() const
      requires(!is_async_alias_v<T> && std::constructible_from<T, T const&>)
    {
      return *require_value();
    }
 
    T const& unsafe_value_ref() const { return *require_value(); }
 
    T& unsafe_value_ref()
      requires(!is_async_alias_v<T>)
    {
      return *require_value();
    }
 
    EpochQueue const& queue() const { return *queue_; }
    shared_storage<T> const& storage() const { return storage_; }
 
    [[nodiscard]] T* storage_address() noexcept
      requires(!is_async_alias_v<T>)
    {
      return storage_.storage_address();
    }
 
    [[nodiscard]] T const* storage_address() const noexcept { return storage_.storage_address(); }
 
    std::shared_ptr<T> value_ptr()
      requires(!is_async_alias_v<T>)
    {
      if (!storage_.valid()) return {};
      return std::shared_ptr<T>(storage_.get(), [storage = storage_](T*) mutable { storage.reset(); });
    }
 
    std::shared_ptr<T const> value_ptr() const
    {
      if (!storage_.valid()) return {};
      return std::shared_ptr<T const>(storage_.get(), [storage = storage_](T const*) mutable { storage.reset(); });
    }
 
  private:
    Async(shared_storage<T> storage, std::shared_ptr<EpochQueue> queue)
        : storage_(std::move(storage)), queue_(std::move(queue))
    {
      CHECK(storage_.has_lifetime_owner());
      CHECK(queue_);
    }
 
    T* try_get_value() const
    {
      if (!storage_.valid()) throw async_storage_missing{};
      return storage_.get();
    }
 
    T* require_value() const
    {
      if (auto* ptr = try_get_value()) return ptr;
      throw async_value_uninitialized{};
    }
 
    friend class ReverseValue<T>;
    template <typename Alias, typename Parent, typename... Args>
    friend Async<Alias> make_async_alias(Async<Parent> const& parent, Args&&... args);
    template <typename Alias, typename Parent, typename... Args>
    friend Async<Alias> make_async_alias_from_parent(Async<Parent>& parent, Args&&... args);
    template <typename Alias, typename Parent, typename... Args>
    friend Async<Alias> make_async_alias_from_parent(Async<Parent> const& parent, Args&&... args);
 
    mutable shared_storage<T> storage_;
    std::shared_ptr<EpochQueue> queue_;
};
 
template <typename Alias, typename Parent, typename... Args>
[[nodiscard]] Async<Alias> make_async_alias(Async<Parent> const& parent, Args&&... args)
{
  static_assert(is_async_alias_v<Alias>, "Async aliases require is_async_alias_v<Alias>");
  return Async<Alias>(make_shared_storage_alias<Alias>(parent.storage_, std::forward<Args>(args)...), parent.queue_);
}
 
template <typename Alias, typename Parent, typename... Args>
[[nodiscard]] Async<Alias> make_async_alias_from_parent(Async<Parent>& parent, Args&&... args)
{
  static_assert(is_async_alias_v<Alias>, "Async aliases require is_async_alias_v<Alias>");
  return Async<Alias>(
      make_shared_storage_alias<Alias>(parent.storage_, parent.storage_.storage_address(), std::forward<Args>(args)...),
      parent.queue_);
}
 
template <typename Alias, typename Parent, typename... Args>
[[nodiscard]] Async<Alias> make_async_alias_from_parent(Async<Parent> const& parent, Args&&... args)
{
  static_assert(is_async_alias_v<Alias>, "Async aliases require is_async_alias_v<Alias>");
  return Async<Alias>(make_shared_storage_alias<Alias>(parent.storage_,
                                                       std::as_const(parent.storage_).storage_address(),
                                                       std::forward<Args>(args)...),
                      parent.queue_);
}
 
template <typename T> ReadBuffer<T> read(Async<T> const& a) { return a.read(); }
template <typename T> WriteBuffer<T> write(Async<T>& a) { return a.write(); }
 
} // namespace uni20::async
 
#include "async_impl.hpp"