thread_pool.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
 
 // end of thread_pools
 
#include <kcenon/thread/utils/formatter.h>
#include <kcenon/thread/utils/formatter_macros.h>
#include <kcenon/thread/core/job_queue.h>
#include <kcenon/thread/core/thread_worker.h>
#include <kcenon/thread/core/cancellation_token.h>
#include <kcenon/thread/utils/convert_string.h>
#include <kcenon/thread/forward.h>
#include <kcenon/thread/interfaces/thread_context.h>
#include <kcenon/thread/interfaces/pool_queue_adapter.h>
#include <kcenon/thread/metrics/thread_pool_metrics.h>
#include <kcenon/thread/metrics/enhanced_metrics.h>
#include <kcenon/thread/metrics/metrics_service.h>
#include <kcenon/thread/pool_policies/pool_policy.h>
#include <kcenon/thread/core/submit_options.h>
 
// Include unified feature flags from common_system if available
#if __has_include(<kcenon/common/config/feature_flags.h>)
#include <kcenon/common/config/feature_flags.h>
#endif
 
// Common system unified interfaces
// KCENON_HAS_COMMON_EXECUTOR is defined by CMake when common_system is available
// Fallback to 0 if not defined (standalone build without common_system)
#ifndef KCENON_HAS_COMMON_EXECUTOR
#define KCENON_HAS_COMMON_EXECUTOR 0
#endif
 
#if KCENON_HAS_COMMON_EXECUTOR
#include <kcenon/common/interfaces/executor_interface.h>
#endif
 
#include "config.h"
 
#include <tuple>
#include <string>
#include <memory>
#include <mutex>
#include <vector>
#include <chrono>
#include <optional>
#include <future>
#include <type_traits>
 
// Forward declarations for thread_pool-specific types
// Core forward declarations are in <kcenon/thread/forward.h>
namespace kcenon::thread::diagnostics {
    class thread_pool_diagnostics;
    struct thread_info;
}
 
namespace kcenon::thread {
    // Type aliases for common policy_queue configurations
    // Uses forward declarations from forward.h
    using standard_queue = policy_queue<
        policies::mutex_sync_policy,
        policies::unbounded_policy,
        policies::overflow_reject_policy
    >;
 
    using policy_lockfree_queue = policy_queue<
        policies::lockfree_sync_policy,
        policies::unbounded_policy,
        policies::overflow_reject_policy
    >;
}
 
namespace kcenon::thread
{
    // Support both old (namespace common) and new (namespace kcenon::common) versions
    // When inside namespace kcenon::thread, 'common' resolves to kcenon::common
#if KCENON_HAS_COMMON_EXECUTOR
    namespace common_ns = common;
#endif
 
    class thread_pool : public std::enable_shared_from_this<thread_pool>
    {
        // Allow numa_thread_pool to access protected/private members for NUMA-specific functionality
        friend class numa_thread_pool;
        // Allow autoscaler to access remove_workers_internal for scaling operations
        friend class autoscaler;
 
    public:
        thread_pool(const std::string& thread_title = "thread_pool",
                   const thread_context& context = thread_context());
 
        thread_pool(const std::string& thread_title,
                   std::shared_ptr<job_queue> custom_queue,
                   const thread_context& context = thread_context());
 
        thread_pool(const std::string& thread_title,
                   std::unique_ptr<pool_queue_adapter_interface> queue_adapter,
                   const thread_context& context = thread_context());
 
        virtual ~thread_pool(void);
 
        [[nodiscard]] auto get_ptr(void) -> std::shared_ptr<thread_pool>;
 
        auto start(void) -> common::VoidResult;
 
        [[nodiscard]] auto get_job_queue(void) -> std::shared_ptr<job_queue>;
 
        [[nodiscard]] const metrics::ThreadPoolMetrics& metrics() const noexcept;
 
        void reset_metrics();
 
        void set_enhanced_metrics_enabled(bool enabled);
 
        [[nodiscard]] bool is_enhanced_metrics_enabled() const;
 
        [[nodiscard]] const metrics::EnhancedThreadPoolMetrics& enhanced_metrics() const;
 
        [[nodiscard]] metrics::EnhancedSnapshot enhanced_metrics_snapshot() const;
 
        auto enqueue(std::unique_ptr<job>&& job) -> common::VoidResult;
 
        auto enqueue_batch(std::vector<std::unique_ptr<job>>&& jobs) -> common::VoidResult;
 
        auto enqueue(std::unique_ptr<thread_worker>&& worker) -> common::VoidResult;
 
        auto enqueue_batch(std::vector<std::unique_ptr<thread_worker>>&& workers)
            -> common::VoidResult;
 
