autoscaler.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2024, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include "autoscaling_policy.h"
#include "scaling_metrics.h"
 
#include <kcenon/thread/core/error_handling.h>
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <deque>
#include <memory>
#include <mutex>
#include <thread>
 
namespace kcenon::thread
{
    // Forward declaration
    class thread_pool;
 
    class autoscaler
    {
    public:
        explicit autoscaler(thread_pool& pool, autoscaling_policy policy = {});
 
        ~autoscaler();
 
        // Non-copyable, non-movable
        autoscaler(const autoscaler&) = delete;
        autoscaler& operator=(const autoscaler&) = delete;
        autoscaler(autoscaler&&) = delete;
        autoscaler& operator=(autoscaler&&) = delete;
 
        // ============================================
        // Control
        // ============================================
 
        auto start() -> void;
 
        auto stop() -> void;
 
        [[nodiscard]] auto is_active() const -> bool;
 
        // ============================================
        // Manual Triggers
        // ============================================
 
        [[nodiscard]] auto evaluate_now() -> scaling_decision;
 
        auto scale_to(std::size_t target_workers) -> common::VoidResult;
 
        auto scale_up() -> common::VoidResult;
 
        auto scale_down() -> common::VoidResult;
 
        // ============================================
        // Configuration
        // ============================================
 
        auto set_policy(autoscaling_policy policy) -> void;
 
        [[nodiscard]] auto get_policy() const -> const autoscaling_policy&;
 
        // ============================================
        // Metrics & Statistics
        // ============================================
 
        [[nodiscard]] auto get_current_metrics() const -> scaling_metrics_sample;
 
        [[nodiscard]] auto get_metrics_history(std::size_t count = 60) const
            -> std::vector<scaling_metrics_sample>;
 
        [[nodiscard]] auto get_stats() const -> autoscaling_stats;
 
        auto reset_stats() -> void;
 
    private:
        auto monitor_loop() -> void;
 
        [[nodiscard]] auto collect_metrics() const -> scaling_metrics_sample;
 
        [[nodiscard]] auto make_decision(const std::vector<scaling_metrics_sample>& samples) const
            -> scaling_decision;
 
        auto execute_scaling(const scaling_decision& decision) -> void;
 
        [[nodiscard]] auto can_scale_up() const -> bool;
 
        [[nodiscard]] auto can_scale_down() const -> bool;
 
        auto add_workers(std::size_t count) -> common::VoidResult;
 
        auto remove_workers(std::size_t count) -> common::VoidResult;
 
        thread_pool& pool_;
        autoscaling_policy policy_;
 
        std::atomic<bool> running_{false};
        std::unique_ptr<std::thread> monitor_thread_;
 
        mutable std::mutex mutex_;
        std::condition_variable cv_;
 
        std::deque<scaling_metrics_sample> metrics_history_;
        mutable 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_;
        mutable std::mutex stats_mutex_;
 
        // Cached values from previous sample for delta calculations
        std::uint64_t last_jobs_completed_{0};
        std::uint64_t last_jobs_submitted_{0};
        std::chrono::steady_clock::time_point last_sample_time_;
    };
 
} // namespace kcenon::thread