kcenon::thread::thread_pool Class Reference

A thread pool for concurrent execution of jobs using multiple worker threads. More...

#include <thread_pool.h>

Inheritance diagram for kcenon::thread::thread_pool:

Collaboration diagram for kcenon::thread::thread_pool:

Public Member Functions
	thread_pool (const std::string &thread_title="thread_pool", const thread_context &context=thread_context())
	Constructs a new thread_pool instance.
	thread_pool (const std::string &thread_title, std::shared_ptr< job_queue > custom_queue, const thread_context &context=thread_context())
	Constructs a new thread_pool instance with a custom job queue.
	thread_pool (const std::string &thread_title, std::unique_ptr< pool_queue_adapter_interface > queue_adapter, const thread_context &context=thread_context())
	Constructs a new thread_pool instance with a policy_queue adapter.
virtual	~thread_pool (void)
	Virtual destructor. Cleans up resources used by the thread pool.
auto	get_ptr (void) -> std::shared_ptr< thread_pool >
	Retrieves a std::shared_ptr to this thread_pool instance.
auto	start (void) -> common::VoidResult
	Starts the thread pool and all associated workers.
auto	get_job_queue (void) -> std::shared_ptr< job_queue >
	Returns the shared job_queue used by this thread pool.
const metrics::ThreadPoolMetrics &	metrics () const noexcept
	Access aggregated runtime metrics (read-only reference).
void	reset_metrics ()
	Reset accumulated metrics.
void	set_enhanced_metrics_enabled (bool enabled)
	Enable or disable enhanced metrics collection.
bool	is_enhanced_metrics_enabled () const
	Check if enhanced metrics is enabled.
const metrics::EnhancedThreadPoolMetrics &	enhanced_metrics () const
	Access enhanced metrics (read-only reference).
metrics::EnhancedSnapshot	enhanced_metrics_snapshot () const
	Get enhanced metrics snapshot.
auto	enqueue (std::unique_ptr< job > &&job) -> common::VoidResult
	Enqueues a new job into the shared job_queue.
auto	enqueue_batch (std::vector< std::unique_ptr< job > > &&jobs) -> common::VoidResult
	Enqueues a batch of jobs into the shared job_queue.
auto	enqueue (std::unique_ptr< thread_worker > &&worker) -> common::VoidResult
	Adds a thread_worker to the thread pool for specialized or additional processing.
auto	enqueue_batch (std::vector< std::unique_ptr< thread_worker > > &&workers) -> common::VoidResult
	Adds a batch of thread_worker objects to the thread pool.
auto	stop (const bool &immediately_stop=false) -> common::VoidResult
	Stops the thread pool and all worker threads.
auto	to_string (void) const -> std::string
	Provides a string representation of this thread_pool.
std::uint32_t	get_pool_instance_id () const
	Get the pool instance id.
void	report_metrics ()
	Collect and report current thread pool metrics.
std::size_t	get_idle_worker_count () const
	Get the number of idle workers.
auto	get_context (void) const -> const thread_context &
	Gets the thread context for this pool.
template<typename F , typename R = std::invoke_result_t<std::decay_t<F>>>
auto	submit (F &&callable, const submit_options &opts={}) -> std::future< R >
template<typename F , typename R = std::invoke_result_t<std::decay_t<F>>>
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>>>
auto	submit_wait_all (std::vector< F > &&callables, const submit_options &opts={}) -> std::vector< R >
	Submit a batch and wait for all results.
template<typename F , typename R = std::invoke_result_t<std::decay_t<F>>>
auto	submit_wait_any (std::vector< F > &&callables, const submit_options &opts={}) -> common::Result< R >
	Submit a batch and return first completed result.
auto	is_running () const -> bool
	Check if the thread pool is currently running.
auto	get_pending_task_count () const -> std::size_t
	Get the number of pending tasks in the queue.
auto	check_worker_health (bool restart_failed=true) -> std::size_t
	Check health of all worker threads and restart failed workers.
auto	get_active_worker_count () const -> std::size_t
	Get the current number of active (running) workers.
void	add_policy (std::unique_ptr< pool_policy > policy)
	Add a policy to the pool.
auto	get_policies () const -> const std::vector< std::unique_ptr< pool_policy > > &
	Get all registered policies.
template<typename T = pool_policy>
auto	find_policy (const std::string &name) -> T *
	Find a policy by name.
auto	remove_policy (const std::string &name) -> bool
	Remove a policy by name.
auto	diagnostics () -> diagnostics::thread_pool_diagnostics &
	Get the diagnostics interface for this pool.
auto	diagnostics () const -> const diagnostics::thread_pool_diagnostics &
	Get the diagnostics interface for this pool (const version).
auto	collect_worker_diagnostics () const -> std::vector< diagnostics::thread_info >
	Collects diagnostics information from all workers.

Private Member Functions
auto	stop_unsafe () noexcept -> common::VoidResult
	Stops the thread pool without logging (for use during static destruction).
auto	remove_workers_internal (std::size_t count, std::size_t min_workers=1) -> common::VoidResult
	Internal method to remove workers from the pool.

Private Attributes
std::string	thread_title_
	A title or name for this thread pool, useful for identification and logging.
std::uint32_t	pool_instance_id_ {0}
	Unique instance ID for this pool (for multi-pool scenarios).
std::atomic< bool >	start_pool_
	Indicates whether the pool is currently running.
std::shared_ptr< job_queue >	job_queue_
	The shared job queue where jobs (job objects) are enqueued.
std::unique_ptr< pool_queue_adapter_interface >	queue_adapter_
	Queue adapter for unified access to different queue types.
std::vector< std::unique_ptr< thread_worker > >	workers_
	A collection of worker threads associated with this pool.
std::mutex	workers_mutex_
	Mutex protecting concurrent access to the workers_ vector.
thread_context	context_
	The thread context providing access to logging and monitoring services.
cancellation_token	pool_cancellation_token_
	Pool-level cancellation token.
std::shared_ptr< metrics::metrics_service >	metrics_service_
	Centralized metrics service for all pool and worker metrics.
std::unique_ptr< diagnostics::thread_pool_diagnostics >	diagnostics_
	Diagnostics interface for this pool.
std::once_flag	diagnostics_init_flag_
	Once flag for thread-safe lazy initialization of diagnostics_.
std::vector< std::unique_ptr< pool_policy > >	policies_
	Registered pool policies for extending thread pool behavior.
std::mutex	policies_mutex_
	Mutex protecting concurrent access to the policies_ vector.

Static Private Attributes
static std::atomic< std::uint32_t >	next_pool_instance_id_ {0}
	Static counter for generating unique pool instance IDs.

Friends
class	numa_thread_pool
class	autoscaler
class	diagnostics::thread_pool_diagnostics

Detailed Description

A thread pool for concurrent execution of jobs using multiple worker threads.

The thread_pool class manages a group of worker threads that process jobs from a shared job_queue. This implementation provides:

• Efficient reuse of threads to reduce thread creation/destruction overhead
• Controlled concurrency through a fixed or dynamic thread count
• A simple interface for submitting jobs of various types
• Graceful handling of thread startup, execution, and shutdown

The thread pool is designed for scenarios where many short-lived tasks need to be executed asynchronously without creating a new thread for each task.

Design Principles

• Worker Thread Model: Each worker runs in its own thread, processing jobs from the shared queue.
• Shared Job Queue: A single, thread-safe queue holds all pending jobs.
• Job-Based Work Units: Jobs encapsulate work to be executed.
• Non-Blocking Submission: Adding jobs to the pool never blocks the caller thread.
• Cooperative Shutdown: Workers can complete current jobs before stopping.

Thread Safety

All public methods of this class are thread-safe and can be called from any thread. The underlying job_queue is also thread-safe, allowing multiple workers to dequeue jobs concurrently.

Performance Considerations

• The number of worker threads should typically be close to the number of available CPU cores for CPU-bound tasks.
• For I/O-bound tasks, more threads may be beneficial to maximize throughput while some threads are blocked on I/O.
• Very large thread pools (significantly more threads than cores) may degrade performance due to context switching overhead.

IExecutor Interface

For IExecutor compatibility, use thread_pool_executor_adapter:

#include <kcenon/thread/adapters/common_executor_adapter.h>
 
auto pool = std::make_shared<thread_pool>("my_pool");
auto executor = std::make_shared<adapters::thread_pool_executor_adapter>(pool);

See also

thread_worker The worker thread class used by the pool

job_queue The shared queue for storing pending jobs

adapters::thread_pool_executor_adapter For IExecutor compatibility

typed_kcenon::thread::typed_thread_pool For a priority-based version

Examples

crash_protection/main.cpp, thread_pool_sample.cpp, typed_thread_pool_sample.cpp, and typed_thread_pool_sample_2.cpp.

