future.hpp

Go to the documentation of this file.

/*
    Copyright 2015 Adobe
    Distributed under the Boost Software License, Version 1.0.
    (See accompanying file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
*/
 
/**************************************************************************************************/
 
#ifndef STLAB_CONCURRENCY_FUTURE_HPP
#define STLAB_CONCURRENCY_FUTURE_HPP
 
#include <stlab/config.hpp>
 
#include <algorithm>
#include <array>
#include <atomic>
#include <cassert>
#include <cstddef>
#include <cstdint>
#include <exception>
#include <functional>
#include <initializer_list>
#include <iterator>
#include <limits>
#include <memory>
#include <mutex>
#include <optional>
#include <stdexcept>
#include <tuple>
#include <type_traits>
#include <utility>
#include <variant> // for std::monostate
#include <vector>
 
#if STLAB_STD_COROUTINES()
#include <coroutine>
#endif
 
#include <stlab/concurrency/executor_base.hpp>
#include <stlab/concurrency/immediate_executor.hpp>
#include <stlab/concurrency/task.hpp>
#include <stlab/concurrency/traits.hpp>
#include <stlab/concurrency/tuple_algorithm.hpp>
#include <stlab/functional.hpp>
#include <stlab/memory.hpp>
#include <stlab/utility.hpp>
 
/**************************************************************************************************/

 
/**************************************************************************************************/
 
namespace stlab {
STLAB_VERSION_NAMESPACE_BEGIN()
 

 
/**************************************************************************************************/
 

template <class F, class... Args>
auto invoke_void_to_monostate_result(F&& f, Args&&... args) {
    if constexpr (std::is_void_v<std::invoke_result_t<F, Args...>>) {
        std::forward<F>(f)(std::forward<Args>(args)...);
        return std::monostate{};
    } else {
        return std::forward<F>(f)(std::forward<Args>(args)...);
    }
}

template <class T>
struct void_to_monostate {
    using type = std::conditional_t<std::is_void_v<T>, std::monostate, T>;
};

template <class T>
using void_to_monostate_t = typename void_to_monostate<T>::type;

template <class T>
inline constexpr bool is_monostate_v = std::is_same_v<T, std::monostate>;

template <class T>
auto optional_monostate_to_bool(std::optional<T>&& o) {
    if constexpr (is_monostate_v<T>) {
        return o.has_value();
    } else {
        return std::move(o);
    }
}

template <class T>
auto monostate_to_void(T&& a) {
    if constexpr (is_monostate_v<std::decay_t<T>>) {
        return;
    } else {
        return std::forward<T>(a);
    }
}

template <class T>
auto monostate_to_empty_tuple(T&& a) {
    if constexpr (is_monostate_v<T>) {
        return std::tuple{};
    } else {
        return std::forward_as_tuple(std::forward<T>(a));
    }
}

template <class F, class... Args>
auto invoke_remove_monostate_arguments(F&& f, Args&&... args) {
    return std::apply(
        [&](auto&&... bargs) {
            return std::forward<F>(f)(std::forward<decltype(bargs)>(bargs)...);
        },
        std::apply(
            [&](auto&&... bargs) {
                return std::tuple_cat(monostate_to_empty_tuple(std::forward<Args>(bargs))...);
            },
            std::forward_as_tuple(std::forward<Args>(args)...)));
}
 
/**************************************************************************************************/

enum class future_error_codes : std::uint8_t {
    broken_promise = 1, 
    no_state            
};
 
/**************************************************************************************************/
 
namespace detail {
 
inline auto Future_error_map(future_error_codes code) noexcept -> const
    char* {         // convert to name of future error
    switch (code) { // switch on error code value
        case future_error_codes::broken_promise:
            return "broken promise";
 
        case future_error_codes::no_state:
            return "no state";
 
        default:
            return nullptr;
    }
}
 
/**************************************************************************************************/
 
// This could be lifted into a common header if needed in other places
#if STLAB_CPP_VERSION_AT_LEAST(17)
template <class F, class... Args>
using result_t = std::invoke_result_t<F, Args...>;
#else
template <class F, class... Args>
using result_t = std::result_of_t<F(Args...)>;
#endif
 
/**************************************************************************************************/
 
} // namespace detail
 
/**************************************************************************************************/

class future_error : public std::logic_error {
public:
    explicit future_error(future_error_codes code) : logic_error(""), _code(code) {}

    [[nodiscard]] auto code() const noexcept -> const future_error_codes& { return _code; }

    [[nodiscard]] auto what() const noexcept -> const char* override {
        return detail::Future_error_map(_code);
    }
 
private:
    const future_error_codes _code; // the stored error code
};
 
/**************************************************************************************************/
 
namespace detail {
 
/**************************************************************************************************/
 
template <class>
struct result_of_;
 
template <class R, class... Args>
struct result_of_<R(Args...)> {
    using type = R;
};
 
template <class F>
using result_of_t_ = typename result_of_<F>::type;
 
template <class F, class T>
struct result_of_when_all_t;
 
template <class F>
struct result_of_when_all_t<F, void> {
    using result_type = detail::result_t<F>;
};
 
template <class F, class T>
struct result_of_when_all_t {
    using result_type = detail::result_t<F, const std::vector<T>&>;
};
 
template <class F, class T>
struct result_of_when_any_t;
 
template <class F>
struct result_of_when_any_t<F, void> {
    using result_type = detail::result_t<F, size_t>;
};
 
template <class F, class R>
struct result_of_when_any_t {
    using result_type = detail::result_t<F, R, size_t>;
};
 
template <class T>
auto unique_usage(const std::shared_ptr<T>& p) -> bool {
    return p.use_count() == 1;
}
 
/**************************************************************************************************/
 
} // namespace detail
 
/**************************************************************************************************/
 
template <class...>
class packaged_task;

template <class, class = void>
class future;
 
/**************************************************************************************************/
 
namespace detail {
 
/**************************************************************************************************/
 
template <class>
struct packaged_task_from_signature;
 
template <class R, class... Args>
struct packaged_task_from_signature<R(Args...)> {
    using type = packaged_task<Args...>;
};
 
template <class T>
using packaged_task_from_signature_t = typename packaged_task_from_signature<T>::type;
 
/**************************************************************************************************/
 
template <class>
struct reduced_signature;
 
template <class R, class... Args>
struct reduced_signature<R(Args...)> {
    using type = R(Args...);
};
 
template <class R, class... Args>
struct reduced_signature<future<R>(Args...)> {
    using type = R(Args...);
};
 
template <class T>
using reduced_signature_t = typename reduced_signature<T>::type;
 
/**************************************************************************************************/
 
template <class T>
inline constexpr bool is_future_v = false;
 
template <class T>
inline constexpr bool is_future_v<future<T>> = true;
 
template <class T>
using reduced_t = std::conditional_t<is_future_v<T>, T, future<T>>;
 
template <class Sig>
using reduced_result_t = reduced_t<result_of_t_<Sig>>;
 
/**************************************************************************************************/
 
template <class T, class = void>
struct value_;
 
} // namespace detail
 
/**************************************************************************************************/

template <class Sig, class E, class F>
auto package(E&&, F&&)
    -> std::pair<detail::packaged_task_from_signature_t<Sig>, detail::reduced_result_t<Sig>>;
 
/**************************************************************************************************/
 
namespace detail {
 
/**************************************************************************************************/
 
template <class, class>
struct shared;
template <class, class = void>
struct shared_base;
 
/**************************************************************************************************/
 
template <class... Args>
struct shared_task {
    void* _co_handle{nullptr}; // storing as void* for ABI stability.
 
    virtual ~shared_task() {
#if STLAB_STD_COROUTINES()
        if (_co_handle) std::coroutine_handle<>::from_address(_co_handle).destroy();
#endif
    }
 
