future.hpp

Go to the documentation of this file.

#pragma once
 
 
#include <threadman/exceptions.hpp>
#include <threadman/log.hpp>
#include <threadman/metrics.hpp>
#include <threadman/stats.hpp>
 
#include <commons/display_info.hpp>
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <exception>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <type_traits>
#include <utility>
#include <vector>
 
namespace threadman {
 
class IExecutor;  // forward — defined in <threadman/executor.hpp>
 
namespace detail {
 
template <class T>
struct FutureSlot {
    std::optional<T> value;
};
 
template <>
struct FutureSlot<void> {};
 
template <class T>
class FutureState : public std::enable_shared_from_this<FutureState<T>> {
public:
    using value_type = T;
    using Continuation = std::function<void()>;
 
    bool ready() const noexcept {
        return ready_.load(std::memory_order_acquire);
    }
    bool has_exception() const noexcept {
        return exception_set_.load(std::memory_order_acquire);
    }
    std::uint32_t continuation_count() const noexcept {
        std::scoped_lock guard(mtx_);
        return static_cast<std::uint32_t>(continuations_.size());
    }
 
    FutureSnapshot snapshot(const bool shared) const {
        FutureSnapshot snap;
        snap.ready = ready();
        snap.has_exception = has_exception();
        snap.continuation_count = continuation_count();
        snap.shared = shared;
        return snap;
    }
 
    template <class U = T>
        requires(!std::is_void_v<U>)
    void set_value(U&& v) {
        std::vector<Continuation> to_run;
        {
            std::unique_lock lock(mtx_);
            if (ready_.load(std::memory_order_relaxed)) {
                throw PromiseAlreadySatisfiedError("promise already satisfied");
            }
            slot_.value.emplace(std::forward<U>(v));
            ready_.store(true, std::memory_order_release);
            to_run.swap(continuations_);
            cv_.notify_all();
        }
        metrics::futures_satisfied_value().inc();
        for (auto& c : to_run) {
            c();
        }
    }
 
    template <class U = T>
        requires std::is_void_v<U>
    void set_value() {
        std::vector<Continuation> to_run;
        {
            std::unique_lock lock(mtx_);
            if (ready_.load(std::memory_order_relaxed)) {
                throw PromiseAlreadySatisfiedError("promise already satisfied");
            }
            ready_.store(true, std::memory_order_release);
            to_run.swap(continuations_);
            cv_.notify_all();
        }
        metrics::futures_satisfied_value().inc();
        for (auto& c : to_run) {
            c();
        }
    }
 
    void set_exception(const std::exception_ptr& ex) {
        std::vector<Continuation> to_run;
        {
            std::unique_lock lock(mtx_);
            if (ready_.load(std::memory_order_relaxed)) {
                throw PromiseAlreadySatisfiedError("promise already satisfied");
            }
            exception_ = ex;
            exception_set_.store(true, std::memory_order_release);
            ready_.store(true, std::memory_order_release);
            to_run.swap(continuations_);
            cv_.notify_all();
        }
        metrics::futures_satisfied_exception().inc();
        for (auto& c : to_run) {
            c();
        }
    }
 
    void wait() const {
        std::unique_lock lock(mtx_);
        cv_.wait(lock, [this] { return ready_.load(std::memory_order_acquire); });
    }
 
    template <class Rep, class Period>
    bool wait_for(const std::chrono::duration<Rep, Period>& dur) const {
        std::unique_lock lock(mtx_);
        return cv_.wait_for(lock, dur, [this] { return ready_.load(std::memory_order_acquire); });
    }
 
    template <class Clock, class Duration>
    bool wait_until(const std::chrono::time_point<Clock, Duration>& tp) const {
        std::unique_lock lock(mtx_);
        return cv_.wait_until(lock, tp, [this] { return ready_.load(std::memory_order_acquire); });
    }
 
    template <class U = T>
        requires(!std::is_void_v<U>)
    U get_value_move() {
        wait();
        if (has_exception()) {
            std::rethrow_exception(exception_);
        }
        return std::move(*slot_.value);
    }
 
    template <class U = T>
        requires(!std::is_void_v<U>)
    const U& get_value_ref() const {
        wait();
        if (has_exception()) {
            std::rethrow_exception(exception_);
        }
        return *slot_.value;
    }
 
