adaptive_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 <kcenon/thread/core/job_queue.h>
#include <kcenon/thread/lockfree/lockfree_job_queue.h>
#include <kcenon/thread/interfaces/queue_capabilities_interface.h>
 
#include <atomic>
#include <chrono>
#include <memory>
#include <mutex>
#include <type_traits>
 
namespace kcenon::thread {
 
class adaptive_job_queue : public scheduler_interface,
                           public queue_capabilities_interface {
public:
    enum class mode {
        mutex,     
        lock_free  
    };
 
    enum class policy {
        accuracy_first,    
        performance_first, 
        balanced,          
        manual             
    };
 
    // ============================================
    // Construction
    // ============================================
 
    explicit adaptive_job_queue(policy p = policy::balanced);
 
    ~adaptive_job_queue();
 
    // Non-copyable
    adaptive_job_queue(const adaptive_job_queue&) = delete;
    adaptive_job_queue& operator=(const adaptive_job_queue&) = delete;
 
    // Non-movable (contains shared state)
    adaptive_job_queue(adaptive_job_queue&&) = delete;
    adaptive_job_queue& operator=(adaptive_job_queue&&) = delete;
 
    // ============================================
    // scheduler_interface implementation
    // ============================================
 
    auto schedule(std::unique_ptr<job>&& work) -> common::VoidResult override;
 
    auto get_next_job() -> common::Result<std::unique_ptr<job>> override;
 
    // ============================================
    // Standard queue operations
    // ============================================
 
    [[nodiscard]] auto enqueue(std::unique_ptr<job>&& j) -> common::VoidResult;
 
    template<typename JobType, typename = std::enable_if_t<std::is_base_of_v<job, JobType>>>
    [[nodiscard]] auto enqueue(std::unique_ptr<JobType>&& value) -> common::VoidResult
    {
        return enqueue(std::unique_ptr<job>(std::move(value)));
    }
 
    [[nodiscard]] auto dequeue() -> common::Result<std::unique_ptr<job>>;
 
    [[nodiscard]] auto try_dequeue() -> common::Result<std::unique_ptr<job>>;
 
    [[nodiscard]] auto empty() const -> bool;
 
    [[nodiscard]] auto size() const -> std::size_t;
 
    auto clear() -> void;
 
    auto stop() -> void;
 
    [[nodiscard]] auto is_stopped() const -> bool;
 
    // ============================================
    // queue_capabilities_interface implementation
    // ============================================
 
    [[nodiscard]] auto get_capabilities() const -> queue_capabilities override;
 
    // ============================================
    // Adaptive-specific API
    // ============================================
 
    [[nodiscard]] auto current_mode() const -> mode;
 
    [[nodiscard]] auto current_policy() const -> policy;
 
    auto switch_mode(mode m) -> common::VoidResult;
 
    struct stats {
        uint64_t mode_switches{0};        
        uint64_t time_in_mutex_ms{0};     
        uint64_t time_in_lockfree_ms{0};  
        uint64_t enqueue_count{0};        
        uint64_t dequeue_count{0};        
    };
 
    [[nodiscard]] auto get_stats() const -> stats;
 
    // ============================================
    // RAII Guard for temporary accuracy
    // ============================================
 
    class accuracy_guard;
 
    [[nodiscard]] auto require_accuracy() -> accuracy_guard;
 
    class accuracy_guard {
    public:
        explicit accuracy_guard(adaptive_job_queue& queue);
 
        ~accuracy_guard();
 
        // Move-only
        accuracy_guard(accuracy_guard&& other) noexcept;
        accuracy_guard& operator=(accuracy_guard&&) = delete;
 
        // Non-copyable
        accuracy_guard(const accuracy_guard&) = delete;
        accuracy_guard& operator=(const accuracy_guard&) = delete;
 
    private:
        adaptive_job_queue* queue_;
        mode previous_mode_;
        bool active_;
    };
 
private:
    // Policy and mode
    policy policy_;
    std::atomic<mode> current_mode_;
    std::atomic<bool> stopped_{false};
 
    // Wrapped queues (existing classes, unchanged)
    std::shared_ptr<job_queue> mutex_queue_;
    std::unique_ptr<detail::lockfree_job_queue> lockfree_queue_;
 
    // Synchronization for mode switching
    mutable std::mutex migration_mutex_;
 
    // Accuracy guard reference count
    std::atomic<int> accuracy_guard_count_{0};
 
    // Statistics tracking
    mutable std::mutex stats_mutex_;
    stats stats_;
    std::chrono::steady_clock::time_point mode_start_time_;
 
    // Private methods
    void migrate_to_mode(mode target);
    void update_mode_time();
    auto determine_mode_for_balanced() const -> mode;
};
 
} // namespace kcenon::thread