future.h

#pragma once
/**
 * Chainable and Coroutine compatible std::future extensions
 * Copyright (c) 2017-2018, 2023, Jorma Rebane
 * Distributed under MIT Software License
 */
#include "thread_pool.h"
#include "future_types.h"
#include "traits.h"
#include "debugging.h" // __assertion_failure
#if RPP_HAS_COROUTINES
#  include <optional> // for cfuture<T>::promise_type (coroutine awaiter)
#endif

namespace rpp
{
    ////////////////////////////////////////////////////////////////////////////////


    /**
     * @brief Alias to std::promise<T>
     */
    template<typename T>
    using cpromise = std::promise<T>;

    /**
     * Runs any task delegate on the rpp::thread_pool using rpp::parallel_task()
     * @param task The task to run in background thread
     * @returns Composable future with return value set to task() return value.
     */
    template<typename Task>
    RPP_CORO_WRAPPER auto async_task(Task task) noexcept -> cfuture<task_return_t<Task>>
    {
        using T = task_return_t<Task>; // decay_t on the return type
        cpromise<T> p;
        cfuture<T> f = p.get_future();
        rpp::parallel_task_detached([move_args(p, task)]() mutable noexcept
        {
            try {
                if constexpr (std::is_same_v<T, void>)
                {
                    task();
                    // run task destructor before calling continuation
                    // provides deterministic sequencing: DownloadAndSaveFile().then(OpenAndParseFile);
                    if constexpr (!std::is_trivially_destructible_v<Task>) {
                        Task t = std::move(task);
                        (void)t;
                    }
                    // notify the awaiters that the value is set
                    p.set_value();
                }
                else
                {
                    T value = task();
                    // run task destructor before calling continuation
                    // provides deterministic sequencing: DownloadAndSaveFile().then(OpenAndParseFile);
                    if constexpr (!std::is_trivially_destructible_v<Task>) {
                        Task t = std::move(task);
                        (void)t;
                    }
                    // notify the awaiters that the value is set
                    p.set_value(std::move(value));
                }
                // Release the promise reference to the shared state NOW, so that
                // the pool_worker's generic.reset() only destroys a moved-from promise (no-op).
                // This avoids a TSAN race between operator delete in ~promise()
                // and pthread_cond_wait in future::get() on another thread.
                cpromise<T> release = std::move(p);
                (void)release;
            } catch (...) {
                // Release the promise reference to the shared state NOW, so that
                // the pool_worker's generic.reset() only destroys a moved-from promise (no-op).
                // This avoids a TSAN race between operator delete in ~promise()
                // and pthread_cond_wait in future::get() on another thread.
                cpromise<T> release = std::move(p);
                release.set_exception(std::current_exception());
            }
        });
        return f;
    }


    ////////////////////////////////////////////////////////////////////////////////


    /**
     * Composable Futures with Coroutine support
     * 
     * Provides composable facilities for continuations:
     *   - then(): allows chaining futures together by passing the result to the next future
     *             and resuming it using rpp::async_task()
     *   - continue_with(): allows to continue execution without returning a future,
     *                      which is useful for the final step in chaining.
     *                      NOTE: *this will be moved into the rpp::async_task(),
     *                            which is needed to avoid having to block in the destructor
     *   - detach(): allows to abandon this future by moving the state into rpp::async_task()
     *               and waiting for finish there
     * 
     * Example with classic composable futures using callbacks
     * @code
     *     rpp::async_task([=]{
     *         return downloadZipFile(url);
     *     }).then([=](std::string zipPath) {
     *         return extractContents(zipPath);
     *     }).continue_with([=](std::string extractedDir) {
     *         callOnUIThread([=]{ jobComplete(extractedDir); });
     *     });
     * @endcode
     * 
     * Example with C++20 coroutines. This uses co_await <lambda> to run each lambda
     * in a background thread. This is useful if you are upgrading legacy Synchronous
     * code which doesn't have awaitable IO.
     * 
     * The lambdas are run in background thread using rpp::thread_pool
     * @code
     *     using namespace rpp::coro_operators;
     *     rpp::cfuture<void> downloadAndUnzipDataAsync(std::string url) {
     *         std::string zipPath = co_await [&]{ return downloadZipFile(url); };
     *         std::string extractedDir = co_await [&]{ return extractContents(zipPath); };
     *         co_await callOnUIThread([=]{ jobComplete(extractedDir); });
     *         co_return;
     *     }
     * @endcode
     */
    template<typename T>
    class NODISCARD RPP_CORO_RETURN_TYPE cfuture : public std::future<T>
    {
    public:
        using super = std::future<T>;
        using value_type = T;
        cfuture() noexcept = default;
        cfuture(std::future<T>&& f) noexcept : super{std::move(f)} {}
        cfuture(cfuture&& f)        noexcept : super{std::move(f)} {}
        cfuture& operator=(std::future<T>&& f) noexcept { super::operator=(std::move(f)); return *this; }
        cfuture& operator=(cfuture&& f)        noexcept { super::operator=(std::move(f)); return *this; }

