experimental_async.H

Go to the documentation of this file.

/*
                          Aleph_w
 
  Data structures & Algorithms
  version 2.0.0b
  https://github.com/lrleon/Aleph-w
 
  This file is part of Aleph-w library
 
  Copyright (c) 2002-2026 Leandro Rabindranath Leon
1
  Permission is hereby granted, free of charge, to any person obtaining a copy
  of this software and associated documentation files (the "Software"), to deal
  in the Software without restriction, including without limitation the rights
  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  copies of the Software, and to permit persons to whom the Software is
  furnished to do so, subject to the following conditions:
 
  The above copyright notice and this permission notice shall be included in all
  copies or substantial portions of the Software.
 
  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  SOFTWARE.
*/
 
#ifndef ALEPH_EXPERIMENTAL_ASYNC_H
#define ALEPH_EXPERIMENTAL_ASYNC_H
 
#include <ah-errors.H>
#include <thread_pool.H>
 
#ifndef ALEPH_ENABLE_EXPERIMENTAL_ASYNC
#  define ALEPH_ENABLE_EXPERIMENTAL_ASYNC 0
#endif
 
#if ALEPH_ENABLE_EXPERIMENTAL_ASYNC && __cplusplus >= 202002L
#  if defined(__has_include)
#    if __has_include(<coroutine>)
#      include <coroutine>
#      define ALEPH_DETAIL_HAS_COROUTINE_HEADER 1
#    else
#      define ALEPH_DETAIL_HAS_COROUTINE_HEADER 0
#    endif
#  else
#    define ALEPH_DETAIL_HAS_COROUTINE_HEADER 0
#  endif
#else
#  define ALEPH_DETAIL_HAS_COROUTINE_HEADER 0
#endif
 
#if ALEPH_ENABLE_EXPERIMENTAL_ASYNC && \
    __cplusplus >= 202002L && \
    ALEPH_DETAIL_HAS_COROUTINE_HEADER && \
    defined(__cpp_impl_coroutine) && \
    __cpp_impl_coroutine >= 201902L
#  define ALEPH_HAS_EXPERIMENTAL_ASYNC 1
#else
#  define ALEPH_HAS_EXPERIMENTAL_ASYNC 0
#endif
 
#if ALEPH_HAS_EXPERIMENTAL_ASYNC
 
#include <condition_variable>
#include <exception>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <system_error>
#include <tuple>
#include <type_traits>
#include <utility>
 
namespace Aleph
{
  namespace experimental
  {
    namespace detail
    {
      [[noreturn]] inline void throw_no_state()
      {
        ah_runtime_error() << "scheduled_operation has no state";
        std::terminate();
      }
 
      class ContinuationRegistration
      {
        mutable std::mutex mutex_;
        std::coroutine_handle<> handle_{};
        bool active_ = true;
 
      public:
        explicit ContinuationRegistration(std::coroutine_handle<> handle) noexcept
          : handle_(handle) {}
 
        [[nodiscard]] bool is_active() const noexcept
        {
          std::lock_guard<std::mutex> lock(mutex_);
          return active_ and static_cast<bool>(handle_);
        }
 
        void unregister() noexcept
        {
          std::lock_guard<std::mutex> lock(mutex_);
          active_ = false;
          handle_ = std::coroutine_handle<>{};
        }
 
        void resume_if_active()
        {
          std::coroutine_handle<> handle;
          {
            std::lock_guard<std::mutex> lock(mutex_);
            if (not active_ or not handle_)
              return;
            active_ = false;
            handle = std::exchange(handle_, std::coroutine_handle<>{});
          }
          handle.resume();
        }
      };
 
      template <typename F, typename... Args>
      auto make_invocable(F && f, Args && ... args)
      {
        return [func = std::forward<F>(f),
                args_tuple = std::make_tuple(std::forward<Args>(args)...)]() mutable
          {
            return std::apply([&func](auto && ... a)
                                {
                                  return std::invoke(std::move(func),
                                                     std::forward<decltype(a)>(a)...);
                                }, std::move(args_tuple));
          };
      }
 