    virtual void operator()(Args...) = 0;
    virtual void set_exception(const std::exception_ptr&) noexcept = 0;
};
 
/**************************************************************************************************/
 
template <class T>
struct shared_base<T, enable_if_copyable<void_to_monostate_t<T>>>
    : std::enable_shared_from_this<shared_base<T>> {
    using then_t = std::vector<std::pair<executor_t, task<void() noexcept>>>;
 
    using type = void_to_monostate_t<T>;
 
    executor_t _executor;
    std::optional<type> _result;
    std::exception_ptr _exception;
    std::mutex _mutex;
    std::atomic_bool _ready{false};
    then_t _then;
 
    explicit shared_base(executor_t s) : _executor(std::move(s)) {}
 
    template <class F>
    auto recover(future<T>&& p, F&& f) {
        return recover(std::move(p), _executor, std::forward<F>(f));
    }
 
    template <class E, class F>
    auto recover(future<T>&& p, E executor, F&& f) {
        using result_type = detail::result_t<F, future<T>>;
 
        auto [pro, fut] = package<result_type()>(
            executor, [_f = std::forward<F>(f), _p = std::move(p)]() mutable {
                return std::move(_f)(std::move(_p));
            });
 
        bool ready{false};
        {
            std::unique_lock<std::mutex> lock(_mutex);
            ready = _ready;
            if (!ready) _then.emplace_back(std::move(executor), std::move(pro));
        }
 
        if (ready) executor(std::move(pro)); // cannot reference this after here
 
        return std::move(fut);
    }
 
    template <class F>
    void _detach(future<T>&& p, F&& f) {
        auto pro = [_f = std::forward<F>(f), _p = std::move(p)]() mutable noexcept {
            std::move(_f)(std::move(_p));
        };
 
        bool ready{false};
        {
            std::unique_lock<std::mutex> lock(_mutex);
            ready = _ready;
            if (!ready) _then.emplace_back(immediate_executor, std::move(pro));
        }
 
        if (ready) std::move(pro)(); // cannot reference this after here
    }
 
    void _detach(future<T>&& p) noexcept {
        std::unique_lock<std::mutex> lock(_mutex);
        if (!_ready) _then.emplace_back([](auto&&) {}, [_p = std::move(p)]() noexcept {});
    }
 
    template <class F>
    void _on_completion(F&& f) {
        task<void() noexcept> t(std::forward<F>(f));
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (!_ready) {
                _then.emplace_back(immediate_executor, std::move(t));
                return;
            }
        }
        std::move(t)();
    }
 
    template <class E, class F>
    void _on_completion(E executor, F&& f) {
        task<void() noexcept> t(std::forward<F>(f));
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (!_ready) {
                _then.emplace_back(std::move(executor), std::move(t));
                return;
            }
        }
        executor(std::move(t));
    }
 
    void _set_exception(const std::exception_ptr& error) noexcept {
        _exception = error;
        then_t then;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            then = std::move(_then);
            _ready = true;
        }
        // propagate exception with scheduling
        for (auto& e : then) {
            e.first(std::move(e.second));
        }
    }
 
    template <class A>
    void set_value(A&& a);
 
    auto is_ready() const& -> bool { return _ready; }
 
    // get_ready() is called internally on continuations when we know _ready is true;
    auto get_ready() -> const type& {
        assert(is_ready() && "FATAL (sean.parent) : get_ready() called but not ready!");
 
        if (_exception) std::rethrow_exception(_exception);
        return *_result;
    }
 
    auto get_ready_r(bool unique) -> type {
        if (!unique) return get_ready();
 
        assert(is_ready() && "FATAL (sean.parent) : get_ready() called but not ready!");
 
        if (_exception) std::rethrow_exception(_exception);
        return std::move(*_result);
    }
 
    auto get_try() -> std::optional<type> {
        bool ready = false;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            ready = _ready;
        }
        if (ready) {
            if (_exception) std::rethrow_exception(_exception);
            return _result;
        }
        return std::nullopt;
    }
 
    auto get_try_r(bool unique) -> std::optional<type> {
        if (!unique) return get_try();
 
        bool ready = false;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            ready = _ready;
        }
        if (ready) {
            if (_exception) std::rethrow_exception(_exception);
            return std::move(_result);
        }
        return std::nullopt;
    }
};
 
/**************************************************************************************************/
 
template <class T>
struct shared_base<T, enable_if_not_copyable<void_to_monostate_t<T>>>
    : std::enable_shared_from_this<shared_base<T>> {
    using then_t = std::pair<executor_t, task<void() noexcept>>;
 
    executor_t _executor;
    std::optional<T> _result;
    std::exception_ptr _exception;
    std::mutex _mutex;
    std::atomic_bool _ready{false};
    then_t _then;
 
    explicit shared_base(executor_t s) : _executor(std::move(s)) {}
 
    template <class F>
    auto recover(future<T>&& p, F&& f) {
        return recover(std::move(p), _executor, std::forward<F>(f));
    }
 
    /*
        NOTE : executor cannot be a reference type here. When invoked it could
        cause _this_ to be deleted, and the executor passed in may be
        this->_executor
    */
    template <class E, class F>
    auto recover(future<T>&& p, E executor, F&& f) {
        using result_type = detail::result_t<F, future<T>>;
 
        auto [pro, fut] = package<result_type()>(
            executor, [_f = std::forward<F>(f), _p = std::move(p)]() mutable {
                return std::move(_f)(std::move(_p));
            });
 
        bool ready{false};
        {
            std::unique_lock<std::mutex> lock(_mutex);
            ready = _ready;
            if (!ready) {
                assert(!_then.second && "recover: _then slot already occupied");
                _then = {std::move(executor), std::move(pro)};
            }
        }
        if (ready) executor(std::move(pro)); // cannot reference this after here
 
        return std::move(fut);
    }
 
    template <class F>
    void _detach(future<T>&& p, F&& f) {
        auto pro = [_f = std::forward<F>(f), _p = std::move(p)]() mutable noexcept {
            std::move(_f)(std::move(_p));
        };
 
        bool ready{false};
        {
            std::unique_lock<std::mutex> lock(_mutex);
            ready = _ready;
            if (!ready) _then = {immediate_executor, std::move(pro)};
        }
 
        if (ready) std::move(pro)(); // cannot reference this after here
    }
 
    void _detach(future<T>&& p) noexcept {
        std::unique_lock<std::mutex> lock(_mutex);
        if (!_ready) _then = then_t([](auto&&) {}, [_p = std::move(p)]() noexcept {});
    }
 
    template <class F>
    void _on_completion(F&& f) {
        task<void() noexcept> t(std::forward<F>(f));
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (!_ready) {
                assert(!_then.second && "on_completion: _then slot already occupied");
                _then = {immediate_executor, std::move(t)};
                return;
            }
        }
        std::move(t)();
    }
 
    template <class E, class F>
    void _on_completion(E executor, F&& f) {
        task<void() noexcept> t(std::forward<F>(f));
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (!_ready) {
                assert(!_then.second && "on_completion: _then slot already occupied");
                _then = {std::move(executor), std::move(t)};
                return;
            }
        }
        executor(std::move(t));
    }
 
    void _set_exception(const std::exception_ptr& error) noexcept {
        _exception = error;
        then_t then;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_then.second) then = std::move(_then);
            _ready = true;
        }
        // propagate exception with scheduling
        if (then.second) then.first(std::move(then.second));
    }
 
    template <class A>
    void set_value(A&& a);
 
    auto is_ready() const -> bool { return _ready; }

    auto get_ready() -> T {
        assert(is_ready() && "FATAL (sean.parent) : get_ready() called but not ready!");
 
        if (_exception) std::rethrow_exception(_exception);
        return std::move(*_result);
    }

    auto get_try() -> std::optional<T> {
        bool ready = false;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            ready = _ready;
        }
        if (ready) {
            if (_exception) std::rethrow_exception(_exception);
            return std::move(_result);
        }
        return std::nullopt;
    }
};
 
/**************************************************************************************************/

template <class R>
class promise {
    using type = void_to_monostate_t<R>;
    std::weak_ptr<shared_base<R>> _p;
 
public:
    explicit promise(const std::shared_ptr<shared_base<R>>& p) noexcept : _p{p} {}
 
    ~promise() {
        if (auto p = _p.lock()) {
            p->_set_exception(
                std::make_exception_ptr(future_error(future_error_codes::broken_promise)));
        }
    }
 
