kcenon::thread::autoscaler Class Reference

Manages automatic scaling of thread pool workers based on load metrics. More...

#include <autoscaler.h>

Collaboration diagram for kcenon::thread::autoscaler:

Public Member Functions
	autoscaler (thread_pool &pool, autoscaling_policy policy={})
	Constructs an autoscaler for the given thread pool.
	~autoscaler ()
	Destructor. Stops the monitor thread if running.
	autoscaler (const autoscaler &)=delete
autoscaler &	operator= (const autoscaler &)=delete
	autoscaler (autoscaler &&)=delete
autoscaler &	operator= (autoscaler &&)=delete
auto	start () -> void
	Starts the autoscaling monitor thread.
auto	stop () -> void
	Stops the autoscaling monitor thread.
auto	is_active () const -> bool
	Checks if the autoscaler is currently active.
auto	evaluate_now () -> scaling_decision
	Manually triggers a scaling evaluation.
auto	scale_to (std::size_t target_workers) -> common::VoidResult
	Manually scales to a specific worker count.
auto	scale_up () -> common::VoidResult
	Manually scales up by the configured increment.
auto	scale_down () -> common::VoidResult
	Manually scales down by the configured increment.
auto	set_policy (autoscaling_policy policy) -> void
	Updates the autoscaling policy.
auto	get_policy () const -> const autoscaling_policy &
	Gets the current autoscaling policy.
auto	get_current_metrics () const -> scaling_metrics_sample
	Collects current metrics from the thread pool.
auto	get_metrics_history (std::size_t count=60) const -> std::vector< scaling_metrics_sample >
	Gets historical metrics samples.
auto	get_stats () const -> autoscaling_stats
	Gets autoscaling statistics.
auto	reset_stats () -> void
	Resets autoscaling statistics.

Private Member Functions
auto	monitor_loop () -> void
	Main monitoring loop running in the background thread.
auto	collect_metrics () const -> scaling_metrics_sample
	Collects current metrics from the pool.
auto	make_decision (const std::vector< scaling_metrics_sample > &samples) const -> scaling_decision
	Makes a scaling decision based on recent samples.
auto	execute_scaling (const scaling_decision &decision) -> void
	Executes a scaling decision.
auto	can_scale_up () const -> bool
	Checks if scale-up cooldown has elapsed.
auto	can_scale_down () const -> bool
	Checks if scale-down cooldown has elapsed.
auto	add_workers (std::size_t count) -> common::VoidResult
	Adds workers to the pool.
auto	remove_workers (std::size_t count) -> common::VoidResult
	Removes workers from the pool.

Private Attributes
thread_pool &	pool_
autoscaling_policy	policy_
std::atomic< bool >	running_ {false}
std::unique_ptr< std::thread >	monitor_thread_
std::mutex	mutex_
std::condition_variable	cv_
std::deque< scaling_metrics_sample >	metrics_history_
std::mutex	history_mutex_
std::chrono::steady_clock::time_point	last_scale_up_time_
std::chrono::steady_clock::time_point	last_scale_down_time_
autoscaling_stats	stats_
std::mutex	stats_mutex_
std::uint64_t	last_jobs_completed_ {0}
std::uint64_t	last_jobs_submitted_ {0}
std::chrono::steady_clock::time_point	last_sample_time_

Detailed Description

Manages automatic scaling of thread pool workers based on load metrics.

The autoscaler monitors thread pool metrics and automatically adjusts the number of workers to match workload demands. It uses a background monitor thread to periodically collect metrics and make scaling decisions.

Design Principles

• Non-intrusive: Scaling decisions are made asynchronously
• Configurable: All thresholds and behaviors are customizable
• Graceful: Scale-down removes workers only when safe
• Observable: Provides statistics and callbacks for monitoring

State Machine