        /**
         * @warning If the future is not waited before destruction, program will std::terminate()
         */
        ~cfuture() noexcept
        {
            if (this->valid())
            {
                // check if we have waited for the future before destruction
                if (await_ready())
                {
                    try { (void)this->get(); }
                    catch (const std::exception& e) { __assertion_failure("cfuture<T> threw exception in destructor: %s", e.what()); }
                }
                // NOTE: This is a fail-fast strategy to catch programming bugs
                //       and this happens if you forget to await a future before destruction.
                //       The default behavior of std::future<T> is to silently block in the destructor,
                //       however in ReCpp we terminate immediately to force the programmer to fix the bug.
                //       https://stackoverflow.com/questions/23455104/why-is-the-destructor-of-a-future-returned-from-stdasync-blocking
                __assertion_failure("cfuture<T> was not awaited before destruction");
                std::terminate(); // always terminate
            }
        }

        /**
         * @brief Downcasts cfuture<T> into cfuture<void>, which allows to simply
         *        wait for a chain of futures to complete without getting a return value.
         * @code
         *     using namespace rpp::coro_operators;
         *     co_await async_task(operation1).then(operation2).then();
         * @endcode
         */
        RPP_CORO_WRAPPER cfuture<void> then() noexcept;

        /**
         * Continuation: after this future is complete, forwards the result to `task`.
         * The continuation is executed on a worker thread via rpp::async_task()
         * 
         * @code 
         *     auto f = async_task([=] {
         *         return downloadZip(url, onProgress);
         *     }).then([=](std::string tempFile) {
         *         return unzipContents(tempFile, onProgress);
         *     }).then([=](std::string filesDir) {
         *         loadComponents(filesDir, onProgress);
         *     });
         * @endcode
         */
        template<typename Task>
        RPP_CORO_WRAPPER auto then(Task task) noexcept -> cfuture<cont_return_t<Task, T>>
        {
            return rpp::async_task([f=std::move(*this), task=std::move(task)]() mutable -> cont_return_t<Task, T> {
                return task(f.get());
            });
        }

        /**
         * Continuation with exception handlers: allows to forward the task result
         * to the next task in chain, or handle an exception of the given type.
         * 
         * This allows for more complicated task chains where exceptions can be used
         * to recover the task chain and return a usable value.
         * 
         * @code
         *     async_task([=] {
         *         return loadCachedScene(getCachePath(file));
         *     }, [=](invalid_cache_state& e) {
         *         return loadCachedScene(downloadAndCacheFile(file)); // recover
         *     }).then([=](std::shared_ptr<SceneData> sceneData) {
         *         setCurrentScene(sceneData);
         *     }, [=](scene_load_failed& e) {
         *         loadDefaultScene(); // recover
         *     });
         * @endcode
         */
        template<typename Task, class ExceptHA>
        RPP_CORO_WRAPPER auto then(Task task, ExceptHA exhA) noexcept -> cfuture<cont_return_t<Task, T>>
        {
            using ExceptA = first_arg_type<ExceptHA>;
            return async_task([f=std::move(*this), move_args(task, exhA)]() mutable -> cont_return_t<Task, T> {
                try { return task(f.get()); } 
                catch (ExceptA& a) { return exhA(a); }
            });
        }

        template<typename Task, class ExceptHA, class ExceptHB>
        RPP_CORO_WRAPPER auto then(Task task, ExceptHA exhA, ExceptHB exhB) noexcept -> cfuture<cont_return_t<Task, T>>
        {
            using ExceptA = first_arg_type<ExceptHA>;
            using ExceptB = first_arg_type<ExceptHB>;
            return async_task([f=std::move(*this), move_args(task, exhA, exhB)]() mutable -> cont_return_t<Task, T> {
                try { return task(f.get()); }
                catch (ExceptA& a) { return exhA(a); }
                catch (ExceptB& b) { return exhB(b); }
            });
        }

        template<typename Task, class ExceptHA, class ExceptHB, class ExceptHC>
        RPP_CORO_WRAPPER auto then(Task task, ExceptHA exhA, ExceptHB exhB, ExceptHC exhC) noexcept -> cfuture<cont_return_t<Task, T>>
        {
            using ExceptA = first_arg_type<ExceptHA>;
            using ExceptB = first_arg_type<ExceptHB>;
            using ExceptC = first_arg_type<ExceptHC>;
            return async_task([f=std::move(*this), move_args(task, exhA, exhB, exhC)]() mutable -> cont_return_t<Task, T> {
                try { return task(f.get()); }
                catch (ExceptA& a) { return exhA(a); }
                catch (ExceptB& b) { return exhB(b); }
                catch (ExceptC& c) { return exhC(c); }
            });
        }

