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 "job.h"
#include <kcenon/thread/utils/formatter.h>
#include "callback_job.h"
#include <kcenon/thread/utils/convert_string.h>
#include "error_handling.h"
#include <kcenon/thread/interfaces/scheduler_interface.h>
#include <kcenon/thread/interfaces/queue_capabilities_interface.h>
#include <kcenon/thread/diagnostics/job_info.h>
 
#include <mutex>
#include <deque>
#include <tuple>
#include <atomic>
#include <optional>
#include <string_view>
#include <type_traits>
#include <condition_variable>
#include <memory>
#include <functional>
 
namespace kcenon::thread
{
    class job_queue : public std::enable_shared_from_this<job_queue>,
                      public scheduler_interface,
                      public queue_capabilities_interface
    {
    public:
        explicit job_queue(std::optional<std::size_t> max_size = std::nullopt);
 
        virtual ~job_queue(void);
 
        [[nodiscard]] auto get_ptr(void) -> std::shared_ptr<job_queue>;
 
        [[nodiscard]] auto is_stopped() const -> bool;
 
        auto set_notify(bool notify) -> void;
 
        auto schedule(std::unique_ptr<job>&& value) -> common::VoidResult override { return enqueue(std::move(value)); }
 
        auto get_next_job() -> common::Result<std::unique_ptr<job>> override { return dequeue(); }
 
        [[nodiscard]] virtual auto enqueue(std::unique_ptr<job>&& value) -> 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]] virtual auto enqueue_batch(std::vector<std::unique_ptr<job>>&& jobs) -> common::VoidResult;
 
        [[nodiscard]] virtual auto dequeue(void) -> common::Result<std::unique_ptr<job>>;
 
        [[nodiscard]] virtual auto try_dequeue(void) -> common::Result<std::unique_ptr<job>>;
 
        [[nodiscard]] virtual auto dequeue_batch(void) -> std::deque<std::unique_ptr<job>>;
 
        [[nodiscard]] virtual auto dequeue_batch_limited(std::size_t max_count)
            -> std::deque<std::unique_ptr<job>>;
 
        virtual auto clear(void) -> void;
 
        [[nodiscard]] auto empty(void) const -> bool;
 
        [[nodiscard]] auto size(void) const -> std::size_t;
 
        struct memory_stats {
            std::size_t queue_size_bytes;      // Estimated queue memory usage
            std::size_t pending_job_count;     // Number of jobs in queue
            std::size_t node_overhead_bytes;   // Deque node overhead estimate
        };
 
        [[nodiscard]] auto get_memory_stats() const -> memory_stats;
 
        auto stop(void) -> void;
 
        [[nodiscard]] virtual auto to_string(void) const -> std::string;
 
        [[nodiscard]] auto get_capabilities() const -> queue_capabilities override {
            return queue_capabilities{
                .exact_size = true,
                .atomic_empty_check = true,
                .lock_free = false,
                .wait_free = false,
                .supports_batch = true,
                .supports_blocking_wait = true,
                .supports_stop = true
            };
        }
 
        // ============================================
        // Bounded queue functionality (merged from bounded_job_queue)
        // ============================================
 
        [[nodiscard]] auto is_bounded() const -> bool;
 
        [[nodiscard]] auto get_max_size() const -> std::optional<std::size_t>;
 
        auto set_max_size(std::optional<std::size_t> max_size) -> void;
 
        [[nodiscard]] auto is_full() const -> bool;
 
        // ============================================
        // Diagnostics support
        // ============================================
 
        [[nodiscard]] virtual auto inspect_pending_jobs(std::size_t limit = 100) const
            -> std::vector<diagnostics::job_info>;
 
    protected:
        std::atomic_bool notify_;
 
        std::atomic_bool stop_;
 
        mutable std::mutex mutex_;
 
        std::condition_variable condition_;
 
    private:
        std::deque<std::unique_ptr<job>> queue_;
 
        std::optional<std::size_t> max_size_;
 
        std::atomic<std::size_t> atomic_size_{0};
    };
}
 
// ----------------------------------------------------------------------------
// Formatter specializations for job_queue
// ----------------------------------------------------------------------------
 
#include <kcenon/thread/utils/formatter_macros.h>
 
// Generate formatter specializations for kcenon::thread::job_queue
DECLARE_FORMATTER(kcenon::thread::job_queue)