aging_typed_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
 
#include "aging_typed_job.h"
#include "priority_aging_config.h"
#include "typed_job.h"
#include "job_types.h"
#include <kcenon/thread/policies/policy_queue.h>
#include <kcenon/thread/policies/sync_policies.h>
#include <kcenon/thread/policies/bound_policies.h>
#include <kcenon/thread/utils/span.h>
 
#include <thread>
#include <atomic>
#include <vector>
#include <mutex>
#include <shared_mutex>
#include <condition_variable>
#include <unordered_map>
#include <optional>
#include <sstream>
 
namespace kcenon::thread
{
    struct aging_stats
    {
        std::size_t total_boosts_applied{0};
        std::size_t jobs_reached_max_boost{0};
        std::size_t starvation_alerts{0};
        std::chrono::milliseconds max_wait_time{0};
        std::chrono::milliseconds avg_wait_time{0};
        double boost_rate{0.0};  // Boosts per second
    };
 
    template <typename job_type = job_types>
    class aging_typed_job_queue_t
    {
    public:
        using queue_type = policy_queue<policies::mutex_sync_policy,
                                        policies::unbounded_policy,
                                        policies::overflow_reject_policy>;
 
        explicit aging_typed_job_queue_t(priority_aging_config config = {});
 
        ~aging_typed_job_queue_t();
 
        // Non-copyable
        aging_typed_job_queue_t(const aging_typed_job_queue_t&) = delete;
        auto operator=(const aging_typed_job_queue_t&) -> aging_typed_job_queue_t& = delete;
 
        // Non-movable (due to thread)
        aging_typed_job_queue_t(aging_typed_job_queue_t&&) = delete;
        auto operator=(aging_typed_job_queue_t&&) -> aging_typed_job_queue_t& = delete;
 
        // ============================================
        // Aging Control
        // ============================================
 
        auto start_aging() -> void;
 
        auto stop_aging() -> void;
 
        [[nodiscard]] auto is_aging_running() const -> bool;
 
        auto apply_aging() -> std::size_t;
 
        // ============================================
        // Starvation Detection
        // ============================================
 
        [[nodiscard]] auto get_starving_jobs() const -> std::vector<job_info>;
 
        // ============================================
        // Statistics
        // ============================================
 
        [[nodiscard]] auto get_aging_stats() const -> aging_stats;
 
        auto reset_aging_stats() -> void;
 
        // ============================================
        // Configuration
        // ============================================
 
        auto set_aging_config(priority_aging_config config) -> void;
 
        [[nodiscard]] auto get_aging_config() const -> const priority_aging_config&;
 
        // ============================================
        // Job Tracking
        // ============================================
 
        [[nodiscard]] auto enqueue(std::unique_ptr<aging_typed_job_t<job_type>>&& value)
            -> common::VoidResult;
 
        auto register_aging_job(aging_typed_job_t<job_type>* job) -> void;
 
        auto unregister_aging_job(aging_typed_job_t<job_type>* job) -> void;
 
        // ============================================
        // typed_job_queue_t compatible API
        // ============================================
 
        [[nodiscard]] auto enqueue(std::unique_ptr<job>&& value) -> common::VoidResult;
 
        [[nodiscard]] auto enqueue(std::unique_ptr<typed_job_t<job_type>>&& value)
            -> common::VoidResult;
 
        template<typename DerivedJob, typename = std::enable_if_t<std::is_base_of_v<typed_job_t<job_type>, DerivedJob>>>
        [[nodiscard]] auto enqueue(std::unique_ptr<DerivedJob>&& value) -> common::VoidResult
        {
            return enqueue(std::unique_ptr<typed_job_t<job_type>>(std::move(value)));
        }
 
        [[nodiscard]] auto enqueue_batch(std::vector<std::unique_ptr<typed_job_t<job_type>>>&& jobs)
            -> common::VoidResult;
 
        [[nodiscard]] auto dequeue() -> common::Result<std::unique_ptr<job>>;
 
        [[nodiscard]] auto dequeue(const std::vector<job_type>& types)
            -> common::Result<std::unique_ptr<typed_job_t<job_type>>>;
 
        [[nodiscard]] auto dequeue(utility_module::span<const job_type> types)
            -> common::Result<std::unique_ptr<typed_job_t<job_type>>>;
 
        auto clear() -> void;
 
        [[nodiscard]] auto empty(const std::vector<job_type>& types) const -> bool;
 
        [[nodiscard]] auto empty(utility_module::span<const job_type> types) const -> bool;
 
        [[nodiscard]] auto to_string() const -> std::string;
 
        auto stop() -> void;
 
        [[nodiscard]] auto size() const -> std::size_t;
 
    private:
        priority_aging_config config_;
        std::unique_ptr<std::thread> aging_thread_;
        std::atomic<bool> aging_running_{false};
        std::atomic<bool> stopped_{false};
        std::condition_variable aging_cv_;
        mutable std::mutex aging_mutex_;
 
        // Type-based job storage using policy_queue for each type
        std::unordered_map<job_type, std::unique_ptr<queue_type>> job_queues_;
        mutable std::shared_mutex queues_mutex_;
 
        // Tracked aging jobs
        std::vector<aging_typed_job_t<job_type>*> aging_jobs_;
        mutable std::mutex jobs_mutex_;
 
        // Statistics
        aging_stats stats_;
        mutable std::mutex stats_mutex_;
        std::chrono::steady_clock::time_point stats_start_time_;
 
        auto aging_loop() -> void;
 
        [[nodiscard]] auto calculate_boost(std::chrono::milliseconds wait_time) const -> int;
 
        auto check_starvation() -> void;
 
        auto update_stats(std::size_t boosts_applied,
                          std::chrono::milliseconds max_wait,
                          std::chrono::milliseconds total_wait,
                          std::size_t job_count) -> void;
 
        auto get_or_create_queue(const job_type& type) -> queue_type*;
 
        [[nodiscard]] auto try_dequeue_from_priority(const job_type& priority)
            -> std::optional<std::unique_ptr<typed_job_t<job_type>>>;
    };
 
    using aging_typed_job_queue = aging_typed_job_queue_t<job_types>;
 
} // namespace kcenon::thread