        template<typename Task, class ExceptHA, class ExceptHB, class ExceptHC, class ExceptHD>
        RPP_CORO_WRAPPER auto then(Task task, ExceptHA exhA, ExceptHB exhB, ExceptHC exhC, ExceptHD exhD) noexcept -> cfuture<cont_return_t<Task, T>>
        {
            using ExceptA = first_arg_type<ExceptHA>;
            using ExceptB = first_arg_type<ExceptHB>;
            using ExceptC = first_arg_type<ExceptHC>;
            using ExceptD = first_arg_type<ExceptHD>;
            return async_task([f=std::move(*this), move_args(task, exhA, exhB, exhC, exhD)]() mutable -> cont_return_t<Task, T> {
                try { return task(f.get()); }
                catch (ExceptA& a) { return exhA(a); }
                catch (ExceptB& b) { return exhB(b); }
                catch (ExceptC& c) { return exhC(c); }
                catch (ExceptD& d) { return exhD(d); }
            });
        }

        template<typename U>
        RPP_CORO_WRAPPER auto then(cfuture<U>&& next) noexcept -> cfuture<U>
        {
            return rpp::async_task([f=std::move(*this), n=std::move(next)]() mutable {
                { (void)f.get(); } // wait for this future to complete
                return n.get(); // return the result of the next future
            });
        }

        /**
         * Similar to .then() continuation, but doesn't return an std::future
         * and detaches this future.
         * 
         * @warning This future will be empty after calling `continue_with()`.
         *          `*this` will be moved into the background thread because of being detached.
         */
        template<typename Task>
        void continue_with(Task task) noexcept
        {
            rpp::parallel_task_detached([f=std::move(*this), move_args(task)]() mutable {
                (void)task(f.get());
            });
        }

        template<typename Task, class ExceptHA>
        void continue_with(Task task, ExceptHA exhA) noexcept
        {
            using ExceptA = first_arg_type<ExceptHA>;
            rpp::parallel_task_detached([f=std::move(*this), move_args(task, exhA)]() mutable {
                try { (void)task(f.get()); }
                catch (ExceptA& a) { (void)exhA(a); }
            });
        }

        template<typename Task, class ExceptHA, class ExceptHB>
        void continue_with(Task task, ExceptHA exhA, ExceptHB exhB) noexcept
        {
            using ExceptA = first_arg_type<ExceptHA>;
            using ExceptB = first_arg_type<ExceptHB>;
            rpp::parallel_task_detached([f=std::move(*this), move_args(task, exhA, exhB)]() mutable {
                try { (void)task(f.get()); }
                catch (ExceptA& a) { (void)exhA(a); }
                catch (ExceptB& b) { (void)exhB(b); }
            });
        }

        template<typename Task, class ExceptHA, class ExceptHB, class ExceptHC>
        void continue_with(Task task, ExceptHA exhA, ExceptHB exhB, ExceptHC exhC) noexcept
        {
            using ExceptA = first_arg_type<ExceptHA>;
            using ExceptB = first_arg_type<ExceptHB>;
            using ExceptC = first_arg_type<ExceptHC>;
            rpp::parallel_task_detached([f=std::move(*this), move_args(task, exhA, exhB, exhC)]() mutable {
                try { (void)task(f.get()); }
                catch (ExceptA& a) { (void)exhA(a); }
                catch (ExceptB& b) { (void)exhB(b); }
                catch (ExceptC& c) { (void)exhC(c); }
            });
        }

        template<typename Task, class ExceptHA, class ExceptHB, class ExceptHC, class ExceptHD>
        void continue_with(Task task, ExceptHA exhA, ExceptHB exhB, ExceptHC exhC, ExceptHD exhD) noexcept
        {
            using ExceptA = first_arg_type<ExceptHA>;
            using ExceptB = first_arg_type<ExceptHB>;
            using ExceptC = first_arg_type<ExceptHC>;
            using ExceptD = first_arg_type<ExceptHD>;
            rpp::parallel_task_detached([f=std::move(*this), move_args(task, exhA, exhB, exhC, exhD)]() mutable {
                try { (void)task(f.get()); }
                catch (ExceptA& a) { (void)exhA(a); }
                catch (ExceptB& b) { (void)exhB(b); }
                catch (ExceptC& c) { (void)exhC(c); }
                catch (ExceptD& d) { (void)exhD(d); }
            });
        }

