thread_system_config.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 "../core/backpressure_config.h"
#include <kcenon/common/resilience/circuit_breaker_config.h>
#include "../dag/dag_config.h"
#include "../impl/typed_pool/priority_aging_config.h"
 
#include <chrono>
#include <cstddef>
#include <thread>
 
namespace kcenon::thread
{
    using common::resilience::circuit_breaker_config;
 
    // Forward declaration for builder pattern
    class config_builder;
 
    struct thread_system_config
    {
        struct pool_config
        {
            std::size_t worker_count = std::thread::hardware_concurrency();
 
            std::size_t queue_capacity = 10000;
 
            backpressure_config backpressure{};
 
            std::chrono::milliseconds wake_interval{100};
 
            std::chrono::seconds shutdown_timeout{5};
 
            std::chrono::seconds worker_idle_timeout{30};
 
            bool yield_on_idle = true;
 
            bool enable_work_stealing = false;
 
            std::size_t max_steal_attempts = 3;
 
            std::chrono::microseconds steal_backoff{50};
        } pool;
 
        struct resilience_config
        {
            circuit_breaker_config circuit_breaker{};
        } resilience;
 
        dag_config dag{};
 
        struct scaling_config
        {
            bool auto_scaling_enabled = false;
 
            priority_aging_config priority_aging{};
        } scaling;
 
        [[nodiscard]] auto is_valid() const -> bool
        {
            // Pool validation
            if (pool.worker_count == 0)
            {
                return false;
            }
 
            // Backpressure validation
            if (!pool.backpressure.is_valid())
            {
                return false;
            }
 
            // DAG validation
            if (dag.max_retries > 10)
            {
                return false; // Sanity check to prevent infinite retries
            }
 
            return true;
        }
 
        static auto builder() -> config_builder;
    };
 
    class config_builder
    {
    public:
        config_builder() = default;
 
        auto with_worker_count(std::size_t count) -> config_builder&
        {
            config_.pool.worker_count = count;
            return *this;
        }
 
        auto with_queue_capacity(std::size_t capacity) -> config_builder&
        {
            config_.pool.queue_capacity = capacity;
            return *this;
        }
 
        auto enable_backpressure() -> config_builder&
        {
            config_.pool.backpressure.policy = backpressure_policy::block;
            return *this;
        }
 
        auto with_backpressure_policy(backpressure_policy policy) -> config_builder&
        {
            config_.pool.backpressure.policy = policy;
            return *this;
        }
 
        auto with_watermarks(double low, double high) -> config_builder&
        {
            config_.pool.backpressure.low_watermark = low;
            config_.pool.backpressure.high_watermark = high;
            return *this;
        }
 
        auto enable_circuit_breaker() -> config_builder&
        {
            config_.resilience.circuit_breaker.failure_threshold = 5;
            return *this;
        }
 
        auto with_failure_threshold(std::size_t threshold) -> config_builder&
        {
            config_.resilience.circuit_breaker.failure_threshold = threshold;
            return *this;
        }
 
        auto with_open_duration(std::chrono::seconds duration) -> config_builder&
        {
            // Note: common_system uses 'timeout' instead of 'open_duration'
            config_.resilience.circuit_breaker.timeout = duration;
            return *this;
        }
 
        auto enable_work_stealing() -> config_builder&
        {
            config_.pool.enable_work_stealing = true;
            return *this;
        }
 
        auto with_work_stealing_params(std::size_t max_attempts,
                                        std::chrono::microseconds backoff) -> config_builder&
        {
            config_.pool.max_steal_attempts = max_attempts;
            config_.pool.steal_backoff = backoff;
            return *this;
        }
 
        auto enable_priority_aging() -> config_builder&
        {
            config_.scaling.priority_aging.enabled = true;
            return *this;
        }
 
        auto with_priority_aging_params(std::chrono::milliseconds interval,
                                         int boost,
                                         int max_boost) -> config_builder&
        {
            config_.scaling.priority_aging.aging_interval = interval;
            config_.scaling.priority_aging.priority_boost_per_interval = boost;
            config_.scaling.priority_aging.max_priority_boost = max_boost;
            return *this;
        }
 
        auto enable_auto_scaling() -> config_builder&
        {
            config_.scaling.auto_scaling_enabled = true;
            return *this;
        }
 
        auto with_dag_failure_policy(dag_failure_policy policy) -> config_builder&
        {
            config_.dag.failure_policy = policy;
            return *this;
        }
 
        auto with_dag_retry_params(std::size_t max_retries,
                                    std::chrono::milliseconds delay) -> config_builder&
        {
            config_.dag.max_retries = max_retries;
            config_.dag.retry_delay = delay;
            return *this;
        }
 
        auto build() -> thread_system_config
        {
            if (!config_.is_valid())
            {
                throw std::invalid_argument("Invalid thread_system_config");
            }
            return config_;
        }
 
    private:
        thread_system_config config_;
    };
 
    inline auto thread_system_config::builder() -> config_builder
    {
        return config_builder{};
    }
 
} // namespace kcenon::thread