┌─────────────────────────────────────────────────────────────┐
│                    Autoscaler Loop                          │
│                                                             │
│  ┌─────────────┐    ┌─────────────┐    ┌─────────────┐     │
│  │ Collect     │───▶│ Aggregate   │───▶│ Make        │     │
│  │ Metrics     │    │ Samples     │    │ Decision    │     │
│  └─────────────┘    └─────────────┘    └─────────────┘     │
│         │                                    │              │
│         │                                    ▼              │
│         │                         ┌─────────────────┐       │
│         │                         │ Check Cooldown  │       │
│         │                         └─────────────────┘       │
│         │                                    │              │
│         │                                    ▼              │
│         │                         ┌─────────────────┐       │
│         │                         │ Execute Scale   │       │
│         │                         └─────────────────┘       │
│         │                                    │              │
│         ▼                                    │              │
│  ┌─────────────┐                            │              │
│  │   Sleep     │◀───────────────────────────┘              │
│  │  Interval   │                                           │
│  └─────────────┘                                           │
└─────────────────────────────────────────────────────────────┘

Thread Safety

All public methods are thread-safe and can be called from any thread.

Usage Example

auto pool = std::make_shared<thread_pool>("MyPool");
auto scaler = std::make_shared<autoscaler>(*pool, autoscaling_policy{
    .min_workers = 2,
    .max_workers = 16,
    .scaling_mode = autoscaling_policy::mode::automatic
});
 
scaler->start();
// ... pool automatically scales ...
scaler->stop();

See also

autoscaling_policy

scaling_metrics_sample

Definition at line 94 of file autoscaler.h.

Constructor & Destructor Documentation

autoscaler() [1/3]

kcenon::thread::autoscaler::autoscaler ( thread_pool & pool, autoscaling_policy policy = {} )

explicit

Constructs an autoscaler for the given thread pool.

Parameters

pool	Reference to the thread pool to manage.
policy	Autoscaling policy configuration.

Definition at line 16 of file autoscaler.cpp.

    : pool_(pool)
    , policy_(std::move(policy))
    , last_sample_time_(std::chrono::steady_clock::now())
{
    // Initialize stats with current worker count
    std::scoped_lock<std::mutex> lock(stats_mutex_);
    stats_.min_workers = pool_.get_active_worker_count();
    stats_.peak_workers = stats_.min_workers;
}

References kcenon::thread::thread_pool::get_active_worker_count(), kcenon::thread::autoscaling_stats::min_workers, kcenon::thread::autoscaling_stats::peak_workers, pool_, stats_, and stats_mutex_.

Here is the call graph for this function:

~autoscaler()

kcenon::thread::autoscaler::~autoscaler

(

)

Destructor. Stops the monitor thread if running.

Definition at line 27 of file autoscaler.cpp.

{
    stop();
}

References stop().

Here is the call graph for this function:

autoscaler() [2/3]

kcenon::thread::autoscaler::autoscaler ( const autoscaler & )

delete

autoscaler() [3/3]

kcenon::thread::autoscaler::autoscaler ( autoscaler && )

delete

Member Function Documentation

add_workers()

auto kcenon::thread::autoscaler::add_workers ( std::size_t count ) -> common::VoidResult

private

Adds workers to the pool.

Parameters

count

Number of workers to add.

Returns

Error if operation fails.

Definition at line 542 of file autoscaler.cpp.

{
    if (count == 0)
    {
        return common::ok();
    }
 
    // Get current context from pool
    const auto& context = pool_.get_context();
 
    for (std::size_t i = 0; i < count; ++i)
    {
        auto worker = std::make_unique<thread_worker>(true, context);
        auto result = pool_.enqueue(std::move(worker));
        if (result.is_err())
        {
            return result;
        }
    }
 
    return common::ok();
}

can_scale_down()

auto kcenon::thread::autoscaler::can_scale_down ( ) const -> bool

nodiscardprivate

Checks if scale-down cooldown has elapsed.

Returns

true if scale-down is allowed.

Definition at line 528 of file autoscaler.cpp.

{
    if (pool_.get_active_worker_count() <= policy_.min_workers)
    {
        return false;
    }
 
    auto now = std::chrono::steady_clock::now();
    auto since_last = std::chrono::duration_cast<std::chrono::seconds>(
        now - last_scale_down_time_);
 
    return since_last >= policy_.scale_down_cooldown;
}

