thread_pool.hpp

Go to the documentation of this file.

#pragma once
 
 
#include <threadman/config.hpp>
#include <threadman/display.hpp>
#include <threadman/exceptions.hpp>
#include <threadman/executor.hpp>
#include <threadman/future.hpp>
#include <threadman/log.hpp>
#include <threadman/metrics.hpp>
#include <threadman/stats.hpp>
#include <threadman/task.hpp>
#include <threadman/thread.hpp>
 
#include <commons/display_info.hpp>
 
#include <algorithm>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <deque>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <stdexcept>
#include <stop_token>
#include <string>
#include <thread>
#include <unordered_map>
#include <utility>
#include <vector>
 
namespace threadman {
 
class ThreadManager;  // forward — defined in <threadman/manager.hpp>
 
namespace detail {
inline std::uint64_t next_pool_id() noexcept {
    static std::atomic<std::uint64_t> counter{1};
    return counter.fetch_add(1, std::memory_order_relaxed);
}
inline std::uint64_t next_task_id() noexcept {
    static std::atomic<std::uint64_t> counter{1};
    return counter.fetch_add(1, std::memory_order_relaxed);
}
}  // namespace detail
 
struct ThreadPoolOptions {
    std::string name = "tm::pool";
 
    // Scaling --------------------------------------------------------------
    std::size_t min_workers = 1;
    std::size_t max_workers = std::thread::hardware_concurrency();
    std::chrono::milliseconds scale_up_wait{THREADMAN_DEFAULT_SCALE_UP_WAIT_MS};
    std::size_t scale_up_when_queue_exceeds = THREADMAN_DEFAULT_SCALE_UP_QUEUE_THRESHOLD;
    std::chrono::milliseconds idle_timeout{THREADMAN_DEFAULT_IDLE_TIMEOUT_MS};
    std::chrono::milliseconds scale_check_interval{THREADMAN_DEFAULT_SCALE_CHECK_INTERVAL_MS};
 
    // Back-pressure --------------------------------------------------------
    std::size_t max_queue_size = 0;  // 0 = unbounded.
 
    // Observability --------------------------------------------------------
    std::chrono::milliseconds stuck_task_threshold{THREADMAN_DEFAULT_STUCK_TASK_THRESHOLD_MS};
    std::size_t recent_tasks_capacity = THREADMAN_RECENT_TASKS_CAPACITY;
 
    // Attachment -----------------------------------------------------------
    ThreadManager* manager = nullptr;
};
 
class ThreadPool final : public IExecutor {
public:
    explicit ThreadPool(ThreadPoolOptions opts = {});
    ~ThreadPool() override;
 
    ThreadPool(const ThreadPool&) = delete;
    ThreadPool& operator=(const ThreadPool&) = delete;
    ThreadPool(ThreadPool&&) = delete;
    ThreadPool& operator=(ThreadPool&&) = delete;
 
    // IExecutor
    void execute(std::function<void()> task) override;
    [[nodiscard]] std::string_view name() const noexcept override {
        return opts_.name;
    }
 
    // Native overloads
    template <class Fn>
        requires std::invocable<Fn>
    auto submit(Fn&& f) -> Future<std::invoke_result_t<Fn>>;
 
    template <class Fn>
        requires std::invocable<Fn>
    auto submit_named(std::string task_name, Fn&& f) -> Future<std::invoke_result_t<Fn>>;
 
    template <class Fn>
        requires std::invocable<Fn, std::stop_token>
    auto submit_stoppable(Fn&& f) -> Future<std::invoke_result_t<Fn, std::stop_token>>;
 
    // Lifecycle
    void shutdown();
    void shutdown_now();
    void join();
    [[nodiscard]] bool is_shutting_down() const noexcept;
    [[nodiscard]] bool is_terminated() const noexcept;
 