    template <class U = T>
        requires std::is_void_v<U>
    void get_value_void() const {
        wait();
        if (has_exception()) {
            std::rethrow_exception(exception_);
        }
    }
 
    enum class AddResult { Scheduled, AlreadyReady };
 
    AddResult add_continuation(Continuation cb) {
        std::unique_lock lock(mtx_);
        if (ready_.load(std::memory_order_relaxed)) {
            return AddResult::AlreadyReady;
        }
        continuations_.push_back(std::move(cb));
        return AddResult::Scheduled;
    }
 
    std::exception_ptr exception() const noexcept {
        return exception_;
    }
 
private:
    mutable std::mutex mtx_;
    mutable std::condition_variable cv_;
    std::atomic<bool> ready_{false};
    std::atomic<bool> exception_set_{false};
    FutureSlot<T> slot_;
    std::exception_ptr exception_;
    std::vector<Continuation> continuations_;
};
 
// Forward reference to the dispatch helper defined in the higher-tier
// <threadman/executor.hpp>; future.hpp must not include executor.hpp (it sits
// below it in the dependency order), so the declaration is intentional.
// NOLINTNEXTLINE(readability-redundant-declaration)
void execute_on(IExecutor& exec, std::function<void()> task);
 
}  // namespace detail
 
template <class T>
class Promise;
template <class T>
class SharedFuture;
 
// ---------------------------------------------------------------------------
// Future<T> — one-shot, move-only.
// ---------------------------------------------------------------------------
template <class T>
class Future {
public:
    using value_type = T;
 
    Future() = default;
    Future(const Future&) = delete;
    Future& operator=(const Future&) = delete;
    Future(Future&&) noexcept = default;
    Future& operator=(Future&&) noexcept = default;
    ~Future() = default;
 
    [[nodiscard]] bool valid() const noexcept {
        return static_cast<bool>(state_);
    }
    [[nodiscard]] bool is_ready() const noexcept {
        return state_ && state_->ready();
    }
 
    void wait() const {
        check_valid();
        state_->wait();
    }
 
    template <class Rep, class Period>
    [[nodiscard]] bool wait_for(const std::chrono::duration<Rep, Period>& dur) const {
        check_valid();
        return state_->wait_for(dur);
    }
 
    template <class Clock, class Duration>
    [[nodiscard]] bool wait_until(const std::chrono::time_point<Clock, Duration>& tp) const {
        check_valid();
        return state_->wait_until(tp);
    }
 
    template <class U = T>
        requires(!std::is_void_v<U>)
    U get() {
        check_valid();
        auto s = std::move(state_);
        return s->get_value_move();
    }
 
    template <class U = T>
        requires std::is_void_v<U>
    void get() {
        check_valid();
        auto s = std::move(state_);
        s->get_value_void();
    }
 
    [[nodiscard]] SharedFuture<T> share();
 
    template <class Exec, class Fn>
    auto then(Exec& exec, Fn&& fn);
 
    template <class Exec, class Fn>
    Future<T> on_error(Exec& exec, Fn&& fn);
 
    [[nodiscard]] FutureSnapshot snapshot() const {
        return state_ ? state_->snapshot(false) : FutureSnapshot{};
    }
 
    [[nodiscard]] static const comms::DisplayInfo& display_info() {
        static const comms::DisplayInfo info{
            .name = "Future",
            .description = "One-shot, move-only async result with single .get and single .then.",
            .icon = comms::Icon::from("mdi:clock-fast"),
        };
        return info;
    }
 
private:
    template <class U>
    friend class Promise;
    template <class U>
    friend class SharedFuture;
    template <class U>
    friend class Future;
 
    explicit Future(std::shared_ptr<detail::FutureState<T>> s) : state_(std::move(s)) {}
 
    void check_valid() const {
        if (!state_) {
            throw FutureError("future has no state (default-constructed or moved-from)");
        }
    }
 
    std::shared_ptr<detail::FutureState<T>> state_;
    bool has_continuation_{false};
};
 
// ---------------------------------------------------------------------------
// SharedFuture<T> — multi-shot, copyable.
// ---------------------------------------------------------------------------
template <class T>
class SharedFuture {
public:
    using value_type = T;
 
