resource_manager.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2021-2025, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <atomic>
#include <chrono>
#include <functional>
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>
 
#include "kcenon/monitoring/core/result_types.h"
 
namespace kcenon::monitoring {
 
// ============================================================================
// Enums
// ============================================================================
 
enum class throttling_strategy {
    reject,  
    delay    
};
 
enum class resource_type {
    memory,  
    cpu      
};
 
// ============================================================================
// Metrics Types
// ============================================================================
 
struct resource_metrics {
    std::atomic<size_t> current_usage{0};
    std::atomic<size_t> total_allocations{0};
    std::atomic<size_t> peak_usage{0};
    std::atomic<size_t> rejected_operations{0};
    std::atomic<size_t> delayed_operations{0};
 
    resource_metrics() = default;
    resource_metrics(const resource_metrics& other)
        : current_usage(other.current_usage.load())
        , total_allocations(other.total_allocations.load())
        , peak_usage(other.peak_usage.load())
        , rejected_operations(other.rejected_operations.load())
        , delayed_operations(other.delayed_operations.load()) {}
 
    resource_metrics& operator=(const resource_metrics& other) {
        if (this != &other) {
            current_usage.store(other.current_usage.load());
            total_allocations.store(other.total_allocations.load());
            peak_usage.store(other.peak_usage.load());
            rejected_operations.store(other.rejected_operations.load());
            delayed_operations.store(other.delayed_operations.load());
        }
        return *this;
    }
};
 
// ============================================================================
// Configuration Types
// ============================================================================
 
struct rate_limit_config {
    double rate_per_second = 100.0;   
    size_t burst_capacity = 10;       
    throttling_strategy strategy = throttling_strategy::reject;
 
    bool validate() const {
        return rate_per_second > 0 && burst_capacity > 0;
    }
};
 
struct resource_quota {
    resource_type type = resource_type::memory;
    size_t max_value = 0;              
    size_t warning_threshold = 0;      
    size_t critical_threshold = 0;     
    throttling_strategy strategy = throttling_strategy::reject;
 
    resource_quota() = default;
    resource_quota(resource_type t, size_t max, throttling_strategy s = throttling_strategy::reject)
        : type(t), max_value(max), warning_threshold(max * 70 / 100),
          critical_threshold(max * 90 / 100), strategy(s) {}
 
    bool validate() const {
        if (max_value == 0) {
            return false;
        }
        if (warning_threshold > max_value) {
            return false;
        }
        if (critical_threshold > max_value) {
            return false;
        }
        if (warning_threshold > critical_threshold) {
            return false;
        }
        return true;
    }
};
 
struct cpu_throttle_config {
    double max_cpu_usage = 0.8;        
    double warning_threshold = 0.7;    
    throttling_strategy strategy = throttling_strategy::reject;
    std::chrono::milliseconds check_interval = std::chrono::milliseconds(100);
 
    bool validate() const {
        if (max_cpu_usage <= 0.0 || max_cpu_usage > 1.0) {
            return false;
        }
        if (warning_threshold <= 0.0 || warning_threshold > 1.0) {
            return false;
        }
        if (warning_threshold > max_cpu_usage) {
            return false;
        }
        return true;
    }
};
 
// ============================================================================
// Rate Limiter Interface
// ============================================================================
 
class rate_limiter {
public:
    virtual ~rate_limiter() = default;
 
    virtual bool try_acquire(size_t count = 1) = 0;
 
    template<typename Func>
    auto execute(Func&& func) -> decltype(func()) {
        using result_type = decltype(func());
        using value_type = typename result_type::value_type;
 
        if (!try_acquire(1)) {
            return common::make_error<value_type>(static_cast<int>(monitoring_error_code::resource_exhausted),
                                          "Rate limit exceeded for '" + get_name() + "'");
        }
        return func();
    }
 
    virtual const std::string& get_name() const = 0;
};
 
// ============================================================================
// Token Bucket Rate Limiter
// ============================================================================
 
class token_bucket_limiter : public rate_limiter {
public:
    using clock = std::chrono::steady_clock;
 
    token_bucket_limiter(const std::string& name, double rate, size_t capacity,
                         throttling_strategy /*strategy*/ = throttling_strategy::reject)
        : name_(name)
        , rate_(rate)
        , capacity_(capacity)
        , tokens_(static_cast<double>(capacity))
        , last_refill_(clock::now()) {}
 
    bool try_acquire(size_t count = 1) override {
        std::lock_guard<std::mutex> lock(mutex_);
        refill();
 
        if (tokens_ >= static_cast<double>(count)) {
            tokens_ -= static_cast<double>(count);
            return true;
        }
        return false;
    }
 
