job_queue.cpp

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.
 
#include <kcenon/thread/core/job_queue.h>
 
namespace kcenon::thread
{
    job_queue::job_queue(std::optional<std::size_t> max_size)
        : notify_(true), stop_(false), mutex_(), condition_(), queue_(), max_size_(max_size) {}
 
    job_queue::~job_queue(void) {}
 
    auto job_queue::get_ptr(void) -> std::shared_ptr<job_queue> { return shared_from_this(); }
 
    auto job_queue::is_stopped() const -> bool { return stop_.load(); }
 
    auto job_queue::set_notify(bool notify) -> void { notify_.store(notify); }
 
    auto job_queue::enqueue(std::unique_ptr<job>&& value) -> common::VoidResult
    {
        // Early validation: check if queue is still accepting jobs
        if (stop_.load())
        {
            return common::error_info{static_cast<int>(error_code::queue_stopped), "Job queue is stopped", "thread_system"};
        }
 
        // Validate input: null jobs are not allowed
        if (value == nullptr)
        {
            return common::error_info{static_cast<int>(error_code::invalid_argument), "cannot enqueue null job", "thread_system"};
        }
 
        // Critical section: modify queue with proper synchronization
        std::scoped_lock<std::mutex> lock(mutex_);
 
        // Check size limit if bounded
        if (max_size_.has_value() && queue_.size() >= max_size_.value())
        {
            return common::error_info{static_cast<int>(error_code::queue_full), "Job queue is full", "thread_system"};
        }
 
        // Move job into queue (efficient transfer of ownership)
        queue_.push_back(std::move(value));
        atomic_size_.fetch_add(1, std::memory_order_relaxed);
 
        // Conditionally notify waiting consumers
        if (notify_)
        {
            condition_.notify_one();  // Wake exactly one waiting thread
        }
 
        return common::ok();
    }
 
    auto job_queue::enqueue_batch(std::vector<std::unique_ptr<job>>&& jobs) -> common::VoidResult
    {
        // Early validation: check if queue is still accepting jobs
        if (stop_.load())
        {
            return common::error_info{static_cast<int>(error_code::queue_stopped), "Job queue is stopped", "thread_system"};
        }
 
        // Validate batch: empty batches are not allowed
        if (jobs.empty())
        {
            return common::error_info{static_cast<int>(error_code::invalid_argument), "cannot enqueue empty batch", "thread_system"};
        }
 
        // Pre-validate all jobs before modifying queue (ensures atomicity)
        for (auto& job : jobs)
        {
            if (job == nullptr)
            {
                return common::error_info{static_cast<int>(error_code::invalid_argument), "cannot enqueue null job in batch", "thread_system"};
            }
        }
 
        // Critical section: add entire batch with single lock acquisition
        std::scoped_lock<std::mutex> lock(mutex_);
 
        // Check size limit if bounded
        if (max_size_.has_value())
        {
            std::size_t available_space = max_size_.value() - queue_.size();
            if (jobs.size() > available_space)
            {
                return common::error_info{static_cast<int>(error_code::queue_full),
                    "Job queue cannot fit entire batch", "thread_system"};
            }
        }
 
        // Move all jobs into queue efficiently
        for (auto& job : jobs)
        {
            queue_.push_back(std::move(job));
        }
        atomic_size_.fetch_add(jobs.size(), std::memory_order_relaxed);
 
        // Single notification for entire batch (performance optimization)
        if (notify_)
        {
            condition_.notify_one();  // Wake one thread to process batch
        }
 
        return common::ok();
    }
 
    auto job_queue::dequeue() -> common::Result<std::unique_ptr<job>>
    {
        // Use unique_lock for condition variable operations
        std::unique_lock<std::mutex> lock(mutex_);
 
        // Block until job available OR queue stopped
        condition_.wait(lock, [this]() { return !queue_.empty() || stop_.load(); });
 
        // CRITICAL: Re-check queue state while still holding lock
        // This prevents race with clear() that could empty the queue
        // between wait() return and queue access (TOCTOU bug)
        if (queue_.empty())
        {
            return common::error_info{static_cast<int>(error_code::queue_empty), "there are no jobs to dequeue", "thread_system"};
        }
 
        // Efficiently extract first job from queue
        // At this point, we're guaranteed queue is not empty because:
        // 1. We hold the mutex continuously from wait() return
        // 2. We just verified !queue_.empty() above
        auto value = std::move(queue_.front());
        queue_.pop_front();
        atomic_size_.fetch_sub(1, std::memory_order_relaxed);
 
        return value;  // Return moved job (caller takes ownership)
    }
 
    auto job_queue::try_dequeue() -> common::Result<std::unique_ptr<job>>
    {
        // Early validation: check if queue is stopped
        if (stop_.load())
        {
            return common::error_info{static_cast<int>(error_code::queue_stopped), "Job queue is stopped", "thread_system"};
        }
 
        // Critical section: check and potentially extract job
        std::scoped_lock<std::mutex> lock(mutex_);
 
        // Non-blocking check: return error if empty
        if (queue_.empty())
        {
            return common::error_info{static_cast<int>(error_code::queue_empty), "there are no jobs to dequeue", "thread_system"};
        }
 
        // Efficiently extract first job from queue
        auto value = std::move(queue_.front());
        queue_.pop_front();
        atomic_size_.fetch_sub(1, std::memory_order_relaxed);
 
        return value;  // Return moved job (caller takes ownership)
    }
 