    SharedFuture() = default;
    SharedFuture(const SharedFuture&) = default;
    SharedFuture(SharedFuture&&) noexcept = default;
    SharedFuture& operator=(const SharedFuture&) = default;
    SharedFuture& operator=(SharedFuture&&) noexcept = default;
    ~SharedFuture() = default;
 
    [[nodiscard]] bool valid() const noexcept {
        return static_cast<bool>(state_);
    }
    [[nodiscard]] bool is_ready() const noexcept {
        return state_ && state_->ready();
    }
 
    void wait() const {
        check_valid();
        state_->wait();
    }
 
    template <class Rep, class Period>
    [[nodiscard]] bool wait_for(const std::chrono::duration<Rep, Period>& dur) const {
        check_valid();
        return state_->wait_for(dur);
    }
 
    template <class U = T>
        requires(!std::is_void_v<U>)
    const U& get() const {
        check_valid();
        return state_->get_value_ref();
    }
 
    template <class U = T>
        requires std::is_void_v<U>
    void get() const {
        check_valid();
        state_->get_value_void();
    }
 
    template <class Exec, class Fn>
    auto then(Exec& exec, Fn&& fn) const;
 
    template <class Exec, class Fn>
    SharedFuture<T> on_error(Exec& exec, Fn&& fn) const;
 
    [[nodiscard]] FutureSnapshot snapshot() const {
        return state_ ? state_->snapshot(true) : FutureSnapshot{};
    }
 
    [[nodiscard]] static const comms::DisplayInfo& display_info() {
        static const comms::DisplayInfo info{
            .name = "SharedFuture",
            .description = "Multi-shot, copyable async result supporting many .get and .then.",
            .icon = comms::Icon::from("mdi:clock-multiple"),
        };
        return info;
    }
 
private:
    template <class U>
    friend class Future;
    template <class U>
    friend class SharedFuture;
    template <class U>
    friend class Promise;
 
    explicit SharedFuture(std::shared_ptr<detail::FutureState<T>> s) : state_(std::move(s)) {}
 
    void check_valid() const {
        if (!state_) {
            throw FutureError("shared_future has no state (default-constructed)");
        }
    }
 
    std::shared_ptr<detail::FutureState<T>> state_;
};
 
template <class T>
SharedFuture<T> Future<T>::share() {
    check_valid();
    return SharedFuture<T>{std::move(state_)};
}
 
// ---------------------------------------------------------------------------
// Promise<T>.
// ---------------------------------------------------------------------------
template <class T>
class Promise {
public:
    using value_type = T;
 
    Promise() : state_(std::make_shared<detail::FutureState<T>>()) {
        metrics::futures_created().inc();
    }
    Promise(const Promise&) = delete;
    Promise& operator=(const Promise&) = delete;
    Promise(Promise&&) noexcept = default;
    Promise& operator=(Promise&&) noexcept = default;
 
    ~Promise() {
        if (state_ && !state_->ready()) {
            try {
                state_->set_exception(std::make_exception_ptr(
                    BrokenPromiseError("promise destroyed before satisfaction")));
            } catch (...) {
                // dtor must not propagate
            }
        }
    }
 
    [[nodiscard]] bool valid() const noexcept {
        return static_cast<bool>(state_);
    }
 
    [[nodiscard]] Future<T> get_future() {
        if (!state_) {
            throw FutureError("promise has no state");
        }
        if (future_retrieved_) {
            throw FutureAlreadyRetrievedError("future already retrieved from promise");
        }
        future_retrieved_ = true;
        return Future<T>{state_};
    }
 
    template <class U = T>
        requires(!std::is_void_v<U>)
    void set_value(U&& v) {
        if (!state_) {
            throw FutureError("promise has no state");
        }
        state_->set_value(std::forward<U>(v));
    }
 
    template <class U = T>
        requires std::is_void_v<U>
    void set_value() {
        if (!state_) {
            throw FutureError("promise has no state");
        }
        state_->set_value();
    }
 
    void set_exception(const std::exception_ptr& ex) {
        if (!state_) {
            throw FutureError("promise has no state");
        }
        state_->set_exception(ex);
    }
 