        /**
         * Abandons this future and prevents any waiting in destructor
         * @warning This will swallow any exceptions
         */
        void detach() noexcept
        {
            if (this->valid())
            {
                rpp::parallel_task_detached([f=std::move(*this)]() mutable {
                    try { (void)f.get(); }
                    catch (...) {}
                });
            }
        }

        /**
         * @brief Asynchronously chains several tasks together by using rpp::async_task().
         *        if this future is valid, then a REGULAR continuation is scheduled, to avoid running
         *        tasks in parallel.
         *        if this future is invalid, it will be initialized into a valid async future.
         * @note If a future throws an exception in the chain, the exception is swallowed !!!
         * @returns Updated reference to this future
         * 
         * @code
         * cfuture<void> tasks;
         * tasks.chain_async([&]{
         *     task1();
         * }).chain_async([&]{
         *     throw runtime_error("task2 failed");
         * }).chain_async([&]{
         *     task3();
         * });
         * tasks.get(); // waits until task3 is completed to be completed in sequence
         * @endcode
         */
        template<typename Task>
        cfuture& chain_async(Task task) noexcept
        {
            if (!this->valid()) {
                *this = rpp::async_task(std::move(task));
            } else {
                *this = rpp::async_task([f=std::move(*this), task=std::move(task)]() mutable {
                    try { (void)f.get(); } catch (...) {} // swallow exceptions before running next task
                    return task();
                });
            }
            return *this;
        }

        cfuture& chain_async(cfuture&& next) noexcept
        {
            if (!this->valid()) {
                *this = std::move(next);
            } else {
                *this = rpp::async_task([f=std::move(*this), next=std::move(next)]() mutable {
                    try { (void)f.get(); } catch (...) {} // swallow exceptions before running next task
                    return next.get();
                });
            }
            return *this;
        }

        // checks if the future is already finished
        bool await_ready() const noexcept
        {
            return this->valid() && super::wait_for(std::chrono::microseconds{0}) != std::future_status::timeout;
        }

        // keep original wait_for / wait_until aliases for build compatibility
        using super::wait_for;
        using super::wait_until;

        // convenience wrapper for rpp::Duration
        std::future_status wait_for(rpp::Duration timeout) const
        {
            return super::wait_for(std::chrono::nanoseconds(timeout.nsec));
        }

        // convenience wrapper for rpp::TimePoint
        std::future_status wait_until(const rpp::TimePoint& until) const
        {
            auto until_tp = std::chrono::steady_clock::time_point(std::chrono::nanoseconds(until.to_epoch_ns()));
            return super::wait_until(until_tp);
        }

        /**
         * @brief Checks if the future is already finished and collects the result (non-blocking)
         *        if the future is valid. Otherwise returns false. Can throw [get()].
         * Equivalent to:
         * @code
         * if (f.await_ready())
         *    *result = f.get();
         * @endcode
         */
        bool collect_ready(T* result = nullptr)
        {
            if (!this->await_ready())
                return false;
            if (result) *result = this->get();
            else        (void)this->get();
            return true;
        }

        /**
         * @brief Waits for the future to complete and collects the result (blocking)
         *        if the future is valid. Otherwise returns false. Can throw [get()].
         * Equivalent to:
         * @code
         * if (f.valid())
         *   *result = f.get();
         * @endcode
         */
        bool collect_wait(T* result = nullptr)
        {
            if (!this->valid())
                return false;
            if (result) *result = this->get();
            else        (void)this->get();
            return true;
        }

    #if RPP_HAS_COROUTINES // C++20 coro support

        // suspension point that launches the background async task
        void await_suspend(rpp::coro_handle<> cont) noexcept
        {
            // if this future is invalid, resume immediately without spawning a background task
            if (!this->valid())
            {
                cont.resume();
                return;
            }
            rpp::parallel_task_detached([this, cont]() /*clang-12 compat:*/mutable
            {
                if (this->valid())
                    this->wait();
                cont.resume(); // call await_resume() and continue on this background thread
            });
        }

        /**
         * @returns The result and will rethrow any exceptions from the cfuture
         */
        [[nodiscard]] inline T await_resume()
        {
            return super::get();
        }

        /**
         * Enable the use of rpp::cfuture<T> as a coroutine type
         * by using a rpp::cpromise<T> as the promise type.
         *
         * This does not provide any async behaviors - simply allows easier interop
         * between existing rpp::cfuture<T> async functions
         *
         * @code
         *     // enables creating coroutines using rpp::cfuture<T>
         *     rpp::cfuture<std::string> doWorkAsync()
         *     {
         *         co_return "hello from suspendable method";
         *     }
         * @endcode
         */
        struct promise_type : rpp::cpromise<T>
        {
            /**
             * For `rpp::cfuture<T> my_coroutine() {}` this is the hidden future object
             */
            RPP_CORO_WRAPPER rpp::cfuture<T> get_return_object() noexcept { return this->get_future(); }
            RPP_CORO_STD::suspend_never initial_suspend() const noexcept { return {}; }
            RPP_CORO_STD::suspend_never final_suspend() const noexcept { return {}; }

