async_ops.hpp

Go to the documentation of this file.

 
#pragma once
 
#include "async.hpp"
#include "async_task.hpp"
#include "awaiters.hpp"
#include "debug_scheduler.hpp"
#include <uni20/common/trace.hpp>
#include <concepts>
#include <functional>
#include <memory>
#include <type_traits>
#include <utility>
 
namespace uni20::async
{
namespace detail
{
 
template <class Output, class Input>
void validate_distinct_async_operand_queues(Output const& output, Input const& input)
{
  if constexpr (requires {
                  output.queue();
                  input.queue();
                })
  {
    ERROR_IF(std::addressof(output.queue()) == std::addressof(input.queue()),
             "async output must not share an epoch queue with an input");
  }
}
 
} // namespace detail
 
 
template <typename Awaitable>
concept read_buffer_awaitable = requires(Awaitable& a) {
  typename std::remove_cvref_t<decltype(get_awaiter(a))>::value_type;
  { get_awaiter(a).await_resume() };
};
 
template <typename Awaitable>
concept write_buffer_awaitable = requires(Awaitable& a, typename Awaitable::value_type v) {
  typename std::remove_cvref_t<decltype(get_awaiter(a))>::value_type;
  { get_awaiter(a).await_resume() = std::move(v) };
};
 
template <typename Awaitable>
concept emplace_buffer_awaitable = requires(Awaitable& a, typename Awaitable::value_type v) {
  typename Awaitable::value_type;
  get_awaiter(a).await_resume().emplace(std::move(v));
};
 
template <typename Awaitable, typename T>
concept emplace_buffer_awaitable_of =
    emplace_buffer_awaitable<Awaitable> && std::constructible_from<typename Awaitable::value_type, T>;
 
template <typename Awaitable, typename T>
concept read_buffer_awaitable_of =
    requires(Awaitable a) { requires std::constructible_from<T, decltype(get_awaiter(a).await_resume())>; };
 
template <typename Awaitable, typename T>
concept write_buffer_awaitable_of = requires(Awaitable a) {
  { get_awaiter(a).await_resume() = std::declval<T>() };
};
 
template <typename Awaitable>
concept read_write_buffer_awaitable = read_buffer_awaitable<Awaitable> && write_buffer_awaitable<Awaitable>;
 
template <typename Awaitable, typename T>
concept read_write_buffer_awaitable_of =
    read_buffer_awaitable_of<Awaitable, T> && write_buffer_awaitable_of<Awaitable, T>;
 
template <typename T>
concept async_reader = requires(T t) { requires read_buffer_awaitable<decltype(t.read())>; };
 
template <typename AsyncLike> using write_buffer_t = decltype(std::declval<AsyncLike&>().write());
template <typename AsyncLike> using take_awaiter_t = decltype(std::declval<write_buffer_t<AsyncLike>&>().take());
 
template <typename AsyncLike>
concept async_writer = requires(AsyncLike t) { requires emplace_buffer_awaitable<decltype(t.write())>; };
 
template <typename AsyncLike, typename T>
concept async_reader_of =
    requires { requires read_buffer_awaitable_of<decltype(std::declval<AsyncLike&>().read()), T>; };
 
template <typename AsyncLike, typename T>
concept async_writer_to =
    requires { requires emplace_buffer_awaitable_of<decltype(std::declval<AsyncLike&>().write()), T>; };
 
static_assert(async_writer_to<Async<int>, int>);
static_assert(!async_writer_to<Async<int> const, int>);
 
template <typename AsyncLike, typename T>
concept async_movable_to = requires {
  requires emplace_buffer_awaitable_of<write_buffer_t<AsyncLike>, T>;
  requires read_buffer_awaitable_of<take_awaiter_t<AsyncLike>, T>;
};
 
template <typename AsyncLike, typename T>
concept async_mutable_writer_of = requires {
  requires emplace_buffer_awaitable_of<write_buffer_t<AsyncLike>, T>;
  requires read_buffer_awaitable_of<take_awaiter_t<AsyncLike>, T>;
};
 
template <typename T>
concept async_read_writer = requires(T t) {
  requires async_reader<T>;
  requires async_writer<T>;
  requires emplace_buffer_awaitable<decltype(t.write())>;
};
 
template <typename T>
concept async_like = async_reader<T> && async_writer<T>;
 
template <typename T, typename U, typename Op>
using async_binary_result_t =
    decltype(std::declval<Op>()(std::declval<async_value_t<T>>(), std::declval<async_value_t<U>>()));
 
template <typename T, typename U, typename Op>
concept async_binary_applicable = (async_reader<std::remove_cvref_t<T>> || async_reader<std::remove_cvref_t<U>>) &&
                                  requires(Op op, async_value_t<T> t, async_value_t<U> u) {
                                    { op(t, u) };
                                  };
 
template <typename A, typename B, typename Writer, typename Op>
concept async_binary_assignable = async_writer<Writer> && requires(Op op, async_value_t<A> a, async_value_t<B> b,
                                                                   Writer& out) {
  requires std::constructible_from<typename std::remove_cvref_t<decltype(out.write())>::value_type, decltype(op(a, b))>;
};
 
template <typename T, typename U, typename Op>
concept is_async_compound_applicable =
    async_like<std::remove_reference_t<T>> &&
    requires(async_value_t<T> lhs, async_value_t<U>&& rhs, Op op) { op(lhs, std::forward<async_value_t<U>>(rhs)); };
 
// Example using move semantics
// template <typename A, typename B, typename R, typename Op> void async_binary_op(Async<R>& out, A&& a, B&& b, Op op)
// {
//   schedule([](auto a_, auto b_, WriteBuffer<R> out_, Op op_) -> AsyncTask {
//     auto [va, vb] = co_await all(std::move(a_), std::move(b_));
//     a_.release();
//     b_.release();
//     R tmp = op_(std::move(va), std::move(vb)); // move values into op
//     auto& outval = co_await out_;
//     outval = std::move(tmp);
//     co_return;
//   }(read(std::forward<A>(a)), read(std::forward<B>(b)), out.write(), std::move(op)));
// }
 
template <typename A, typename B, async_writer Writer, typename Op>
  requires async_binary_applicable<A, B, Op>
void async_binary_op(Writer& out, A&& a, B&& b, Op op)
{
  detail::validate_distinct_async_operand_queues(out, a);
  detail::validate_distinct_async_operand_queues(out, b);
 
  auto a_buf = read(std::forward<A>(a));
  auto b_buf = read(std::forward<B>(b));
  auto out_buf = out.write();
 
  schedule([](auto a_, auto b_, auto out_, Op op_) static -> AsyncTask {
    auto ab = co_await all(a_, b_);
    auto tmp = op_(std::get<0>(ab), std::get<1>(ab));
    a_.release();
    b_.release();
    co_await out_ = std::move(tmp);
    co_return;
  }(std::move(a_buf), std::move(b_buf), std::move(out_buf), std::move(op)));
}
 
template <typename T, typename U, typename Op> void async_compound_op(Async<T>& lhs, U&& rhs, Op op)
{
  detail::validate_distinct_async_operand_queues(lhs, rhs);
 
  auto rhs_buf = read(std::forward<U>(rhs));
  auto lhs_out = lhs.write();
 
  schedule([](WriteBuffer<T> out_, auto rhs_, Op op_) static -> AsyncTask {
    auto output_storage = out_.storage();
    auto awaited = co_await all(output_storage, rhs_);
    auto& storage = std::get<0>(awaited);
    auto& rhs_value = std::get<1>(awaited);
    if (!storage.constructed()) throw buffer_write_uninitialized{};
    op_(*storage, rhs_value);
    co_return;
  }(std::move(lhs_out), std::move(rhs_buf), std::move(op)));
}
 
// This version is correct, but does not use all() awaiter
// template <typename A, typename B, typename R, typename Op> void async_binary_op(Async<R>& out, A&& a, B&& b, Op op)
// {
//   auto a_buf = read(std::forward<A>(a));
//   auto b_buf = read(std::forward<B>(b));
//   auto out_buf = out.write();
//
//   schedule([](auto a_, auto b_, WriteBuffer<R> out_, Op op_) -> AsyncTask {
//     auto const& va = co_await a_;
//     auto const& vb = co_await b_;
//     auto tmp = op_(va, vb);
//     a_.release();
//     b_.release();
//     co_await out_ = std::move(tmp); // Suspend *after* releasing readers
//     co_return;
//   }(read(std::forward<A>(a)), read(std::forward<B>(b)), out.write(), std::move(op)));
// }
 
namespace detail
{
 
template <typename Awaitable, typename Target>
concept async_initialization_awaitable =
    read_buffer_awaitable<Awaitable> &&
    std::constructible_from<Target, decltype(get_awaiter(std::declval<Awaitable&>()).await_resume())>;
 
template <typename Target, typename Source>
concept current_value_assignment_source =
    !is_async_alias_v<Target> && std::constructible_from<Target, async_value_t<Source> const&> &&
    write_proxy_assignable_source<Target, Target>;
 
template <typename Awaitable, typename Target>
concept async_alias_assignment_awaitable =
    read_buffer_awaitable<Awaitable> &&
    async_alias_assignable_from<Target, decltype(get_awaiter(std::declval<Awaitable&>()).await_resume())>;
 
} // namespace detail
 
template <typename T, typename U>
  requires(!is_async_alias_v<T> && detail::async_initialization_awaitable<U, T>)
void async_initialize(WriteBuffer<T> destination, U&& source)
{
  schedule([](auto source_, WriteBuffer<T> destination_) static -> AsyncTask {
    auto&& value = co_await source_;
    auto target = co_await destination_;
    target.emplace(value);
    source_.release();
    co_return;
  }(std::forward<U>(source), std::move(destination)));
}
 
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)
{
  async_initialize(destination.write(), read(std::forward<U>(source)));
}
 