        auto stop(const bool& immediately_stop = false) -> common::VoidResult;
 
        [[nodiscard]] auto to_string(void) const -> std::string;
 
        [[nodiscard]] std::uint32_t get_pool_instance_id() const;
 
        void report_metrics();
 
        [[nodiscard]] std::size_t get_idle_worker_count() const;
 
        [[nodiscard]] auto get_context(void) const -> const thread_context&;
 
        // ============================================================================
        // Unified Submit API
        // ============================================================================
        // These methods provide a unified interface for job submission with
        // configurable behavior through submit_options.
 
        template<typename F, typename R = std::invoke_result_t<std::decay_t<F>>>
        [[nodiscard]] auto submit(F&& callable, const submit_options& opts = {})
            -> std::future<R>;
 
        template<typename F, typename R = std::invoke_result_t<std::decay_t<F>>>
        [[nodiscard]] auto submit(std::vector<F>&& callables, const submit_options& opts = {})
            -> std::vector<std::future<R>>;
 
        template<typename F, typename R = std::invoke_result_t<std::decay_t<F>>>
        [[nodiscard]] auto submit_wait_all(std::vector<F>&& callables,
                                           const submit_options& opts = {})
            -> std::vector<R>;
 
        template<typename F, typename R = std::invoke_result_t<std::decay_t<F>>>
        [[nodiscard]] auto submit_wait_any(std::vector<F>&& callables,
                                           const submit_options& opts = {})
            -> common::Result<R>;
 
        auto is_running() const -> bool;
 
        auto get_pending_task_count() const -> std::size_t;
 
        auto check_worker_health(bool restart_failed = true) -> std::size_t;
 
        auto get_active_worker_count() const -> std::size_t;
 
        // =========================================================================
        // Pool Policies
        // =========================================================================
 
        void add_policy(std::unique_ptr<pool_policy> policy);
 
        [[nodiscard]] auto get_policies() const -> const std::vector<std::unique_ptr<pool_policy>>&;
 
        template<typename T = pool_policy>
        [[nodiscard]] auto find_policy(const std::string& name) -> T*;
 
        auto remove_policy(const std::string& name) -> bool;
 
        // =========================================================================
        // Diagnostics
        // =========================================================================
 
        [[nodiscard]] auto diagnostics() -> diagnostics::thread_pool_diagnostics&;
 
        [[nodiscard]] auto diagnostics() const -> const diagnostics::thread_pool_diagnostics&;
 
        [[nodiscard]] auto collect_worker_diagnostics() const
            -> std::vector<diagnostics::thread_info>;
 
    private:
        // Allow diagnostics to access internal state
        friend class diagnostics::thread_pool_diagnostics;
        static std::atomic<std::uint32_t> next_pool_instance_id_;
 
        std::string thread_title_;
 
        std::uint32_t pool_instance_id_{0};
 
        std::atomic<bool> start_pool_;
 
        std::shared_ptr<job_queue> job_queue_;
 
        std::unique_ptr<pool_queue_adapter_interface> queue_adapter_;
 
        std::vector<std::unique_ptr<thread_worker>> workers_;
 
        mutable std::mutex workers_mutex_;
 
        thread_context context_;
 
        cancellation_token pool_cancellation_token_;
 
        auto stop_unsafe() noexcept -> common::VoidResult;
 
        std::shared_ptr<metrics::metrics_service> metrics_service_;
 
 
        mutable std::unique_ptr<diagnostics::thread_pool_diagnostics> diagnostics_;
 
        mutable std::once_flag diagnostics_init_flag_;
 
        std::vector<std::unique_ptr<pool_policy>> policies_;
 
        mutable std::mutex policies_mutex_;
 
        [[nodiscard]] auto remove_workers_internal(std::size_t count, std::size_t min_workers = 1)
            -> common::VoidResult;
    };
} // namespace kcenon::thread
 
// ----------------------------------------------------------------------------
// Template method implementations for thread_pool
// ----------------------------------------------------------------------------
// Separated into thread_pool_impl.h for improved compilation times.
// See thread_pool_impl.h for submit_async, submit_batch_async, submit_all, submit_any.
 
#include <kcenon/thread/core/thread_pool_impl.h>
 
// ----------------------------------------------------------------------------
// Formatter specializations for thread_pool
// ----------------------------------------------------------------------------
// Separated into thread_pool_fmt.h for improved compilation times.
// See thread_pool_fmt.h for std::formatter specializations.
// For unified formatter access, include <kcenon/thread/formatters.h> instead.
 
#define KCENON_THREAD_INTERNAL_INCLUDE
#include <kcenon/thread/core/thread_pool_fmt.h>
#undef KCENON_THREAD_INTERNAL_INCLUDE