            // Deferred return value and exception: stored here so the future is only
            // signalled in ~promise_type(), AFTER the coroutine frame (and all its local
            // variables) are fully destroyed.  This prevents a race where a thread waiting
            // on the future observes shared state before local destructors have run.
            std::optional<T>   deferred_value {};
            std::exception_ptr deferred_ex    {};

            void return_value(const T& value) noexcept(std::is_nothrow_copy_constructible_v<T>)
            {
                deferred_value = value;
            }
            void return_value(T&& value) noexcept(std::is_nothrow_move_constructible_v<T>)
            {
                deferred_value = std::move(value);
            }
            void unhandled_exception() noexcept { deferred_ex = std::current_exception(); }

            // Signal the future here, after all coroutine locals have been destroyed.
            // The C++ runtime destroys coroutine body locals before the promise object,
            // so any side-effects in those destructors are guaranteed visible to the
            // thread that wakes up on the future.
            ~promise_type() noexcept
            {
                if (deferred_ex)    rpp::cpromise<T>::set_exception(deferred_ex);
                else if (deferred_value) rpp::cpromise<T>::set_value(std::move(*deferred_value));
            }
        };
    #endif // RPP_HAS_COROUTINES
    }; // cfuture<T>


    /**
     * Composable Futures with Coroutine support. See docs in cfuture<T>.
     */
    template<>
    class NODISCARD RPP_CORO_RETURN_TYPE cfuture<void> : public std::future<void>
    {
    public:
        using super = future<void>;
        using value_type = void;
        cfuture() noexcept = default;
        cfuture(std::future<void>&& f) noexcept : super{std::move(f)} {}
        cfuture(cfuture&& f)           noexcept : super{std::move(f)} {}
        cfuture& operator=(std::future<void>&& f) noexcept { super::operator=(std::move(f)); return *this; }
        cfuture& operator=(cfuture&& f)           noexcept { super::operator=(std::move(f)); return *this; }

        /**
         * @warning If the future is not waited before destruction, program will std::terminate()
         */
        ~cfuture() noexcept
        {
            if (this->valid())
            {
                // check if we have waited for the future before destruction
                if (await_ready())
                {
                    try { (void)this->get(); }
                    catch (std::exception& e) { __assertion_failure("cfuture<void> threw exception in destructor: %s", e.what()); }
                }
                // NOTE: This is a fail-fast strategy to catch programming bugs
                //       and this happens if you forget to await a future before destruction.
                //       The default behavior of std::future<T> is to silently block in the destructor,
                //       however in ReCpp we terminate immediately to force the programmer to fix the bug.
                //       https://stackoverflow.com/questions/23455104/why-is-the-destructor-of-a-future-returned-from-stdasync-blocking
                __assertion_failure("cfuture<void> was not awaited before destruction");
                std::terminate(); // always terminate
            }
        }

        /**
         * @brief This is a no-op on cfuture<void>
         */
        RPP_CORO_WRAPPER cfuture<void> then() noexcept { return { std::move(*this) }; }

        /**
         * Standard continuation of a void future:
         * connectToServer(auth).then([=] {
         *     auto r = sendUserCredentials();
         *     return getLoginResponse(r);
         * });
         */
        template<typename Task>
        RPP_CORO_WRAPPER auto then(Task task) noexcept -> cfuture<task_return_t<Task>>
        {
            return rpp::async_task([f=std::move(*this), move_args(task)]() mutable -> task_return_t<Task> {
                f.get();
                return task();
            });
        }

        /**
         * Continuation with exception handlers: allows cfuture<void> to be followed
         * by another cfuture<T> or to recover on specific exceptions.
         * 
         * This allows for more complicated task chains where exceptions can be used
         * to recover the task chain.
         * 
         * @code
         *     // start with a void future
         *     async_task([=] {
         *         loadScene(persistentStorage().getLastScene());
         *     }, [=](scene_load_failed& e) {
         *         loadDefaultScene(); // recover
         *     }).then([=]() -> SceneComponents { // it can be continued by cfuture<T>
         *         return loadPluginComponents(plugin);
         *     }, [=](component_load_failed& e) {
         *         return {}; // ignore bad plugins
         *     });
         * @endcode
         */
        template<typename Task, class ExceptHA>
        RPP_CORO_WRAPPER auto then(Task task, ExceptHA exhA) noexcept -> cfuture<task_return_t<Task>>
        {
            using ExceptA = first_arg_type<ExceptHA>;
            return async_task([f=std::move(*this), move_args(task, exhA)]() mutable -> task_return_t<Task> {
                try { f.get(); return task(); }
                catch (ExceptA& a) { return exhA(a); }
            });
        }