Definition at line 181 of file thread_pool.h.

Constructor & Destructor Documentation

thread_pool() [1/3]

kcenon::thread::thread_pool::thread_pool	(	const std::string &	thread_title = "thread_pool",
		const thread_context &	context = thread_context() )

Constructs a new thread_pool instance.

Constructs a thread pool with adaptive job queue.

Parameters

thread_title	An optional title or identifier for the thread pool (defaults to "thread_pool").
context	Optional thread context for logging and monitoring (defaults to empty context).

This title can be used for logging or debugging purposes. The context provides access to logging and monitoring services.

Implementation details:

• Initializes with provided thread title for identification
• Creates adaptive job queue that automatically optimizes based on contention
• Pool starts in stopped state (start_pool_ = false)
• No workers are initially assigned (workers_ is empty)
• Stores thread context for logging and monitoring
• Creates pool-level cancellation token for hierarchical cancellation

Adaptive Queue Strategy:

• ADAPTIVE mode automatically switches between mutex and lock-free implementations
• Provides optimal performance across different contention levels
• Eliminates need for manual queue strategy selection

Parameters

thread_title	Descriptive name for this thread pool instance
context	Thread context providing logging and monitoring services

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 51 of file thread_pool.cpp.

    : thread_title_(thread_title)
    , pool_instance_id_(next_pool_instance_id_.fetch_add(1))
    , start_pool_(false)
    , job_queue_(std::make_shared<kcenon::thread::job_queue>())
    , context_(context)
    , pool_cancellation_token_(cancellation_token::create())
    , metrics_service_(std::make_shared<metrics::metrics_service>()) {
    // Report initial pool registration if monitoring is available
    if (context_.monitoring()) {
        common::interfaces::thread_pool_metrics initial_metrics(thread_title_, pool_instance_id_);
        initial_metrics.worker_threads.value = 0;
        context_.update_thread_pool_metrics(thread_title_, pool_instance_id_, initial_metrics);
    }
}

References context_, and kcenon::thread::thread_context::monitoring().

Here is the call graph for this function:

thread_pool() [2/3]

kcenon::thread::thread_pool::thread_pool	(	const std::string &	thread_title,
		std::shared_ptr< job_queue >	custom_queue,
		const thread_context &	context = thread_context() )

Constructs a new thread_pool instance with a custom job queue.

Constructs a thread pool with a custom job queue.

Parameters

thread_title	A title or identifier for the thread pool.
custom_queue	A custom job queue implementation (e.g., backpressure_job_queue).
context	Optional thread context for logging and monitoring (defaults to empty context).

This constructor allows injecting a specialized job queue such as backpressure_job_queue for rate limiting and flow control.

Implementation details:

• Initializes with provided thread title for identification
• Uses the provided custom job queue (e.g., backpressure_job_queue)
• Pool starts in stopped state (start_pool_ = false)
• No workers are initially assigned (workers_ is empty)
• Stores thread context for logging and monitoring
• Creates pool-level cancellation token for hierarchical cancellation

Parameters

thread_title	Descriptive name for this thread pool instance
custom_queue	Custom job queue implementation
context	Thread context providing logging and monitoring services

Definition at line 82 of file thread_pool.cpp.

    : thread_title_(thread_title)
    , pool_instance_id_(next_pool_instance_id_.fetch_add(1))
    , start_pool_(false)
    , job_queue_(std::move(custom_queue))
    , context_(context)
    , pool_cancellation_token_(cancellation_token::create())
    , metrics_service_(std::make_shared<metrics::metrics_service>()) {
    // Report initial pool registration if monitoring is available
    if (context_.monitoring()) {
        common::interfaces::thread_pool_metrics initial_metrics(thread_title_, pool_instance_id_);
        initial_metrics.worker_threads.value = 0;
        context_.update_thread_pool_metrics(thread_title_, pool_instance_id_, initial_metrics);
    }
}

References context_, and kcenon::thread::thread_context::monitoring().

Here is the call graph for this function:

thread_pool() [3/3]

kcenon::thread::thread_pool::thread_pool	(	const std::string &	thread_title,
		std::unique_ptr< pool_queue_adapter_interface >	queue_adapter,
		const thread_context &	context = thread_context() )

Constructs a new thread_pool instance with a policy_queue adapter.

Constructs a thread pool with a policy_queue adapter.

Parameters

thread_title	A title or identifier for the thread pool.
queue_adapter	A queue adapter wrapping a policy_queue.
context	Optional thread context for logging and monitoring (defaults to empty context).

This constructor allows using the new policy-based queue system with thread_pool. Use adapters from <kcenon/thread/adapters/policy_queue_adapter.h>.

Example

#include <kcenon/thread/adapters/policy_queue_adapter.h>
 
// Create thread_pool with standard_queue (mutex-based, unbounded)
auto pool = std::make_shared<thread_pool>(
    "my_pool",
    make_standard_queue_adapter());
 
// Or with lock-free queue
auto fast_pool = std::make_shared<thread_pool>(
    "fast_pool",
    make_lockfree_queue_adapter());

Implementation details:

• Initializes with provided thread title for identification
• Uses the provided queue adapter for queue operations
• If adapter wraps a job_queue, extracts it for backward compatibility
• Pool starts in stopped state (start_pool_ = false)
• No workers are initially assigned (workers_ is empty)
• Stores thread context for logging and monitoring
• Creates pool-level cancellation token for hierarchical cancellation

Parameters

thread_title	Descriptive name for this thread pool instance
queue_adapter	Queue adapter wrapping job_queue or policy_queue
context	Thread context providing logging and monitoring services

Definition at line 116 of file thread_pool.cpp.

    : thread_title_(thread_title)
    , pool_instance_id_(next_pool_instance_id_.fetch_add(1))
    , start_pool_(false)
    , job_queue_(queue_adapter ? queue_adapter->get_job_queue() : nullptr)
    , queue_adapter_(std::move(queue_adapter))
    , context_(context)
    , pool_cancellation_token_(cancellation_token::create())
    , metrics_service_(std::make_shared<metrics::metrics_service>()) {
    // Report initial pool registration if monitoring is available
    if (context_.monitoring()) {
        common::interfaces::thread_pool_metrics initial_metrics(thread_title_, pool_instance_id_);
        initial_metrics.worker_threads.value = 0;
        context_.update_thread_pool_metrics(thread_title_, pool_instance_id_, initial_metrics);
    }
}

References context_, and kcenon::thread::thread_context::monitoring().

Here is the call graph for this function:

~thread_pool()

kcenon::thread::thread_pool::~thread_pool ( void )

virtual

Virtual destructor. Cleans up resources used by the thread pool.

Destroys the thread pool, ensuring all workers are stopped.

If the pool is still running, this typically calls stop() internally to ensure all worker threads are properly shut down.

Implementation details:

• Checks if static destruction is in progress to avoid SDOF
• Uses stop_unsafe() during static destruction (no logging)
• Uses regular stop() during normal destruction
• Workers will complete current jobs before terminating
• Prevents resource leaks from running threads

See also

stop_unsafe() for the logging-free shutdown variant

thread_logger::is_shutting_down() for shutdown detection

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 148 of file thread_pool.cpp.

                          {
    // Check if we're in static destruction phase
    // During static destruction, logger/monitoring singletons may already be destroyed
    if (thread_logger::is_shutting_down()) {
        // Minimal cleanup without logging to avoid SDOF
        stop_unsafe();
    } else {
        stop();
    }
}

References kcenon::thread::thread_logger::is_shutting_down(), stop(), and stop_unsafe().

Here is the call graph for this function:

Member Function Documentation

add_policy()

void kcenon::thread::thread_pool::add_policy

(

std::unique_ptr< pool_policy >

policy

)

Add a policy to the pool.

Parameters

policy

Unique pointer to the policy to add.

Policies provide a way to extend thread pool behavior without modifying the thread_pool class. Multiple policies can be added and they will be called in order of addition.

See also

pool_policy

circuit_breaker_policy

Example

auto pool = std::make_shared<thread_pool>("my_pool");
 
// Add circuit breaker protection
pool->add_policy(std::make_unique<circuit_breaker_policy>(config));
 
pool->start();

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 834 of file thread_pool.cpp.

                                                              {
    if (!policy) {
        return;
    }
 
    std::scoped_lock<std::mutex> lock(policies_mutex_);
    policies_.push_back(std::move(policy));
}

check_worker_health()

auto kcenon::thread::thread_pool::check_worker_health

(

bool

restart_failed = true

)

-> std::size_t

Check health of all worker threads and restart failed workers.

This method performs health monitoring on all worker threads:

• Detects workers that have stopped unexpectedly (consecutive failures)
• Removes dead workers from the pool
• Optionally restarts failed workers to maintain pool capacity

Use Cases:

• Periodic health checks (e.g., from a watchdog thread)
• Recovery from worker failures in long-running processes
• Maintaining consistent thread pool capacity

Thread Safety:

• Thread-safe: can be called from any thread
• Acquires workers_mutex_ for safe access to workers vector

Parameters

restart_failed

If true, creates new workers to replace failed ones

Returns

Number of workers that were detected as failed/unhealthy

Example:

// Periodic health check with auto-restart
auto failed_count = pool.check_worker_health(true);
if (failed_count > 0) {
    LOG_WARNING("Restarted {} failed workers", failed_count);
}

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 706 of file thread_pool.cpp.

                                                                      {
    std::scoped_lock<std::mutex> lock(workers_mutex_);
 
    std::size_t failed_count = 0;
 
    // Remove dead workers using erase-remove idiom
    auto remove_iter =
        std::remove_if(workers_.begin(), workers_.end(),
                       [&failed_count](const std::unique_ptr<thread_worker>& worker) {
                           if (!worker || !worker->is_running()) {
                               ++failed_count;
                               return true;  // Remove this worker
                           }
                           return false;  // Keep this worker
                       });
 
    workers_.erase(remove_iter, workers_.end());
 
    // Restart workers if requested and pool is running
    if (restart_failed && failed_count > 0 && is_running()) {
        // Get diagnostics pointer and config for new workers
        auto* diag = &diagnostics();
        const auto sample_rate = diag->get_config().event_sample_rate;
 
        // Create new workers to replace failed ones
        for (std::size_t i = 0; i < failed_count; ++i) {
            // Create worker with default settings and context
            auto worker = std::make_unique<thread_worker>(true, context_);
 
            // Set job queue and diagnostics
            worker->set_job_queue(job_queue_);
            worker->set_metrics(metrics_service_->get_basic_metrics());
            worker->set_diagnostics(diag);
            worker->set_diagnostics_sample_rate(sample_rate);
 
            // Start the new worker
            auto start_result = worker->start();
            if (start_result.is_err()) {
                // Failed to start, skip this worker
                continue;
            }
 
            workers_.push_back(std::move(worker));
        }
    }
 
    return failed_count;
}

collect_worker_diagnostics()

auto kcenon::thread::thread_pool::collect_worker_diagnostics ( ) const -> std::vector<diagnostics::thread_info>

nodiscard

Collects diagnostics information from all workers.

Returns

Vector of thread_info for each worker.

Thread Safety:

• Acquires workers_mutex_ for safe access
• Returns snapshot of current worker states

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 783 of file thread_pool.cpp.

                                         {
    std::scoped_lock<std::mutex> lock(workers_mutex_);
 
    std::vector<diagnostics::thread_info> result;
    result.reserve(workers_.size());
 
    for (std::size_t i = 0; i < workers_.size(); ++i) {
        const auto& worker = workers_[i];
        if (!worker) {
            continue;
        }
 
        diagnostics::thread_info info;
        info.thread_id = worker->get_thread_id();
        info.thread_name = "Worker-" + std::to_string(i);
        info.worker_id = worker->get_worker_id();
 
        // Determine worker state
        if (!worker->is_running()) {
            info.state = diagnostics::worker_state::stopped;
        } else if (worker->is_idle()) {
            info.state = diagnostics::worker_state::idle;
        } else {
            info.state = diagnostics::worker_state::active;
        }
 
        info.state_since = worker->get_state_since();
 
        // Get current job info if active
        info.current_job = worker->get_current_job_info();
 
        // Get statistics
        info.jobs_completed = worker->get_jobs_completed();
        info.jobs_failed = worker->get_jobs_failed();
        info.total_busy_time = worker->get_total_busy_time();
        info.total_idle_time = worker->get_total_idle_time();
 
        // Calculate utilization
        info.update_utilization();
 
        result.push_back(std::move(info));
    }
 
    return result;
}

References kcenon::thread::info.

Referenced by kcenon::thread::diagnostics::thread_pool_diagnostics::detect_bottlenecks(), and kcenon::thread::diagnostics::thread_pool_diagnostics::dump_thread_states().

Here is the caller graph for this function:

diagnostics() [1/2]

auto kcenon::thread::thread_pool::diagnostics ( ) -> diagnostics::thread_pool_diagnostics&

nodiscard

Get the diagnostics interface for this pool.

Returns

Reference to the diagnostics object.

The diagnostics interface provides:

• Thread dump capabilities
• Job inspection
• Bottleneck detection
• Health check integration
• Event tracing

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 768 of file thread_pool.cpp.

                                                                     {
    std::call_once(diagnostics_init_flag_, [this]() {
        diagnostics_ = std::make_unique<diagnostics::thread_pool_diagnostics>(*this);
    });
    return *diagnostics_;
}

diagnostics() [2/2]

auto kcenon::thread::thread_pool::diagnostics ( ) const -> const diagnostics::thread_pool_diagnostics&

nodiscard

Get the diagnostics interface for this pool (const version).

Returns

Const reference to the diagnostics object.

Definition at line 775 of file thread_pool.cpp.

                                                                                 {
    std::call_once(diagnostics_init_flag_, [this]() {
        diagnostics_ = std::make_unique<diagnostics::thread_pool_diagnostics>(
            const_cast<thread_pool&>(*this));
    });
    return *diagnostics_;
}

enhanced_metrics()

const metrics::EnhancedThreadPoolMetrics & kcenon::thread::thread_pool::enhanced_metrics ( ) const

nodiscard

Access enhanced metrics (read-only reference).

Returns

Reference to enhanced metrics with histograms and percentiles.

Exceptions

std::runtime_error

if enhanced metrics is not enabled.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 294 of file thread_pool.cpp.

                                                                            {
    return metrics_service_->enhanced_metrics();
}

References metrics_service_.

enhanced_metrics_snapshot()

metrics::EnhancedSnapshot kcenon::thread::thread_pool::enhanced_metrics_snapshot ( ) const

nodiscard

Get enhanced metrics snapshot.

Returns

EnhancedSnapshot with all current metric values.

Returns empty snapshot if enhanced metrics is not enabled.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 298 of file thread_pool.cpp.

                                                                     {
    return metrics_service_->enhanced_snapshot();
}

References metrics_service_.

Referenced by kcenon::thread::autoscaler::collect_metrics().

Here is the caller graph for this function:

enqueue() [1/2]

auto kcenon::thread::thread_pool::enqueue

(

std::unique_ptr< job > &&

job

)

-> common::VoidResult

Enqueues a new job into the shared job_queue.

Adds a single job to the thread pool for processing.

Parameters

job	A std::unique_ptr<job> representing the work to be done.

Returns

common::VoidResult containing an error on failure, or success value on success.

Implementation details:

• Validates job pointer before submission
• Validates queue availability
• Delegates to adaptive job queue for optimal scheduling
• Job will be processed by next available worker thread

Queue Behavior:

• Adaptive queue automatically selects best strategy (mutex/lock-free)
• Jobs are processed in FIFO order within the selected strategy
• Workers are notified when jobs become available

Thread Safety:

• Safe to call from multiple threads simultaneously
• Adaptive queue handles contention efficiently

Parameters

job	Unique pointer to job (ownership transferred)

Returns

std::nullopt on success, error message on failure

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h, and crash_protection/main.cpp.

Definition at line 323 of file thread_pool.cpp.

                                                                      {
    // Validate inputs
    if (job == nullptr) {
        return common::error_info{static_cast<int>(error_code::invalid_argument), "job is null", "thread_system"};
    }
 
    // Check queue availability and stopped state
    // Supports both queue_adapter_ (policy_queue) and job_queue_ (legacy)
    if (queue_adapter_) {
        if (queue_adapter_->is_stopped()) {
            return common::error_info{static_cast<int>(error_code::queue_stopped), "thread pool is stopped", "thread_system"};
        }
    } else if (job_queue_ == nullptr) {
        return common::error_info{static_cast<int>(error_code::resource_allocation_failed), "job queue is null", "thread_system"};
    } else if (job_queue_->is_stopped()) {
        return common::error_info{static_cast<int>(error_code::queue_stopped), "thread pool is stopped", "thread_system"};
    }
 
    // Record metrics and enqueue job
    metrics_service_->record_submission();
 
    auto start_time = std::chrono::steady_clock::now();
 
    if (queue_adapter_) {
        auto enqueue_result = queue_adapter_->enqueue(std::move(job));
        if (enqueue_result.is_err()) {
            return enqueue_result.error();
        }
    } else {
        auto enqueue_result = job_queue_->enqueue(std::move(job));
        if (enqueue_result.is_err()) {
            return enqueue_result.error();
        }
    }
 
    auto end_time = std::chrono::steady_clock::now();
    auto latency = std::chrono::duration_cast<std::chrono::nanoseconds>(end_time - start_time);
    metrics_service_->record_enqueue_with_latency(latency);
 
    auto queue_size = queue_adapter_ ? queue_adapter_->size() : job_queue_->size();
    metrics_service_->record_queue_depth(queue_size);
 
    return common::ok();
}

