kcenon::thread::detail::lockfree_job_queue Class Reference

Lock-free Multi-Producer Multi-Consumer (MPMC) job queue (Internal implementation) More...

#include <lockfree_job_queue.h>

Inheritance diagram for kcenon::thread::detail::lockfree_job_queue:

Collaboration diagram for kcenon::thread::detail::lockfree_job_queue:

Classes
struct	node
	Internal queue node structure. More...
class	node_pool
	Lock-free node freelist (Treiber stack) for node recycling. More...

Public Member Functions
	lockfree_job_queue ()
	Constructs an empty lock-free job queue.
	~lockfree_job_queue ()
	Destructor.
	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
auto	enqueue (std::unique_ptr< job > &&job) -> common::VoidResult
	Enqueues a job into the queue (thread-safe)
auto	dequeue () -> common::Result< std::unique_ptr< job > >
	Dequeues a job from the queue (thread-safe)
auto	try_dequeue () -> common::Result< std::unique_ptr< job > >
	Tries to dequeue a job without blocking.
auto	empty () const -> bool
	Checks if the queue is empty.
auto	size () const -> std::size_t
	Gets approximate queue size.
auto	schedule (std::unique_ptr< job > &&work) -> common::VoidResult override
	Schedule a job (delegates to enqueue)
auto	get_next_job () -> common::Result< std::unique_ptr< job > > override
	Get next job (delegates to dequeue)
auto	get_capabilities () const -> queue_capabilities override
	Returns capabilities of lockfree_job_queue.
Public Member Functions inherited from kcenon::thread::scheduler_interface
virtual	~scheduler_interface ()=default
Public Member Functions inherited from kcenon::thread::queue_capabilities_interface
virtual	~queue_capabilities_interface ()=default
auto	has_exact_size () const -> bool
	Check if size() returns exact values.
auto	has_atomic_empty () const -> bool
	Check if empty() check is atomic.
auto	is_lock_free () const -> bool
	Check if this is a lock-free implementation.
auto	is_wait_free () const -> bool
	Check if this is a wait-free implementation.
auto	supports_batch () const -> bool
	Check if batch operations are supported.
auto	supports_blocking_wait () const -> bool
	Check if blocking wait is supported.
auto	supports_stop () const -> bool
	Check if stop signaling is supported.

Private Types
using	node_hp_domain = typed_safe_hazard_domain<node>

Private Member Functions
void	retire_node (node *n)
	Retire a node through hazard pointers, recycling via pool on reclamation.

Private Attributes
std::atomic< node * >	head_
std::atomic< node * >	tail_
std::shared_ptr< node_pool >	pool_
std::atomic< std::size_t >	approximate_size_ {0}
std::atomic< bool >	shutdown_ {false}

Detailed Description

Lock-free Multi-Producer Multi-Consumer (MPMC) job queue (Internal implementation)

This class implements a lock-free MPMC queue using the Michael-Scott algorithm with Safe Hazard Pointers for memory reclamation. It uses explicit memory ordering to ensure correctness on weak memory model architectures (ARM, etc.)

Algorithm: Michael-Scott Queue (1996) Memory Reclamation: Safe Hazard Pointers with explicit memory ordering

Key Features:

• True lock-free operation (no mutexes, no locks)
• Safe concurrent access from multiple producers and consumers
• Automatic memory reclamation using Safe Hazard Pointers
• Correct memory ordering for weak memory model architectures (ARM)
• No TLS node pool (eliminates destructor ordering issues)
• ABA problem prevention through HP-based protection

Performance Characteristics:

• Enqueue: O(1) amortized, wait-free
• Dequeue: O(1) amortized, lock-free
• Memory overhead: ~256 bytes per thread (hazard pointers)

Thread Safety:

• All methods are thread-safe
• Can be called concurrently from any number of threads
• Uses atomic operations with acquire/release semantics

Note

This implementation is production-safe and resolves TICKET-001 (TLS bug) and TICKET-002 (weak memory model safety).

See also

lockfree_job_queue_test.cpp for usage examples

Examples

queue_capabilities_sample.cpp.

Definition at line 63 of file lockfree_job_queue.h.

Member Typedef Documentation

node_hp_domain

using kcenon::thread::detail::lockfree_job_queue::node_hp_domain = typed_safe_hazard_domain<node>