    promise(promise&&) noexcept = default;
    auto operator=(promise&& x) noexcept -> promise& {
        if (this == &x) return *this;
        if (auto p = _p.lock()) {
            p->_set_exception(
                std::make_exception_ptr(future_error(future_error_codes::broken_promise)));
        }
        _p = std::move(x._p);
        return *this;
    }
 
    promise(const promise&) = delete;
    auto operator=(const promise&) -> promise& = delete;

    void set_value(type&& value) && noexcept {
        if (auto p = _p.lock()) {
            _p.reset();
            p->set_value(std::move(value));
        }
    }

    auto set_value() && noexcept { set_value(std::monostate{}); }

    void set_exception(const std::exception_ptr& error) && noexcept {
        if (auto p = _p.lock()) {
            _p.reset();
            p->_set_exception(error);
        }
    }

    [[nodiscard]] auto canceled() const -> bool { return _p.expired(); }
};

template <class R>
promise(std::shared_ptr<shared_base<R>>) -> promise<R>;
 
template <class F, class R, class... Args>
struct shared<F, R(Args...)> final : shared_base<R>, shared_task<Args...> {
    std::optional<F> _f;
 
    shared(executor_t s, F&& f) : shared_base<R>(std::move(s)), _f(std::move(f)) {}
 
    void operator()(Args... args) noexcept override {
        static_assert(
            std::is_nothrow_invocable_v<F, decltype(promise{this->shared_from_this()}), Args...>);
        assert(!this->weak_from_this().expired());
        std::move (*_f)(promise{this->shared_from_this()}, std::move(args)...);
        // After invoking `_f`, it is not destructed because it could be satisfying the promise
        // asynchronously. `_f` is responsible for any cleanup prior to the future being released
        // and must be aware that it may be destructed at any time after this point.
    }
 
    void set_exception(const std::exception_ptr& error) noexcept override {
        shared_base<R>::_set_exception(error);
        _f.reset(); // _f will not be invoked, release it and any resources it holds
    }
};
 
template <class Sig, class F>
auto make_shared_state(executor_t s, F&& f) -> std::shared_ptr<shared<F, Sig>> {
    return std::make_shared<shared<F, Sig>>(std::move(s), std::forward<F>(f));
}
 
template <class... Args>
[[nodiscard]] auto weak_state(const packaged_task<Args...>& p)
    -> std::weak_ptr<shared_task<Args...>>;
 
/**************************************************************************************************/
 
} // namespace detail
 
/**************************************************************************************************/

template <class... Args>
class packaged_task {
    using ptr_t = std::weak_ptr<detail::shared_task<Args...>>;
    ptr_t _p;
 
    explicit packaged_task(ptr_t&& p) : _p(std::move(p)) {}
 
    template <class Sig, class E, class F>
    friend auto package(E&&, F&&)
        -> std::pair<detail::packaged_task_from_signature_t<Sig>, detail::reduced_result_t<Sig>>;
 
    template <class T, class E>
    friend auto future_with_broken_promise(E&&) -> detail::reduced_t<T>;
 
    template <class... Brgs>
    friend auto detail::weak_state(const packaged_task<Brgs...>& p)
        -> std::weak_ptr<detail::shared_task<Brgs...>>;
 
public:
    packaged_task() = default;
    ~packaged_task() {
        if (auto p = _p.lock()) {
            p->set_exception(
                std::make_exception_ptr(future_error(future_error_codes::broken_promise)));
        }
    }
 
    packaged_task(const packaged_task&) = delete;
    auto operator=(const packaged_task&) -> packaged_task& = delete;
 
    packaged_task(packaged_task&&) noexcept = default;
    auto operator=(packaged_task&& x) noexcept -> packaged_task& {
        if (this == &x) return *this;
        if (auto p = _p.lock()) {
            p->set_exception(
                std::make_exception_ptr(future_error(future_error_codes::broken_promise)));
        }
        _p = std::move(x._p);
        return *this;
    }

    template <class... A>
    void operator()(A&&... args) noexcept {
        if (auto p = _p.lock()) {
            p->_co_handle = nullptr;
            _p.reset();
            (*p)(std::forward<A>(args)...);
        }
    }

    [[nodiscard]] auto canceled() const -> bool { return _p.expired(); }

    void set_exception(const std::exception_ptr& error) noexcept {
        if (auto p = _p.lock()) {
            p->_co_handle = nullptr;
            _p.reset();
            p->set_exception(error);
        }
    }
};
 
namespace detail {
 
template <class... Args>
[[nodiscard]] auto weak_state(const packaged_task<Args...>& p)
    -> std::weak_ptr<detail::shared_task<Args...>> {
    return p._p;
}
 
} // namespace detail
 
/**************************************************************************************************/

template <class T>
class STLAB_NODISCARD() future<T, enable_if_copyable<void_to_monostate_t<T>>> {
    using type = void_to_monostate_t<T>;
    using ptr_t = std::shared_ptr<detail::shared_base<T>>;
    ptr_t _p;
 
    explicit future(ptr_t p) : _p(std::move(p)) {}
 
    template <class Sig, class E, class F>
    friend auto package(E&&, F&&)
        -> std::pair<detail::packaged_task_from_signature_t<Sig>, detail::reduced_result_t<Sig>>;
 
    template <class U, class E>
    friend auto future_with_broken_promise(E&&) -> detail::reduced_t<U>;

    friend struct detail::shared_base<type>;
    template <class, class>
    friend struct detail::value_;
 
public:
    using result_type = T;
 
    future() = default;
    future(const future&) = default;
    future(future&&) noexcept = default;
    auto operator=(const future&) -> future& = default;
    auto operator=(future&&) noexcept -> future& = default;

    void swap(future& x) noexcept { std::swap(_p, x._p); }

    inline friend void swap(future& x, future& y) noexcept { x.swap(y); }


    inline friend auto operator==(const future& x, const future& y) -> bool { return x._p == y._p; }
    inline friend auto operator!=(const future& x, const future& y) -> bool { return !(x == y); }

    [[nodiscard]] auto valid() const -> bool { return static_cast<bool>(_p); }

    template <class F>
    auto then(F&& f) const& {
        return recover([_f = std::forward<F>(f)](future<result_type>&& p) mutable {
            return invoke_remove_monostate_arguments(
                std::move(_f),
                invoke_void_to_monostate_result([&] { return std::move(p).get_ready(); }));
        });
    }


    template <class F>
    auto operator|(F&& f) const& {
        return then(std::forward<F>(f));
    }

    template <class E, class F>
    auto then(E&& executor, F&& f) const& {
        return recover(
            std::forward<E>(executor), [_f = std::forward<F>(f)](future<result_type>&& p) mutable {
                return invoke_remove_monostate_arguments(
                    std::move(_f),
                    invoke_void_to_monostate_result([&] { return std::move(p).get_ready(); }));
            });
    }

    template <class F>
    auto operator|(executor_task_pair<F> etp) const& {
        return then(std::move(etp)._executor, std::move(etp)._f);
    }

    template <class F>
    auto then(F&& f) && {
        return std::move(*this).recover([_f = std::forward<F>(f)](future<result_type>&& p) mutable {
            return invoke_remove_monostate_arguments(
                std::move(_f),
                invoke_void_to_monostate_result([&] { return std::move(p).get_ready(); }));
        });
    }

    template <class F>
    auto operator|(F&& f) && {
        return std::move(*this).then(std::forward<F>(f));
    }

    template <class E, class F>
    auto then(E&& executor, F&& f) && {
        return std::move(*this).recover(
            std::forward<E>(executor), [_f = std::forward<F>(f)](future<result_type>&& p) mutable {
                return invoke_remove_monostate_arguments(
                    std::move(_f),
                    invoke_void_to_monostate_result([&] { return std::move(p).get_ready(); }));
            });
    }

    template <class F>
    auto operator|(executor_task_pair<F> etp) && {
        return std::move(*this).then(std::move(etp)._executor, std::move(etp)._f);
    }

    template <class F>
    auto recover(F&& f) const& {
        return _p->recover(copy(*this), std::forward<F>(f));
    }

    template <class F>
    auto operator^(F&& f) const& {
        return recover(std::forward<F>(f));
    }

    template <class E, class F>
    auto recover(E&& executor, F&& f) const& {
        return _p->recover(copy(*this), std::forward<E>(executor), std::forward<F>(f));
    }