References kcenon::thread::invalid_argument, kcenon::thread::latency, kcenon::thread::queue_stopped, and kcenon::thread::resource_allocation_failed.

Referenced by main().

Here is the caller graph for this function:

enqueue() [2/2]

auto kcenon::thread::thread_pool::enqueue

(

std::unique_ptr< thread_worker > &&

worker

)

-> common::VoidResult

Adds a thread_worker to the thread pool for specialized or additional processing.

Parameters

worker

A std::unique_ptr<thread_worker> object.

Returns

common::VoidResult containing an error on failure, or success value on success.

Definition at line 412 of file thread_pool.cpp.

                                                                                   {
    if (worker == nullptr) {
        return common::error_info{static_cast<int>(error_code::invalid_argument), "worker is null", "thread_system"};
    }
 
    // Get job_queue from adapter if available, otherwise use direct job_queue_
    std::shared_ptr<job_queue> worker_queue;
    if (queue_adapter_) {
        worker_queue = queue_adapter_->get_job_queue();
        if (!worker_queue) {
            // policy_queue adapter without job_queue backend
            // Workers currently require job_queue; this limitation may be lifted in future versions
            return common::error_info{
                static_cast<int>(error_code::resource_allocation_failed),
                "policy_queue adapter without job_queue backend not yet supported for workers",
                "thread_system"};
        }
    } else if (job_queue_ == nullptr) {
        return common::error_info{static_cast<int>(error_code::resource_allocation_failed), "job queue is null", "thread_system"};
    } else {
        worker_queue = job_queue_;
    }
 
    worker->set_job_queue(worker_queue);
    worker->set_context(context_);
    worker->set_metrics(metrics_service_->get_basic_metrics());
    worker->set_diagnostics(&diagnostics());
    worker->set_diagnostics_sample_rate(diagnostics().get_config().event_sample_rate);
 
    // Acquire lock before checking start_pool_ and adding worker
    // This prevents race condition with stop():
    // - stop() acquires workers_mutex_ after atomically setting start_pool_ to false
    // - If we check start_pool_ while holding the lock, we ensure consistent state
    std::scoped_lock<std::mutex> lock(workers_mutex_);
 
    // Use memory_order_acquire to ensure we see all previous modifications
    // made by the thread that set start_pool_ to true (in start())
    bool is_running = start_pool_.load(std::memory_order_acquire);
 
    // Add worker to vector first, before starting
    // This ensures stop() will see and stop this worker if called concurrently
    workers_.emplace_back(std::move(worker));
 
    // Only start the worker if pool is running
    // Since we hold workers_mutex_, stop() cannot proceed until we release it
    if (is_running) {
        auto start_result = workers_.back()->start();
        if (start_result.is_err()) {
            // Remove the worker we just added since it failed to start
            workers_.pop_back();
            return start_result.error();
        }
    }
 
    return common::ok();
}

References kcenon::thread::invalid_argument, and kcenon::thread::resource_allocation_failed.

enqueue_batch() [1/2]

auto kcenon::thread::thread_pool::enqueue_batch

(

std::vector< std::unique_ptr< job > > &&

jobs

)

-> common::VoidResult

Enqueues a batch of jobs into the shared job_queue.

Parameters

jobs	A vector of std::unique_ptr<job> objects to be added.

Returns

common::VoidResult containing an error on failure, or success value on success.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h, crash_protection/main.cpp, thread_pool_sample.cpp, typed_thread_pool_sample.cpp, and typed_thread_pool_sample_2.cpp.

Definition at line 368 of file thread_pool.cpp.

                                                                                        {
    if (jobs.empty()) {
        return common::error_info{static_cast<int>(error_code::invalid_argument), "jobs are empty", "thread_system"};
    }
 
    // Check queue availability and stopped state
    // Supports both queue_adapter_ (policy_queue) and job_queue_ (legacy)
    if (queue_adapter_) {
        if (queue_adapter_->is_stopped()) {
            return common::error_info{static_cast<int>(error_code::queue_stopped), "thread pool is stopped", "thread_system"};
        }
    } else if (job_queue_ == nullptr) {
        return common::error_info{static_cast<int>(error_code::resource_allocation_failed), "job queue is null", "thread_system"};
    } else if (job_queue_->is_stopped()) {
        return common::error_info{static_cast<int>(error_code::queue_stopped), "thread pool is stopped", "thread_system"};
    }
 
    const auto batch_size = jobs.size();
    metrics_service_->record_submission(batch_size);
 
    auto start_time = std::chrono::steady_clock::now();
 
    if (queue_adapter_) {
        auto enqueue_result = queue_adapter_->enqueue_batch(std::move(jobs));
        if (enqueue_result.is_err()) {
            return enqueue_result.error();
        }
    } else {
        auto enqueue_result = job_queue_->enqueue_batch(std::move(jobs));
        if (enqueue_result.is_err()) {
            return enqueue_result.error();
        }
    }
 
    auto end_time = std::chrono::steady_clock::now();
    auto latency = std::chrono::duration_cast<std::chrono::nanoseconds>(end_time - start_time);
    metrics_service_->record_enqueue_with_latency(latency, batch_size);
 
    auto queue_size = queue_adapter_ ? queue_adapter_->size() : job_queue_->size();
    metrics_service_->record_queue_depth(queue_size);
 
    return common::ok();
}

References kcenon::thread::invalid_argument, kcenon::thread::latency, kcenon::thread::queue_stopped, and kcenon::thread::resource_allocation_failed.

Referenced by main(), store_job(), store_job(), and store_job().

Here is the caller graph for this function:

enqueue_batch() [2/2]

auto kcenon::thread::thread_pool::enqueue_batch

(

std::vector< std::unique_ptr< thread_worker > > &&

workers

)

-> common::VoidResult

Adds a batch of thread_worker objects to the thread pool.

Parameters

workers

A vector of std::unique_ptr<thread_worker> objects.

Returns

common::VoidResult containing an error on failure, or success value on success.

Definition at line 469 of file thread_pool.cpp.

                        {
    if (workers.empty()) {
        return common::error_info{static_cast<int>(error_code::invalid_argument), "workers are empty", "thread_system"};
    }
 
    if (job_queue_ == nullptr) {
        return common::error_info{static_cast<int>(error_code::resource_allocation_failed), "job queue is null", "thread_system"};
    }
 
    // Acquire lock before processing workers
    // This ensures atomic check-and-add operation with respect to stop()
    std::scoped_lock<std::mutex> lock(workers_mutex_);
 
    // Check pool running state once with acquire semantics
    bool is_running = start_pool_.load(std::memory_order_acquire);
 
    // Track the starting index for rollback in case of error
    std::size_t start_index = workers_.size();
 
    // Get diagnostics pointer and config once outside the loop for efficiency
    auto* diag = &diagnostics();
    const auto sample_rate = diag->get_config().event_sample_rate;
 
    for (auto& worker : workers) {
        worker->set_job_queue(job_queue_);
        worker->set_context(context_);
        worker->set_metrics(metrics_service_->get_basic_metrics());
        worker->set_diagnostics(diag);
        worker->set_diagnostics_sample_rate(sample_rate);
 
        // Add worker to vector first
        workers_.emplace_back(std::move(worker));
 
        // Only start if pool is running
        if (is_running) {
            auto start_result = workers_.back()->start();
            if (start_result.is_err()) {
                // Rollback: remove all workers added in this batch
                workers_.erase(workers_.begin() + static_cast<std::ptrdiff_t>(start_index),
                               workers_.end());
                return start_result.error();
            }
        }
    }
 
    return common::ok();
}

References kcenon::thread::invalid_argument, and kcenon::thread::resource_allocation_failed.

find_policy()

template<typename T >

auto kcenon::thread::thread_pool::find_policy ( const std::string & name ) -> T*

nodiscard

Find a policy by name.

Parameters

name	The policy name to search for.

Returns

Pointer to the policy, or nullptr if not found.

Example

auto* cb = pool->find_policy<circuit_breaker_policy>("circuit_breaker_policy");
if (cb) {
    std::cout << "Circuit state: " << to_string(cb->get_state()) << std::endl;
}

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 118 of file thread_pool_impl.h.

{
    std::scoped_lock<std::mutex> lock(policies_mutex_);
 
    for (auto& policy : policies_) {
        if (policy && policy->get_name() == name) {
            return dynamic_cast<T*>(policy.get());
        }
    }
 
    return nullptr;
}