        template<typename Task, class ExceptHA, class ExceptHB>
        RPP_CORO_WRAPPER auto then(Task task, ExceptHA exhA, ExceptHB exhB) noexcept -> cfuture<task_return_t<Task>>
        {
            using ExceptA = first_arg_type<ExceptHA>;
            using ExceptB = first_arg_type<ExceptHB>;
            return async_task([f=std::move(*this), move_args(task, exhA, exhB)]() mutable -> task_return_t<Task> {
                try { f.get(); return task(); }
                catch (ExceptA& a) { return exhA(a); }
                catch (ExceptB& b) { return exhB(b); }
            });
        }

        template<typename Task, class ExceptHA, class ExceptHB, class ExceptHC>
        RPP_CORO_WRAPPER auto then(Task task, ExceptHA exhA, ExceptHB exhB, ExceptHC exhC) noexcept -> cfuture<task_return_t<Task>>
        {
            using ExceptA = first_arg_type<ExceptHA>;
            using ExceptB = first_arg_type<ExceptHB>;
            using ExceptC = first_arg_type<ExceptHC>;
            return async_task([f=std::move(*this), move_args(task, exhA, exhB, exhC)]() mutable -> task_return_t<Task> {
                try { f.get(); return task(); }
                catch (ExceptA& a) { return exhA(a); }
                catch (ExceptB& b) { return exhB(b); }
                catch (ExceptC& c) { return exhC(c); }
            });
        }

        template<typename Task, class ExceptHA, class ExceptHB, class ExceptHC, class ExceptHD>
        RPP_CORO_WRAPPER auto then(Task task, ExceptHA exhA, ExceptHB exhB, ExceptHC exhC, ExceptHD exhD) noexcept -> cfuture<task_return_t<Task>>
        {
            using ExceptA = first_arg_type<ExceptHA>;
            using ExceptB = first_arg_type<ExceptHB>;
            using ExceptC = first_arg_type<ExceptHC>;
            using ExceptD = first_arg_type<ExceptHD>;
            return async_task([f=std::move(*this), move_args(task, exhA, exhB, exhC, exhD)]() mutable -> task_return_t<Task> {
                try { f.get(); return task(); }
                catch (ExceptA& a) { return exhA(a); }
                catch (ExceptB& b) { return exhB(b); }
                catch (ExceptC& c) { return exhC(c); }
                catch (ExceptD& d) { return exhD(d); }
            });
        }

        template<typename U>
        RPP_CORO_WRAPPER auto then(cfuture<U>&& next) noexcept -> cfuture<U>
        {
            return rpp::async_task([f=std::move(*this), n=std::move(next)]() mutable {
                { f.get(); } // wait for this future to complete
                return n.get(); // return the result of the next future
            });
        }

        /**
         * Similar to .then() continuation, but doesn't return an std::future
         * and detaches this future.
         * 
         * @warning This future will be empty after calling `continue_with()`.
         *          `*this` will be moved into the background thread because of being detached.
         */
        template<typename Task>
        void continue_with(Task task) noexcept
        {
            rpp::parallel_task_detached([f=std::move(*this), move_args(task)]() mutable {
                f.get();
                (void)task();
            });
        }

        template<typename Task, class ExceptHA>
        void continue_with(Task task, ExceptHA exhA) noexcept
        {
            using ExceptA = first_arg_type<ExceptHA>;
            rpp::parallel_task_detached([f=std::move(*this), move_args(task, exhA)]() mutable noexcept {
                try { f.get(); (void)task(); }
                catch (ExceptA& a) { (void)exhA(a); }
            });
        }

        template<typename Task, class ExceptHA, class ExceptHB>
        void continue_with(Task task, ExceptHA exhA, ExceptHB exhB) noexcept
        {
            using ExceptA = first_arg_type<ExceptHA>;
            using ExceptB = first_arg_type<ExceptHB>;
            rpp::parallel_task_detached([f=std::move(*this), move_args(task, exhA, exhB)]() mutable {
                try { f.get(); (void)task(); }
                catch (ExceptA& a) { (void)exhA(a); }
                catch (ExceptB& b) { (void)exhB(b); }
            });
        }

        template<typename Task, class ExceptHA, class ExceptHB, class ExceptHC>
        void continue_with(Task task, ExceptHA exhA, ExceptHB exhB, ExceptHC exhC) noexcept
        {
            using ExceptA = first_arg_type<ExceptHA>;
            using ExceptB = first_arg_type<ExceptHB>;
            using ExceptC = first_arg_type<ExceptHC>;
            rpp::parallel_task_detached([f=std::move(*this), move_args(task, exhA, exhB, exhC)]() mutable {
                try { f.get(); (void)task(); }
                catch (ExceptA& a) { (void)exhA(a); }
                catch (ExceptB& b) { (void)exhB(b); }
                catch (ExceptC& c) { (void)exhC(c); }
            });
        }

