lockfree_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.h>
#include <kcenon/thread/core/safe_hazard_pointer.h>
#include <kcenon/thread/core/error_handling.h>
#include <kcenon/thread/interfaces/scheduler_interface.h>
#include <kcenon/thread/interfaces/queue_capabilities_interface.h>
 
#include <atomic>
#include <memory>
#include <optional>
 
namespace kcenon::thread {
 
namespace detail {
 
class lockfree_job_queue : public scheduler_interface,
                           public queue_capabilities_interface {
public:
    lockfree_job_queue();
 
    ~lockfree_job_queue();
 
    // Non-copyable and non-movable (queue has shared global state)
    lockfree_job_queue(const lockfree_job_queue&) = delete;
    lockfree_job_queue& operator=(const lockfree_job_queue&) = delete;
    lockfree_job_queue(lockfree_job_queue&&) = delete;
    lockfree_job_queue& operator=(lockfree_job_queue&&) = delete;
 
    [[nodiscard]] auto enqueue(std::unique_ptr<job>&& job) -> common::VoidResult;
 
    [[nodiscard]] auto dequeue() -> common::Result<std::unique_ptr<job>>;
 
    [[nodiscard]] auto try_dequeue() -> common::Result<std::unique_ptr<job>> {
        return dequeue();
    }
 
    [[nodiscard]] auto empty() const -> bool;
 
    [[nodiscard]] auto size() const -> std::size_t;
 
    // ============================================
    // 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_capabilities_interface implementation
    // ============================================
 
    [[nodiscard]] auto get_capabilities() const -> queue_capabilities override {
        return queue_capabilities{
            .exact_size = false,             // Approximate only
            .atomic_empty_check = false,     // Non-atomic
            .lock_free = true,               // Lock-free implementation
            .wait_free = false,              // Not wait-free
            .supports_batch = false,         // No batch operations
            .supports_blocking_wait = false, // Spin-wait only
            .supports_stop = false           // No stop() method
        };
    }
 
private:
    struct node {
        std::unique_ptr<job> data;          // Payload (nullptr for dummy node)
        std::atomic<node*> next{nullptr};   // Next node in queue
 
        node() : data(nullptr), next(nullptr) {}
 
        explicit node(std::unique_ptr<job>&& job_data)
            : data(std::move(job_data))
            , next(nullptr) {}
    };
 
    class node_pool {
    public:
        node_pool() = default;
 
        ~node_pool() {
            // Drain freelist and delete all pooled nodes
            node* head = free_list_.load(std::memory_order_relaxed);
            while (head) {
                node* next = head->next.load(std::memory_order_relaxed);
                delete head;
                head = next;
            }
        }
 
        // Non-copyable, non-movable
        node_pool(const node_pool&) = delete;
        node_pool& operator=(const node_pool&) = delete;
 
        node* acquire(std::unique_ptr<job>&& job_data) {
            node* head = free_list_.load(std::memory_order_acquire);
            while (head) {
                if (free_list_.compare_exchange_weak(
                        head, head->next.load(std::memory_order_relaxed),
                        std::memory_order_acq_rel,
                        std::memory_order_acquire)) {
                    // Reuse pooled node — reset fields
                    head->data = std::move(job_data);
                    head->next.store(nullptr, std::memory_order_relaxed);
                    return head;
                }
                // head is updated by CAS on failure, retry
            }
            // Pool empty — allocate fresh node
            return new node(std::move(job_data));
        }
 
        void release(node* n) {
            if (!n) return;
            // Clear payload before pooling
            n->data.reset();
            node* old_head = free_list_.load(std::memory_order_relaxed);
            do {
                n->next.store(old_head, std::memory_order_relaxed);
            } while (!free_list_.compare_exchange_weak(
                old_head, n,
                std::memory_order_release,
                std::memory_order_relaxed));
        }
 
    private:
        std::atomic<node*> free_list_{nullptr};
    };
 
    void retire_node(node* n);
 
    // Queue state: head and tail pointers
    std::atomic<node*> head_;  // Dequeue end (with acquire/release)
    std::atomic<node*> tail_;  // Enqueue end (with acquire/release)
 
    // Node pool for recycling retired nodes (shared_ptr for capture safety)
    std::shared_ptr<node_pool> pool_;
 
    // Hazard pointer domain for safe memory reclamation
    // Uses safe_hazard_pointer with explicit memory ordering (safe on ARM/weak memory models)
    using node_hp_domain = typed_safe_hazard_domain<node>;
 
    // Statistics (for monitoring, relaxed memory order)
    mutable std::atomic<std::size_t> approximate_size_{0};
 
    // Shutdown flag for safe destruction
    std::atomic<bool> shutdown_{false};
};
 
}  // namespace detail
 
}  // namespace kcenon::thread