template <typename T, typename U>
  requires(!is_async_alias_v<T> && read_buffer_awaitable_of<U, T> && detail::write_proxy_assignable_source<T, T>)
void async_assign(WriteBuffer<T> destination, U&& source)
{
  schedule([](auto source_, WriteBuffer<T> destination_) static -> AsyncTask {
    T value(co_await source_);
    source_.release();
    co_await destination_ = std::move(value);
    co_return;
  }(std::forward<U>(source), std::move(destination)));
}
 
template <typename T, typename U>
  requires(is_async_alias_v<T> && detail::async_alias_assignment_awaitable<U, T>)
void async_assign(WriteBuffer<T> destination, U&& source)
{
  schedule([](auto source_, WriteBuffer<T> destination_) static -> AsyncTask {
    auto&& value = co_await source_;
    co_await destination_ = value;
    source_.release();
    co_return;
  }(std::forward<U>(source), std::move(destination)));
}
 
template <typename T, typename U>
void async_assign(Async<T>& lhs, U&& rhs)
  requires(detail::current_value_assignment_source<T, U> ||
           (is_async_alias_v<T> && detail::async_alias_assignable_from<T, async_value_t<U> const&>))
{
  async_assign(lhs.write(), read(std::forward<U>(rhs)));
}
 
template <typename T, typename U>
void async_assign(WriteBuffer<T> lhs, U&& rhs)
  requires(!read_buffer_awaitable<std::remove_cvref_t<U>>) &&
          (detail::current_value_assignment_source<T, U> ||
           (is_async_alias_v<T> && detail::async_alias_assignable_from<T, async_value_t<U> const&>))
{
  async_assign(std::move(lhs), read(std::forward<U>(rhs)));
}
 