get_active_worker_count()

auto kcenon::thread::thread_pool::get_active_worker_count

(

)

const -> std::size_t

Get the current number of active (running) workers.

Returns

Number of workers currently in running state

This differs from the total worker count as it only counts workers that are actively running, not stopped or stopping.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 755 of file thread_pool.cpp.

                                                             {
    std::scoped_lock<std::mutex> lock(workers_mutex_);
 
    return static_cast<std::size_t>(std::count_if(workers_.begin(), workers_.end(),
                                                  [](const std::unique_ptr<thread_worker>& worker) {
                                                      return worker && worker->is_running();
                                                  }));
}

Referenced by kcenon::thread::autoscaler::autoscaler(), kcenon::thread::autoscaler::can_scale_down(), kcenon::thread::autoscaler::can_scale_up(), kcenon::thread::diagnostics::thread_pool_diagnostics::check_worker_health(), kcenon::thread::autoscaler::collect_metrics(), kcenon::thread::diagnostics::thread_pool_diagnostics::detect_bottlenecks(), kcenon::thread::diagnostics::thread_pool_diagnostics::format_thread_dump(), kcenon::thread::diagnostics::thread_pool_diagnostics::health_check(), and kcenon::thread::adapters::thread_pool_executor_adapter::worker_count().

Here is the caller graph for this function:

get_context()

auto kcenon::thread::thread_pool::get_context ( void ) const -> const thread_context&

nodiscard

Gets the thread context for this pool.

Returns

The thread context providing access to logging and monitoring services.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 645 of file thread_pool.cpp.

                                                                 {
    return context_;
}

get_idle_worker_count()

std::size_t kcenon::thread::thread_pool::get_idle_worker_count ( ) const

nodiscard

Get the number of idle workers.

Returns

Number of workers currently not processing jobs.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 679 of file thread_pool.cpp.

                                                   {
    // Count idle workers by checking each worker's idle state
    // Thread safety: workers_mutex_ protects access to workers_ vector
    std::scoped_lock<std::mutex> lock(workers_mutex_);
 
    return static_cast<std::size_t>(std::count_if(
        workers_.begin(), workers_.end(),
        [](const std::unique_ptr<thread_worker>& worker) { return worker && worker->is_idle(); }));
}

References workers_, and workers_mutex_.

Referenced by kcenon::thread::diagnostics::thread_pool_diagnostics::check_worker_health(), kcenon::thread::autoscaler::collect_metrics(), kcenon::thread::diagnostics::thread_pool_diagnostics::detect_bottlenecks(), kcenon::thread::diagnostics::thread_pool_diagnostics::format_thread_dump(), and report_metrics().

Here is the caller graph for this function:

get_job_queue()

auto kcenon::thread::thread_pool::get_job_queue ( void ) -> std::shared_ptr<job_queue>

nodiscard

Returns the shared job_queue used by this thread pool.

Returns the shared job queue used by all workers.

Returns

A std::shared_ptr<job_queue> that stores the queued jobs.

The returned queue is shared among all worker threads in the pool, which can concurrently dequeue and process jobs.

Implementation details:

• Provides access to the adaptive job queue for external job submission
• Queue is shared among all workers for load balancing
• Adaptive queue automatically optimizes based on contention patterns

Returns

Shared pointer to the job queue

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 269 of file thread_pool.cpp.

                                                                {
    return job_queue_;
}

Referenced by kcenon::thread::diagnostics::thread_pool_diagnostics::check_queue_health(), kcenon::thread::diagnostics::thread_pool_diagnostics::detect_bottlenecks(), and kcenon::thread::diagnostics::thread_pool_diagnostics::health_check().

Here is the caller graph for this function:

get_pending_task_count()

auto kcenon::thread::thread_pool::get_pending_task_count

(

)

const -> std::size_t

Get the number of pending tasks in the queue.

Returns

Number of tasks waiting to be processed

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 695 of file thread_pool.cpp.

                                                            {
    // Supports both queue_adapter_ and job_queue_
    if (queue_adapter_) {
        return queue_adapter_->size();
    }
    if (job_queue_) {
        return job_queue_->size();
    }
    return 0;
}

References job_queue_, and queue_adapter_.

Referenced by kcenon::thread::diagnostics::thread_pool_diagnostics::check_queue_health(), kcenon::thread::autoscaler::collect_metrics(), kcenon::thread::diagnostics::thread_pool_diagnostics::detect_bottlenecks(), kcenon::thread::diagnostics::thread_pool_diagnostics::health_check(), and kcenon::thread::adapters::thread_pool_executor_adapter::pending_tasks().

Here is the caller graph for this function:

get_policies()

auto kcenon::thread::thread_pool::get_policies ( ) const -> const std::vector<std::unique_ptr<pool_policy>>&

nodiscard

Get all registered policies.

Returns

Const reference to the vector of policies.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 843 of file thread_pool.cpp.

                                                                                     {
    return policies_;
}

get_pool_instance_id()

std::uint32_t kcenon::thread::thread_pool::get_pool_instance_id ( ) const

nodiscard

Get the pool instance id.

Returns

Returns the unique instance id for this pool.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 649 of file thread_pool.cpp.

                                                    {
    return pool_instance_id_;
}

References pool_instance_id_.

get_ptr()

auto kcenon::thread::thread_pool::get_ptr ( void ) -> std::shared_ptr<thread_pool>

nodiscard

Retrieves a std::shared_ptr to this thread_pool instance.

Returns a shared pointer to this thread pool instance.

Returns

A shared pointer to the current thread_pool object.

By inheriting from std::enable_shared_from_this, you can call get_ptr() within member functions to avoid storing a separate shared pointer.

Implementation details:

• Uses std::enable_shared_from_this for safe shared_ptr creation
• Required for passing pool reference to workers or other components
• Ensures proper lifetime management in multi-threaded environment

Returns

Shared pointer to this thread_pool

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 169 of file thread_pool.cpp.

                                                            {
    return this->shared_from_this();
}

is_enhanced_metrics_enabled()

bool kcenon::thread::thread_pool::is_enhanced_metrics_enabled ( ) const

nodiscard

Check if enhanced metrics is enabled.

Returns

True if enhanced metrics collection is enabled.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 290 of file thread_pool.cpp.

                                                    {
    return metrics_service_->is_enhanced_metrics_enabled();
}

References metrics_service_.

Referenced by kcenon::thread::autoscaler::collect_metrics().

Here is the caller graph for this function:

is_running()

auto kcenon::thread::thread_pool::is_running

(

)

const -> bool

Check if the thread pool is currently running.

Returns

true if the pool is active, false otherwise

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 689 of file thread_pool.cpp.

                                           {
    // Use acquire to ensure we see the latest pool state
    // This is important for callers making decisions based on running state
    return start_pool_.load(std::memory_order_acquire);
}

References start_pool_.

Referenced by kcenon::thread::diagnostics::thread_pool_diagnostics::check_worker_health(), kcenon::thread::diagnostics::thread_pool_diagnostics::is_healthy(), and kcenon::thread::adapters::thread_pool_executor_adapter::is_running().

Here is the caller graph for this function:

metrics()

const metrics::ThreadPoolMetrics & kcenon::thread::thread_pool::metrics ( ) const

nodiscardnoexcept

Access aggregated runtime metrics (read-only reference).

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 273 of file thread_pool.cpp.

                                                                    {
    return metrics_service_->basic_metrics();
}

References metrics_service_.

Referenced by kcenon::thread::autoscaler::collect_metrics(), kcenon::thread::diagnostics::thread_pool_diagnostics::detect_bottlenecks(), kcenon::thread::diagnostics::thread_pool_diagnostics::health_check(), and report_metrics().

Here is the caller graph for this function:

remove_policy()

auto kcenon::thread::thread_pool::remove_policy

(

const std::string &

name

)

-> bool

Remove a policy by name.

Parameters

name	The policy name to remove.

Returns

True if policy was found and removed.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 847 of file thread_pool.cpp.

                                                             {
    std::scoped_lock<std::mutex> lock(policies_mutex_);
 
    auto it = std::remove_if(policies_.begin(), policies_.end(),
        [&name](const std::unique_ptr<pool_policy>& policy) {
            return policy && policy->get_name() == name;
        });
 
    if (it != policies_.end()) {
        policies_.erase(it, policies_.end());
        return true;
    }
 
    return false;
}

remove_workers_internal()

auto kcenon::thread::thread_pool::remove_workers_internal ( std::size_t count, std::size_t min_workers = 1 ) -> common::VoidResult

nodiscardprivate

Internal method to remove workers from the pool.

Parameters