References kcenon::thread::thread_pool::get_active_worker_count(), last_scale_down_time_, kcenon::thread::autoscaling_policy::min_workers, policy_, pool_, and kcenon::thread::autoscaling_policy::scale_down_cooldown.

Here is the call graph for this function:

can_scale_up()

auto kcenon::thread::autoscaler::can_scale_up ( ) const -> bool

nodiscardprivate

Checks if scale-up cooldown has elapsed.

Returns

true if scale-up is allowed.

Definition at line 514 of file autoscaler.cpp.

{
    if (pool_.get_active_worker_count() >= policy_.max_workers)
    {
        return false;
    }
 
    auto now = std::chrono::steady_clock::now();
    auto since_last = std::chrono::duration_cast<std::chrono::seconds>(
        now - last_scale_up_time_);
 
    return since_last >= policy_.scale_up_cooldown;
}

References kcenon::thread::thread_pool::get_active_worker_count(), last_scale_up_time_, kcenon::thread::autoscaling_policy::max_workers, policy_, pool_, and kcenon::thread::autoscaling_policy::scale_up_cooldown.

Here is the call graph for this function:

collect_metrics()

auto kcenon::thread::autoscaler::collect_metrics ( ) const -> scaling_metrics_sample

nodiscardprivate

Collects current metrics from the pool.

Returns

Collected metrics sample.

Definition at line 289 of file autoscaler.cpp.

{
    auto now = std::chrono::steady_clock::now();
 
    scaling_metrics_sample sample;
    sample.timestamp = now;
    sample.worker_count = pool_.get_active_worker_count();
    sample.active_workers = sample.worker_count - pool_.get_idle_worker_count();
    sample.queue_depth = pool_.get_pending_task_count();
 
    // Calculate utilization
    if (sample.worker_count > 0)
    {
        sample.utilization = static_cast<double>(sample.active_workers) /
            static_cast<double>(sample.worker_count);
        sample.queue_depth_per_worker = static_cast<double>(sample.queue_depth) /
            static_cast<double>(sample.worker_count);
    }
 
    // Get metrics from pool
    auto metrics_snapshot = pool_.metrics().snapshot();
    sample.jobs_completed = metrics_snapshot.tasks_executed;
    sample.jobs_submitted = metrics_snapshot.tasks_submitted;
 
    // Calculate throughput if we have a previous sample
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(
        now - last_sample_time_).count();
    if (duration > 0 && sample.jobs_completed >= last_jobs_completed_)
    {
        std::uint64_t jobs_delta = sample.jobs_completed - last_jobs_completed_;
        sample.throughput_per_second = static_cast<double>(jobs_delta) * 1000.0 /
            static_cast<double>(duration);
    }
 
    // Get P95 latency from enhanced metrics if available
    // Note: Using P99 wait time as closest approximation to P95
    if (pool_.is_enhanced_metrics_enabled())
    {
        auto enhanced_snapshot = pool_.enhanced_metrics_snapshot();
        // Convert from microseconds to milliseconds
        sample.p95_latency_ms = enhanced_snapshot.wait_time_p99_us / 1000.0;
    }
 
    // Update cached values for next sample
    const_cast<autoscaler*>(this)->last_jobs_completed_ = sample.jobs_completed;
    const_cast<autoscaler*>(this)->last_jobs_submitted_ = sample.jobs_submitted;
    const_cast<autoscaler*>(this)->last_sample_time_ = now;
 
    return sample;
}

References kcenon::thread::scaling_metrics_sample::active_workers, kcenon::thread::thread_pool::enhanced_metrics_snapshot(), kcenon::thread::thread_pool::get_active_worker_count(), kcenon::thread::thread_pool::get_idle_worker_count(), kcenon::thread::thread_pool::get_pending_task_count(), kcenon::thread::thread_pool::is_enhanced_metrics_enabled(), kcenon::thread::scaling_metrics_sample::jobs_completed, kcenon::thread::scaling_metrics_sample::jobs_submitted, last_jobs_completed_, last_jobs_submitted_, last_sample_time_, kcenon::thread::thread_pool::metrics(), kcenon::thread::scaling_metrics_sample::p95_latency_ms, pool_, kcenon::thread::scaling_metrics_sample::queue_depth, kcenon::thread::scaling_metrics_sample::queue_depth_per_worker, kcenon::thread::metrics::ThreadPoolMetrics::snapshot(), kcenon::thread::scaling_metrics_sample::throughput_per_second, kcenon::thread::scaling_metrics_sample::timestamp, kcenon::thread::scaling_metrics_sample::utilization, and kcenon::thread::scaling_metrics_sample::worker_count.