template <typename T, typename U>
  requires(!is_async_alias_v<T>) && async_mutable_writer_of<U, T> && async_movable_to<U, T>
void async_move(WriteBuffer<T> lhs, U&& rhs)
{
  auto src = std::forward<U>(rhs).write();
  schedule([](auto src, WriteBuffer<T> dst) static -> AsyncTask {
    T movable(co_await src.take());
    src.release();
    co_await dst = std::move(movable);
    co_return;
  }(std::move(src), std::move(lhs)));
}
 
template <typename T, typename U>
  requires(!is_async_alias_v<T>) && (!async_writer<std::remove_cvref_t<U>>) && std::constructible_from<T, U&&>
void async_move(WriteBuffer<T> lhs, U&& rhs)
{
  schedule([](T val, WriteBuffer<T> dst) static -> AsyncTask {
    co_await dst = std::move(val);
    co_return;
  }(std::move(rhs), std::move(lhs)));
}
 
template <typename T, typename U>
  requires(!is_async_alias_v<T>)
void async_move(Async<T>& lhs, U&& rhs)
{
  async_move(lhs.write(), std::forward<U>(rhs));
}
 
// +, -, *, /
#define UNI20_DEFINE_BINARY_OP(OPNAME, OP)                                                                             \
  template <typename T, typename U>                                                                                    \
    requires async_binary_applicable<T, U, OP>                                                                         \
  auto operator OPNAME(T&& a, U&& b)                                                                                   \
  {                                                                                                                    \
    using R = async_binary_result_t<T, U, OP>;                                                                         \
    Async<R> result;                                                                                                   \
    async_binary_op(result, std::forward<T>(a), std::forward<U>(b), OP{});                                             \
    return result;                                                                                                     \
  }
 