      template <typename T>
      class OperationState
      {
        mutable std::mutex mutex_;
        mutable std::condition_variable ready_cv_;
        bool ready_ = false;
        bool consumed_ = false;
        std::exception_ptr exception_;
        std::optional<T> value_;
        std::weak_ptr<ContinuationRegistration> continuation_;
 
        template <typename F>
        void complete(F && fill)
        {
          std::weak_ptr<ContinuationRegistration> continuation;
          {
            std::lock_guard<std::mutex> lock(mutex_);
            if (ready_)
              return;
            fill();
            ready_ = true;
            continuation = std::exchange(continuation_, {});
          }
          ready_cv_.notify_all();
          if (auto registration = continuation.lock(); registration != nullptr)
            registration->resume_if_active();
        }
 
      public:
        OperationState() = default;
 
        [[nodiscard]] bool is_ready() const
        {
          std::lock_guard<std::mutex> lock(mutex_);
          return ready_;
        }
 
        void wait() const
        {
          std::unique_lock<std::mutex> lock(mutex_);
          ready_cv_.wait(lock, [this] { return ready_; });
        }
 
        template <typename U>
        void set_value(U && value)
        {
          complete([this, &value]
                   {
                     value_.emplace(std::forward<U>(value));
                   });
        }
 
        void set_exception(std::exception_ptr exception)
        {
          complete([this, &exception]
                   {
                     exception_ = exception;
                   });
        }
 
        [[nodiscard]] bool attach_continuation(
            const std::shared_ptr<ContinuationRegistration> & continuation)
        {
          std::lock_guard<std::mutex> lock(mutex_);
          if (ready_)
            return false;
          if (auto registered = continuation_.lock();
              registered != nullptr and registered->is_active())
            ah_logic_error_if(true)
              << "scheduled_operation already has a waiting coroutine";
          continuation_ = continuation;
          return true;
        }
 
        void detach_continuation(
            const std::shared_ptr<ContinuationRegistration> & continuation) noexcept
        {
          std::lock_guard<std::mutex> lock(mutex_);
          if (auto registered = continuation_.lock(); registered == continuation)
            continuation_.reset();
        }
 
        T take()
        {
          std::unique_lock<std::mutex> lock(mutex_);
          ready_cv_.wait(lock, [this] { return ready_; });
          if (consumed_)
            throw_no_state();
          consumed_ = true;
          if (exception_ != nullptr)
            std::rethrow_exception(exception_);
          return std::move(*value_);
        }
      };
 
      template <>
      class OperationState<void>
      {
        mutable std::mutex mutex_;
        mutable std::condition_variable ready_cv_;
        bool ready_ = false;
        bool consumed_ = false;
        std::exception_ptr exception_;
        std::weak_ptr<ContinuationRegistration> continuation_;
 
        template <typename F>
        void complete(F && fill)
        {
          std::weak_ptr<ContinuationRegistration> continuation;
          {
            std::lock_guard<std::mutex> lock(mutex_);
            if (ready_)
              return;
            fill();
            ready_ = true;
            continuation = std::exchange(continuation_, {});
          }
          ready_cv_.notify_all();
          if (auto registration = continuation.lock(); registration != nullptr)
            registration->resume_if_active();
        }
 
      public:
        OperationState() = default;
 
        [[nodiscard]] bool is_ready() const
        {
          std::lock_guard<std::mutex> lock(mutex_);
          return ready_;
        }
 
        void wait() const
        {
          std::unique_lock<std::mutex> lock(mutex_);
          ready_cv_.wait(lock, [this] { return ready_; });
        }
 
        void set_value()
        {
          complete([] {});
        }
 
        void set_exception(std::exception_ptr exception)
        {
          complete([this, &exception]
                   {
                     exception_ = exception;
                   });
        }
 