    template <class F>
    auto operator^(executor_task_pair<F> etp) const& {
        return recover(std::move(etp)._executor, std::move(etp)._f);
    }

    template <class F>
    auto recover(F&& f) && {
        auto& self = *_p.get();
        return self.recover(std::move(*this), std::forward<F>(f));
    }

    template <class F>
    auto operator^(F&& f) && {
        return std::move(*this).recover(std::forward<F>(f));
    }

    template <class E, class F>
    auto recover(E&& executor, F&& f) && {
        auto& self = *_p.get();
        return self.recover(std::move(*this), std::forward<E>(executor), std::forward<F>(f));
    }

    template <class F>
    auto operator^(executor_task_pair<F> etp) && {
        return std::move(*this).recover(std::move(etp)._executor, std::move(etp)._f);
    }

    void detach() && {
        auto& self = *_p.get();
        self._detach(std::move(*this));
    }

    template <class F>
    void detach(F&& f) && {
        auto& self = *_p.get();
        self._detach(std::move(*this), std::forward<F>(f));
    }

    template <class F>
    void on_completion(F&& f) {
        _p->_on_completion(std::forward<F>(f));
    }

    template <class E, class F>
    void on_completion(E&& executor, F&& f) {
        _p->_on_completion(std::forward<E>(executor), std::forward<F>(f));
    }

    void reset() noexcept { _p.reset(); }

    [[nodiscard]] auto is_ready() const& -> bool { return _p && _p->is_ready(); }

    [[nodiscard]] auto get_try() const& { return optional_monostate_to_bool(_p->get_try()); }

    auto get_try() && { return optional_monostate_to_bool(_p->get_try_r(unique_usage(_p))); }

    [[nodiscard]] auto get_ready() const& { return monostate_to_void(_p->get_ready()); }

    auto get_ready() && { return monostate_to_void(_p->get_ready_r(unique_usage(_p))); }

    [[nodiscard]] auto exception() const& -> std::exception_ptr {
        assert(is_ready());
        return _p->_exception;
    }
};
 
/**************************************************************************************************/

template <class T>
class STLAB_NODISCARD() future<T, enable_if_not_copyable<void_to_monostate_t<T>>> {
    using ptr_t = std::shared_ptr<detail::shared_base<T>>;
    ptr_t _p;
 
    explicit future(ptr_t p) : _p(std::move(p)) {}
 
    template <class Sig, class E, class F>
    friend auto package(E&&, F&&)
        -> std::pair<detail::packaged_task_from_signature_t<Sig>, detail::reduced_result_t<Sig>>;
 
    template <class U, class E>
    friend auto future_with_broken_promise(E&&) -> detail::reduced_t<U>;

    friend struct detail::shared_base<T>;
    template <class, class>
    friend struct detail::value_;
 
public:
    using result_type = T;
 
    future() = default;
    future(const future&) = delete;
    future(future&&) noexcept = default;
    auto operator=(const future&) -> future& = delete;
    auto operator=(future&&) noexcept -> future& = default;

    void swap(future& x) noexcept { std::swap(_p, x._p); }

    inline friend void swap(future& x, future& y) noexcept { x.swap(y); }


    inline friend auto operator==(const future& x, const future& y) -> bool { return x._p == y._p; }
    inline friend auto operator!=(const future& x, const future& y) -> bool { return !(x == y); }

    [[nodiscard]] auto valid() const -> bool { return static_cast<bool>(_p); }

    template <class F>
    auto then(F&& f) && {
        return std::move(*this).recover([_f = std::forward<F>(f)](future<result_type>&& p) mutable {
            return std::move(_f)(std::move(p).get_ready());
        });
    }


    template <class F>
    auto operator|(F&& f) && {
        return std::move(*this).then(std::forward<F>(f));
    }

    template <class E, class F>
    auto then(E&& executor, F&& f) && {
        return std::move(*this).recover(std::forward<E>(executor),
                                        [_f = std::forward<F>(f)](future<result_type>&& p) mutable {
                                            return std::move(_f)(std::move(p).get_ready());
                                        });
    }

    template <class F>
    auto operator|(executor_task_pair<F> etp) && {
        return std::move(*this).then(std::move(etp)._executor, std::move(etp)._f);
    }

    template <class F>
    auto recover(F&& f) && {
        auto& self = *_p.get();
        return self.recover(std::move(*this), std::forward<F>(f));
    }

    template <class F>
    auto operator^(F&& f) && {
        return std::move(*this).recover(std::forward<F>(f));
    }

    template <class E, class F>
    auto recover(E&& executor, F&& f) && {
        auto& self = *_p.get();
        return self.recover(std::move(*this), std::forward<E>(executor), std::forward<F>(f));
    }

    template <class F>
    auto operator^(executor_task_pair<F> etp) && {
        return std::move(*this).recover(std::move(etp)._executor, std::move(etp)._f);
    }

    void detach() && {
        auto& self = *_p.get();
        self._detach(std::move(*this));
    }

    template <class F>
    void detach(F&& f) && {
        auto& self = *_p.get();
        self._detach(std::move(*this), std::forward<F>(f));
    }

    template <class F>
    void on_completion(F&& f) {
        _p->_on_completion(std::forward<F>(f));
    }

    template <class E, class F>
    void on_completion(E&& executor, F&& f) {
        _p->_on_completion(std::forward<E>(executor), std::forward<F>(f));
    }

    void reset() noexcept { _p.reset(); }

    [[nodiscard]] auto is_ready() const& -> bool { return _p && _p->is_ready(); }

    [[nodiscard]] auto get_try() & -> std::optional<T> { return _p->get_try(); }

    [[nodiscard]] auto get_ready() && -> T { return _p->get_ready(); }

    [[nodiscard]] auto exception() const& -> std::exception_ptr {
        assert(is_ready());
        return _p->_exception;
    }
};

template <class Sig, class E, class F>
auto package(E&& executor, F&& f)
    -> std::pair<detail::packaged_task_from_signature_t<Sig>, detail::reduced_result_t<Sig>> {
    if constexpr (std::is_same_v<std::decay_t<E>, executor_t>) {
        assert(executor && "FATAL (sean.parent) : executor is null!");
    }
 
    using result_t = detail::result_of_t_<Sig>;
 
    if constexpr (detail::is_future_v<result_t>) {
        auto p = detail::make_shared_state<detail::reduced_signature_t<Sig>>(
            std::forward<E>(executor),
            [_f = std::make_optional(std::forward<F>(f)),
             _hold = future<void>{}](auto&& promise, auto&&... args) mutable noexcept {
                assert(_f && "packaged task invoked twice");
                try {
                    auto r = std::move(*_f)(std::forward<decltype(args)>(args)...);
                    try {
                        _hold = std::move(r).recover(
                            immediate_executor,
                            [_p = std::move(promise)](result_t&& f) mutable noexcept {
                                if (auto e = f.exception()) {
                                    std::move(_p).set_exception(std::move(e));
                                } else {
                                    std::move(_p).set_value(invoke_void_to_monostate_result(
                                        [&] { return std::move(f).get_ready(); }));
                                }
                            });
                    } catch (...) {
                        /* NOTE: an exception here is reported as a broken promise. Ideally recover
                         * would be passed the initial promise (it flows through the chain), but the
                         * API isn't there yet. */
                    }
                } catch (...) {
                    std::move(promise).set_exception(std::current_exception());
                }
                _f.reset();
            });
        return {detail::packaged_task_from_signature_t<Sig>{p}, result_t{p}};
    } else {
        auto p = detail::make_shared_state<Sig>(
            std::forward<E>(executor), [_f = std::make_optional(std::forward<F>(f))](
                                           auto&& promise, auto&&... args) mutable noexcept {
                assert(_f && "packaged task invoked twice");
                try {
                    auto tmp = invoke_void_to_monostate_result(
                        std::move(*_f), std::forward<decltype(args)>(args)...);
                    std::move(promise).set_value(std::move(tmp)); // noexcept
                } catch (...) {
                    std::move(promise).set_exception(std::current_exception());
                }
                _f.reset();
            });
        return {detail::packaged_task_from_signature_t<Sig>{p}, future<result_t>{p}};
    }
}