    // Inspection
    [[nodiscard]] std::uint64_t id() const noexcept {
        return pool_id_;
    }
    [[nodiscard]] std::size_t worker_count() const noexcept;
    [[nodiscard]] std::size_t core_worker_count() const noexcept {
        return opts_.min_workers;
    }
    [[nodiscard]] std::size_t max_worker_count() const noexcept {
        return opts_.max_workers;
    }
    [[nodiscard]] ThreadPoolStats stats() const;
    [[nodiscard]] std::vector<ThreadSnapshot> snapshot_workers() const;
    [[nodiscard]] std::vector<TaskSnapshot> snapshot_recent_tasks() const;
    [[nodiscard]] std::vector<TaskSnapshot> snapshot_queued_tasks() const;
 
    [[nodiscard]] static const comms::DisplayInfo& display_info() {
        static const comms::DisplayInfo info{
            .name = "ThreadPool",
            .description = "Dynamic-scaling thread pool executor with task tracking.",
            .icon = comms::Icon::from("mdi:server-network"),
        };
        return info;
    }
 
    int scale_tick(std::chrono::steady_clock::time_point now);
    [[nodiscard]] std::vector<StuckTaskEvent>
    detect_stuck_tasks(std::chrono::steady_clock::time_point now) const;
 
private:
    struct QueuedTask {
        TaskHandle handle;
        std::function<void(std::stop_token)> work;
        std::function<void()> on_cancel;  // satisfies the future with TaskCancelledError
        std::chrono::steady_clock::time_point enqueued_at;
    };
 
    template <class Fn, class R>
    Future<R> submit_impl(std::optional<std::string> task_name, Fn&& f);
    template <class Fn, class R>
    Future<R> submit_stoppable_impl(std::optional<std::string> task_name, Fn&& f);
 
    void enqueue(QueuedTask qt);
    void throw_if_not_running() const {
        if (state_.load(std::memory_order_acquire) == PoolState::Running) {
            return;
        }
        metrics::pool_submit_rejected().labels(pool_labels_).inc();
        throw PoolShuttingDownError("ThreadPool '" + opts_.name + "': shutting down or terminated");
    }
    void worker_loop(std::stop_token tok, std::uint64_t worker_index, bool is_core);
    void spawn_worker(bool is_core);
    void record_recent(TaskHandle h);
    void register_with_manager();
    void unregister_from_manager() noexcept;
 
    ThreadPoolOptions opts_;
    std::uint64_t pool_id_ = detail::next_pool_id();
    ThreadManager* manager_ = nullptr;
 
    mutable std::mutex workers_mtx_;
    std::vector<std::unique_ptr<ManagedThread>> workers_;
    std::vector<std::shared_ptr<ManagedThread::ControlBlock>> worker_cbs_;  // parallel to workers_
 
    mutable std::mutex queue_mtx_;
    std::condition_variable queue_cv_;
    std::deque<QueuedTask> queue_;
 
    std::atomic<PoolState> state_{PoolState::Running};
 
    std::atomic<std::size_t> active_{0};  // worker currently in user work
    std::atomic<std::size_t> idle_{0};    // worker waiting on queue
    std::atomic<std::uint64_t> completed_{0};
    std::atomic<std::uint64_t> failed_{0};
    std::atomic<std::uint64_t> scale_ups_{0};
    std::atomic<std::uint64_t> scale_downs_{0};
    std::atomic<std::uint64_t> total_exec_ns_{0};
 
    mutable std::mutex recent_mtx_;
    std::deque<TaskHandle> recent_;
 
    mutable std::mutex running_mtx_;
    std::unordered_map<std::uint64_t, TaskHandle> running_;  // by task id
 