        [[nodiscard]] bool attach_continuation(
            const std::shared_ptr<ContinuationRegistration> & continuation)
        {
          std::lock_guard<std::mutex> lock(mutex_);
          if (ready_)
            return false;
          if (auto registered = continuation_.lock();
              registered != nullptr and registered->is_active())
            ah_logic_error_if(true)
              << "scheduled_operation already has a waiting coroutine";
          continuation_ = continuation;
          return true;
        }
 
        void detach_continuation(
            const std::shared_ptr<ContinuationRegistration> & continuation) noexcept
        {
          std::lock_guard<std::mutex> lock(mutex_);
          if (auto registered = continuation_.lock(); registered == continuation)
            continuation_.reset();
        }
 
        void take()
        {
          std::unique_lock<std::mutex> lock(mutex_);
          ready_cv_.wait(lock, [this] { return ready_; });
          if (consumed_)
            throw_no_state();
          consumed_ = true;
          if (exception_ != nullptr)
            std::rethrow_exception(exception_);
        }
      };
    } // namespace detail
 
    template <typename T>
    class scheduled_operation
    {
      std::shared_ptr<detail::OperationState<T>> state_;
      std::shared_ptr<detail::ContinuationRegistration> continuation_registration_;
 
      explicit scheduled_operation(std::shared_ptr<detail::OperationState<T>> state)
        : state_(std::move(state)) {}
 
      void unregister_continuation() noexcept
      {
        if (state_ != nullptr and continuation_registration_ != nullptr)
          state_->detach_continuation(continuation_registration_);
        if (continuation_registration_ != nullptr)
          continuation_registration_->unregister();
        continuation_registration_.reset();
      }
 
      std::shared_ptr<detail::OperationState<T>> consume_state()
      {
        unregister_continuation();
        auto state = std::exchange(state_, nullptr);
        if (state == nullptr)
          detail::throw_no_state();
        return state;
      }
 
      detail::OperationState<T> & state()
      {
        if (state_ == nullptr)
          detail::throw_no_state();
        return *state_;
      }
 
      const detail::OperationState<T> & state() const
      {
        if (state_ == nullptr)
          detail::throw_no_state();
        return *state_;
      }
 
      template <typename F, typename... Args>
      friend auto schedule(ThreadPool & pool, F && f, Args && ... args)
        -> scheduled_operation<std::invoke_result_t<F, Args...>>;
 
    public:
      scheduled_operation() = default;
      scheduled_operation(const scheduled_operation &) = delete;
      scheduled_operation & operator = (const scheduled_operation &) = delete;
      scheduled_operation(scheduled_operation &&) noexcept = default;
      scheduled_operation & operator = (scheduled_operation && other) noexcept
      {
        if (this != &other)
          {
            unregister_continuation();
            state_ = std::move(other.state_);
            continuation_registration_ = std::move(other.continuation_registration_);
          }
        return *this;
      }
      ~scheduled_operation() { unregister_continuation(); }
 
      [[nodiscard]] bool valid() const noexcept { return state_ != nullptr; }
 
      [[nodiscard]] bool is_ready() const
      {
        return state().is_ready();
      }
 
      void wait() const
      {
        state().wait();
      }
 
      T get()
      {
        return consume_state()->take();
      }
 
      [[nodiscard]] bool await_ready() const
      {
        return state().is_ready();
      }
 
      bool await_suspend(std::coroutine_handle<> continuation)
      {
        auto registration =
            std::make_shared<detail::ContinuationRegistration>(continuation);
        auto previous = std::move(continuation_registration_);
        continuation_registration_ = registration;
        try
          {
            if (state().attach_continuation(registration))
              {
                if (previous)
                  previous->unregister();
                return true;
              }
          }
        catch (...)
          {
            continuation_registration_ = std::move(previous);
            throw;
          }
        continuation_registration_ = std::move(previous);
        return false;
      }
 
      T await_resume()
      {
        return consume_state()->take();
      }
    };
 
    template <>
    class scheduled_operation<void>
    {
      std::shared_ptr<detail::OperationState<void>> state_;
      std::shared_ptr<detail::ContinuationRegistration> continuation_registration_;
 
      explicit scheduled_operation(std::shared_ptr<detail::OperationState<void>> state)
        : state_(std::move(state)) {}
 