template <class T, class E>
auto future_with_broken_promise(E&& executor) -> detail::reduced_t<T> {
    auto p = std::make_shared<detail::shared_base<typename detail::reduced_t<T>::result_type>>(
        std::forward<E>(executor));
    p->_exception = std::make_exception_ptr(future_error(future_error_codes::broken_promise));
    p->_ready = true;
 
    return detail::reduced_t<T>{p};
}
 
/**************************************************************************************************/
 
namespace detail {
 
template <class F>
struct assign_ready_future {
    template <class T>
    static void assign(T& x, F&& f) {
        x = std::move(f).get_ready();
    }
};
 
template <>
struct assign_ready_future<future<void>> {
    template <class T>
    static void assign(T& x, const future<void>&) {
        x = std::move(typename T::value_type()); // to set the optional
    }
};

template <class... Ts, class F, std::size_t... I>
void for_each_when_input_(std::tuple<Ts...>& inputs, F&& f, std::index_sequence<I...>) {
    (f(std::get<I>(inputs), std::integral_constant<std::size_t, I>{}), ...);
}

template <class... Ts, class F>
void for_each_when_input(std::tuple<Ts...>& inputs, F&& f) {
    for_each_when_input_(inputs, std::forward<F>(f), std::index_sequence_for<Ts...>{});
}

template <class T, class F>
void for_each_when_input(std::vector<future<T>>& inputs, F&& f) {
    std::size_t index = 0;
    // Visit each owned future without copying its shared state.
    for (auto& input : inputs)
        f(input, index++);
}

template <class Inputs, bool Any>
struct when_shared {
    Inputs _inputs;
    executor_t _executor;
    packaged_task<> _f;
    std::mutex _guard;
    std::size_t _remaining;
    std::size_t _index{0};
    std::exception_ptr _exception;
    bool _registering{true};
    bool _decided{false};

    when_shared(Inputs inputs, executor_t executor, std::size_t count) :
        _inputs(std::move(inputs)), _executor(std::move(executor)), _remaining(count) {}

    void dispatch(packaged_task<> ready) {
        if (ready.canceled()) return;
        for_each_when_input(_inputs, [&](auto& input, std::size_t index) {
            if (_exception || (Any && index != _index)) input.reset();
        });
        _executor(std::move(ready));
    }

    template <class T>
    void done(const future<T>& input, std::size_t index) {
        packaged_task<> ready;
        {
            std::unique_lock<std::mutex> lock{_guard};
            if (_decided) return;
            auto error = input.exception();
            if constexpr (Any) {
                if (!error || --_remaining == 0) {
                    _index = index;
                    _exception = std::move(error);
                    _decided = true;
                }
            } else {
                if (error || --_remaining == 0) {
                    _exception = std::move(error);
                    _decided = true;
                }
            }
            if (_decided && !_registering) ready = std::move(_f);
        }
        dispatch(std::move(ready));
    }

    void finish_registration() {
        packaged_task<> ready;
        {
            std::unique_lock<std::mutex> lock{_guard};
            _registering = false;
            if (_decided) ready = std::move(_f);
        }
        dispatch(std::move(ready));
    }
};

template <class P>
void attach_when_inputs(const std::shared_ptr<P>& shared) {
    for_each_when_input(shared->_inputs, [&](auto& input, std::size_t index) {
        input.on_completion([weak = std::weak_ptr<P>(shared), source = &input, index]() noexcept {
            if (auto p = weak.lock()) p->done(*source, index);
        });
    });
    shared->finish_registration();
}

template <class F, class... Ts>
auto apply_when_all_args(F& f, std::tuple<future<Ts>...>& inputs) {
    using args_t = optional_placeholder_tuple<Ts...>;
    args_t args;
    for_each_when_input(inputs, [&](auto& input, auto index) {
        assign_ready_future<std::decay_t<decltype(input)>>::assign(
            std::get<decltype(index)::value>(args), std::move(input));
    });
    return apply_optional_indexed<index_sequence_transform_t<
        std::index_sequence_for<Ts...>, remove_placeholder<args_t>::template function>>(f, args);
}

template <class T, class F, class P>
auto apply_when_any_arg(F& f, P& shared) {
    if constexpr (std::is_void_v<T>) {
        return f(shared->_index);
    } else {
        std::optional<T> arg;
        for_each_when_input(shared->_inputs, [&](auto& input, std::size_t index) {
            if (index == shared->_index) arg.emplace(std::move(input).get_ready());
        });
        return f(std::move(*arg), shared->_index);
    }
}
 
} // namespace detail
 
/**************************************************************************************************/

template <class E, class F, class... Ts>
auto when_all(const E& executor, F f, future<Ts>... args) {
    using vt_t = voidless_tuple<Ts...>;
    using result_t = decltype(apply_ignore_placeholders(std::declval<F>(), std::declval<vt_t>()));
 
    using inputs_t = std::tuple<future<Ts>...>;
    auto shared = std::make_shared<detail::when_shared<inputs_t, false>>(
        inputs_t{std::move(args)...}, executor, sizeof...(Ts));
    auto p = package<result_t()>(executor, [_f = std::move(f), _p = shared]() mutable {
        if (_p->_exception) std::rethrow_exception(_p->_exception);
        return detail::apply_when_all_args(_f, _p->_inputs);
    });
    shared->_f = std::move(p.first);
 
    detail::attach_when_inputs(shared);
 
    return std::move(p.second);
}
 
/**************************************************************************************************/

template <class T>
struct make_when_any {
    template <class E, class F, class... Ts>
    static auto make(const E& executor, F f, future<T> arg, future<Ts>... args) {
        using result_t = detail::result_t<F, T, size_t>;
 
        using inputs_t = std::tuple<future<T>, future<Ts>...>;
        auto shared = std::make_shared<detail::when_shared<inputs_t, true>>(
            inputs_t{std::move(arg), std::move(args)...}, executor, sizeof...(Ts) + 1);
        auto p = package<result_t()>(executor, [_f = std::move(f), _p = shared]() mutable {
            if (_p->_exception) std::rethrow_exception(_p->_exception);
            return detail::apply_when_any_arg<T>(_f, _p);
        });
        shared->_f = std::move(p.first);
 
        detail::attach_when_inputs(shared);
 
        return std::move(p.second);
    }
};
 
/**************************************************************************************************/

template <>
struct make_when_any<void> {
    template <class E, class F, class... Ts>
    static auto make(E&& executor, F&& f, future<Ts>... args) {
        using result_t = detail::result_t<F, size_t>;
 
        using inputs_t = std::tuple<future<Ts>...>;
        auto shared = std::make_shared<detail::when_shared<inputs_t, true>>(
            inputs_t{std::move(args)...}, executor, sizeof...(Ts));
        auto p = package<result_t()>(executor, [_f = std::forward<F>(f), _p = shared]() mutable {
            if (_p->_exception) std::rethrow_exception(_p->_exception);
            return detail::apply_when_any_arg<void>(_f, _p);
        });
        shared->_f = std::move(p.first);
 
        detail::attach_when_inputs(shared);
 
        return std::move(p.second);
    }
};
 
/**************************************************************************************************/

template <class E, class F, class T, class... Ts>
auto when_any(E&& executor, F&& f, future<T>&& arg, future<Ts>&&... args) {
    return make_when_any<T>::make(std::forward<E>(executor), std::forward<F>(f), std::move(arg),
                                  std::move(args)...);
}
 
/**************************************************************************************************/
 
