policy_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 <type_traits>
#include <memory>
 
#include <kcenon/thread/core/job.h>
#include <kcenon/thread/core/error_handling.h>
#include <kcenon/thread/interfaces/scheduler_interface.h>
#include <kcenon/thread/interfaces/queue_capabilities_interface.h>
#include <kcenon/thread/policies/sync_policies.h>
#include <kcenon/thread/policies/bound_policies.h>
#include <kcenon/thread/policies/overflow_policies.h>
 
namespace kcenon::thread {
 
template<
    typename SyncPolicy,
    typename BoundPolicy = policies::unbounded_policy,
    typename OverflowPolicy = policies::overflow_reject_policy
>
class policy_queue : public scheduler_interface,
                     public queue_capabilities_interface {
public:
    using sync_policy_type = SyncPolicy;
    using bound_policy_type = BoundPolicy;
    using overflow_policy_type = OverflowPolicy;
 
    // ============================================
    // Construction
    // ============================================
 
    policy_queue()
        : sync_policy_()
        , bound_policy_()
        , overflow_policy_() {}
 
    explicit policy_queue(BoundPolicy bound_policy)
        : sync_policy_()
        , bound_policy_(std::move(bound_policy))
        , overflow_policy_() {}
 
    policy_queue(SyncPolicy sync_policy,
                 BoundPolicy bound_policy,
                 OverflowPolicy overflow_policy)
        : sync_policy_(std::move(sync_policy))
        , bound_policy_(std::move(bound_policy))
        , overflow_policy_(std::move(overflow_policy)) {}
 
    ~policy_queue() = default;
 
    // Non-copyable, non-movable
    policy_queue(const policy_queue&) = delete;
    policy_queue& operator=(const policy_queue&) = delete;
    policy_queue(policy_queue&&) = delete;
    policy_queue& operator=(policy_queue&&) = delete;
 
    // ============================================
    // scheduler_interface implementation
    // ============================================
 
    auto schedule(std::unique_ptr<job>&& work) -> common::VoidResult override {
        return enqueue(std::move(work));
    }
 
    auto get_next_job() -> common::Result<std::unique_ptr<job>> override {
        return dequeue();
    }
 
    // ============================================
    // Queue operations
    // ============================================
 
    [[nodiscard]] auto enqueue(std::unique_ptr<job>&& value) -> common::VoidResult {
        if (!value) {
            return common::error_info{-105, "cannot enqueue null job", "thread_system"};
        }
 
        // Check bound policy
        if constexpr (BoundPolicy::is_bounded()) {
            if (bound_policy_.is_full(sync_policy_.size())) {
                return handle_overflow(std::move(value));
            }
        }
 
        return sync_policy_.enqueue(std::move(value));
    }
 
    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>> {
        return sync_policy_.dequeue();
    }
 
    [[nodiscard]] auto try_dequeue() -> common::Result<std::unique_ptr<job>> {
        return sync_policy_.try_dequeue();
    }
 
    [[nodiscard]] auto empty() const -> bool {
        return sync_policy_.empty();
    }
 
    [[nodiscard]] auto size() const -> std::size_t {
        return sync_policy_.size();
    }
 
    auto clear() -> void {
        sync_policy_.clear();
    }
 
    auto stop() -> void {
        sync_policy_.stop();
    }
 
    [[nodiscard]] auto is_stopped() const -> bool {
        return sync_policy_.is_stopped();
    }
 
    // ============================================
    // queue_capabilities_interface implementation
    // ============================================
 
    [[nodiscard]] auto get_capabilities() const -> queue_capabilities override {
        return sync_policy_.get_capabilities();
    }
 
    // ============================================
    // Policy access
    // ============================================
 
    [[nodiscard]] auto sync_policy() -> SyncPolicy& {
        return sync_policy_;
    }
 
    [[nodiscard]] auto sync_policy() const -> const SyncPolicy& {
        return sync_policy_;
    }
 
    [[nodiscard]] auto bound_policy() -> BoundPolicy& {
        return bound_policy_;
    }
 
    [[nodiscard]] auto bound_policy() const -> const BoundPolicy& {
        return bound_policy_;
    }
 
    [[nodiscard]] auto overflow_policy() -> OverflowPolicy& {
        return overflow_policy_;
    }
 
    [[nodiscard]] auto overflow_policy() const -> const OverflowPolicy& {
        return overflow_policy_;
    }
 
    // ============================================
    // Bounded queue convenience methods
    // ============================================
 
    [[nodiscard]] auto is_bounded() const -> bool {
        if constexpr (requires { bound_policy_.is_bounded(); }) {
            return bound_policy_.is_bounded();
        }
        return BoundPolicy::is_bounded();
    }
 
    [[nodiscard]] auto is_full() const -> bool {
        if constexpr (BoundPolicy::is_bounded()) {
            return bound_policy_.is_full(sync_policy_.size());
        }
        return false;
    }
 
    [[nodiscard]] auto remaining_capacity() const -> std::size_t {
        return bound_policy_.remaining_capacity(sync_policy_.size());
    }
 
private:
    [[nodiscard]] auto handle_overflow(std::unique_ptr<job>&& value) -> common::VoidResult {
        if constexpr (std::is_same_v<OverflowPolicy, policies::overflow_reject_policy>) {
            return overflow_policy_.handle_overflow(std::move(value));
        } else if constexpr (std::is_same_v<OverflowPolicy, policies::overflow_drop_newest_policy>) {
            return overflow_policy_.handle_overflow(std::move(value));
        } else if constexpr (std::is_same_v<OverflowPolicy, policies::overflow_drop_oldest_policy>) {
            // Drop oldest by dequeuing and discarding
            auto oldest = sync_policy_.try_dequeue();
            (void)oldest;  // Discard
            return sync_policy_.enqueue(std::move(value));
        } else {
            // For blocking policies, just return the error
            // Actual blocking must be implemented at a higher level
            return common::error_info{-120, "queue is full", "thread_system"};
        }
    }
 
    SyncPolicy sync_policy_;
    BoundPolicy bound_policy_;
    OverflowPolicy overflow_policy_;
};
 
// ============================================
// Type aliases for common configurations
// ============================================
 
using standard_queue = policy_queue<
    policies::mutex_sync_policy,
    policies::unbounded_policy,
    policies::overflow_reject_policy
>;
 
using policy_lockfree_queue = policy_queue<
    policies::lockfree_sync_policy,
    policies::unbounded_policy,
    policies::overflow_reject_policy
>;
 
template<std::size_t MaxSize>
using bounded_blocking_queue = policy_queue<
    policies::mutex_sync_policy,
    policies::bounded_policy,
    policies::overflow_block_policy
>;
 
template<std::size_t MaxSize>
using bounded_rejecting_queue = policy_queue<
    policies::mutex_sync_policy,
    policies::bounded_policy,
    policies::overflow_reject_policy
>;
 
template<std::size_t MaxSize>
using ring_buffer_queue = policy_queue<
    policies::mutex_sync_policy,
    policies::bounded_policy,
    policies::overflow_drop_oldest_policy
>;
 
} // namespace kcenon::thread