      void unregister_continuation() noexcept
      {
        if (state_ != nullptr and continuation_registration_ != nullptr)
          state_->detach_continuation(continuation_registration_);
        if (continuation_registration_ != nullptr)
          continuation_registration_->unregister();
        continuation_registration_.reset();
      }
 
      std::shared_ptr<detail::OperationState<void>> consume_state()
      {
        unregister_continuation();
        auto state = std::exchange(state_, nullptr);
        if (state == nullptr)
          detail::throw_no_state();
        return state;
      }
 
      detail::OperationState<void> & state()
      {
        if (state_ == nullptr)
          detail::throw_no_state();
        return *state_;
      }
 
      const detail::OperationState<void> & state() const
      {
        if (state_ == nullptr)
          detail::throw_no_state();
        return *state_;
      }
 
      template <typename F, typename... Args>
      friend auto schedule(ThreadPool & pool, F && f, Args && ... args)
        -> scheduled_operation<std::invoke_result_t<F, Args...>>;
 
    public:
      scheduled_operation() = default;
      scheduled_operation(const scheduled_operation &) = delete;
      scheduled_operation & operator = (const scheduled_operation &) = delete;
      scheduled_operation(scheduled_operation &&) noexcept = default;
      scheduled_operation & operator = (scheduled_operation && other) noexcept
      {
        if (this != &other)
          {
            unregister_continuation();
            state_ = std::move(other.state_);
            continuation_registration_ = std::move(other.continuation_registration_);
          }
        return *this;
      }
      ~scheduled_operation() { unregister_continuation(); }
 
      [[nodiscard]] bool valid() const noexcept { return state_ != nullptr; }
 
      [[nodiscard]] bool is_ready() const
      {
        return state().is_ready();
      }
 
      void wait() const
      {
        state().wait();
      }
 
      void get()
      {
        consume_state()->take();
      }
 
      [[nodiscard]] bool await_ready() const
      {
        return state().is_ready();
      }
 
      bool await_suspend(std::coroutine_handle<> continuation)
      {
        auto registration =
            std::make_shared<detail::ContinuationRegistration>(continuation);
        auto previous = std::move(continuation_registration_);
        continuation_registration_ = registration;
        try
          {
            if (state().attach_continuation(registration))
              {
                if (previous)
                  previous->unregister();
                return true;
              }
          }
        catch (...)
          {
            continuation_registration_ = std::move(previous);
            throw;
          }
        continuation_registration_ = std::move(previous);
        return false;
      }
 
      void await_resume()
      {
        consume_state()->take();
      }
    };
 
    template <typename F, typename... Args>
    [[nodiscard]] auto schedule(ThreadPool & pool, F && f, Args && ... args)
      -> scheduled_operation<std::invoke_result_t<F, Args...>>
    {
      using return_type = std::invoke_result_t<F, Args...>;
      static_assert(not std::is_reference_v<return_type>,
                    "experimental::schedule() does not support reference returns");
 
      auto state = std::make_shared<detail::OperationState<return_type>>();
      auto invocable = detail::make_invocable(std::forward<F>(f),
                                              std::forward<Args>(args)...);
 
      pool.enqueue_detached([state, invocable = std::move(invocable)]() mutable
                            {
                              try
                                {
                                  if constexpr (std::is_void_v<return_type>)
                                    {
                                      invocable();
                                      state->set_value();
                                    }
                                  else
                                    {
                                      state->set_value(invocable());
                                    }
                                }
                              catch (...)
                                {
                                  state->set_exception(std::current_exception());
                                }
                            });
 
      return scheduled_operation<return_type>(std::move(state));
    }
 
    template <typename F, typename... Args>
    [[nodiscard]] auto schedule(F && f, Args && ... args)
      -> scheduled_operation<std::invoke_result_t<F, Args...>>
    {
      return schedule(default_pool(),
                      std::forward<F>(f),
                      std::forward<Args>(args)...);
    }
  } // namespace experimental
} // namespace Aleph
 
#endif
 
#endif