future_extensions.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2024, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <kcenon/thread/forward.h>
#include <kcenon/thread/core/error_handling.h>
#include <future>
#include <memory>
#include <chrono>
#include <functional>
 
namespace kcenon::thread {
 
    template<typename T>
    class pool_future {
    public:
        using value_type = T;
        
        explicit pool_future(std::future<T>&& future) 
            : future_(std::move(future)) {}
        
        bool is_ready() const {
            return future_.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
        }
        
        T get() {
            return future_.get();
        }
        
        template<typename Rep, typename Period>
        std::future_status wait_for(const std::chrono::duration<Rep, Period>& timeout) {
            return future_.wait_for(timeout);
        }
        
        void wait() {
            future_.wait();
        }
        
        bool valid() const {
            return future_.valid();
        }
        
        template<typename F>
        auto then(F&& continuation) -> pool_future<std::invoke_result_t<F, T>>;
        
    private:
        std::future<T> future_;
    };
    
    template<typename T>
    class pool_promise {
    public:
        using value_type = T;
        
        pool_promise() = default;
        
        pool_future<T> get_future() {
            return pool_future<T>{promise_.get_future()};
        }
        
        void set_value(const T& value) {
            promise_.set_value(value);
        }
        
        void set_value(T&& value) {
            promise_.set_value(std::move(value));
        }
        
        void set_exception(std::exception_ptr exception) {
            promise_.set_exception(exception);
        }
        
        void set_value_at_thread_exit(const T& value) {
            promise_.set_value_at_thread_exit(value);
        }
        
        void set_exception_at_thread_exit(std::exception_ptr exception) {
            promise_.set_exception_at_thread_exit(exception);
        }
        
    private:
        std::promise<T> promise_;
    };
    
    template<>
    class pool_promise<void> {
    public:
        pool_future<void> get_future() {
            return pool_future<void>{promise_.get_future()};
        }
        
        void set_value() {
            promise_.set_value();
        }
        
        void set_exception(std::exception_ptr exception) {
            promise_.set_exception(exception);
        }
        
        void set_value_at_thread_exit() {
            promise_.set_value_at_thread_exit();
        }
        
        void set_exception_at_thread_exit(std::exception_ptr exception) {
            promise_.set_exception_at_thread_exit(exception);
        }
        
    private:
        std::promise<void> promise_;
    };
    
    template<typename T>
    pool_future<std::decay_t<T>> make_ready_future(T&& value) {
        pool_promise<std::decay_t<T>> promise;
        auto future = promise.get_future();
        promise.set_value(std::forward<T>(value));
        return future;
    }
    
    inline pool_future<void> make_ready_future() {
        pool_promise<void> promise;
        auto future = promise.get_future();
        promise.set_value();
        return future;
    }
    
    template<typename T>
    pool_future<T> make_exceptional_future(std::exception_ptr exception) {
        pool_promise<T> promise;
        auto future = promise.get_future();
        promise.set_exception(exception);
        return future;
    }
    
    template<typename... Futures>
    void wait_for_all(Futures&... futures) {
        (futures.wait(), ...);
    }
    
    template<typename... Futures>
    size_t wait_for_any(Futures&... futures) {
        // Use short timeout-based polling instead of busy-wait
        // This significantly reduces CPU usage while maintaining responsiveness
        constexpr auto poll_interval = std::chrono::microseconds(100);
 
        size_t index = 0;
        auto check_future = [&](auto& future) {
            if (future.is_ready()) {
                return true;
            }
            ++index;
            return false;
        };
 
        while (true) {
            index = 0;
            if ((check_future(futures) || ...)) {
                return index;
            }
            // Sleep briefly to reduce CPU usage instead of busy-waiting
            std::this_thread::sleep_for(poll_interval);
        }
    }
    
} // namespace kcenon::thread