UNI20_DEFINE_BINARY_OP(+, std::plus<>)
UNI20_DEFINE_BINARY_OP(-, std::minus<>)
UNI20_DEFINE_BINARY_OP(*, std::multiplies<>)
UNI20_DEFINE_BINARY_OP(/, std::divides<>)
 
#undef UNI20_DEFINE_BINARY_OP
 
#define UNI20_DEFINE_ASSIGN_OP(NAME, OP)                                                                               \
  struct NAME                                                                                                          \
  {                                                                                                                    \
      template <typename LHS, typename RHS> void operator()(LHS& lhs, RHS&& rhs) const                                 \
      {                                                                                                                \
        lhs OP std::forward<RHS>(rhs);                                                                                 \
      }                                                                                                                \
  };
 
UNI20_DEFINE_ASSIGN_OP(plus_assign, +=)
UNI20_DEFINE_ASSIGN_OP(minus_assign, -=)
UNI20_DEFINE_ASSIGN_OP(multiplies_assign, *=)
UNI20_DEFINE_ASSIGN_OP(divides_assign, /=)
 
#undef UNI20_DEFINE_ASSIGN_OP
 
#define UNI20_DEFINE_ASYNC_COMPOUND_OPERATOR(OPSYM, FUNCTOR)                                                           \
  template <typename T, typename U>                                                                                    \
    requires uni20::async::is_async_compound_applicable<Async<T>&, U, uni20::async::FUNCTOR>                           \
  Async<T>& operator OPSYM(Async<T>& lhs, U && rhs)                                                                    \
  {                                                                                                                    \
    uni20::async::async_compound_op(lhs, std::forward<U>(rhs), uni20::async::FUNCTOR{});                               \
    return lhs;                                                                                                        \
  }
 
UNI20_DEFINE_ASYNC_COMPOUND_OPERATOR(+=, plus_assign)
UNI20_DEFINE_ASYNC_COMPOUND_OPERATOR(-=, minus_assign)
UNI20_DEFINE_ASYNC_COMPOUND_OPERATOR(*=, multiplies_assign)
UNI20_DEFINE_ASYNC_COMPOUND_OPERATOR(/=, divides_assign)
 
#undef UNI20_DEFINE_ASYNC_COMPOUND_OPERATOR
 
// unary operators
 
template <typename T, typename U>
void async_negate(WriteBuffer<T> lhs, U&& rhs)
  requires read_buffer_awaitable_of<U, T>
{
  schedule([](auto in_, WriteBuffer<T> out_) static -> AsyncTask {
    auto result = -co_await in_;
    in_.release();
    co_await out_ = std::move(result);
    co_return;
  }(std::move(rhs), std::move(lhs)));
}
 
template <typename T>
auto operator-(Async<T> const& x)
  requires requires(T t) { -t; }
{
  Async<T> result;
  async_negate(result.write(), x.read());
  return result;
}
 
 
} // namespace uni20::async