    auto job_queue::dequeue_batch(void) -> std::deque<std::unique_ptr<job>>
    {
        std::deque<std::unique_ptr<job>> all_items;
        {
            // Critical section: atomically transfer all queue contents
            std::scoped_lock<std::mutex> lock(mutex_);
 
            // Efficient O(1) transfer of all jobs
            std::swap(queue_, all_items);
            atomic_size_.store(0, std::memory_order_relaxed);
 
            // Wake all waiting threads since queue is now empty
            condition_.notify_all();
        }
 
        return all_items;  // Return all extracted jobs
    }
 
    auto job_queue::dequeue_batch_limited(std::size_t max_count)
        -> std::deque<std::unique_ptr<job>>
    {
        std::deque<std::unique_ptr<job>> batch_items;
        {
            // Critical section: atomically extract up to max_count jobs
            std::scoped_lock<std::mutex> lock(mutex_);
 
            // Extract jobs up to max_count or queue size, whichever is smaller
            std::size_t count = std::min(max_count, queue_.size());
            for (std::size_t i = 0; i < count; ++i)
            {
                batch_items.push_back(std::move(queue_.front()));
                queue_.pop_front();
            }
            atomic_size_.fetch_sub(count, std::memory_order_relaxed);
 
            // Only notify if queue is now empty (to wake waiting workers)
            // If jobs remain, other workers can continue dequeuing without wakeup overhead
            if (queue_.empty())
            {
                condition_.notify_all();
            }
        }
 
        return batch_items;  // Return extracted batch
    }
 
    auto job_queue::clear(void) -> void
    {
        // Critical section: atomically clear all jobs
        std::scoped_lock<std::mutex> lock(mutex_);
 
        // Destroy all jobs in queue
        queue_.clear();
        atomic_size_.store(0, std::memory_order_relaxed);
 
        // Wake all waiting threads since queue is now empty
        condition_.notify_all();
    }
 
    auto job_queue::empty(void) const -> bool
    {
        return atomic_size_.load(std::memory_order_relaxed) == 0;
    }
 
    auto job_queue::stop(void) -> void
    {
        // Critical section: set stop flag and notify waiters
        std::scoped_lock<std::mutex> lock(mutex_);
 
        // Prevent new jobs from being added
        stop_.store(true);
 
        // Wake all threads waiting in dequeue()
        condition_.notify_all();
    }
 
 
    auto job_queue::size(void) const -> std::size_t
    {
        return atomic_size_.load(std::memory_order_relaxed);
    }
 
    auto job_queue::get_memory_stats() const -> memory_stats
    {
        std::scoped_lock<std::mutex> lock(mutex_);
 
        std::size_t job_count = queue_.size();
 
        // Estimate memory usage
        // std::unique_ptr<job> typically 8 bytes (pointer size)
        // std::deque node overhead varies by platform, typically 32-48 bytes
        constexpr std::size_t ptr_size = sizeof(std::unique_ptr<job>);
        constexpr std::size_t node_overhead = 40;  // Conservative estimate
 
        // Use traditional aggregate initialization for C++17 compatibility
        memory_stats stats;
        stats.queue_size_bytes = (ptr_size + node_overhead) * job_count;
        stats.pending_job_count = job_count;
        stats.node_overhead_bytes = node_overhead * job_count;
        return stats;
    }
 
    auto job_queue::to_string(void) const -> std::string
    {
        return utility_module::formatter::format("contained {} jobs", size());
    }
 
    // ============================================
    // Bounded queue functionality
    // ============================================
 
    auto job_queue::is_bounded() const -> bool
    {
        return max_size_.has_value();
    }
 
    auto job_queue::get_max_size() const -> std::optional<std::size_t>
    {
        return max_size_;
    }
 
    auto job_queue::set_max_size(std::optional<std::size_t> max_size) -> void
    {
        std::scoped_lock<std::mutex> lock(mutex_);
        max_size_ = max_size;
    }
 
    auto job_queue::is_full() const -> bool
    {
        if (!max_size_.has_value())
        {
            return false;  // Unbounded queue is never full
        }
        return atomic_size_.load(std::memory_order_relaxed) >= max_size_.value();
    }
 
    // ============================================
    // Diagnostics support
    // ============================================
 
    auto job_queue::inspect_pending_jobs(std::size_t limit) const
        -> std::vector<diagnostics::job_info>
    {
        std::scoped_lock<std::mutex> lock(mutex_);
 
        std::vector<diagnostics::job_info> result;
        auto count = (limit == 0) ? queue_.size() : std::min(limit, queue_.size());
        result.reserve(count);
 
        auto now = std::chrono::steady_clock::now();
        std::size_t index = 0;
 
        for (const auto& job_ptr : queue_)
        {
            if (limit > 0 && index >= limit)
            {
                break;
            }
 
            if (job_ptr == nullptr)
            {
                ++index;
                continue;
            }
 
            diagnostics::job_info info;
            info.job_id = job_ptr->get_job_id();
            info.job_name = job_ptr->get_name();
            info.status = diagnostics::job_status::pending;
            info.enqueue_time = job_ptr->get_enqueue_time();
            info.start_time = job_ptr->get_enqueue_time(); // Not started yet
 
            // Calculate wait time (time since enqueue)
            info.wait_time = std::chrono::duration_cast<std::chrono::nanoseconds>(
                now - job_ptr->get_enqueue_time()
            );
            info.execution_time = std::chrono::nanoseconds{0}; // Not started yet
 
            result.push_back(std::move(info));
            ++index;
        }
 
        return result;
    }
 
} // namespace kcenon::thread