private

Definition at line 312 of file lockfree_job_queue.h.

Constructor & Destructor Documentation

lockfree_job_queue() [1/3]

kcenon::thread::detail::lockfree_job_queue::lockfree_job_queue

(

)

Constructs an empty lock-free job queue.

Initializes the queue with a dummy node to simplify the algorithm. The dummy node is never removed, allowing concurrent enqueue/dequeue.

Definition at line 41 of file lockfree_job_queue.cpp.

    : pool_(std::make_shared<node_pool>()) {
    // Create dummy node (Michael-Scott algorithm requires one dummy node)
    // This simplifies the algorithm by ensuring head and tail are never null
    node* dummy = new node();
 
    head_.store(dummy, std::memory_order_relaxed);
    tail_.store(dummy, std::memory_order_relaxed);
    approximate_size_.store(0, std::memory_order_relaxed);
}

References approximate_size_, head_, and tail_.

~lockfree_job_queue()

kcenon::thread::detail::lockfree_job_queue::~lockfree_job_queue

(

)

Destructor.

Drains the queue and reclaims all nodes. Thread-safe even if other threads are still accessing the queue (they will get errors).

Definition at line 53 of file lockfree_job_queue.cpp.

                                        {
    // Signal shutdown to prevent new operations
    shutdown_.store(true, std::memory_order_release);
 
    // Drain remaining jobs (release ownership)
    while (true) {
        auto result = dequeue();
        if (result.is_err()) {
            break;
        }
        // Jobs are destroyed when unique_ptr goes out of scope
    }
 
    // Safe cleanup: acquire semantics ensure we see all writes
    node* dummy = head_.load(std::memory_order_acquire);
 
    // Retire dummy node through pool-aware reclamation
    // This ensures the node is only deleted when no other thread
    // holds a hazard pointer to it (uses explicit memory ordering)
    retire_node(dummy);
}

References dequeue(), head_, retire_node(), and shutdown_.

Here is the call graph for this function:

lockfree_job_queue() [2/3]

kcenon::thread::detail::lockfree_job_queue::lockfree_job_queue ( const lockfree_job_queue & )

delete

lockfree_job_queue() [3/3]

kcenon::thread::detail::lockfree_job_queue::lockfree_job_queue ( lockfree_job_queue && )

delete

Member Function Documentation

dequeue()

auto kcenon::thread::detail::lockfree_job_queue::dequeue ( ) -> common::Result<std::unique_ptr<job>>

nodiscard

Dequeues a job from the queue (thread-safe)

Returns

common::Result<std::unique_ptr<job>> The dequeued job or error

Note

Lock-free operation (system-wide progress guaranteed)

Returns empty result if queue is empty (not an error)

Uses Hazard Pointers to protect nodes from premature deletion

Retired nodes are eventually reclaimed by the HP domain

Time Complexity: O(1) amortized Memory Ordering: acquire/release semantics