Referenced by get_current_metrics().

Here is the call graph for this function:

Here is the caller graph for this function:

evaluate_now()

auto kcenon::thread::autoscaler::evaluate_now ( ) -> scaling_decision

nodiscard

Manually triggers a scaling evaluation.

Returns

The scaling decision that would be made.

This does not actually execute the scaling; use scale_to() or scale_up()/scale_down() to actually modify worker count.

Definition at line 75 of file autoscaler.cpp.

{
    // Collect current metrics
    auto sample = collect_metrics();
 
    // Add to history
    {
        std::scoped_lock<std::mutex> lock(history_mutex_);
        metrics_history_.push_back(sample);
        if (metrics_history_.size() > 60)
        {
            metrics_history_.pop_front();
        }
    }
 
    // Get recent samples for decision
    std::vector<scaling_metrics_sample> samples;
    {
        std::scoped_lock<std::mutex> lock(history_mutex_);
        std::size_t count = std::min(metrics_history_.size(), policy_.samples_for_decision);
        samples.reserve(count);
        auto it = metrics_history_.end();
        std::advance(it, -static_cast<std::ptrdiff_t>(count));
        for (; it != metrics_history_.end(); ++it)
        {
            samples.push_back(*it);
        }
    }
 
    // Make decision
    return make_decision(samples);
}

execute_scaling()

auto kcenon::thread::autoscaler::execute_scaling ( const scaling_decision & decision ) -> void

private

Executes a scaling decision.

Parameters

decision

The decision to execute.

Definition at line 465 of file autoscaler.cpp.

{
    std::size_t current_workers = pool_.get_active_worker_count();
    auto now = std::chrono::steady_clock::now();
 
    if (decision.direction == scaling_direction::up)
    {
        auto result = add_workers(decision.target_workers - current_workers);
        if (result.is_ok())
        {
            last_scale_up_time_ = now;
 
            std::scoped_lock<std::mutex> lock(stats_mutex_);
            ++stats_.scale_up_count;
            stats_.last_scale_up = now;
 
            if (policy_.scaling_callback)
            {
                policy_.scaling_callback(
                    scaling_direction::up,
                    decision.reason,
                    current_workers,
                    decision.target_workers);
            }
        }
    }
    else if (decision.direction == scaling_direction::down)
    {
        auto result = remove_workers(current_workers - decision.target_workers);
        if (result.is_ok())
        {
            last_scale_down_time_ = now;
 
            std::scoped_lock<std::mutex> lock(stats_mutex_);
            ++stats_.scale_down_count;
            stats_.last_scale_down = now;
 
            if (policy_.scaling_callback)
            {
                policy_.scaling_callback(
                    scaling_direction::down,
                    decision.reason,
                    current_workers,
                    decision.target_workers);
            }
        }
    }
}

References kcenon::thread::down, kcenon::thread::result< T >::is_ok(), and kcenon::thread::up.

Here is the call graph for this function:

get_current_metrics()

auto kcenon::thread::autoscaler::get_current_metrics ( ) const -> scaling_metrics_sample

nodiscard

Collects current metrics from the thread pool.

Returns

Current metrics sample.

Definition at line 165 of file autoscaler.cpp.

{
    return collect_metrics();
}

References collect_metrics().

Here is the call graph for this function:

get_metrics_history()

auto kcenon::thread::autoscaler::get_metrics_history ( std::size_t count = 60 ) const -> std::vector<scaling_metrics_sample>

nodiscard

Gets historical metrics samples.

Parameters

count

Maximum number of samples to return.

Returns

Vector of recent metrics samples.