count	Number of workers to remove.
min_workers	Minimum workers to keep.

Returns

Error if operation fails.

Used internally by autoscaler for scale-down operations. Gracefully stops and removes idle workers.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 867 of file thread_pool.cpp.

{
    if (count == 0) {
        return common::ok();
    }
 
    std::scoped_lock<std::mutex> lock(workers_mutex_);
 
    if (workers_.size() <= min_workers) {
        return common::error_info{
            static_cast<int>(error_code::invalid_argument),
            "Cannot remove workers: already at minimum",
            "thread_system"
        };
    }
 
    // Calculate how many we can actually remove
    std::size_t max_removable = workers_.size() - min_workers;
    count = std::min(count, max_removable);
 
    std::size_t removed = 0;
 
    // First pass: remove idle workers
    auto it = workers_.begin();
    while (it != workers_.end() && removed < count) {
        if (*it && (*it)->is_idle()) {
            // Stop the worker
            (*it)->stop();
            it = workers_.erase(it);
            ++removed;
        } else {
            ++it;
        }
    }
 
    // If we still need to remove more, wait briefly for workers to become idle
    if (removed < count) {
        // Just return success with what we removed
        // Remaining workers will be removed on subsequent calls
        context_.log(common::interfaces::log_level::info,
            formatter::format("Removed {} of {} requested workers (remaining are busy)",
                              removed, count));
    }
 
    return common::ok();
}

References kcenon::thread::utils::formatter::format().

Here is the call graph for this function:

report_metrics()

void kcenon::thread::thread_pool::report_metrics

(

)

Collect and report current thread pool metrics.

This method gathers current metrics from the pool and reports them through the monitoring interface if available.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 653 of file thread_pool.cpp.

                                 {
    if (!context_.monitoring()) {
        return;
    }
 
    common::interfaces::thread_pool_metrics metrics(thread_title_, pool_instance_id_);
 
    // Protect workers_ access with lock
    {
        std::scoped_lock<std::mutex> lock(workers_mutex_);
        metrics.worker_threads.value = static_cast<double>(workers_.size());
    }
 
    metrics.idle_threads.value = static_cast<double>(get_idle_worker_count());
 
    // Get pending jobs count (supports both queue_adapter_ and job_queue_)
    if (queue_adapter_) {
        metrics.jobs_pending.value = static_cast<double>(queue_adapter_->size());
    } else if (job_queue_) {
        metrics.jobs_pending.value = static_cast<double>(job_queue_->size());
    }
 
    // Report metrics with pool identification
    context_.update_thread_pool_metrics(thread_title_, pool_instance_id_, metrics);
}

References context_, get_idle_worker_count(), job_queue_, metrics(), kcenon::thread::thread_context::monitoring(), pool_instance_id_, queue_adapter_, thread_title_, kcenon::thread::thread_context::update_thread_pool_metrics(), workers_, and workers_mutex_.

Here is the call graph for this function:

reset_metrics()

void kcenon::thread::thread_pool::reset_metrics

(

)

Reset accumulated metrics.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 277 of file thread_pool.cpp.

                                {
    metrics_service_->reset();
}

References metrics_service_, and kcenon::thread::metrics::ThreadPoolMetrics::reset().

Here is the call graph for this function:

set_enhanced_metrics_enabled()

void kcenon::thread::thread_pool::set_enhanced_metrics_enabled

(

bool

enabled

)

Enable or disable enhanced metrics collection.

Parameters

enabled

True to enable enhanced metrics (histograms, percentiles).

When enabled, additional metrics like latency histograms and throughput counters are collected. This has minimal overhead (< 100ns per operation) but can be disabled for maximum performance.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 281 of file thread_pool.cpp.

                                                           {
    std::size_t worker_count = 0;
    {
        std::scoped_lock<std::mutex> lock(workers_mutex_);
        worker_count = workers_.size();
    }
    metrics_service_->set_enhanced_metrics_enabled(enabled, worker_count);
}

References metrics_service_, workers_, and workers_mutex_.

start()

auto kcenon::thread::thread_pool::start

(

void

)

-> common::VoidResult

Starts the thread pool and all associated workers.

Starts all worker threads in the pool.

Returns

common::VoidResult containing an error on failure, or success value on success.

If the pool is already running, a subsequent call to start() may return an error. On success, each thread_worker in workers_ is started, enabling them to process jobs from the job_queue_.

Implementation details:

• Validates that workers have been added to the pool
• Starts each worker thread individually
• If any worker fails to start, stops all workers and returns error
• Sets start_pool_ flag to true on successful startup
• Workers begin processing jobs from the shared queue immediately

Startup Sequence:

1. Check that workers exist
2. Start each worker sequentially
3. On failure: stop all workers and return error message
4. On success: mark pool as started

Error Handling:

• All-or-nothing startup (if one fails, all stop)
• Provides detailed error message from failed worker
• Ensures consistent pool state (either all running or all stopped)

Returns

std::nullopt on success, error message on failure

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h, crash_protection/main.cpp, thread_pool_sample.cpp, typed_thread_pool_sample.cpp, and typed_thread_pool_sample_2.cpp.

Definition at line 196 of file thread_pool.cpp.

                                                {
    // Acquire lock to check workers_ safely
    std::scoped_lock<std::mutex> lock(workers_mutex_);
 
    // Check if pool is already running
    // Use acquire to ensure we see all previous modifications to pool state
    if (start_pool_.load(std::memory_order_acquire)) {
        return common::error_info{static_cast<int>(error_code::thread_already_running), "thread pool is already running", "thread_system"};
    }
 
    // Validate that workers have been added
    if (workers_.empty()) {
        return common::error_info{static_cast<int>(error_code::invalid_argument), "no workers to start", "thread_system"};
    }
 
    // Handle queue initialization for restart scenarios
    // The approach differs based on whether we're using job_queue or queue_adapter
    if (queue_adapter_) {
        // Using queue_adapter (policy_queue or wrapped job_queue)
        if (queue_adapter_->is_stopped()) {
            // For policy_queue adapters, we can't easily recreate,
            // so we return an error suggesting pool recreation
            return common::error_info{
                static_cast<int>(error_code::queue_stopped),
                "queue is stopped; create a new thread_pool instance for restart",
                "thread_system"};
        }
        // Update job_queue_ reference if adapter wraps a job_queue
        job_queue_ = queue_adapter_->get_job_queue();
    } else if (job_queue_ == nullptr || job_queue_->is_stopped()) {
        // Create fresh job queue for restart scenarios
        // Stopped queues cannot accept new jobs, so we must create a new instance
        job_queue_ = std::make_shared<kcenon::thread::job_queue>();
 
        // Update all workers with the new queue reference
        for (auto& worker : workers_) {
            worker->set_job_queue(job_queue_);
        }
    }
 
    // Create fresh pool cancellation token for restart scenarios
    // This ensures workers start with a non-cancelled token
    pool_cancellation_token_ = cancellation_token::create();
    metrics_service_->reset();
 
    // Attempt to start each worker
    for (auto& worker : workers_) {
        auto start_result = worker->start();
        if (start_result.is_err()) {
            // If any worker fails, stop all and return error
            stop();
            return start_result.error();
        }
    }
 
    // Mark pool as successfully started
    // Use release to ensure all previous modifications (worker starts, queue setup)
    // are visible to other threads before they see start_pool_ == true
    start_pool_.store(true, std::memory_order_release);
 
    return common::ok();
}

References kcenon::thread::cancellation_token::create(), kcenon::thread::invalid_argument, kcenon::thread::queue_stopped, and kcenon::thread::thread_already_running.

Referenced by main(), and main().

Here is the call graph for this function:

Here is the caller graph for this function:

stop()

auto kcenon::thread::thread_pool::stop

(

const bool &

immediately_stop = false

)

-> common::VoidResult

Stops the thread pool and all worker threads.

Parameters

immediately_stop

If true, any ongoing jobs may be interrupted; if false (default), each worker attempts to finish its current job before stopping.

Returns

common::VoidResult containing an error on failure, or success value on success.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h, crash_protection/main.cpp, thread_pool_sample.cpp, typed_thread_pool_sample.cpp, and typed_thread_pool_sample_2.cpp.