Definition at line 144 of file lockfree_job_queue.cpp.

                                                                     {
    // Acquire hazard pointer guards for protecting nodes (uses safe memory ordering)
    safe_hazard_guard hp_head;
    safe_hazard_guard hp_next;
 
    // Limit retries to prevent infinite loop during concurrent mode switching
    // This is important for adaptive_job_queue which may switch modes while
    // dequeue is in progress
    constexpr int MAX_OUTER_RETRIES = 100;
    constexpr int MAX_INNER_RETRIES = 10;
 
    for (int outer_retry = 0; outer_retry < MAX_OUTER_RETRIES; ++outer_retry) {
        // Protect head from reclamation
        node* head = head_.load(std::memory_order_acquire);
        hp_head.protect(head);
 
        // Verify head hasn't changed (ABA protection)
        if (head != head_.load(std::memory_order_acquire)) {
            backoff(outer_retry);
            continue;  // Head changed, retry
        }
 
        node* tail = tail_.load(std::memory_order_acquire);
 
        // Protect next node using loop until stable (with retry limit)
        node* next = nullptr;
        bool next_stable = false;
        for (int inner_retry = 0; inner_retry < MAX_INNER_RETRIES; ++inner_retry) {
            next = head->next.load(std::memory_order_acquire);
            if (next == nullptr) {
                next_stable = true;
                break;  // No next node
            }
 
            hp_next.protect(next);
 
            // Verify next is still the same after protection
            if (next == head->next.load(std::memory_order_acquire)) {
                next_stable = true;
                break;  // Stable, protected
            }
            backoff(inner_retry);
            // Next changed, retry protection
        }
 
        // If we couldn't stabilize next pointer, retry outer loop
        if (!next_stable) {
            backoff(outer_retry);
            continue;
        }
 
        // Check if head is still consistent
        if (head == head_.load(std::memory_order_acquire)) {
            if (head == tail) {
                if (next == nullptr) {
                    // Queue is empty
                    return common::error_info{static_cast<int>(error_code::queue_empty), "Queue is empty", "thread_system"};
                }
 
                // Tail is behind, try to advance it
                tail_.compare_exchange_weak(
                    tail, next,
                    std::memory_order_release,
                    std::memory_order_relaxed);
            } else {
                if (next == nullptr) {
                    // Inconsistent state, retry
                    backoff(outer_retry);
                    continue;
                }
 
                // Try to swing head to next
                if (head_.compare_exchange_weak(
                        head, next,
                        std::memory_order_release,
                        std::memory_order_relaxed)) {
 
                    // Successfully dequeued - now safe to read data
                    // We now own the old head node exclusively
                    std::unique_ptr<job> job_data = std::move(next->data);
 
                    // Retire the old head node for later reclamation (safe memory ordering)
                    // Reclaimed nodes are returned to the pool instead of deleted
                    retire_node(head);
 
                    // Update size (relaxed - just for monitoring)
                    approximate_size_.fetch_sub(1, std::memory_order_relaxed);
 
                    // Return the job data
                    return std::move(job_data);
                }
                backoff(outer_retry);
            }
        } else {
            backoff(outer_retry);
        }
    }
 
    // If we exhausted retries, report queue as empty
    // This is safe because the caller will retry if needed
    return common::error_info{static_cast<int>(error_code::queue_empty), "Queue is empty", "thread_system"};
}

References kcenon::thread::detail::lockfree_job_queue::node::data, kcenon::thread::detail::lockfree_job_queue::node::next, kcenon::thread::safe_hazard_guard::protect(), and kcenon::thread::queue_empty.

Referenced by get_next_job(), try_dequeue(), and ~lockfree_job_queue().

Here is the call graph for this function:

Here is the caller graph for this function:

empty()

auto kcenon::thread::detail::lockfree_job_queue::empty ( ) const -> bool

nodiscard

Checks if the queue is empty.

Returns

true if queue appears empty, false otherwise

Note

This is a snapshot view; queue may change immediately after

Use for hints only, not for synchronization

Definition at line 248 of file lockfree_job_queue.cpp.

                                             {
    // Use hazard pointer protection to safely access head node
    // This prevents UAF if another thread retires the head during our check
    safe_hazard_guard hp_head;
 
    // Try to get a stable read of head
    // If head keeps changing due to concurrent modifications, retry a few times
    constexpr int MAX_RETRIES = 10;
    for (int retry = 0; retry < MAX_RETRIES; ++retry) {
        node* head = head_.load(std::memory_order_acquire);
        hp_head.protect(head);
 
        // Verify head hasn't changed after protection
        if (head != head_.load(std::memory_order_acquire)) {
            continue;  // Head changed, retry
        }
 
        node* next = head->next.load(std::memory_order_acquire);
 
        // Queue is empty if head->next is null
        return next == nullptr;
    }
 
    // If we exhausted retries, do one final check without verification
    // This handles the edge case where head keeps changing but we need a definitive answer
    node* head = head_.load(std::memory_order_acquire);
    hp_head.protect(head);
    node* next = head->next.load(std::memory_order_acquire);
    return next == nullptr;
}

References head_, kcenon::thread::detail::lockfree_job_queue::node::next, kcenon::thread::safe_hazard_guard::protect(), and kcenon::thread::retry.

Here is the call graph for this function:

enqueue()

auto kcenon::thread::detail::lockfree_job_queue::enqueue ( std::unique_ptr< job > && job ) -> common::VoidResult

nodiscard

Enqueues a job into the queue (thread-safe)

Parameters