Definition at line 170 of file autoscaler.cpp.

{
    std::scoped_lock<std::mutex> lock(history_mutex_);
 
    std::vector<scaling_metrics_sample> result;
    std::size_t actual_count = std::min(count, metrics_history_.size());
    result.reserve(actual_count);
 
    auto it = metrics_history_.end();
    std::advance(it, -static_cast<std::ptrdiff_t>(actual_count));
    for (; it != metrics_history_.end(); ++it)
    {
        result.push_back(*it);
    }
 
    return result;
}

get_policy()

auto kcenon::thread::autoscaler::get_policy ( ) const -> const autoscaling_policy&

nodiscard

Gets the current autoscaling policy.

Returns

Const reference to the policy.

Definition at line 160 of file autoscaler.cpp.

{
    return policy_;
}

References policy_.

get_stats()

auto kcenon::thread::autoscaler::get_stats ( ) const -> autoscaling_stats

nodiscard

Gets autoscaling statistics.

Returns

Statistics about scaling operations.

Definition at line 189 of file autoscaler.cpp.

{
    std::scoped_lock<std::mutex> lock(stats_mutex_);
    return stats_;
}

References stats_, and stats_mutex_.

is_active()

auto kcenon::thread::autoscaler::is_active ( ) const -> bool

nodiscard

Checks if the autoscaler is currently active.

Returns

true if the monitor thread is running.

Definition at line 70 of file autoscaler.cpp.

{
    return running_.load(std::memory_order_acquire);
}

References running_.

make_decision()

auto kcenon::thread::autoscaler::make_decision ( const std::vector< scaling_metrics_sample > & samples ) const -> scaling_decision

nodiscardprivate

Makes a scaling decision based on recent samples.

Parameters

samples

Recent metrics samples.

Returns

Scaling decision.

Definition at line 340 of file autoscaler.cpp.

{
    if (samples.empty())
    {
        return scaling_decision{};
    }
 
    // Calculate average metrics from samples
    double avg_utilization = 0.0;
    double avg_queue_depth_per_worker = 0.0;
    double avg_latency = 0.0;
    std::size_t avg_queue_depth = 0;
 
    for (const auto& sample : samples)
    {
        avg_utilization += sample.utilization;
        avg_queue_depth_per_worker += sample.queue_depth_per_worker;
        avg_latency += sample.p95_latency_ms;
        avg_queue_depth += sample.queue_depth;
    }
 
    auto sample_count = static_cast<double>(samples.size());
    avg_utilization /= sample_count;
    avg_queue_depth_per_worker /= sample_count;
    avg_latency /= sample_count;
    avg_queue_depth /= samples.size();
 
    std::size_t current_workers = pool_.get_active_worker_count();
 
    // Check scale-up triggers (ANY trigger)
    if (can_scale_up())
    {
        if (avg_utilization > policy_.scale_up.utilization_threshold)
        {
            std::size_t target = std::min(
                current_workers + policy_.scale_up_increment,
                policy_.max_workers);
 
            return scaling_decision{
                .direction = scaling_direction::up,
                .reason = scaling_reason::worker_utilization,
                .target_workers = target,
                .explanation = utility_module::formatter::format(
                    "Utilization {:.1f}% exceeds threshold {:.1f}%",
                    avg_utilization * 100, policy_.scale_up.utilization_threshold * 100)
            };
        }
 
        if (avg_queue_depth_per_worker > policy_.scale_up.queue_depth_threshold)
        {
            std::size_t target = std::min(
                current_workers + policy_.scale_up_increment,
                policy_.max_workers);
 
            return scaling_decision{
                .direction = scaling_direction::up,
                .reason = scaling_reason::queue_depth,
                .target_workers = target,
                .explanation = utility_module::formatter::format(
                    "Queue depth per worker {:.1f} exceeds threshold {:.1f}",
                    avg_queue_depth_per_worker, policy_.scale_up.queue_depth_threshold)
            };
        }
 
        if (avg_latency > policy_.scale_up.latency_threshold_ms && avg_latency > 0)
        {
            std::size_t target = std::min(
                current_workers + policy_.scale_up_increment,
                policy_.max_workers);
 
            return scaling_decision{
                .direction = scaling_direction::up,
                .reason = scaling_reason::latency,
                .target_workers = target,
                .explanation = utility_module::formatter::format(
                    "P95 latency {:.1f}ms exceeds threshold {:.1f}ms",
                    avg_latency, policy_.scale_up.latency_threshold_ms)
            };
        }
 
        if (avg_queue_depth > policy_.scale_up.pending_jobs_threshold)
        {
            std::size_t target = std::min(
                current_workers + policy_.scale_up_increment,
                policy_.max_workers);
 
            return scaling_decision{
                .direction = scaling_direction::up,
                .reason = scaling_reason::queue_depth,
                .target_workers = target,
                .explanation = utility_module::formatter::format(
                    "Queue depth {} exceeds threshold {}",
                    avg_queue_depth, policy_.scale_up.pending_jobs_threshold)
            };
        }
    }
 
    // Check scale-down triggers (ALL triggers)
    if (can_scale_down() && current_workers > policy_.min_workers)
    {
        bool utilization_ok = avg_utilization < policy_.scale_down.utilization_threshold;
        bool queue_depth_ok = avg_queue_depth_per_worker < policy_.scale_down.queue_depth_threshold;
 
        if (utilization_ok && queue_depth_ok)
        {
            std::size_t target = std::max(
                current_workers - policy_.scale_down_increment,
                policy_.min_workers);
 
            return scaling_decision{
                .direction = scaling_direction::down,
                .reason = scaling_reason::worker_utilization,
                .target_workers = target,
                .explanation = utility_module::formatter::format(
                    "Utilization {:.1f}% below threshold {:.1f}%, queue depth {:.1f} below {:.1f}",
                    avg_utilization * 100, policy_.scale_down.utilization_threshold * 100,
                    avg_queue_depth_per_worker, policy_.scale_down.queue_depth_threshold)
            };
        }
    }
 
    return scaling_decision{};
}