Definition at line 518 of file thread_pool.cpp.

                                                                       {
    // Use compare_exchange_strong to atomically check and set state
    // This prevents TOCTOU (Time-Of-Check-Time-Of-Use) race conditions
    // where multiple threads might call stop() simultaneously
    bool expected = true;
    if (!start_pool_.compare_exchange_strong(expected, false, std::memory_order_acq_rel,
                                             std::memory_order_acquire)) {
        // Pool is already stopped or being stopped by another thread
        return common::ok();
    }
 
    // At this point, we've atomically transitioned from running to stopped
    // and only this thread will execute the shutdown sequence
 
    // Cancel pool-level token to propagate cancellation to all workers and jobs
    // This triggers hierarchical cancellation:
    // 1. Pool token cancelled → linked worker tokens cancelled
    // 2. Worker tokens cancelled → running jobs receive cancellation signal
    pool_cancellation_token_.cancel();
 
    // Stop the queue (supports both queue_adapter_ and job_queue_)
    if (queue_adapter_) {
        queue_adapter_->stop();
        if (immediately_stop) {
            queue_adapter_->clear();
        }
    } else if (job_queue_ != nullptr) {
        job_queue_->stop();
        if (immediately_stop) {
            job_queue_->clear();
        }
    }
 
    // Stop workers while holding lock to ensure consistent iteration
    // This is safe because worker->stop() only signals and joins threads,
    // it does not call back into thread_pool methods
    std::scoped_lock<std::mutex> lock(workers_mutex_);
    for (auto& worker : workers_) {
        auto stop_result = worker->stop();
        if (stop_result.is_err()) {
            context_.log(common::interfaces::log_level::error, formatter::format("error stopping worker: {}",
                                                             stop_result.error().message));
        }
    }
 
    return common::ok();
}

References kcenon::thread::utils::formatter::format().

Referenced by main(), main(), kcenon::thread::adapters::thread_pool_executor_adapter::shutdown(), and ~thread_pool().

Here is the call graph for this function:

Here is the caller graph for this function:

stop_unsafe()

auto kcenon::thread::thread_pool::stop_unsafe ( ) -> common::VoidResult

privatenoexcept

Stops the thread pool without logging (for use during static destruction).

This method performs the same shutdown sequence as stop() but without any logging operations. It is specifically designed to be called from the destructor when static destruction is in progress, preventing Static Destruction Order Fiasco (SDOF) by avoiding access to potentially destroyed logger/monitoring singletons.

Returns

common::VoidResult containing an error on failure, or success value on success.

Note

This method should only be called from the destructor during static destruction.

This method performs the same shutdown sequence as stop() but without any logging operations. It is specifically designed to be called from the destructor when static destruction is in progress, preventing Static Destruction Order Fiasco (SDOF) by avoiding access to potentially destroyed logger/monitoring singletons.

Implementation details:

• Same atomic state transition as stop()
• Cancels pool-level token for hierarchical cancellation
• Stops job queue without clearing (graceful shutdown)
• Stops all workers without logging errors
• noexcept guarantee for safe destructor use

Returns

result_void containing an error on failure, or success value on success.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 584 of file thread_pool.cpp.

                                                           {
    // Use compare_exchange_strong to atomically check and set state
    // Same atomic transition as stop() to prevent race conditions
    bool expected = true;
    if (!start_pool_.compare_exchange_strong(expected, false, std::memory_order_acq_rel,
                                             std::memory_order_acquire)) {
        // Pool is already stopped or being stopped by another thread
        return common::ok();
    }
 
    // Cancel pool-level token to propagate cancellation to all workers and jobs
    pool_cancellation_token_.cancel();
 
    // Stop the queue (supports both queue_adapter_ and job_queue_)
    if (queue_adapter_) {
        queue_adapter_->stop();
    } else if (job_queue_ != nullptr) {
        job_queue_->stop();
    }
 
    // Stop workers while holding lock to ensure consistent iteration
    // No logging to avoid accessing potentially destroyed singletons
    std::scoped_lock<std::mutex> lock(workers_mutex_);
    for (auto& worker : workers_) {
        // Stop worker without checking result to avoid any potential exceptions
        // during static destruction
        worker->stop();
    }
 
    return common::ok();
}

References kcenon::thread::cancellation_token::cancel(), job_queue_, pool_cancellation_token_, queue_adapter_, start_pool_, workers_, and workers_mutex_.

Referenced by ~thread_pool().

Here is the call graph for this function:

Here is the caller graph for this function:

submit() [1/2]

template<typename F , typename R >

auto kcenon::thread::thread_pool::submit ( F && callable, const submit_options & opts = {} ) -> std::future<R>

nodiscard

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 35 of file thread_pool_impl.h.

{
    auto job_ptr = std::make_unique<future_job<R>>(
        std::forward<F>(callable),
        opts.name.empty() ? "async_job" : opts.name
    );
 
    auto future = job_ptr->get_future();
 
    auto result = enqueue(std::move(job_ptr));
    if (result.is_err()) {
        std::promise<R> error_promise;
        error_promise.set_exception(
            std::make_exception_ptr(
                std::runtime_error(result.error().message)
            )
        );
        return error_promise.get_future();
    }
 
    return future;
}

submit() [2/2]

template<typename F , typename R >

auto kcenon::thread::thread_pool::submit ( std::vector< F > && callables, const submit_options & opts = {} ) -> std::vector<std::future<R>>

nodiscard

Definition at line 60 of file thread_pool_impl.h.

{
    return detail::batch_apply(std::move(callables), [this, &opts](auto&& callable) {
        submit_options single_opts;
        single_opts.name = opts.name;
        return submit<F, R>(std::move(callable), single_opts);
    });
}

References kcenon::thread::detail::batch_apply(), and kcenon::thread::submit_options::name.

Here is the call graph for this function:

submit_wait_all()

template<typename F , typename R >

auto kcenon::thread::thread_pool::submit_wait_all ( std::vector< F > && callables, const submit_options & opts = {} ) -> std::vector<R>

nodiscard

Submit a batch and wait for all results.

Convenience overload that blocks until all tasks complete. Equivalent to submit(callables, submit_options::all()).

Template Parameters

F	Callable type
R	Return type (automatically deduced)

Parameters

callables	Vector of functions to execute
opts	Submit options (wait_all is implicitly true)

Returns

Vector of results

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 71 of file thread_pool_impl.h.

{
    auto futures = submit<F, R>(std::move(callables), opts);
    return detail::collect_all(futures);
}

References kcenon::thread::detail::collect_all().

Here is the call graph for this function:

submit_wait_any()

template<typename F , typename R >

auto kcenon::thread::thread_pool::submit_wait_any ( std::vector< F > && callables, const submit_options & opts = {} ) -> common::Result<R>

nodiscard

Submit a batch and return first completed result.

Convenience overload that returns when any task completes. Equivalent to submit(callables, submit_options::any()).

Template Parameters

F	Callable type
R	Return type (automatically deduced)

Parameters

callables	Vector of functions to execute
opts	Submit options (wait_any is implicitly true)

Returns

common::Result<R> containing the first completed result, or an error if callables is empty or execution failed

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 79 of file thread_pool_impl.h.

{
    if (callables.empty()) {
        return make_error_result<R>(error_code::invalid_argument, "Empty callables vector");
    }
 
    auto futures = submit<F, R>(std::move(callables), opts);
    auto completed = std::make_shared<std::atomic<bool>>(false);
    auto result_promise = std::make_shared<std::promise<R>>();
    auto result_future = result_promise->get_future();
 
    for (std::size_t i = 0; i < futures.size(); ++i) {
        std::thread([completed, result_promise, fut = std::move(futures[i])]() mutable {
            try {
                R result = fut.get();
                bool expected = false;
                if (completed->compare_exchange_strong(expected, true)) {
                    result_promise->set_value(std::move(result));
                }
            } catch (...) {
                bool expected = false;
                if (completed->compare_exchange_strong(expected, true)) {
                    result_promise->set_exception(std::current_exception());
                }
            }
        }).detach();
    }
 
    try {
        return common::Result<R>::ok(result_future.get());
    } catch (const std::exception& e) {
        return make_error_result<R>(error_code::job_execution_failed, e.what());
    } catch (...) {
        return make_error_result<R>(error_code::unknown_error, "Unknown exception in submit_wait_any");
    }
}

References kcenon::thread::completed, kcenon::thread::invalid_argument, kcenon::thread::job_execution_failed, kcenon::thread::make_error_result(), and kcenon::thread::unknown_error.

Here is the call graph for this function:

to_string()

auto kcenon::thread::thread_pool::to_string ( void ) const -> std::string

nodiscard

Provides a string representation of this thread_pool.

Returns

A string describing the pool, including its title and other optional details.

Derived classes may override this to include more diagnostic or state-related info.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h, thread_pool_sample.cpp, typed_thread_pool_sample.cpp, and typed_thread_pool_sample_2.cpp.

