backpressure_job_queue.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
 
 // end of backpressure
 
#include "job_queue.h"
#include "backpressure_config.h"
#include "token_bucket.h"
 
#include <memory>
#include <mutex>
 
namespace kcenon::thread
{
    class backpressure_job_queue : public job_queue
    {
    public:
        explicit backpressure_job_queue(
            std::size_t max_size,
            backpressure_config config = backpressure_config{});
 
        ~backpressure_job_queue() override;
 
        // Non-copyable, non-movable (inherits shared_ptr semantics from job_queue)
        backpressure_job_queue(const backpressure_job_queue&) = delete;
        backpressure_job_queue& operator=(const backpressure_job_queue&) = delete;
        backpressure_job_queue(backpressure_job_queue&&) = delete;
        backpressure_job_queue& operator=(backpressure_job_queue&&) = delete;
 
        // =========================================================================
        // Overridden Queue Operations
        // =========================================================================
 
        [[nodiscard]] auto enqueue(std::unique_ptr<job>&& value) -> common::VoidResult override;
 
        [[nodiscard]] auto enqueue_batch(std::vector<std::unique_ptr<job>>&& jobs)
            -> common::VoidResult override;
 
        // =========================================================================
        // Backpressure-Specific Methods
        // =========================================================================
 
        [[nodiscard]] auto get_pressure_level() const -> pressure_level;
 
        [[nodiscard]] auto get_pressure_ratio() const -> double;
 
        auto set_backpressure_config(backpressure_config config) -> void;
 
        [[nodiscard]] auto get_backpressure_config() const -> const backpressure_config&;
 
        // =========================================================================
        // Rate Limiting
        // =========================================================================
 
        [[nodiscard]] auto is_rate_limited() const -> bool;
 
        [[nodiscard]] auto get_available_tokens() const -> std::size_t;
 
        // =========================================================================
        // Statistics
        // =========================================================================
 
        [[nodiscard]] auto get_backpressure_stats() const -> backpressure_stats_snapshot;
 
        auto reset_stats() -> void;
 
        // =========================================================================
        // String Representation
        // =========================================================================
 
        [[nodiscard]] auto to_string() const -> std::string override;
 
    private:
        [[nodiscard]] auto apply_backpressure(std::unique_ptr<job>&& value) -> common::VoidResult;
 
        [[nodiscard]] auto apply_rate_limiting() -> bool;
 
        auto update_pressure_state() -> void;
 
        [[nodiscard]] auto handle_block_policy(std::unique_ptr<job>&& value) -> common::VoidResult;
 
        [[nodiscard]] auto handle_drop_oldest_policy(std::unique_ptr<job>&& value) -> common::VoidResult;
 
        [[nodiscard]] auto handle_callback_policy(std::unique_ptr<job>&& value) -> common::VoidResult;
 
        [[nodiscard]] auto handle_adaptive_policy(std::unique_ptr<job>&& value) -> common::VoidResult;
 
        [[nodiscard]] auto direct_enqueue(std::unique_ptr<job>&& value) -> common::VoidResult;
 
        backpressure_config config_;
 
        mutable std::mutex config_mutex_;
 
        std::unique_ptr<token_bucket> rate_limiter_;
 
        std::atomic<pressure_level> current_pressure_{pressure_level::none};
 
        std::atomic<double> current_pressure_ratio_{0.0};
 
        backpressure_stats stats_;
 
        std::condition_variable space_available_;
    };
 
} // namespace kcenon::thread