References kcenon::thread::down, utility_module::formatter::format(), kcenon::thread::latency, kcenon::thread::queue_depth, kcenon::thread::up, and kcenon::thread::worker_utilization.

Here is the call graph for this function:

monitor_loop()

auto kcenon::thread::autoscaler::monitor_loop ( ) -> void

private

Main monitoring loop running in the background thread.

Definition at line 203 of file autoscaler.cpp.

{
    while (running_.load(std::memory_order_acquire))
    {
        // Wait for sample interval
        {
            std::unique_lock<std::mutex> lock(mutex_);
            cv_.wait_for(lock, policy_.sample_interval, [this]() {
                return !running_.load(std::memory_order_acquire);
            });
        }
 
        if (!running_.load(std::memory_order_acquire))
        {
            break;
        }
 
        // Skip if pool is not running
        if (!pool_.is_running())
        {
            continue;
        }
 
        // Collect metrics
        auto sample = collect_metrics();
 
        // Add to history
        {
            std::scoped_lock<std::mutex> lock(history_mutex_);
            metrics_history_.push_back(sample);
 
            // Keep max 60 samples (1 minute at 1s interval)
            while (metrics_history_.size() > 60)
            {
                metrics_history_.pop_front();
            }
        }
 
        // Only auto-scale in automatic mode
        if (policy_.scaling_mode != autoscaling_policy::mode::automatic)
        {
            continue;
        }
 
        // Collect samples for decision
        std::vector<scaling_metrics_sample> samples;
        {
            std::scoped_lock<std::mutex> lock(history_mutex_);
            std::size_t count = std::min(metrics_history_.size(), policy_.samples_for_decision);
            if (count < policy_.samples_for_decision)
            {
                // Not enough samples yet
                continue;
            }
 
            samples.reserve(count);
            auto it = metrics_history_.end();
            std::advance(it, -static_cast<std::ptrdiff_t>(count));
            for (; it != metrics_history_.end(); ++it)
            {
                samples.push_back(*it);
            }
        }
 
        // Make and execute decision
        auto decision = make_decision(samples);
        if (decision.should_scale())
        {
            execute_scaling(decision);
        }
 
        // Update stats
        {
            std::scoped_lock<std::mutex> lock(stats_mutex_);
            ++stats_.decisions_evaluated;
 
            std::size_t current = pool_.get_active_worker_count();
            stats_.peak_workers = std::max(stats_.peak_workers, current);
            if (stats_.min_workers == 0 || current < stats_.min_workers)
            {
                stats_.min_workers = current;
            }
        }
    }
}