        template<typename Task, class ExceptHA, class ExceptHB, class ExceptHC, class ExceptHD>
        void continue_with(Task task, ExceptHA exhA, ExceptHB exhB, ExceptHC exhC, ExceptHD exhD) noexcept
        {
            using ExceptA = first_arg_type<ExceptHA>;
            using ExceptB = first_arg_type<ExceptHB>;
            using ExceptC = first_arg_type<ExceptHC>;
            using ExceptD = first_arg_type<ExceptHD>;
            rpp::parallel_task_detached([f=std::move(*this), move_args(task, exhA, exhB, exhC, exhD)]() mutable {
                try { f.get(); (void)task(); }
                catch (ExceptA& a) { (void)exhA(a); }
                catch (ExceptB& b) { (void)exhB(b); }
                catch (ExceptC& c) { (void)exhC(c); }
                catch (ExceptD& d) { (void)exhD(d); }
            });
        }

        /**
         * @brief Asynchronously chains several tasks together by using rpp::async_task().
         *        if this future is valid, then a REGULAR continuation is scheduled, to avoid running
         *        tasks in parallel.
         *        if this future is invalid, it will be initialized into a valid async future.
         * @note If a future throws an exception in the chain, the exception is swallowed !!!
         * @returns Updated reference to this future
         * 
         * @code
         * cfuture<void> tasks;
         * tasks.chain_async([&]{
         *     task1();
         * }).chain_async([&]{
         *     throw runtime_error("task2 failed");
         * }).chain_async([&]{
         *     task3();
         * });
         * tasks.get(); // waits until task3 is completed to be completed in sequence
         * @endcode
         */
        template<typename Task>
        cfuture& chain_async(Task task) noexcept
        {
            if (!this->valid()) {
                *this = rpp::async_task(std::move(task));
            } else {
                *this = rpp::async_task([f=std::move(*this), task=std::move(task)]() mutable {
                    try { (void)f.get(); } catch (...) {} // swallow exceptions before running next task
                    return task();
                });
            }
            return *this;
        }

        cfuture& chain_async(cfuture&& next) noexcept
        {
            if (!this->valid()) {
                *this = std::move(next);
            } else {
                *this = rpp::async_task([f=std::move(*this), next=std::move(next)]() mutable {
                    try { (void)f.get(); } catch (...) {} // swallow exceptions before running next task
                    return next.get();
                });
            }
            return *this;
        }

        /**
         * Abandons this future and prevents any waiting in destructor
         * @warning This will swallow any exceptions
         */
        void detach()
        {
            if (this->valid())
            {
                rpp::parallel_task_detached([f=std::move(*this)]() mutable noexcept
                {
                    try { f.get(); }
                    catch (...) {}
                });
            }
        }

        // checks if the future is already finished
        bool await_ready() const noexcept
        {
            return this->valid() && super::wait_for(std::chrono::microseconds{0}) != std::future_status::timeout;
        }

        // keep original wait_for / wait_until aliases for build compatibility
        using super::wait_for;
        using super::wait_until;

        // convenience wrapper for rpp::Duration
        std::future_status wait_for(rpp::Duration timeout) const
        {
            return super::wait_for(std::chrono::nanoseconds(timeout.nsec));
        }

        // convenience wrapper for rpp::TimePoint
        std::future_status wait_until(const rpp::TimePoint& until) const
        {
            auto until_tp = std::chrono::steady_clock::time_point(std::chrono::nanoseconds(until.to_epoch_ns()));
            return super::wait_until(until_tp);
        }

        /**
         * @brief Checks if the future is already finished and collects the result (non-blocking)
         *        if the future is valid. Otherwise returns false. Can throw [get()].
         * Equivalent to:
         * @code
         * if (f.await_ready())
         *    *result = f.get();
         * @endcode
         */
        bool collect_ready()
        {
            if (!this->await_ready())
                return false;
            this->get();
            return true;
        }

        /**
         * @brief Waits for the future to complete and collects the result (blocking)
         *        if the future is valid. Otherwise returns false. Can throw [get()].
         * Equivalent to:
         * @code
         * if (f.valid())
         *     f.get();
         * @endcode
         */
        bool collect_wait()
        {
            if (!this->valid())
                return false;
            this->get();
            return true;
        }

    #if RPP_HAS_COROUTINES // C++20 coroutine support