namespace detail {
template <bool Any, class R, class E, class F, class I>
auto create_range_of_futures(const E& executor, F&& f, I first, I last) {
    using future_t = typename std::iterator_traits<I>::value_type;
    using value_t = typename future_t::result_type;
    std::vector<future_t> inputs;
    inputs.reserve(std::distance(first, last));
    std::transform(first, last, std::back_inserter(inputs), [](auto&& input) -> future_t {
        if constexpr (std::is_copy_constructible_v<future_t>)
            return std::forward<decltype(input)>(input);
        else
            return std::move(input);
    });
    auto count = inputs.size();
    auto shared =
        std::make_shared<when_shared<decltype(inputs), Any>>(std::move(inputs), executor, count);
    auto p = package<R()>(executor, [_f = std::forward<F>(f), _p = shared]() mutable -> R {
        if (_p->_exception) std::rethrow_exception(_p->_exception);
        if constexpr (Any) {
            if constexpr (std::is_void_v<value_t>)
                return _f(_p->_index);
            else
                return _f(std::move(_p->_inputs[_p->_index]).get_ready(), _p->_index);
        } else {
            if constexpr (std::is_void_v<value_t>) {
                return _f();
            } else {
                std::vector<value_t> values;
                values.reserve(_p->_inputs.size());
                std::transform(_p->_inputs.begin(), _p->_inputs.end(), std::back_inserter(values),
                               [](auto& input) { return std::move(input).get_ready(); });
                return _f(std::move(values));
            }
        }
    });
    shared->_f = std::move(p.first);
    attach_when_inputs(shared);
    return std::move(p.second);
}
 
/**************************************************************************************************/
 
} // namespace detail
 
/**************************************************************************************************/

template <class E, class F, class I>
auto when_all(const E& executor, F f, std::pair<I, I> range) {
    using param_t = typename std::iterator_traits<I>::value_type::result_type;
    using result_t = typename detail::result_of_when_all_t<F, param_t>::result_type;
    if (range.first == range.second) {
        auto p = package<result_t()>(executor, [_f = std::move(f)]() mutable -> result_t {
            if constexpr (std::is_void_v<param_t>)
                return _f();
            else
                return _f(std::vector<param_t>{});
        });
        executor(std::move(p.first));
        return std::move(p.second);
    }
 
    return detail::create_range_of_futures<false, result_t>(executor, std::move(f), range.first,
                                                            range.second);
}
 
/**************************************************************************************************/

template <class E, class F, class I>
auto when_any(const E& executor, F&& f, std::pair<I, I> range) {
    using param_t = typename std::iterator_traits<I>::value_type::result_type;
    using result_t = std::decay_t<typename detail::result_of_when_any_t<F, param_t>::result_type>;
    if (range.first == range.second) {
        return future_with_broken_promise<std::decay_t<result_t>>(executor);
    }
 
    return detail::create_range_of_futures<true, result_t>(std::move(executor), std::forward<F>(f),
                                                           range.first, range.second);
}
 
/**************************************************************************************************/

template <class E, class F, class... Args>
auto async(const E& executor, F&& f, Args&&... args)
    -> detail::reduced_t<detail::result_t<std::decay_t<F>, std::decay_t<Args>...>> {
    using result_type = detail::result_t<std::decay_t<F>, std::decay_t<Args>...>;
 
    auto [pro, fut] = package<result_type()>(
        executor,
        [f = std::forward<F>(f), args = std::make_tuple(std::forward<Args>(args)...)]() mutable
            -> result_type { return std::apply(std::move(f), std::move(args)); });
 
    executor(std::move(pro));
 
    return std::move(fut);
}
 
/**************************************************************************************************/
 
namespace detail {
 
/**************************************************************************************************/
 
template <class T>
struct value_<T, enable_if_copyable<void_to_monostate_t<T>>> {
    template <class C>
    static void proceed(C& sb) {
        typename C::then_t then;
        {
            std::unique_lock<std::mutex> lock(sb._mutex);
            sb._ready = true;
            then = std::move(sb._then);
        }
        for (auto& e : then)
            e.first(std::move(e.second));
    }
 
    template <class R, class A>
    static void set(shared_base<R>& sb, A&& a) {
        sb._result = std::forward<A>(a);
        proceed(sb);
    }
};
 
template <class T>
struct value_<T, enable_if_not_copyable<void_to_monostate_t<T>>> {
    template <class C>
    static void proceed(C& sb) {
        typename C::then_t then;
        {
            std::unique_lock<std::mutex> lock(sb._mutex);
            sb._ready = true;
            then = std::move(sb._then);
        }
        if (then.first) then.first(std::move(then.second));
    }
 
    template <class R, class A>
    static void set(shared_base<R>& sb, A&& a) {
        sb._result = std::forward<A>(a);
        proceed(sb);
    }
};
 
/**************************************************************************************************/
 
template <class T>
template <class A>
void shared_base<T, enable_if_copyable<void_to_monostate_t<T>>>::set_value(A&& a) {
    value_<T>::set(*this, std::forward<A>(a));
}
 
template <class T>
template <class A>
void shared_base<T, enable_if_not_copyable<void_to_monostate_t<T>>>::set_value(A&& a) {
    value_<T>::set(*this, std::forward<A>(a));
}
 
/**************************************************************************************************/
 
} // namespace detail
 
/**************************************************************************************************/

 
STLAB_VERSION_NAMESPACE_END()
} // namespace stlab
 
/**************************************************************************************************/
 
#if STLAB_STD_COROUTINES()
 
/**************************************************************************************************/
 
namespace stlab {
STLAB_VERSION_NAMESPACE_BEGIN()
 

 
namespace detail {
 
// --- Generic awaitable support for resume_on(executor, any_awaitable) ---
// C++ coroutine protocol: awaitable is (1) has member operator co_await, (2) has free
// operator co_await, or (3) is directly an awaiter (await_ready/await_suspend/await_resume).
template <class A, class = void>
struct has_member_co_await : std::false_type {};
template <class A>
struct has_member_co_await<A, std::void_t<decltype(std::declval<A>().operator co_await())>>
    : std::true_type {};
 
template <class A, class = void>
struct has_free_co_await : std::false_type {};
template <class A>
struct has_free_co_await<A, std::void_t<decltype(operator co_await(std::declval<A>()))>>
    : std::true_type {};
 
template <class A>
auto get_awaiter(A&& a) {
    if constexpr (has_member_co_await<A>::value) {
        return std::forward<A>(a).operator co_await();
    } else if constexpr (has_free_co_await<A>::value) {
        return operator co_await(std::forward<A>(a));
    } else {
        return std::forward<A>(a); // direct awaiter (e.g. std::suspend_always)
    }
}
 
template <class A>
using awaiter_t = decltype(get_awaiter(std::declval<A&>()));
 
template <class A>
using await_result_t = decltype(std::declval<awaiter_t<A>>().await_resume());
 
template <class A>
using await_result_storage_t = void_to_monostate_t<await_result_t<A>>;
 
// Fire-and-forget return type for the proxy coroutine. No handle is retained after launch;
// the proxy owns itself and self-destroys at final_suspend (suspend_never).
struct proxy_fire_and_forget {
    struct promise_type {
        proxy_fire_and_forget get_return_object() { return {}; }
        std::suspend_never initial_suspend() { return {}; }
        std::suspend_never final_suspend() noexcept { return {}; }
        void return_void() {}
        void unhandled_exception() { std::terminate(); }
    };
};

template <class A, class Inner, class E, class F>
proxy_fire_and_forget resume_on_proxy_coro(E executor,
                                           F resume_cb,
                                           Inner inner,
                                           std::optional<await_result_storage_t<A>>* result_ptr,
                                           std::exception_ptr* exception_ptr) {
    try {
        if constexpr (std::is_void_v<await_result_t<A>>) {
            co_await std::move(inner);
            result_ptr->emplace(std::monostate{});
        } else {
            result_ptr->emplace(co_await std::move(inner));
        }
    } catch (...) {
        if (exception_ptr) *exception_ptr = std::current_exception();
    }
    executor(std::move(resume_cb));
}

template <class A, class Inner, class E, class F, class WeakPtr>
proxy_fire_and_forget resume_on_proxy_coro_controlled(
    E executor,
    F resume_cb,
    Inner inner,
    WeakPtr weak,
    std::optional<await_result_storage_t<A>>* result_ptr,
    std::exception_ptr* exception_ptr) {
    // proceed with co_await; if weak_state is destroyed during suspension, the proxy is
    // responsible for not writing to awaiter storage and not invoking the resume callback.
    bool was_alive = false;
    try {
        if constexpr (std::is_void_v<await_result_t<A>>) {
            co_await std::move(inner);
            if (auto keepalive = weak.lock()) {
                (void)keepalive;
                result_ptr->emplace(std::monostate{});
                was_alive = true;
            }
        } else {
            auto tmp = co_await std::move(inner);
            if (auto keepalive = weak.lock()) {
                (void)keepalive;
                result_ptr->emplace(std::move(tmp));
                was_alive = true;
            }
        }
    } catch (...) {
        if (auto keepalive = weak.lock()) {
            (void)keepalive;
            if (exception_ptr) *exception_ptr = std::current_exception();
            was_alive = true;
        }
    }
    if (was_alive) executor(std::move(resume_cb));
}
 
template <class... Args>
struct final_awaiter {
    bool await_ready() noexcept { return false; }
    void await_resume() noexcept {}
    void await_suspend(std::coroutine_handle<> h) noexcept { h.destroy(); }
};

template <class R, class E>
struct resume_on_awaiter {
    E _executor;
    future<R> _input;
 
