future_wait_pool.hpp

Go to the documentation of this file.

#pragma once
 
 
#include <threadman/config.hpp>
#include <threadman/executor.hpp>
#include <threadman/future.hpp>
#include <threadman/stats.hpp>
#include <threadman/thread_pool.hpp>
 
#include <commons/display_info.hpp>
 
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <exception>
#include <functional>
#include <future>
#include <memory>
#include <string>
#include <string_view>
#include <thread>
#include <type_traits>
#include <utility>
#include <vector>
 
namespace threadman {
 
class ThreadManager;  // forward — defined in <threadman/manager.hpp>
 
struct FutureWaitPoolOptions {
    std::string name = "tm::future-wait";
    std::size_t min_workers = 1;
    std::size_t max_workers = std::thread::hardware_concurrency();
 
    std::chrono::milliseconds scale_up_wait{THREADMAN_DEFAULT_FUTURE_WAIT_SCALE_UP_WAIT_MS};
    std::size_t scale_up_when_queue_exceeds = 0;
    std::chrono::milliseconds idle_timeout{THREADMAN_DEFAULT_IDLE_TIMEOUT_MS};
    std::chrono::milliseconds scale_check_interval{THREADMAN_DEFAULT_SCALE_CHECK_INTERVAL_MS};
 
    std::size_t max_pending = 0;
};
 
class FutureWaitPool final : public IExecutor {
public:
    explicit FutureWaitPool(FutureWaitPoolOptions opts = {}, ThreadManager* manager = nullptr) {
        ThreadPoolOptions po;
        po.name = std::move(opts.name);
        po.min_workers = opts.min_workers;
        po.max_workers = opts.max_workers;
        po.scale_up_wait = opts.scale_up_wait;
        po.scale_up_when_queue_exceeds = opts.scale_up_when_queue_exceeds;
        po.idle_timeout = opts.idle_timeout;
        po.scale_check_interval = opts.scale_check_interval;
        po.max_queue_size = opts.max_pending;
        po.manager = manager;
        pool_ = std::make_unique<ThreadPool>(std::move(po));
    }
 
    FutureWaitPool(const FutureWaitPool&) = delete;
    FutureWaitPool& operator=(const FutureWaitPool&) = delete;
    FutureWaitPool(FutureWaitPool&&) noexcept = default;
    FutureWaitPool& operator=(FutureWaitPool&&) noexcept = default;
    ~FutureWaitPool() override = default;
 
    // --- IExecutor --------------------------------------------------------
    void execute(std::function<void()> task) override {
        pool_->execute(std::move(task));
    }
    [[nodiscard]] std::string_view name() const noexcept override {
        return pool_->name();
    }
 
    // --- Adoption ---------------------------------------------------------
 
    template <class T>
    [[nodiscard]] Future<T> add(std::future<T> fut) {
        auto promise = std::make_shared<Promise<T>>();
        auto future = promise->get_future();
        // std::future is move-only, so it can't be captured directly into the
        // std::function the pool wants — hold it behind a shared_ptr so the
        // closure stays copyable.
        auto fut_sp = std::make_shared<std::future<T>>(std::move(fut));
        pool_->execute([promise, fut_sp]() mutable {
            try {
                if constexpr (std::is_void_v<T>) {
                    fut_sp->get();
                    promise->set_value();
                } else {
                    promise->set_value(fut_sp->get());
                }
            } catch (...) {
                promise->set_exception(std::current_exception());
            }
        });
        return future;
    }
 
    template <class T>
    [[nodiscard]] Future<T> add(std::shared_future<T> fut) {
        auto promise = std::make_shared<Promise<T>>();
        auto future = promise->get_future();
        pool_->execute([promise, fut = std::move(fut)]() mutable {
            try {
                if constexpr (std::is_void_v<T>) {
                    fut.get();
                    promise->set_value();
                } else {
                    promise->set_value(fut.get());
                }
            } catch (...) {
                promise->set_exception(std::current_exception());
            }
        });
        return future;
    }
 
    template <class Fn>
        requires std::invocable<Fn>
    [[nodiscard]] auto add_blocking(Fn&& fn) -> Future<std::invoke_result_t<Fn>> {
        return pool_->submit(std::forward<Fn>(fn));
    }
 
    // --- Inspection / lifecycle (delegated to the backing pool) -----------
    [[nodiscard]] std::uint64_t id() const noexcept {
        return pool_->id();
    }
    [[nodiscard]] std::size_t worker_count() const noexcept {
        return pool_->worker_count();
    }
    [[nodiscard]] std::size_t core_worker_count() const noexcept {
        return pool_->core_worker_count();
    }
    [[nodiscard]] std::size_t max_worker_count() const noexcept {
        return pool_->max_worker_count();
    }
    [[nodiscard]] ThreadPoolStats stats() const {
        return pool_->stats();
    }
    [[nodiscard]] std::vector<ThreadSnapshot> snapshot_workers() const {
        return pool_->snapshot_workers();
    }
    [[nodiscard]] ThreadPool& pool() const noexcept {
        return *pool_;
    }
 
    void shutdown() const {
        pool_->shutdown();
    }
    void shutdown_now() const {
        pool_->shutdown_now();
    }
    void join() const {
        pool_->join();
    }
    [[nodiscard]] bool is_shutting_down() const noexcept {
        return pool_->is_shutting_down();
    }
    [[nodiscard]] bool is_terminated() const noexcept {
        return pool_->is_terminated();
    }
 
    [[nodiscard]] static const comms::DisplayInfo& display_info() {
        static const comms::DisplayInfo info{
            .name = "FutureWaitPool",
            .description = "Elastic executor for blocking std::future waits; scales 1..N under "
                           "wait pressure.",
            .icon = comms::Icon::from("mdi:timer-sand"),
        };
        return info;
    }
 
private:
    std::unique_ptr<ThreadPool> pool_;
};
 
}  // namespace threadman