Definition at line 616 of file thread_pool.cpp.

                                                   {
    std::string format_string;
 
    // Use relaxed memory order for diagnostic/logging purposes
    // Exact state ordering is not critical for debug output
    formatter::format_to(std::back_inserter(format_string), "{} is {},\n", thread_title_,
                         start_pool_.load(std::memory_order_relaxed) ? "running" : "stopped");
 
    // Get queue string representation (supports both queue_adapter_ and job_queue_)
    std::string queue_str;
    if (queue_adapter_) {
        queue_str = queue_adapter_->to_string();
    } else if (job_queue_ != nullptr) {
        queue_str = job_queue_->to_string();
    } else {
        queue_str = "nullptr";
    }
    formatter::format_to(std::back_inserter(format_string), "\tjob_queue: {}\n\n", queue_str);
 
    // Protect workers_ access with lock
    std::scoped_lock<std::mutex> lock(workers_mutex_);
    formatter::format_to(std::back_inserter(format_string), "\tworkers: {}\n", workers_.size());
    for (const auto& worker : workers_) {
        formatter::format_to(std::back_inserter(format_string), "\t{}\n", worker->to_string());
    }
 
    return format_string;
}

References kcenon::thread::utils::formatter::format_to().

Referenced by std::formatter< kcenon::thread::thread_pool >::format(), std::formatter< kcenon::thread::thread_pool, wchar_t >::format(), kcenon::thread::diagnostics::thread_pool_diagnostics::format_thread_dump(), main(), and kcenon::thread::diagnostics::thread_pool_diagnostics::to_prometheus().

Here is the call graph for this function:

Here is the caller graph for this function:

Friends And Related Symbol Documentation

autoscaler

friend class autoscaler

friend

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 186 of file thread_pool.h.

diagnostics::thread_pool_diagnostics

friend class diagnostics::thread_pool_diagnostics

friend

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 643 of file thread_pool.h.

numa_thread_pool

friend class numa_thread_pool

friend

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 184 of file thread_pool.h.

Member Data Documentation

context_

thread_context kcenon::thread::thread_pool::context_

private

The thread context providing access to logging and monitoring services.

This context is shared with all worker threads created by this pool, enabling consistent logging and monitoring throughout the pool.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 721 of file thread_pool.h.

Referenced by report_metrics(), thread_pool(), thread_pool(), and thread_pool().

diagnostics_

std::unique_ptr<diagnostics::thread_pool_diagnostics> kcenon::thread::thread_pool::diagnostics_

mutableprivate

Diagnostics interface for this pool.

Lazily initialized on first access to diagnostics().

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 776 of file thread_pool.h.

diagnostics_init_flag_

std::once_flag kcenon::thread::thread_pool::diagnostics_init_flag_

mutableprivate

Once flag for thread-safe lazy initialization of diagnostics_.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 781 of file thread_pool.h.

job_queue_

std::shared_ptr<job_queue> kcenon::thread::thread_pool::job_queue_

private

The shared job queue where jobs (job objects) are enqueued.

Worker threads dequeue jobs from this queue to perform tasks. The queue persists for the lifetime of the pool or until no more references exist.

Note

When queue_adapter_ is set, this may be nullptr for policy_queue adapters. Use get_queue_adapter() for unified access.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 676 of file thread_pool.h.

Referenced by get_pending_task_count(), report_metrics(), and stop_unsafe().

metrics_service_

std::shared_ptr<metrics::metrics_service> kcenon::thread::thread_pool::metrics_service_

private

Centralized metrics service for all pool and worker metrics.

Consolidates metrics management that was previously spread across thread_pool and thread_worker. Provides:

• Basic metrics (ThreadPoolMetrics)
• Enhanced metrics with histograms and percentiles (lazily initialized)
• Unified recording interface
• Clear ownership semantics (pool owns, workers borrow)

See also

metrics::metrics_service

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 768 of file thread_pool.h.

Referenced by enhanced_metrics(), enhanced_metrics_snapshot(), is_enhanced_metrics_enabled(), metrics(), reset_metrics(), and set_enhanced_metrics_enabled().

next_pool_instance_id_

std::atomic< std::uint32_t > kcenon::thread::thread_pool::next_pool_instance_id_ {0}

staticprivate

Static counter for generating unique pool instance IDs.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 647 of file thread_pool.h.

policies_

std::vector<std::unique_ptr<pool_policy> > kcenon::thread::thread_pool::policies_

private

Registered pool policies for extending thread pool behavior.

Policies are called in order of addition for each lifecycle event. Protected by policies_mutex_ for thread-safe access.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 789 of file thread_pool.h.

policies_mutex_

std::mutex kcenon::thread::thread_pool::policies_mutex_

mutableprivate

Mutex protecting concurrent access to the policies_ vector.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 794 of file thread_pool.h.

pool_cancellation_token_

cancellation_token kcenon::thread::thread_pool::pool_cancellation_token_

private

Pool-level cancellation token.

This token is used to propagate cancellation to all workers and jobs when the pool is stopped. Each worker receives a linked token that combines this pool token with its own worker token, creating a hierarchical cancellation structure.

Cancellation Hierarchy:

• Pool stop() → cancels pool_cancellation_token_
• Pool token cancellation → propagates to all linked worker tokens
• Worker token cancellation → propagates to running jobs

Note

This token is reset when the pool is restarted to allow fresh job execution without stale cancellation state.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 739 of file thread_pool.h.

Referenced by stop_unsafe().

pool_instance_id_

std::uint32_t kcenon::thread::thread_pool::pool_instance_id_ {0}

private

Unique instance ID for this pool (for multi-pool scenarios).

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 657 of file thread_pool.h.

{0};

Referenced by get_pool_instance_id(), and report_metrics().

queue_adapter_

std::unique_ptr<pool_queue_adapter_interface> kcenon::thread::thread_pool::queue_adapter_

private

Queue adapter for unified access to different queue types.

When set, provides a unified interface for both job_queue and policy_queue. This enables thread_pool to work with the new policy-based queue system while maintaining backward compatibility.

The adapter wraps either:

• job_queue: Legacy queue with full feature support
• policy_queue: New policy-based queue with compile-time configuration

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 689 of file thread_pool.h.

Referenced by get_pending_task_count(), report_metrics(), and stop_unsafe().

start_pool_

std::atomic<bool> kcenon::thread::thread_pool::start_pool_

private

Indicates whether the pool is currently running.

Set to true after a successful call to start(), and reset to false after stop(). Used internally to prevent multiple active starts or erroneous state transitions.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 665 of file thread_pool.h.

Referenced by is_running(), and stop_unsafe().

thread_title_

std::string kcenon::thread::thread_pool::thread_title_

private

A title or name for this thread pool, useful for identification and logging.

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 652 of file thread_pool.h.

Referenced by report_metrics().

workers_

std::vector<std::unique_ptr<thread_worker> > kcenon::thread::thread_pool::workers_

private

A collection of worker threads associated with this pool.

Each thread_worker typically runs in its own thread context, processing jobs from job_queue_ or performing specialized logic. They are started together when thread_pool::start() is called.

Thread Safety:

• Protected by workers_mutex_ to prevent concurrent modification
• Accessed during enqueue_worker() and stop() operations

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 702 of file thread_pool.h.

Referenced by kcenon::thread::diagnostics::thread_pool_diagnostics::check_worker_health(), kcenon::thread::diagnostics::thread_pool_diagnostics::detect_bottlenecks(), kcenon::thread::diagnostics::thread_pool_diagnostics::format_thread_dump(), get_idle_worker_count(), kcenon::thread::diagnostics::thread_pool_diagnostics::health_check(), kcenon::thread::diagnostics::thread_pool_diagnostics::is_healthy(), report_metrics(), set_enhanced_metrics_enabled(), and stop_unsafe().

workers_mutex_

std::mutex kcenon::thread::thread_pool::workers_mutex_

mutableprivate

Mutex protecting concurrent access to the workers_ vector.

Synchronization Strategy:

• Guards all workers_ modifications (enqueue, stop)
• Prevents iterator invalidation during concurrent operations
• Uses scoped_lock for RAII-based protection
• Held briefly to minimize contention

Examples

/home/runner/work/thread_system/thread_system/include/kcenon/thread/core/thread_pool.h.

Definition at line 713 of file thread_pool.h.

Referenced by kcenon::thread::diagnostics::thread_pool_diagnostics::check_worker_health(), kcenon::thread::diagnostics::thread_pool_diagnostics::detect_bottlenecks(), kcenon::thread::diagnostics::thread_pool_diagnostics::format_thread_dump(), get_idle_worker_count(), kcenon::thread::diagnostics::thread_pool_diagnostics::health_check(), kcenon::thread::diagnostics::thread_pool_diagnostics::is_healthy(), report_metrics(), set_enhanced_metrics_enabled(), and stop_unsafe().

---

The documentation for this class was generated from the following files:

• include/kcenon/thread/core/thread_pool.h
• src/impl/thread_pool/thread_pool.cpp
• include/kcenon/thread/core/thread_pool_impl.h