    bool await_ready() const noexcept { return false; }
 
    auto await_resume() { return std::move(_input).get_ready(); }
 
    void await_suspend(std::coroutine_handle<> ch) {
        _input.on_completion(std::move(_executor), [ch]() noexcept { ch.resume(); });
    }
};
 
template <class R, class E>
resume_on_awaiter(E, future<R>) -> resume_on_awaiter<R, std::decay_t<E>>;

template <class R, class E, class WeakPtr, bool AllowSkipSuspend>
struct resume_on_awaiter_with_control {
    E _executor;
    future<R> _input;
    WeakPtr _weak;
 
    bool await_ready() const noexcept {
        if constexpr (AllowSkipSuspend) {
            return _input.is_ready();
        } else {
            return false;
        }
    }
    auto await_resume() { return std::move(_input).get_ready(); }
    void await_suspend(std::coroutine_handle<> ch) {
        assert(_weak.lock() && "await_suspend: weak_state is gone");
        _weak.lock()->_co_handle = ch.address();
        _input.on_completion(std::move(_executor), [weak = std::move(_weak)]() noexcept {
            if (auto state = weak.lock())
                std::coroutine_handle<>::from_address(state->_co_handle).resume();
        });
    }
};

template <class E>
struct resume_on_executor_awaiter {
    E _executor;
 
    bool await_ready() const noexcept { return false; }
    void await_resume() noexcept {}
    void await_suspend(std::coroutine_handle<> ch) {
        std::move(_executor)([ch]() noexcept { ch.resume(); });
    }
};

template <class E, class WeakPtr>
struct resume_on_executor_awaiter_with_control {
    E _executor;
    WeakPtr _weak;
 
    [[nodiscard]] bool await_ready() const noexcept { return false; }
    void await_resume() noexcept {}
    void await_suspend(std::coroutine_handle<> ch) {
        assert(_weak.lock() && "await_suspend: weak_state is gone");
        _weak.lock()->_co_handle = ch.address();
        std::move(_executor)([weak = std::move(_weak)]() noexcept {
            if (auto state = weak.lock())
                std::coroutine_handle<>::from_address(state->_co_handle).resume();
        });
    }
};

template <class E, class A>
struct resume_on_any_awaiter {
    E _executor;
    A _awaitable;
    std::optional<await_result_storage_t<A>> _result;
    std::exception_ptr _exception;
 
    bool await_ready() const noexcept { return false; }
 
    await_result_t<A> await_resume() {
        if (_exception) std::rethrow_exception(_exception);
        if constexpr (std::is_void_v<await_result_t<A>>) {
            return;
        } else {
            return std::move(*_result);
        }
    }
 
    void await_suspend(std::coroutine_handle<> ch) {
        auto inner = get_awaiter(std::move(_awaitable));
        if (inner.await_ready()) {
            try {
                if constexpr (std::is_void_v<await_result_t<A>>) {
                    inner.await_resume();
                    _result.emplace(std::monostate{});
                } else {
                    _result.emplace(inner.await_resume());
                }
            } catch (...) {
                _exception = std::current_exception();
            }
            std::move(_executor)([ch]() noexcept { ch.resume(); });
            return;
        }
        resume_on_proxy_coro<std::decay_t<A>>(
            std::move(_executor), [ch]() noexcept { ch.resume(); }, std::move(inner), &_result,
            &_exception);
    }
};

template <class E, class A, class WeakPtr>
struct resume_on_any_awaiter_with_control {
    E _executor;
    A _awaitable;
    WeakPtr _weak;
    std::optional<await_result_storage_t<A>> _result;
    std::exception_ptr _exception;
 
    bool await_ready() const noexcept { return false; }
 
    await_result_t<A> await_resume() {
        if (_exception) std::rethrow_exception(_exception);
        if constexpr (std::is_void_v<await_result_t<A>>) {
            return;
        } else {
            return std::move(*_result);
        }
    }
 
    void await_suspend(std::coroutine_handle<> ch) {
        assert(_weak.lock() && "await_suspend: weak_state is gone");
        _weak.lock()->_co_handle = ch.address();
        auto inner = get_awaiter(std::move(_awaitable));
        if (inner.await_ready()) {
            try {
                if constexpr (std::is_void_v<await_result_t<A>>) {
                    inner.await_resume();
                    _result.emplace(std::monostate{});
                } else {
                    _result.emplace(inner.await_resume());
                }
            } catch (...) {
                _exception = std::current_exception();
            }
            std::move(_executor)([weak = _weak]() noexcept {
                if (auto state = weak.lock())
                    std::coroutine_handle<>::from_address(state->_co_handle).resume();
            });
            return;
        }
        resume_on_proxy_coro_controlled<std::decay_t<A>>(
            std::move(_executor),
            [weak = _weak]() noexcept {
                if (auto state = weak.lock())
                    std::coroutine_handle<>::from_address(state->_co_handle).resume();
            },
            std::move(inner), _weak, &_result, &_exception);
    }
};

template <class A>
struct cancelable_requires_stlab_future_coroutine {
    A _awaitable;
};
 
template <>
struct cancelable_requires_stlab_future_coroutine<void> {};

template <class WeakPtr>
struct cancelable_unique_checkpoint_awaiter {
    WeakPtr _weak;
 
    [[nodiscard]] bool await_ready() const noexcept { return !_weak.expired(); }
    void await_resume() noexcept {}
    void await_suspend(std::coroutine_handle<> ch) { ch.destroy(); }
};
 
} // namespace detail

template <class E, class R>
auto resume_on(E&& executor, future<R>&& f) -> detail::resume_on_awaiter<R, std::decay_t<E>> {
    return detail::resume_on_awaiter{std::forward<E>(executor), std::move(f)};
}

template <class E, class A>
auto resume_on(E&& executor, A&& awaitable)
    -> detail::resume_on_any_awaiter<std::decay_t<E>, std::decay_t<A>> {
    return detail::resume_on_any_awaiter<std::decay_t<E>, std::decay_t<A>>{
        std::forward<E>(executor), std::forward<A>(awaitable), {}, {}};
}

template <class E>
auto resume_on(E&& executor) -> detail::resume_on_executor_awaiter<std::decay_t<E>> {
    return detail::resume_on_executor_awaiter<std::decay_t<E>>{std::forward<E>(executor)};
}

template <class A, std::enable_if_t<!detail::is_future_v<std::decay_t<A>>, int> = 0>
auto cancelable(A&& awaitable)
    -> detail::cancelable_requires_stlab_future_coroutine<std::decay_t<A>> {
    return {std::forward<A>(awaitable)};
}
 
template <class R>
auto cancelable(future<R>&) // Use co_await std::move(f); it is already cancelable.
    -> future<R> = delete;
 
template <class R>
auto cancelable(future<R>&&) // Use co_await std::move(f); it is already cancelable.
    -> future<R> = delete;
 
template <class R>
auto cancelable(const future<R>&) // Use co_await std::move(f); it is already cancelable.
    -> future<R> = delete;

[[nodiscard]] inline auto cancelable() -> detail::cancelable_requires_stlab_future_coroutine<void> {
    return {};
}

 
STLAB_VERSION_NAMESPACE_END()
} // namespace stlab
 