References kcenon::thread::autoscaling_policy::automatic.

operator=() [1/2]

autoscaler & kcenon::thread::autoscaler::operator= ( autoscaler && )

delete

operator=() [2/2]

autoscaler & kcenon::thread::autoscaler::operator= ( const autoscaler & )

delete

remove_workers()

auto kcenon::thread::autoscaler::remove_workers ( std::size_t count ) -> common::VoidResult

private

Removes workers from the pool.

Parameters

count

Number of workers to remove.

Returns

Error if operation fails.

Definition at line 565 of file autoscaler.cpp.

{
    if (count == 0)
    {
        return common::ok();
    }
 
    // Request pool to remove workers using internal method
    // This will gracefully stop idle workers
    auto result = pool_.remove_workers_internal(count, policy_.min_workers);
    return result;
}

reset_stats()

auto kcenon::thread::autoscaler::reset_stats

(

)

-> void

Resets autoscaling statistics.

Definition at line 195 of file autoscaler.cpp.

{
    std::scoped_lock<std::mutex> lock(stats_mutex_);
    stats_ = autoscaling_stats{};
    stats_.min_workers = pool_.get_active_worker_count();
    stats_.peak_workers = stats_.min_workers;
}

References kcenon::thread::autoscaling_stats::min_workers.

scale_down()

auto kcenon::thread::autoscaler::scale_down

(

)

-> common::VoidResult

Manually scales down by the configured increment.

Returns

Error if scaling fails.

Definition at line 143 of file autoscaler.cpp.

{
    std::size_t current = pool_.get_active_worker_count();
    std::size_t target = current > policy_.scale_down_increment
        ? current - policy_.scale_down_increment
        : policy_.min_workers;
 
    target = std::max(target, policy_.min_workers);
    return scale_to(target);
}

scale_to()

auto kcenon::thread::autoscaler::scale_to

(

std::size_t

target_workers

)

-> common::VoidResult

Manually scales to a specific worker count.

Parameters

target_workers

Desired number of workers.

Returns

Error if scaling fails.

The target is clamped to [min_workers, max_workers] from the policy.

Definition at line 108 of file autoscaler.cpp.

{
    // Clamp to policy bounds
    target_workers = std::clamp(target_workers, policy_.min_workers, policy_.max_workers);
 
    std::size_t current_workers = pool_.get_active_worker_count();
 
    if (target_workers > current_workers)
    {
        return add_workers(target_workers - current_workers);
    }
    else if (target_workers < current_workers)
    {
        return remove_workers(current_workers - target_workers);
    }
 
    return common::ok();
}

scale_up()

auto kcenon::thread::autoscaler::scale_up

(

)

-> common::VoidResult

Manually scales up by the configured increment.

Returns

Error if scaling fails.

Definition at line 127 of file autoscaler.cpp.

{
    std::size_t current = pool_.get_active_worker_count();
    std::size_t increment = policy_.use_multiplicative_scaling
        ? static_cast<std::size_t>(current * (policy_.scale_up_factor - 1.0))
        : policy_.scale_up_increment;
 
    if (increment == 0)
    {
        increment = 1;
    }
 
    std::size_t target = std::min(current + increment, policy_.max_workers);
    return scale_to(target);
}

set_policy()

auto kcenon::thread::autoscaler::set_policy

(

autoscaling_policy

policy

)

-> void

Updates the autoscaling policy.

Parameters

policy

New policy configuration.

Definition at line 154 of file autoscaler.cpp.

{
    std::scoped_lock<std::mutex> lock(mutex_);
    policy_ = std::move(policy);
}

start()

