backpressure_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 "job.h"
 
#include <atomic>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <memory>
#include <string>
 
namespace kcenon::thread
{
    enum class backpressure_policy
    {
        block,        
        drop_oldest,  
        drop_newest,  
        callback,     
        adaptive      
    };
 
    enum class backpressure_decision
    {
        accept,           
        reject,           
        drop_and_accept,  
        delay             
    };
 
    enum class pressure_level
    {
        none,      
        low,       
        high,      
        critical   
    };
 
    [[nodiscard]] inline auto pressure_level_to_string(pressure_level level) -> std::string
    {
        switch (level)
        {
            case pressure_level::none: return "none";
            case pressure_level::low: return "low";
            case pressure_level::high: return "high";
            case pressure_level::critical: return "critical";
            default: return "unknown";
        }
    }
 
    [[nodiscard]] inline auto backpressure_policy_to_string(backpressure_policy policy) -> std::string
    {
        switch (policy)
        {
            case backpressure_policy::block: return "block";
            case backpressure_policy::drop_oldest: return "drop_oldest";
            case backpressure_policy::drop_newest: return "drop_newest";
            case backpressure_policy::callback: return "callback";
            case backpressure_policy::adaptive: return "adaptive";
            default: return "unknown";
        }
    }
 
    struct backpressure_config
    {
        // =========================================================================
        // Policy Selection
        // =========================================================================
 
        backpressure_policy policy = backpressure_policy::block;
 
        // =========================================================================
        // Watermarks
        // =========================================================================
 
        double high_watermark = 0.8;
 
        double low_watermark = 0.5;
 
        // =========================================================================
        // Blocking Behavior
        // =========================================================================
 
        std::chrono::milliseconds block_timeout{5000};
 
        // =========================================================================
        // Rate Limiting (Token Bucket)
        // =========================================================================
 
        bool enable_rate_limiting = false;
 
        std::size_t rate_limit_tokens_per_second = 10000;
 
        std::size_t rate_limit_burst_size = 1000;
 
        // =========================================================================
        // Callbacks
        // =========================================================================
 
        std::function<void(std::size_t queue_depth, double pressure_ratio)> pressure_callback;
 
        std::function<backpressure_decision(std::unique_ptr<job>&)> decision_callback;
 
        // =========================================================================
        // Adaptive Mode Settings
        // =========================================================================
 
        std::chrono::milliseconds adaptive_sample_interval{100};
 
        double adaptive_target_latency_ms = 10.0;
 
        [[nodiscard]] auto is_valid() const -> bool
        {
            // Watermark validation
            if (low_watermark < 0.0 || low_watermark > 1.0)
            {
                return false;
            }
            if (high_watermark < 0.0 || high_watermark > 1.0)
            {
                return false;
            }
            if (low_watermark >= high_watermark)
            {
                return false;
            }
 
            // Callback policy requires decision_callback
            if (policy == backpressure_policy::callback && !decision_callback)
            {
                return false;
            }
 
            // Rate limiting validation
            if (enable_rate_limiting)
            {
                if (rate_limit_tokens_per_second == 0)
                {
                    return false;
                }
                if (rate_limit_burst_size == 0)
                {
                    return false;
                }
            }
 
            return true;
        }
    };
 
    struct backpressure_stats_snapshot
    {
        std::uint64_t jobs_accepted{0};
        std::uint64_t jobs_rejected{0};
        std::uint64_t jobs_dropped{0};
        std::uint64_t rate_limit_waits{0};
        std::uint64_t pressure_events{0};
        std::uint64_t total_block_time_ns{0};
 
        [[nodiscard]] auto acceptance_rate() const -> double
        {
            auto total = jobs_accepted + jobs_rejected;
            if (total == 0)
            {
                return 1.0;
            }
            return static_cast<double>(jobs_accepted) / static_cast<double>(total);
        }
 
        [[nodiscard]] auto avg_block_time_ms() const -> double
        {
            if (rate_limit_waits == 0)
            {
                return 0.0;
            }
            return static_cast<double>(total_block_time_ns) /
                static_cast<double>(rate_limit_waits) / 1e6;
        }
    };
 
    struct backpressure_stats
    {
        std::atomic<std::uint64_t> jobs_accepted{0};
 
        std::atomic<std::uint64_t> jobs_rejected{0};
 
        std::atomic<std::uint64_t> jobs_dropped{0};
 
        std::atomic<std::uint64_t> rate_limit_waits{0};
 
        std::atomic<std::uint64_t> pressure_events{0};
 
        std::atomic<std::uint64_t> total_block_time_ns{0};
 
        auto reset() -> void
        {
            jobs_accepted.store(0, std::memory_order_relaxed);
            jobs_rejected.store(0, std::memory_order_relaxed);
            jobs_dropped.store(0, std::memory_order_relaxed);
            rate_limit_waits.store(0, std::memory_order_relaxed);
            pressure_events.store(0, std::memory_order_relaxed);
            total_block_time_ns.store(0, std::memory_order_relaxed);
        }
 
        [[nodiscard]] auto snapshot() const -> backpressure_stats_snapshot
        {
            backpressure_stats_snapshot snap;
            snap.jobs_accepted = jobs_accepted.load(std::memory_order_relaxed);
            snap.jobs_rejected = jobs_rejected.load(std::memory_order_relaxed);
            snap.jobs_dropped = jobs_dropped.load(std::memory_order_relaxed);
            snap.rate_limit_waits = rate_limit_waits.load(std::memory_order_relaxed);
            snap.pressure_events = pressure_events.load(std::memory_order_relaxed);
            snap.total_block_time_ns = total_block_time_ns.load(std::memory_order_relaxed);
            return snap;
        }
    };
 
} // namespace kcenon::thread