job	Unique pointer to the job to enqueue

Returns

common::VoidResult Success or error

Note

Wait-free operation (bounded number of steps)

Takes ownership of the job pointer

Never blocks, always makes progress

Time Complexity: O(1) amortized Memory Ordering: release semantics for visibility

Definition at line 76 of file lockfree_job_queue.cpp.

                                                                                 {
    if (!job_ptr) {
        return common::error_info{static_cast<int>(error_code::invalid_argument), "Cannot enqueue null job", "thread_system"};
    }
 
    // Acquire node from pool (reuses retired nodes, falls back to new)
    node* new_node = pool_->acquire(std::move(job_ptr));
 
    // Acquire hazard pointer guard for tail protection (uses safe memory ordering)
    safe_hazard_guard hp_tail;
 
    // Limit retries to prevent infinite loop during concurrent mode switching
    constexpr int MAX_RETRIES = 1000;
 
    for (int retry = 0; retry < MAX_RETRIES; ++retry) {
        // Read current tail
        node* tail = tail_.load(std::memory_order_acquire);
 
        // Protect tail to ensure it's not reclaimed while we read next
        hp_tail.protect(tail);
 
        // Verify tail hasn't changed (if it changed, our protection might be on the wrong node)
        if (tail != tail_.load(std::memory_order_acquire)) {
            backoff(retry);
            continue;
        }
 
        node* next = tail->next.load(std::memory_order_acquire);
 
        // Check if tail is still consistent
        if (tail == tail_.load(std::memory_order_acquire)) {
            if (next == nullptr) {
                // Tail is pointing to the last node, try to link new node
                if (tail->next.compare_exchange_weak(
                        next, new_node,
                        std::memory_order_release,
                        std::memory_order_relaxed)) {
 
                    // Successfully linked, try to swing tail (best effort)
                    tail_.compare_exchange_weak(
                        tail, new_node,
                        std::memory_order_release,
                        std::memory_order_relaxed);
 
                    // Update size (relaxed - just for monitoring)
                    approximate_size_.fetch_add(1, std::memory_order_relaxed);
 
                    return common::ok();  // Success
                }
                backoff(retry);
            } else {
                // Tail is behind, try to advance it
                tail_.compare_exchange_weak(
                    tail, next,
                    std::memory_order_release,
                    std::memory_order_relaxed);
            }
        } else {
            backoff(retry);
        }
    }
 
    // If we exhausted retries, return node to pool and report error
    pool_->release(new_node);
    return common::error_info{static_cast<int>(error_code::queue_busy), "Queue is busy, retry later", "thread_system"};
}

References kcenon::thread::invalid_argument, kcenon::thread::detail::lockfree_job_queue::node::next, kcenon::thread::safe_hazard_guard::protect(), kcenon::thread::queue_busy, and kcenon::thread::retry.

Referenced by schedule().

Here is the call graph for this function:

Here is the caller graph for this function:

get_capabilities()

auto kcenon::thread::detail::lockfree_job_queue::get_capabilities ( ) const -> queue_capabilities

inlinenodiscardoverridevirtual

Returns capabilities of lockfree_job_queue.

Returns

queue_capabilities with lock-free characteristics

Warning

size() is APPROXIMATE, empty() is NON-ATOMIC

Capabilities:

• exact_size: false (approximate only due to concurrent modifications)
• atomic_empty_check: false (snapshot view, may change immediately)
• lock_free: true (uses lock-free Michael-Scott algorithm)
• wait_free: false (enqueue is wait-free, dequeue is lock-free)
• supports_batch: false (no batch operations available)
• supports_blocking_wait: false (spin-wait only via try_dequeue)
• supports_stop: false (no stop() method available)

Reimplemented from kcenon::thread::queue_capabilities_interface.

Definition at line 197 of file lockfree_job_queue.h.

                                                                               {
        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
        };
    }

get_next_job()

auto kcenon::thread::detail::lockfree_job_queue::get_next_job ( ) -> common::Result<std::unique_ptr<job>>

inlineoverridevirtual

Get next job (delegates to dequeue)

Returns

common::Result<std::unique_ptr<job>> The dequeued job or error

Note

Part of scheduler_interface

Implements kcenon::thread::scheduler_interface.