    [[nodiscard]] static const comms::DisplayInfo& display_info() {
        static const comms::DisplayInfo info{
            .name = "Promise",
            .description = "Producer side of a Future/SharedFuture pair.",
            .icon = comms::Icon::from("mdi:handshake"),
        };
        return info;
    }
 
private:
    std::shared_ptr<detail::FutureState<T>> state_;
    bool future_retrieved_ = false;
};
 
// ---------------------------------------------------------------------------
// .then() implementations
// ---------------------------------------------------------------------------
namespace detail {
 
template <class Fn, class Arg>
struct invoke_result_helper {
    using type = std::invoke_result_t<Fn, Arg>;
};
 
template <class Fn>
struct invoke_result_helper<Fn, void> {
    using type = std::invoke_result_t<Fn>;
};
 
template <class Fn, class Arg>
using future_invoke_result_t = typename invoke_result_helper<Fn, Arg>::type;
 
template <class T, class Fn, class Sink>
inline void run_continuation(const std::shared_ptr<FutureState<T>>& src,
                             Fn& fn,
                             const std::shared_ptr<Sink>& sink_promise) {
    metrics::future_continuations_dispatched().inc();
    using SinkT = typename Sink::value_type;
    try {
        if (src->has_exception()) {
            sink_promise->set_exception(src->exception());
            return;
        }
        if constexpr (std::is_void_v<T>) {
            if constexpr (std::is_void_v<SinkT>) {
                fn();
                sink_promise->set_value();
            } else {
                sink_promise->set_value(fn());
            }
        } else {
            if constexpr (std::is_void_v<SinkT>) {
                fn(src->get_value_ref());
                sink_promise->set_value();
            } else {
                sink_promise->set_value(fn(src->get_value_ref()));
            }
        }
    } catch (...) {
        sink_promise->set_exception(std::current_exception());
    }
}
 
}  // namespace detail
 
template <class T>
template <class Exec, class Fn>
auto Future<T>::then(Exec& exec, Fn&& fn) {
    check_valid();
    if (has_continuation_) {
        throw ContinuationAlreadyRegisteredError(
            "Future::then already registered; use share() for multi-shot");
    }
    has_continuation_ = true;
    metrics::future_continuations_registered().inc();
    using U = typename detail::invoke_result_helper<std::decay_t<Fn>, T>::type;
 
    auto sink_promise = std::make_shared<Promise<U>>();
    auto sink_future = sink_promise->get_future();
 
    auto src = state_;  // keep state_ alive on this Future too — only the
                        // has_continuation_ flag prevents reuse, so that a
                        // second .then() throws ContinuationAlreadyRegisteredError
                        // rather than the generic "no state" FutureError.
    auto fn_shared = std::make_shared<std::decay_t<Fn>>(std::forward<Fn>(fn));
 
    auto schedule = [&exec, src, fn_shared, sink_promise]() {
        detail::execute_on(exec, [src, fn_shared, sink_promise]() {
            detail::run_continuation<T>(src, *fn_shared, sink_promise);
        });
    };
 
    auto result = src->add_continuation([schedule = std::move(schedule)]() mutable { schedule(); });
    if (result == detail::FutureState<T>::AddResult::AlreadyReady) {
        detail::execute_on(exec, [src, fn_shared, sink_promise]() {
            detail::run_continuation<T>(src, *fn_shared, sink_promise);
        });
    }
    return sink_future;
}
 
template <class T>
template <class Exec, class Fn>
Future<T> Future<T>::on_error(Exec& exec, Fn&& fn) {
    check_valid();
    if (has_continuation_) {
        throw ContinuationAlreadyRegisteredError(
            "Future::on_error already registered (a Future supports a single continuation)");
    }
    has_continuation_ = true;
    metrics::future_continuations_registered().inc();
    auto sink_promise = std::make_shared<Promise<T>>();
    auto sink_future = sink_promise->get_future();
 
    auto src = state_;
    auto fn_shared = std::make_shared<std::decay_t<Fn>>(std::forward<Fn>(fn));
 
    auto run = [src, fn_shared, sink_promise]() {
        metrics::future_continuations_dispatched().inc();
        try {
            if (src->has_exception()) {
                if constexpr (std::is_void_v<T>) {
                    (*fn_shared)(src->exception());
                    sink_promise->set_value();
                } else {
                    auto recovered = (*fn_shared)(src->exception());
                    sink_promise->set_value(std::move(recovered));
                }
            } else {
                if constexpr (std::is_void_v<T>) {
                    sink_promise->set_value();
                } else {
                    sink_promise->set_value(T{src->get_value_ref()});
                }
            }
        } catch (...) {
            sink_promise->set_exception(std::current_exception());
        }
    };
 
    auto schedule = [&exec, run = std::move(run)]() mutable { detail::execute_on(exec, run); };
 
    auto result = src->add_continuation([schedule = std::move(schedule)]() mutable { schedule(); });
    if (result == detail::FutureState<T>::AddResult::AlreadyReady) {
        detail::execute_on(exec, [src, fn_shared, sink_promise]() {
            try {
                if (src->has_exception()) {
                    if constexpr (std::is_void_v<T>) {
                        (*fn_shared)(src->exception());
                        sink_promise->set_value();
                    } else {
                        auto recovered = (*fn_shared)(src->exception());
                        sink_promise->set_value(std::move(recovered));
                    }
                } else {
                    if constexpr (std::is_void_v<T>) {
                        sink_promise->set_value();
                    } else {
                        sink_promise->set_value(T{src->get_value_ref()});
                    }
                }
            } catch (...) {
                sink_promise->set_exception(std::current_exception());
            }
        });
    }
    return sink_future;
}
 
template <class T>
template <class Exec, class Fn>
auto SharedFuture<T>::then(Exec& exec, Fn&& fn) const {
    check_valid();
    metrics::future_continuations_registered().inc();
    using U = typename detail::invoke_result_helper<std::decay_t<Fn>, T>::type;
 
    auto sink_promise = std::make_shared<Promise<U>>();
    auto sink_future = sink_promise->get_future();
 
    auto src = state_;
    auto fn_shared = std::make_shared<std::decay_t<Fn>>(std::forward<Fn>(fn));
 
    auto schedule = [&exec, src, fn_shared, sink_promise]() {
        detail::execute_on(exec, [src, fn_shared, sink_promise]() {
            detail::run_continuation<T>(src, *fn_shared, sink_promise);
        });
    };
 
    auto result = src->add_continuation([schedule = std::move(schedule)]() mutable { schedule(); });
    if (result == detail::FutureState<T>::AddResult::AlreadyReady) {
        detail::execute_on(exec, [src, fn_shared, sink_promise]() {
            detail::run_continuation<T>(src, *fn_shared, sink_promise);
        });
    }
    return sink_future;
}
 
template <class T>
template <class Exec, class Fn>
SharedFuture<T> SharedFuture<T>::on_error(Exec& exec, Fn&& fn) const {
    check_valid();
    metrics::future_continuations_registered().inc();
    auto sink_promise = std::make_shared<Promise<T>>();
    auto sink_future = sink_promise->get_future().share();
 
    auto src = state_;
    auto fn_shared = std::make_shared<std::decay_t<Fn>>(std::forward<Fn>(fn));
 
    auto run = [src, fn_shared, sink_promise]() {
        metrics::future_continuations_dispatched().inc();
        try {
            if (src->has_exception()) {
                if constexpr (std::is_void_v<T>) {
                    (*fn_shared)(src->exception());
                    sink_promise->set_value();
                } else {
                    auto recovered = (*fn_shared)(src->exception());
                    sink_promise->set_value(std::move(recovered));
                }
            } else {
                if constexpr (std::is_void_v<T>) {
                    sink_promise->set_value();
                } else {
                    sink_promise->set_value(T{src->get_value_ref()});
                }
            }
        } catch (...) {
            sink_promise->set_exception(std::current_exception());
        }
    };
 
    auto schedule = [&exec, run = std::move(run)]() mutable { detail::execute_on(exec, run); };
 
    auto result = src->add_continuation([schedule = std::move(schedule)]() mutable { schedule(); });
    if (result == detail::FutureState<T>::AddResult::AlreadyReady) {
        detail::execute_on(exec, run);
    }
    return sink_future;
}
 
}  // namespace threadman