/**************************************************************************************************/
// Coroutine ownership and destruction rules (implementation notes):
//
// We destroy the coroutine only while it is suspended, in final_awaiter::await_suspend
// (h.destroy()). Before invoking the packaged_task (in
// return_value/return_void/unhandled_exception), we clear _co_handle in the shared state so
// ~shared_task() will not try to destroy the handle when the future is later destroyed. When
// suspended on a future we store the handle in _co_handle so we can resume via weak_state. For
// non-future awaitables we clear _co_handle before suspending (give up ownership for that suspend).
// initial_suspend is suspend_never so we never own before the first suspend.
// `co_await cancelable()` with no arguments: if weak_state is expired (last future already
// released), await_suspend destroys the coroutine handle directly — not via _co_handle. If the weak
// is still valid, await_ready avoids suspension so execution continues.
/**************************************************************************************************/

 
template <class T, class... Args>
struct std::coroutine_traits<stlab::future<T>, Args...> { // NOLINT(cert-dcl58-cpp)
    struct promise_type {
        stlab::packaged_task<std::variant<T, std::exception_ptr>> _promise;
 
        stlab::future<T> get_return_object() {
            auto [pro, fut] = stlab::package<T(std::variant<T, std::exception_ptr>)>(
                stlab::immediate_executor,
                [](std::variant<T, std::exception_ptr>&& v) mutable -> T {
                    // Use index (1 = exception) to avoid ambiguity when T is std::exception_ptr.
                    if (auto* ep = std::get_if<1>(&v)) {
                        std::rethrow_exception(*ep);
                    }
                    return std::get<0>(std::move(v));
                });
            _promise = std::move(pro);
            return std::move(fut);
        }
 
        auto initial_suspend() const noexcept { return std::suspend_never{}; }
 
        auto final_suspend() noexcept {
            assert(_promise.canceled() && "final_suspend: promise not fulfilled");
            return stlab::detail::final_awaiter<std::variant<T, std::exception_ptr>>{};
        }
 
        template <class U>
        void return_value(U&& val) {
            _promise(
                std::variant<T, std::exception_ptr>{std::in_place_index<0>, std::forward<U>(val)});
        }
 
        void unhandled_exception() {
            _promise(std::variant<T, std::exception_ptr>{std::in_place_index<1>,
                                                         std::current_exception()});
        }
 
        template <class R>
        auto await_transform(stlab::future<R>&& f) {
            return stlab::detail::resume_on_awaiter_with_control<
                R, decltype(stlab::immediate_executor),
                decltype(stlab::detail::weak_state(_promise)), true>{
                stlab::immediate_executor, std::move(f), stlab::detail::weak_state(_promise)};
        }
        template <class R, class E>
        auto await_transform(stlab::detail::resume_on_awaiter<R, E> a) {
            return stlab::detail::resume_on_awaiter_with_control<
                R, E, decltype(stlab::detail::weak_state(_promise)), false>{
                std::move(a._executor), std::move(a._input), stlab::detail::weak_state(_promise)};
        }
        template <class E>
        auto await_transform(stlab::detail::resume_on_executor_awaiter<E> a) {
            return stlab::detail::resume_on_executor_awaiter_with_control<
                E, decltype(stlab::detail::weak_state(_promise))>{
                std::move(a._executor), stlab::detail::weak_state(_promise)};
        }
        template <class E, class A>
        auto await_transform(stlab::detail::resume_on_any_awaiter<E, A> a) {
            return stlab::detail::resume_on_any_awaiter_with_control<
                E, A, decltype(stlab::detail::weak_state(_promise))>{
                std::move(a._executor), std::move(a._awaitable),
                stlab::detail::weak_state(_promise), std::move(a._result), std::move(a._exception)};
        }
        auto await_transform(stlab::detail::cancelable_requires_stlab_future_coroutine<void>) {
            return stlab::detail::cancelable_unique_checkpoint_awaiter<
                decltype(stlab::detail::weak_state(_promise))>{stlab::detail::weak_state(_promise)};
        }
        template <class A, std::enable_if_t<!std::is_void_v<A>, int> = 0>
        auto await_transform(stlab::detail::cancelable_requires_stlab_future_coroutine<A> c) {
            return stlab::detail::resume_on_any_awaiter_with_control<
                decltype(stlab::immediate_executor), A,
                decltype(stlab::detail::weak_state(_promise))>{stlab::immediate_executor,
                                                               std::move(c._awaitable),
                                                               stlab::detail::weak_state(_promise),
                                                               {},
                                                               {}};
        }
        template <class U>
        U&& await_transform(U&& u) {
            if (auto state = stlab::detail::weak_state(_promise).lock())
                state->_co_handle = nullptr;
            return std::forward<U>(u);
        }
    };
};
 
template <class... Args>
struct std::coroutine_traits<stlab::future<void>, Args...> {
    struct promise_type {
        stlab::packaged_task<std::exception_ptr> _promise;
 
        stlab::future<void> get_return_object() {
            auto [pro, fut] = stlab::package<void(std::exception_ptr)>(
                stlab::immediate_executor, [](const std::exception_ptr& ep) mutable {
                    if (ep) std::rethrow_exception(ep);
                });
            _promise = std::move(pro);
            return std::move(fut);
        }
 
        auto initial_suspend() const noexcept { return std::suspend_never{}; }
 
        auto final_suspend() noexcept {
            assert(_promise.canceled() && "final_suspend: promise not fulfilled");
            return stlab::detail::final_awaiter<std::exception_ptr>{};
        }
 
        void return_void() { _promise(std::exception_ptr{}); }
 
        void unhandled_exception() { _promise(std::current_exception()); }
 
        template <class R>
        auto await_transform(stlab::future<R>&& f) {
            return stlab::detail::resume_on_awaiter_with_control<
                R, decltype(stlab::immediate_executor),
                decltype(stlab::detail::weak_state(_promise)), true>{
                stlab::immediate_executor, std::move(f), stlab::detail::weak_state(_promise)};
        }
        template <class R, class E>
        auto await_transform(stlab::detail::resume_on_awaiter<R, E> a) {
            return stlab::detail::resume_on_awaiter_with_control<
                R, E, decltype(stlab::detail::weak_state(_promise)), false>{
                std::move(a._executor), std::move(a._input), stlab::detail::weak_state(_promise)};
        }
        template <class E>
        auto await_transform(stlab::detail::resume_on_executor_awaiter<E> a) {
            return stlab::detail::resume_on_executor_awaiter_with_control<
                E, decltype(stlab::detail::weak_state(_promise))>{
                std::move(a._executor), stlab::detail::weak_state(_promise)};
        }
        template <class E, class A>
        auto await_transform(stlab::detail::resume_on_any_awaiter<E, A> a) {
            return stlab::detail::resume_on_any_awaiter_with_control<
                E, A, decltype(stlab::detail::weak_state(_promise))>{
                std::move(a._executor), std::move(a._awaitable),
                stlab::detail::weak_state(_promise), std::move(a._result), std::move(a._exception)};
        }
        auto await_transform(stlab::detail::cancelable_requires_stlab_future_coroutine<void>) {
            return stlab::detail::cancelable_unique_checkpoint_awaiter<
                decltype(stlab::detail::weak_state(_promise))>{stlab::detail::weak_state(_promise)};
        }
        template <class A, std::enable_if_t<!std::is_void_v<A>, int> = 0>
        auto await_transform(stlab::detail::cancelable_requires_stlab_future_coroutine<A> c) {
            return stlab::detail::resume_on_any_awaiter_with_control<
                decltype(stlab::immediate_executor), A,
                decltype(stlab::detail::weak_state(_promise))>{stlab::immediate_executor,
                                                               std::move(c._awaitable),
                                                               stlab::detail::weak_state(_promise),
                                                               {},
                                                               {}};
        }
        template <class U>
        U&& await_transform(U&& u) {
            if (auto state = stlab::detail::weak_state(_promise).lock())
                state->_co_handle = nullptr;
            return std::forward<U>(u);
        }
    };
};
 
/**************************************************************************************************/
 
template <class R>
auto operator co_await(stlab::future<R>&& f) {
    return stlab::detail::resume_on_awaiter{stlab::immediate_executor, std::move(f)};
}
 
// co_await on an lvalue future is deleted — use std::move(f) to make cancellation semantics
// explicit
template <class R>
auto operator co_await(stlab::future<R>& f) = delete;

 
#endif // STLAB_STD_COROUTINES()
 
#endif