    const std::string& get_name() const override {
        return name_;
    }
 
private:
    void refill() {
        auto now = clock::now();
        auto elapsed = std::chrono::duration<double>(now - last_refill_).count();
        tokens_ = std::min(static_cast<double>(capacity_), tokens_ + elapsed * rate_);
        last_refill_ = now;
    }
 
    std::string name_;
    double rate_;
    size_t capacity_;
    double tokens_;
    clock::time_point last_refill_;
    mutable std::mutex mutex_;
};
 
// ============================================================================
// Leaky Bucket Rate Limiter
// ============================================================================
 
class leaky_bucket_limiter : public rate_limiter {
public:
    using clock = std::chrono::steady_clock;
 
    leaky_bucket_limiter(const std::string& name, double rate, size_t capacity)
        : name_(name)
        , rate_(rate)
        , capacity_(capacity)
        , water_(0.0)
        , last_leak_(clock::now()) {}
 
    bool try_acquire(size_t count = 1) override {
        std::lock_guard<std::mutex> lock(mutex_);
        leak();
 
        if (water_ + static_cast<double>(count) <= static_cast<double>(capacity_)) {
            water_ += static_cast<double>(count);
            return true;
        }
        return false;
    }
 
    const std::string& get_name() const override {
        return name_;
    }
 
private:
    void leak() {
        auto now = clock::now();
        auto elapsed = std::chrono::duration<double>(now - last_leak_).count();
        water_ = std::max(0.0, water_ - elapsed * rate_);
        last_leak_ = now;
    }
 
    std::string name_;
    double rate_;
    size_t capacity_;
    double water_;
    clock::time_point last_leak_;
    mutable std::mutex mutex_;
};
 
// ============================================================================
// Memory Quota Manager
// ============================================================================
 
class memory_quota_manager {
public:
    memory_quota_manager(const std::string& name, const resource_quota& quota)
        : name_(name), quota_(quota) {}
 
    memory_quota_manager(const std::string& name, size_t max_bytes,
                         throttling_strategy strategy = throttling_strategy::reject)
        : name_(name), quota_(resource_type::memory, max_bytes, strategy) {}
 
    common::Result<bool> allocate(size_t bytes) {
        std::lock_guard<std::mutex> lock(mutex_);
 
        if (metrics_.current_usage.load() + bytes > quota_.max_value) {
            metrics_.rejected_operations++;
            return common::make_error<bool>(static_cast<int>(monitoring_error_code::resource_exhausted),
                                   "Memory quota exceeded for '" + name_ + "'");
        }
 
        metrics_.current_usage += bytes;
        metrics_.total_allocations++;
 
        // Update peak usage
        size_t current = metrics_.current_usage.load();
        size_t peak = metrics_.peak_usage.load();
        while (current > peak && !metrics_.peak_usage.compare_exchange_weak(peak, current)) {
            // Retry until successful
        }
 
        return common::ok(true);
    }
 
    void deallocate(size_t bytes) {
        std::lock_guard<std::mutex> lock(mutex_);
        size_t current = metrics_.current_usage.load();
        metrics_.current_usage.store(current >= bytes ? current - bytes : 0);
    }
 
    size_t current_usage() const {
        return metrics_.current_usage.load();
    }
 
    bool is_over_warning_threshold() const {
        return metrics_.current_usage.load() >= quota_.warning_threshold;
    }
 
    bool is_over_critical_threshold() const {
        return metrics_.current_usage.load() >= quota_.critical_threshold;
    }
 
    resource_metrics get_metrics() const {
        return metrics_;
    }
 
    const std::string& get_name() const {
        return name_;
    }
 
private:
    std::string name_;
    resource_quota quota_;
    mutable resource_metrics metrics_;
    mutable std::mutex mutex_;
};
 
// ============================================================================
// CPU Throttler
// ============================================================================
 
class cpu_throttler {
public:
    cpu_throttler(const std::string& name, const cpu_throttle_config& config)
        : name_(name), config_(config) {}
 
    template<typename Func>
    auto execute(Func&& func) -> decltype(func()) {
        // For simplicity, we just execute the function and track metrics
        // Real CPU monitoring would require platform-specific code
        metrics_.total_allocations++;
 
        return std::forward<Func>(func)();
    }
 
    resource_metrics get_metrics() const {
        return metrics_;
    }
 
    const std::string& get_name() const {
        return name_;
    }
 
private:
    std::string name_;
    cpu_throttle_config config_;
    mutable resource_metrics metrics_;
};
 
// ============================================================================
// Resource Manager
// ============================================================================
 
class resource_manager {
public:
    explicit resource_manager(const std::string& name) : name_(name) {}
 