auto kcenon::thread::autoscaler::start

(

)

-> void

Starts the autoscaling monitor thread.

The monitor thread periodically collects metrics and makes scaling decisions based on the configured policy.

Definition at line 32 of file autoscaler.cpp.

{
    bool expected = false;
    if (!running_.compare_exchange_strong(expected, true))
    {
        // Already running
        return;
    }
 
    // Start monitor thread
    monitor_thread_ = std::make_unique<std::thread>([this]() {
        monitor_loop();
    });
}

stop()

auto kcenon::thread::autoscaler::stop

(

)

-> void

Stops the autoscaling monitor thread.

Waits for the monitor thread to complete before returning.

Definition at line 47 of file autoscaler.cpp.

{
    bool expected = true;
    if (!running_.compare_exchange_strong(expected, false))
    {
        // Already stopped
        return;
    }
 
    // Wake up monitor thread
    {
        std::lock_guard<std::mutex> lock(mutex_);
        cv_.notify_one();
    }
 
    // Wait for thread to complete
    if (monitor_thread_ && monitor_thread_->joinable())
    {
        monitor_thread_->join();
    }
    monitor_thread_.reset();
}

Referenced by ~autoscaler().

Here is the caller graph for this function:

Member Data Documentation

cv_

std::condition_variable kcenon::thread::autoscaler::cv_

private

Definition at line 278 of file autoscaler.h.

history_mutex_

std::mutex kcenon::thread::autoscaler::history_mutex_

mutableprivate

Definition at line 281 of file autoscaler.h.

last_jobs_completed_

std::uint64_t kcenon::thread::autoscaler::last_jobs_completed_ {0}

private

Definition at line 290 of file autoscaler.h.

{0};

Referenced by collect_metrics().

last_jobs_submitted_

std::uint64_t kcenon::thread::autoscaler::last_jobs_submitted_ {0}

private

Definition at line 291 of file autoscaler.h.

{0};

Referenced by collect_metrics().

last_sample_time_

std::chrono::steady_clock::time_point kcenon::thread::autoscaler::last_sample_time_

private

Definition at line 292 of file autoscaler.h.

Referenced by collect_metrics().

last_scale_down_time_

std::chrono::steady_clock::time_point kcenon::thread::autoscaler::last_scale_down_time_

private

Definition at line 284 of file autoscaler.h.

Referenced by can_scale_down().

last_scale_up_time_

std::chrono::steady_clock::time_point kcenon::thread::autoscaler::last_scale_up_time_

private

Definition at line 283 of file autoscaler.h.

Referenced by can_scale_up().

metrics_history_

std::deque<scaling_metrics_sample> kcenon::thread::autoscaler::metrics_history_

private

Definition at line 280 of file autoscaler.h.

monitor_thread_

std::unique_ptr<std::thread> kcenon::thread::autoscaler::monitor_thread_

private

Definition at line 275 of file autoscaler.h.

mutex_

std::mutex kcenon::thread::autoscaler::mutex_

mutableprivate

Definition at line 277 of file autoscaler.h.

policy_

autoscaling_policy kcenon::thread::autoscaler::policy_

private

Definition at line 272 of file autoscaler.h.

Referenced by can_scale_down(), can_scale_up(), and get_policy().

pool_

thread_pool& kcenon::thread::autoscaler::pool_

private

Definition at line 271 of file autoscaler.h.

Referenced by autoscaler(), can_scale_down(), can_scale_up(), and collect_metrics().

running_

std::atomic<bool> kcenon::thread::autoscaler::running_ {false}

private

Definition at line 274 of file autoscaler.h.

{false};

Referenced by is_active().

stats_

autoscaling_stats kcenon::thread::autoscaler::stats_

private

Definition at line 286 of file autoscaler.h.

Referenced by autoscaler(), and get_stats().

stats_mutex_

std::mutex kcenon::thread::autoscaler::stats_mutex_

mutableprivate

Definition at line 287 of file autoscaler.h.

Referenced by autoscaler(), and get_stats().

---

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

• include/kcenon/thread/scaling/autoscaler.h
• src/scaling/autoscaler.cpp