Definition at line 173 of file lockfree_job_queue.h.

                                                                     {
        return dequeue();
    }

References dequeue().

Here is the call graph for this function:

operator=() [1/2]

lockfree_job_queue & kcenon::thread::detail::lockfree_job_queue::operator= ( const lockfree_job_queue & )

delete

operator=() [2/2]

lockfree_job_queue & kcenon::thread::detail::lockfree_job_queue::operator= ( lockfree_job_queue && )

delete

retire_node()

void kcenon::thread::detail::lockfree_job_queue::retire_node ( node * n )

private

Retire a node through hazard pointers, recycling via pool on reclamation.

Definition at line 286 of file lockfree_job_queue.cpp.

                                            {
    if (!n) return;
 
    // Capture pool by shared_ptr so the closure remains valid even if the
    // queue is destroyed before the hazard pointer domain reclaims this node
    std::shared_ptr<node_pool> pool = pool_;
 
    safe_hazard_pointer_domain::instance().retire(
        n,
        [pool](void* ptr) {
            pool->release(static_cast<node*>(ptr));
        }
    );
}

References kcenon::thread::safe_hazard_pointer_domain::instance(), pool_, and kcenon::thread::safe_hazard_pointer_domain::retire().

Referenced by ~lockfree_job_queue().

Here is the call graph for this function:

Here is the caller graph for this function:

schedule()

auto kcenon::thread::detail::lockfree_job_queue::schedule ( std::unique_ptr< job > && work ) -> common::VoidResult

inlineoverridevirtual

Schedule a job (delegates to enqueue)

Parameters

work	Job to schedule

Returns

common::VoidResult Success or error

Note

Part of scheduler_interface

Implements kcenon::thread::scheduler_interface.

Definition at line 162 of file lockfree_job_queue.h.

                                                                          {
        return enqueue(std::move(work));
    }

References enqueue().

Here is the call graph for this function:

size()

auto kcenon::thread::detail::lockfree_job_queue::size ( ) const -> std::size_t

nodiscard

Gets approximate queue size.

Returns

Approximate number of jobs in queue

Note

This is a best-effort estimate due to concurrent modifications

Use for monitoring/debugging, not for correctness

Definition at line 280 of file lockfree_job_queue.cpp.

                                                 {
    // Return cached size (may not be exact due to concurrent modifications)
    return approximate_size_.load(std::memory_order_relaxed);
}

References approximate_size_.

try_dequeue()

auto kcenon::thread::detail::lockfree_job_queue::try_dequeue ( ) -> common::Result<std::unique_ptr<job>>

inlinenodiscard

Tries to dequeue a job without blocking.

Returns

common::Result<std::unique_ptr<job>> The dequeued job or empty

Note

Alias for dequeue() (lock-free queues never block)

Provided for API compatibility with mutex-based queue

Definition at line 126 of file lockfree_job_queue.h.

                                                                         {
        return dequeue();
    }

References dequeue().

Here is the call graph for this function:

Member Data Documentation

approximate_size_

std::atomic<std::size_t> kcenon::thread::detail::lockfree_job_queue::approximate_size_ {0}

mutableprivate

Definition at line 315 of file lockfree_job_queue.h.

{0};

Referenced by lockfree_job_queue(), and size().

head_

std::atomic<node*> kcenon::thread::detail::lockfree_job_queue::head_

private

Definition at line 304 of file lockfree_job_queue.h.

Referenced by empty(), lockfree_job_queue(), and ~lockfree_job_queue().

pool_

std::shared_ptr<node_pool> kcenon::thread::detail::lockfree_job_queue::pool_

private

Definition at line 308 of file lockfree_job_queue.h.

Referenced by retire_node().

shutdown_

std::atomic<bool> kcenon::thread::detail::lockfree_job_queue::shutdown_ {false}

private

Definition at line 318 of file lockfree_job_queue.h.

{false};

Referenced by ~lockfree_job_queue().

tail_

std::atomic<node*> kcenon::thread::detail::lockfree_job_queue::tail_

private

Definition at line 305 of file lockfree_job_queue.h.

Referenced by lockfree_job_queue().

---

The documentation for this class was generated from the following files:

• include/kcenon/thread/lockfree/lockfree_job_queue.h
• src/lockfree/lockfree_job_queue.cpp