    common::Result<bool> add_rate_limiter(const std::string& name, const rate_limit_config& config) {
        std::lock_guard<std::mutex> lock(mutex_);
 
        if (rate_limiters_.find(name) != rate_limiters_.end()) {
            return common::make_error<bool>(static_cast<int>(monitoring_error_code::already_exists),
                                   "Rate limiter '" + name + "' already exists");
        }
 
        rate_limiters_[name] = std::make_unique<token_bucket_limiter>(
            name, config.rate_per_second, config.burst_capacity, config.strategy);
        return common::ok(true);
    }
 
    rate_limiter* get_rate_limiter(const std::string& name) {
        std::lock_guard<std::mutex> lock(mutex_);
        auto it = rate_limiters_.find(name);
        return it != rate_limiters_.end() ? it->second.get() : nullptr;
    }
 
    common::Result<bool> add_memory_quota(const std::string& name, const resource_quota& quota) {
        std::lock_guard<std::mutex> lock(mutex_);
 
        if (memory_quotas_.find(name) != memory_quotas_.end()) {
            return common::make_error<bool>(static_cast<int>(monitoring_error_code::already_exists),
                                   "Memory quota '" + name + "' already exists");
        }
 
        memory_quotas_[name] = std::make_unique<memory_quota_manager>(name, quota);
        return common::ok(true);
    }
 
    memory_quota_manager* get_memory_quota(const std::string& name) {
        std::lock_guard<std::mutex> lock(mutex_);
        auto it = memory_quotas_.find(name);
        return it != memory_quotas_.end() ? it->second.get() : nullptr;
    }
 
    common::Result<bool> add_cpu_throttler(const std::string& name, const cpu_throttle_config& config) {
        std::lock_guard<std::mutex> lock(mutex_);
 
        if (cpu_throttlers_.find(name) != cpu_throttlers_.end()) {
            return common::make_error<bool>(static_cast<int>(monitoring_error_code::already_exists),
                                   "CPU throttler '" + name + "' already exists");
        }
 
        cpu_throttlers_[name] = std::make_unique<cpu_throttler>(name, config);
        return common::ok(true);
    }
 
    cpu_throttler* get_cpu_throttler(const std::string& name) {
        std::lock_guard<std::mutex> lock(mutex_);
        auto it = cpu_throttlers_.find(name);
        return it != cpu_throttlers_.end() ? it->second.get() : nullptr;
    }
 
    common::Result<bool> is_healthy() const {
        std::lock_guard<std::mutex> lock(mutex_);
 
        for (const auto& [name, manager] : memory_quotas_) {
            if (manager->is_over_critical_threshold()) {
                return common::ok(false);
            }
        }
 
        return common::ok(true);
    }
 
    std::unordered_map<std::string, resource_metrics> get_all_metrics() const {
        std::lock_guard<std::mutex> lock(mutex_);
        std::unordered_map<std::string, resource_metrics> all_metrics;
 
        for (const auto& [name, limiter] : rate_limiters_) {
            all_metrics["rate_" + name] = resource_metrics{};
        }
 
        for (const auto& [name, manager] : memory_quotas_) {
            all_metrics["memory_" + name] = manager->get_metrics();
        }
 
        for (const auto& [name, throttler] : cpu_throttlers_) {
            all_metrics["cpu_" + name] = throttler->get_metrics();
        }
 
        return all_metrics;
    }
 
private:
    std::string name_;
    std::unordered_map<std::string, std::unique_ptr<rate_limiter>> rate_limiters_;
    std::unordered_map<std::string, std::unique_ptr<memory_quota_manager>> memory_quotas_;
    std::unordered_map<std::string, std::unique_ptr<cpu_throttler>> cpu_throttlers_;
    mutable std::mutex mutex_;
};
 
// ============================================================================
// Factory Functions
// ============================================================================
 
inline std::unique_ptr<token_bucket_limiter> create_token_bucket_limiter(
    const std::string& name, double rate, size_t capacity,
    throttling_strategy strategy = throttling_strategy::reject) {
    return std::make_unique<token_bucket_limiter>(name, rate, capacity, strategy);
}
 
inline std::unique_ptr<leaky_bucket_limiter> create_leaky_bucket_limiter(
    const std::string& name, double rate, size_t capacity) {
    return std::make_unique<leaky_bucket_limiter>(name, rate, capacity);
}
 
inline std::unique_ptr<memory_quota_manager> create_memory_quota_manager(
    const std::string& name, size_t max_bytes,
    throttling_strategy strategy = throttling_strategy::reject) {
    return std::make_unique<memory_quota_manager>(name, max_bytes, strategy);
}
 
inline std::unique_ptr<resource_manager> create_resource_manager(const std::string& name) {
    return std::make_unique<resource_manager>(name);
}
 
} // namespace kcenon::monitoring