    // Cached `{pool=<name>}` label set reused at every metric call site. Only
    // the Labels are cached, never a labelled child: prom children are pinned to
    // the adapter present at their creation, so building a fresh child per call
    // keeps metrics flowing to whatever backend is installed at that moment.
    prom::Labels pool_labels_{prom::Labels{{"pool", opts_.name}}};
};
 
// ---------------------------------------------------------------------------
// Inline definitions
// ---------------------------------------------------------------------------
 
inline ThreadPool::ThreadPool(ThreadPoolOptions opts) : opts_(std::move(opts)) {
    if (opts_.max_workers == 0) {
        opts_.max_workers = 1;
    }
 
    opts_.min_workers = std::min(opts_.min_workers, opts_.max_workers);
 
    if (opts_.recent_tasks_capacity == 0) {
        opts_.recent_tasks_capacity = 1;
    }
 
    register_with_manager();
 
    for (std::size_t i = 0; i < opts_.min_workers; ++i) {
        spawn_worker(/*is_core=*/true);
    }
 
    if (const auto& lg = log::pool()) {
        lg->debug("pool '{}' created (id={}, core={}, max={})",
                  opts_.name,
                  pool_id_,
                  opts_.min_workers,
                  opts_.max_workers);
    }
}
 
inline ThreadPool::~ThreadPool() {
    if (state_.load(std::memory_order_acquire) == PoolState::Running) {
        try {
            shutdown();
        } catch (...) {
            // dtor must not propagate
        }
    }
    // Join all workers.
    {
        std::scoped_lock guard(workers_mtx_);
        for (const auto& w : workers_) {
            if (w && w->joinable()) {
                w->request_stop();
            }
        }
    }
    {
        std::unique_lock lk(queue_mtx_);
        queue_cv_.notify_all();
    }
    {
        std::scoped_lock guard(workers_mtx_);
        for (const auto& w : workers_) {
            if (w && w->joinable()) {
                w->join();
            }
        }
        workers_.clear();
        worker_cbs_.clear();
    }
    state_.store(PoolState::Terminated, std::memory_order_release);
    unregister_from_manager();
}
 
inline void ThreadPool::execute(std::function<void()> task) {
    if (!task) {
        return;
    }
    throw_if_not_running();
    const TaskHandle handle(detail::next_task_id(), pool_id_);
    QueuedTask qt;
    qt.handle = handle;
    qt.enqueued_at = std::chrono::steady_clock::now();
    qt.work = [t = std::move(task)](const std::stop_token&) { t(); };
    enqueue(std::move(qt));
}
 
template <class Fn, class R>
Future<R> ThreadPool::submit_impl(std::optional<std::string> task_name, Fn&& f) {
    throw_if_not_running();
    auto promise = std::make_shared<Promise<R>>();
    auto future = promise->get_future();
    const TaskHandle handle(detail::next_task_id(), pool_id_, std::move(task_name));
 
    auto fn_shared = std::make_shared<std::decay_t<Fn>>(std::forward<Fn>(f));
 
    QueuedTask qt;
    qt.handle = handle;
    qt.enqueued_at = std::chrono::steady_clock::now();
    qt.work = [promise, fn_shared](const std::stop_token&) {
        try {
            if constexpr (std::is_void_v<R>) {
                (*fn_shared)();
                promise->set_value();
            } else {
                promise->set_value((*fn_shared)());
            }
        } catch (...) {
            promise->set_exception(std::current_exception());
        }
    };
    qt.on_cancel = [promise]() {
        try {
            promise->set_exception(
                std::make_exception_ptr(TaskCancelledError("task cancelled before it could run")));
        } catch (...) {
            // promise may already be satisfied via another path
        }
    };
    enqueue(std::move(qt));
    return future;
}
 
template <class Fn, class R>
Future<R> ThreadPool::submit_stoppable_impl(std::optional<std::string> task_name, Fn&& f) {
    throw_if_not_running();
    auto promise = std::make_shared<Promise<R>>();
    auto future = promise->get_future();
    const TaskHandle handle(detail::next_task_id(), pool_id_, std::move(task_name));
 
    auto fn_shared = std::make_shared<std::decay_t<Fn>>(std::forward<Fn>(f));
 
    QueuedTask qt;
    qt.handle = handle;
    qt.enqueued_at = std::chrono::steady_clock::now();
    // stop_token is taken by value (the std::function<void(std::stop_token)>
    // signature) and forwarded via std::move into the user callable, which may
    // accept it by value or by rvalue-ref — const& would break the latter.
    // NOLINTNEXTLINE(performance-unnecessary-value-param)
    qt.work = [promise, fn_shared](std::stop_token tok) {
        try {
            if constexpr (std::is_void_v<R>) {
                (*fn_shared)(std::move(tok));
                promise->set_value();
            } else {
                promise->set_value((*fn_shared)(std::move(tok)));
            }
        } catch (...) {
            promise->set_exception(std::current_exception());
        }
    };
    qt.on_cancel = [promise]() {
        try {
            promise->set_exception(
                std::make_exception_ptr(TaskCancelledError("task cancelled before it could run")));
        } catch (...) {
            // promise may already be satisfied via another path
        }
    };
    enqueue(std::move(qt));
    return future;
}
 
template <class Fn>
    requires std::invocable<Fn>
auto ThreadPool::submit(Fn&& f) -> Future<std::invoke_result_t<Fn>> {
    using R = std::invoke_result_t<Fn>;
    return submit_impl<Fn, R>(std::nullopt, std::forward<Fn>(f));
}
 
template <class Fn>
    requires std::invocable<Fn>
auto ThreadPool::submit_named(std::string task_name, Fn&& f) -> Future<std::invoke_result_t<Fn>> {
    using R = std::invoke_result_t<Fn>;
    return submit_impl<Fn, R>(std::move(task_name), std::forward<Fn>(f));
}
 
template <class Fn>
    requires std::invocable<Fn, std::stop_token>
auto ThreadPool::submit_stoppable(Fn&& f) -> Future<std::invoke_result_t<Fn, std::stop_token>> {
    using R = std::invoke_result_t<Fn, std::stop_token>;
    return submit_stoppable_impl<Fn, R>(std::nullopt, std::forward<Fn>(f));
}
 
inline void ThreadPool::enqueue(QueuedTask qt) {
    {
        std::unique_lock lk(queue_mtx_);
        if (opts_.max_queue_size != 0 && queue_.size() >= opts_.max_queue_size) {
            // Emit outside the lock — release first, then count the rejection.
            lk.unlock();
            metrics::pool_queue_rejected().labels(pool_labels_).inc();
            throw PoolQueueFullError("ThreadPool '" + opts_.name + "': queue full");
        }
        queue_.push_back(std::move(qt));
        queue_cv_.notify_one();
    }
    metrics::pool_tasks_submitted().labels(pool_labels_).inc();
}
 
inline void ThreadPool::spawn_worker(bool is_core) {
    ManagedThread::Options topts;
    topts.name = opts_.name + "-worker-" + std::to_string(workers_.size());
    topts.pool_id = pool_id_;
    topts.is_core = is_core;
    topts.manager = manager_;
 
    const std::uint64_t worker_index = workers_.size();
    auto worker = std::make_unique<ManagedThread>(
        std::move(topts), [this, worker_index, is_core](std::stop_token tok) {
            worker_loop(std::move(tok), worker_index, is_core);
        });
    worker_cbs_.push_back(worker->control_block());
    workers_.push_back(std::move(worker));
    if (!is_core) {
        scale_ups_.fetch_add(1, std::memory_order_relaxed);
        metrics::pool_scale_ups().labels(pool_labels_).inc();
    }
}
 
inline void ThreadPool::worker_loop(std::stop_token tok, std::uint64_t worker_index, bool is_core) {
    (void)worker_index;
 
    auto& lg_pool = log::pool();
    auto* cb_raw = ManagedThread::current();
 
    // Until stopped, fetch tasks. Track per-iteration last-active time so an
    // overflow worker can age out via idle_timeout.
    auto last_active = std::chrono::steady_clock::now();
 
    while (!tok.stop_requested()) {
        QueuedTask task;
        bool got_task = false;
        bool drain = false;
        std::uint64_t cancelled_now = 0;
 
        {
            std::unique_lock lk(queue_mtx_);
 
            // Park as idle.
            if (cb_raw != nullptr) {
                cb_raw->state.store(ThreadState::Idle, std::memory_order_release);
            }
            idle_.fetch_add(1, std::memory_order_relaxed);
 
            const bool waited = queue_cv_.wait_for(lk, opts_.scale_check_interval, [&] {
                if (tok.stop_requested()) {
                    return true;
                }
                if (!queue_.empty()) {
                    return true;
                }
                const auto s = state_.load(std::memory_order_acquire);
                if (s == PoolState::ShutdownNow) {
                    return true;
                }
                if (s == PoolState::ShuttingDown && queue_.empty()) {
                    return true;
                }
                return false;
            });
            (void)waited;
 
            idle_.fetch_sub(1, std::memory_order_relaxed);
            if (cb_raw != nullptr) {
                cb_raw->state.store(ThreadState::Running, std::memory_order_release);
            }
 
            if (const auto s = state_.load(std::memory_order_acquire);
                s == PoolState::ShutdownNow) {
                // Cancel any remaining queued tasks visible to us.
                while (!queue_.empty()) {
                    auto cancelled = std::move(queue_.front());
                    queue_.pop_front();
                    cancelled.handle.mark_cancelled(std::make_exception_ptr(
                        TaskCancelledError("task cancelled by shutdown_now")));
                    if (cancelled.on_cancel) {
                        cancelled.on_cancel();
                    }
                    failed_.fetch_add(1, std::memory_order_relaxed);
                    ++cancelled_now;
                    record_recent(std::move(cancelled.handle));
                }
                drain = true;
            } else if (!queue_.empty()) {
                task = std::move(queue_.front());
                queue_.pop_front();
                got_task = true;
            } else if (s == PoolState::ShuttingDown) {
                drain = true;
            }
        }
 
        // Emit cancellations outside the queue lock (per metrics policy).
        if (cancelled_now != 0) {
            metrics::pool_tasks_cancelled().labels(pool_labels_).inc(cancelled_now);
        }
 
        if (drain) {
            break;
        }
 
        if (!got_task) {
            // Idle-timeout retirement for non-core workers.
            if (!is_core) {
                if (const auto now = std::chrono::steady_clock::now();
                    now - last_active >= opts_.idle_timeout) {
                    // Self-retire.
                    if (cb_raw != nullptr) {
                        cb_raw->state.store(ThreadState::Retiring, std::memory_order_release);
                    }
                    scale_downs_.fetch_add(1, std::memory_order_relaxed);
                    metrics::pool_scale_downs().labels(pool_labels_).inc();
                    if (auto& lg = log::pool()) {
                        lg->info("pool '{}' retiring idle non-core worker", opts_.name);
                    }
                    return;
                }
            }
            continue;
        }
 
        last_active = std::chrono::steady_clock::now();
        active_.fetch_add(1, std::memory_order_relaxed);
        task.handle.mark_running();
        {
            std::scoped_lock guard(running_mtx_);
            running_.emplace(task.handle.id(), task.handle);
        }
        const auto t0 = std::chrono::steady_clock::now();
        const auto queue_wait_ns =
            std::chrono::duration_cast<std::chrono::nanoseconds>(t0 - task.enqueued_at).count();
        bool task_failed = false;
        try {
            task.work(tok);
            task.handle.mark_completed();
            completed_.fetch_add(1, std::memory_order_relaxed);
        } catch (...) {
            task.handle.mark_failed(std::current_exception());
            failed_.fetch_add(1, std::memory_order_relaxed);
            task_failed = true;
        }
        const auto t1 = std::chrono::steady_clock::now();
        const auto elap = std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count();
        total_exec_ns_.fetch_add(static_cast<std::uint64_t>(elap), std::memory_order_relaxed);
        active_.fetch_sub(1, std::memory_order_relaxed);
        {
            std::scoped_lock guard(running_mtx_);
            running_.erase(task.handle.id());
        }
        record_recent(std::move(task.handle));
        // Emit metrics only after record_recent so the small window callers
        // observe right after Future::get() (recent-task snapshot) is not
        // widened by per-task metric work.
        metrics::pool_task_queue_wait_seconds()
            .labels(pool_labels_)
            .observe(static_cast<double>(queue_wait_ns) / 1e9);
        metrics::pool_task_execution_seconds()
            .labels(pool_labels_)
            .observe(static_cast<double>(elap) / 1e9);
        if (task_failed) {
            metrics::pool_tasks_failed().labels(pool_labels_).inc();
        } else {
            metrics::pool_tasks_completed().labels(pool_labels_).inc();
        }
    }
 
    // Drain on shutdown (drain queued tasks with the worker's own loop).
    std::uint64_t drain_cancelled = 0;
    while (true) {
        QueuedTask task;
        bool got_task = false;
        {
            std::unique_lock lk(queue_mtx_);
            if (state_.load(std::memory_order_acquire) == PoolState::ShutdownNow) {
                while (!queue_.empty()) {
                    auto cancelled = std::move(queue_.front());
                    queue_.pop_front();
                    cancelled.handle.mark_cancelled(std::make_exception_ptr(
                        TaskCancelledError("task cancelled by shutdown_now")));
                    if (cancelled.on_cancel) {
                        cancelled.on_cancel();
                    }
                    failed_.fetch_add(1, std::memory_order_relaxed);
                    ++drain_cancelled;
                    record_recent(std::move(cancelled.handle));
                }
                break;
            }
            if (queue_.empty()) {
                break;
            }
            task = std::move(queue_.front());
            queue_.pop_front();
            got_task = true;
        }
        if (!got_task) {
            break;
        }
        active_.fetch_add(1, std::memory_order_relaxed);
        task.handle.mark_running();
        {
            std::scoped_lock guard(running_mtx_);
            running_.emplace(task.handle.id(), task.handle);
        }
        const auto t0 = std::chrono::steady_clock::now();
        const auto queue_wait_ns =
            std::chrono::duration_cast<std::chrono::nanoseconds>(t0 - task.enqueued_at).count();
        bool task_failed = false;
        try {
            task.work(tok);
            task.handle.mark_completed();
            completed_.fetch_add(1, std::memory_order_relaxed);
        } catch (...) {
            task.handle.mark_failed(std::current_exception());
            failed_.fetch_add(1, std::memory_order_relaxed);
            task_failed = true;
        }
        const auto t1 = std::chrono::steady_clock::now();
        const auto elap = std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count();
        total_exec_ns_.fetch_add(static_cast<std::uint64_t>(elap), std::memory_order_relaxed);
        active_.fetch_sub(1, std::memory_order_relaxed);
        {
            std::scoped_lock guard(running_mtx_);
            running_.erase(task.handle.id());
        }
        record_recent(std::move(task.handle));
        // Emit metrics only after record_recent so the small window callers
        // observe right after Future::get() (recent-task snapshot) is not
        // widened by per-task metric work.
        metrics::pool_task_queue_wait_seconds()
            .labels(pool_labels_)
            .observe(static_cast<double>(queue_wait_ns) / 1e9);
        metrics::pool_task_execution_seconds()
            .labels(pool_labels_)
            .observe(static_cast<double>(elap) / 1e9);
        if (task_failed) {
            metrics::pool_tasks_failed().labels(pool_labels_).inc();
        } else {
            metrics::pool_tasks_completed().labels(pool_labels_).inc();
        }
    }
 
    // Emit drain-time cancellations outside the queue lock (per metrics policy).
    if (drain_cancelled != 0) {
        metrics::pool_tasks_cancelled().labels(pool_labels_).inc(drain_cancelled);
    }
 
    (void)lg_pool;
}
 
inline void ThreadPool::record_recent(TaskHandle h) {
    std::scoped_lock guard(recent_mtx_);
    if (recent_.size() >= opts_.recent_tasks_capacity) {
        recent_.pop_front();
    }
    recent_.push_back(std::move(h));
}
 
inline void ThreadPool::shutdown() {
    if (auto expected = PoolState::Running; !state_.compare_exchange_strong(
            expected, PoolState::ShuttingDown, std::memory_order_acq_rel)) {
        return;  // already shutting down or terminated
    }
    {
        std::unique_lock lk(queue_mtx_);
        queue_cv_.notify_all();
    }
    if (const auto& lg = log::pool()) {
        lg->info("pool '{}' shutdown requested", opts_.name);
    }
}
 
inline void ThreadPool::shutdown_now() {
    auto current = state_.load(std::memory_order_acquire);
    while (current == PoolState::Running || current == PoolState::ShuttingDown) {
        if (state_.compare_exchange_weak(
                current, PoolState::ShutdownNow, std::memory_order_acq_rel)) {
            break;
        }
    }
    if (state_.load(std::memory_order_acquire) != PoolState::ShutdownNow) {
        return;
    }
    {
        std::scoped_lock guard(workers_mtx_);
        for (auto& w : workers_) {
            if (w) {
                w->request_stop();
            }
        }
    }
    {
        std::unique_lock lk(queue_mtx_);
        queue_cv_.notify_all();
    }
    if (const auto& lg = log::pool()) {
        lg->info("pool '{}' shutdown_now requested", opts_.name);
    }
}
 
inline void ThreadPool::join() {
    if (state_.load(std::memory_order_acquire) == PoolState::Running) {
        throw std::logic_error("ThreadPool::join() called without prior shutdown()");
    }
    std::scoped_lock guard(workers_mtx_);
    for (const auto& w : workers_) {
        if (w && w->joinable()) {
            w->join();
        }
    }
    state_.store(PoolState::Terminated, std::memory_order_release);
}
 
inline bool ThreadPool::is_shutting_down() const noexcept {
    const auto s = state_.load(std::memory_order_acquire);
    return s == PoolState::ShuttingDown || s == PoolState::ShutdownNow ||
           s == PoolState::Terminated;
}
 
inline bool ThreadPool::is_terminated() const noexcept {
    return state_.load(std::memory_order_acquire) == PoolState::Terminated;
}
 
inline std::size_t ThreadPool::worker_count() const noexcept {
    std::scoped_lock guard(workers_mtx_);
    return workers_.size();
}
 
inline ThreadPoolStats ThreadPool::stats() const {
    ThreadPoolStats s;
    s.pool_id = pool_id_;
    s.name = opts_.name;
    s.state = state_.load(std::memory_order_acquire);
    s.core_workers = opts_.min_workers;
    s.max_workers = opts_.max_workers;
    s.active = active_.load(std::memory_order_relaxed);
    s.idle = idle_.load(std::memory_order_relaxed);
    s.completed = completed_.load(std::memory_order_relaxed);
    s.failed = failed_.load(std::memory_order_relaxed);
    s.scale_ups = scale_ups_.load(std::memory_order_relaxed);
    s.scale_downs = scale_downs_.load(std::memory_order_relaxed);
    {
        std::scoped_lock guard(workers_mtx_);
        s.workers = workers_.size();
    }
    {
        std::scoped_lock lk(queue_mtx_);
        s.queued = queue_.size();
    }
    if (const auto done = s.completed + s.failed; done > 0) {
        const auto total = total_exec_ns_.load(std::memory_order_relaxed);
        s.avg_execution_duration = std::chrono::nanoseconds{total / done};
    }
    return s;
}
 
inline std::vector<ThreadSnapshot> ThreadPool::snapshot_workers() const {
    std::scoped_lock guard(workers_mtx_);
    std::vector<ThreadSnapshot> out;
    out.reserve(workers_.size());
    for (const auto& w : workers_) {
        if (w) {
            out.push_back(w->snapshot());
        }
    }
    return out;
}
 
inline std::vector<TaskSnapshot> ThreadPool::snapshot_recent_tasks() const {
    std::scoped_lock guard(recent_mtx_);
    std::vector<TaskSnapshot> out;
    out.reserve(recent_.size());
    for (const auto& h : recent_) {
        out.push_back(h.snapshot());
    }
    return out;
}
 
inline std::vector<TaskSnapshot> ThreadPool::snapshot_queued_tasks() const {
    std::scoped_lock lk(queue_mtx_);
    std::vector<TaskSnapshot> out;
    out.reserve(queue_.size());
    for (const auto& q : queue_) {
        out.push_back(q.handle.snapshot());
    }
    return out;
}
 
inline int ThreadPool::scale_tick(const std::chrono::steady_clock::time_point now) {
    if (state_.load(std::memory_order_acquire) != PoolState::Running) {
        return 0;
    }
 
    // Reap workers whose ManagedThread has exited (state Completed/Failed).
    int reaped = 0;
    {
        std::scoped_lock guard(workers_mtx_);
        for (std::size_t i = 0; i < workers_.size();) {
            if (const auto st = workers_[i]->control_block()->state.load(std::memory_order_acquire);
                st == ThreadState::Completed || st == ThreadState::Failed) {
                workers_[i]->join();
                workers_.erase(workers_.begin() + static_cast<std::ptrdiff_t>(i));
                worker_cbs_.erase(worker_cbs_.begin() + static_cast<std::ptrdiff_t>(i));
                ++reaped;
            } else {
                ++i;
            }
        }
    }
 
    int net = -reaped;
 
    // Scale-up decision.
    bool scale_up = false;
    std::size_t queue_size = 0;
    std::optional<std::chrono::steady_clock::time_point> oldest_enqueued;
    {
        std::scoped_lock lk(queue_mtx_);
        queue_size = queue_.size();
        if (!queue_.empty()) {
            oldest_enqueued = queue_.front().enqueued_at;
        }
    }
    const std::size_t idle_now = idle_.load(std::memory_order_relaxed);
    std::size_t workers_now = 0;
    {
        std::scoped_lock guard(workers_mtx_);
        workers_now = workers_.size();
    }
 
    if (workers_now < opts_.max_workers && idle_now == 0) {
        const bool queue_pressure = queue_size > opts_.scale_up_when_queue_exceeds;
        const bool wait_pressure =
            oldest_enqueued.has_value() && (now - *oldest_enqueued) >= opts_.scale_up_wait;
        scale_up = queue_pressure || wait_pressure;
    }
 
    if (scale_up) {
        std::scoped_lock guard(workers_mtx_);
        if (workers_.size() < opts_.max_workers) {
            spawn_worker(/*is_core=*/false);
            ++net;
            if (const auto& lg = log::pool()) {
                lg->info("pool '{}' scaled up to {} workers (idle=0, queued={})",
                         opts_.name,
                         workers_.size(),
                         queue_size);
            }
        }
    }
 
    return net;
}
 
inline std::vector<StuckTaskEvent>
ThreadPool::detect_stuck_tasks(std::chrono::steady_clock::time_point now) const {
    std::vector<StuckTaskEvent> out;
    std::scoped_lock guard(running_mtx_);
    for (const auto& h : running_ | std::views::values) {
        if (const auto snap = h.snapshot();
            snap.state == TaskState::Running && snap.started_at.has_value()) {
            if (const auto running_for = now - *snap.started_at;
                running_for >= opts_.stuck_task_threshold) {
                StuckTaskEvent ev;
                ev.task = snap;
                ev.running_for = std::chrono::duration_cast<std::chrono::nanoseconds>(running_for);
                ev.pool_id = pool_id_;
                out.push_back(std::move(ev));
            }
        }
    }
    return out;
}
 
// ---------------------------------------------------------------------------
// SingleThreadExecutor — convenience wrapper around a ThreadPool with
// min_workers == max_workers == 1. Owns its pool; tasks run FIFO on a single
// thread.
// ---------------------------------------------------------------------------
 
class SingleThreadExecutor final : public IExecutor {
public:
    explicit SingleThreadExecutor(std::string nm = "tm::single") {
        ThreadPoolOptions o;
        o.name = std::move(nm);
        o.min_workers = 1;
        o.max_workers = 1;
        pool_ = std::make_unique<ThreadPool>(std::move(o));
    }
 
    void execute(std::function<void()> task) override {
        pool_->execute(std::move(task));
    }
    [[nodiscard]] std::string_view name() const noexcept override {
        return pool_->name();
    }
 
    template <class Fn>
        requires std::invocable<Fn>
    auto submit(Fn&& f) -> Future<std::invoke_result_t<Fn>> {
        return pool_->submit(std::forward<Fn>(f));
    }
 
    [[nodiscard]] ThreadPool& pool() const noexcept {
        return *pool_;
    }
 
    [[nodiscard]] static const comms::DisplayInfo& display_info() {
        static const comms::DisplayInfo info{
            .name = "SingleThreadExecutor",
            .description = "Executor owning a thread pool of exactly one worker (FIFO).",
            .icon = comms::Icon::from("mdi:speedometer-slow"),
        };
        return info;
    }
 
private:
    std::unique_ptr<ThreadPool> pool_;
};
 
}  // namespace threadman