        // suspension point that launches the background async task
        void await_suspend(rpp::coro_handle<> cont) noexcept
        {
            // if this future is invalid, resume immediately without spawning a background task
            if (!this->valid())
            {
                cont.resume();
                return;
            }
            rpp::parallel_task_detached([this, cont]() /*clang-12 compat:*/mutable
            {
                if (this->valid())
                    this->wait();
                cont.resume(); // call await_resume() and continue on this background thread
            });
        }

        /**
         * @briefWaits for the final result and will rethrow any exceptions from the cfuture
         */
        inline void await_resume()
        {
            this->get();
        }

        /**
         * Enable the use of rpp::cfuture<void> as a coroutine type.
         *
         * This does not provide any async behaviors - simply allows easier interop
         * between existing rpp::cfuture<void> async functions
         *
         * @code
         *     rpp::cfuture<void> doWorkAsync()
         *     {
         *         co_return;
         *     }
         * @endcode
         */
        struct promise_type : rpp::cpromise<void>
        {
            RPP_CORO_WRAPPER rpp::cfuture<void> get_return_object() noexcept { return this->get_future(); }
            RPP_CORO_STD::suspend_never initial_suspend() const noexcept { return {}; }
            RPP_CORO_STD::suspend_never final_suspend() const noexcept { return {}; }

            // Same deferred-signal approach as cfuture<T>::promise_type.
            std::exception_ptr deferred_ex     {};
            bool               deferred_return { false };

            void return_void() noexcept { deferred_return = true; }
            void unhandled_exception() noexcept { deferred_ex = std::current_exception(); }

            ~promise_type() noexcept
            {
                if (deferred_ex)      rpp::cpromise<void>::set_exception(deferred_ex);
                else if (deferred_return) rpp::cpromise<void>::set_value();
            }
        };
    #endif // RPP_HAS_COROUTINES
    }; // cfuture<void>


    /**
     * @brief Downcasts cfuture<T> into cfuture<void>, which allows to simply
     *        wait for a chain of futures to complete without getting a return value.
     * @code
     *     using namespace rpp::coro_operators;
     *     co_await async_task(operation1).then(operation2).then();
     * @endcode
     */
    template<typename T>
    inline cfuture<void> cfuture<T>::then() noexcept
    {
        return rpp::async_task([f=std::move(*this)]() mutable {
            (void)f.get();
        });
    }

    /** @brief Creates a future<T> which is already completed. Useful for some chaining edge cases. */
    template<typename T>
    RPP_CORO_WRAPPER inline auto make_ready_future(T value) -> cfuture<T>
    {
        std::promise<T> p;
        p.set_value(std::move(value));
        return p.get_future();
    }

    /** @brief Creates a future<void> which is already completed. Useful for some chaining edge cases. */
    RPP_CORO_WRAPPER inline auto make_ready_future() -> cfuture<void>
    {
        std::promise<void> p;
        p.set_value();
        return p.get_future();
    }

    /** @brief Creates a future<T> which is already errored with the exception. Useful for some chaining edge cases. */
    template<typename T, typename E>
    RPP_CORO_WRAPPER inline auto make_exceptional_future(E e) -> cfuture<T>
    {
        std::promise<T> p;
        p.set_exception(std::make_exception_ptr(std::forward<E>(e)));
        return p.get_future();
    }

    /** @brief Given a vector of futures, waits blockingly on all of them to be completed, without getting the values. */
    template<typename T>
    inline void wait_all(const std::vector<cfuture<T>>& vf)
    {
        for (const cfuture<T>& f : vf)
        {
            f.wait();
        }
    }

    /**
     * @brief Blocks and gathers the results from all of the futures
     * @throws an exception if any of the futures threw an exception
     */
    template<typename T>
    inline std::vector<T> get_all(std::vector<cfuture<T>>& vf)
    {
        std::vector<T> all;
        all.reserve(vf.size());
        for (cfuture<T>& f : vf)
        {
            all.emplace_back(f.get());
        }
        return all;
    }

    /**
     * @brief Blocks and gets() the void result from all of the futures.
     * @throws an exception if any of the futures threw an exception
     */
    inline void get_all(std::vector<cfuture<void>>& vf)
    {
        for (cfuture<void>& f : vf)
        {
            f.get();
        }
    }

    /**
     * Used to launch multiple parallel Tasks and then wait for
     * the results. It assumes that launcher already started its
     * own future task, which makes it more flexible.
     * @code
     *     rpp::run_tasks(dataList, [&](SomeData& data) {
     *         return rpp::async_task([&]{
     *             heavyComputation(data);
     *         };
     *     });
     * @endcode
     */
    template<typename U, typename Launcher>
    inline void run_tasks(std::vector<U>& items, const Launcher& futureLauncher)
    {
        std::vector<cfuture<void>> futures;
        futures.reserve(items.size());

        for (U& item : items)
        {
            futures.emplace_back(futureLauncher(item));
        }

        return wait_all(futures);
    }
} // namespace rpp