Thread-safe MPMC queue with blocking wait support (Internal implementation) More...

#include <concurrent_queue.h>

Collaboration diagram for kcenon::thread::detail::concurrent_queue< T >:

Classes
struct	node

Public Member Functions
	concurrent_queue ()
	Constructs an empty queue.

	~concurrent_queue ()
	Destructor - signals shutdown and drains the queue.

	concurrent_queue (const concurrent_queue &)=delete

concurrent_queue &	operator= (const concurrent_queue &)=delete

	concurrent_queue (concurrent_queue &&)=delete

concurrent_queue &	operator= (concurrent_queue &&)=delete

void	enqueue (T value)
	Enqueues a value into the queue.

auto	try_dequeue () -> std::optional< T >
	Tries to dequeue a value (non-blocking)

template<typename Rep , typename Period >
auto	wait_dequeue (const std::chrono::duration< Rep, Period > &timeout) -> std::optional< T >
	Dequeues a value with blocking wait.

auto	wait_dequeue () -> std::optional< T >
	Dequeues a value with indefinite blocking wait.

auto	empty () const noexcept -> bool
	Checks if the queue appears empty.

auto	size () const noexcept -> std::size_t
	Gets approximate queue size.

void	notify_one ()
	Wakes one waiting consumer.

void	notify_all ()
	Wakes all waiting consumers.

void	shutdown ()
	Signals shutdown and wakes all waiters.

auto	is_shutdown () const noexcept -> bool
	Checks if shutdown has been signaled.

Private Attributes
node *	head_

node *	tail_

std::mutex	head_mutex_

std::mutex	tail_mutex_

std::atomic< std::size_t >	size_ {0}

std::atomic< bool >	shutdown_ {false}

std::mutex	cv_mutex_

std::condition_variable	cv_

Detailed Description

template<typename T>
class kcenon::thread::detail::concurrent_queue< T >

Thread-safe MPMC queue with blocking wait support (Internal implementation)

This class implements a thread-safe Multi-Producer Multi-Consumer (MPMC) queue using fine-grained locking for simplicity and correctness. It provides both non-blocking and blocking operations for flexible use cases.

Implementation Notes

This implementation uses fine-grained locking with separate head and tail mutexes rather than a true lock-free algorithm. This provides:

Correctness guarantee without complex memory reclamation
Good performance for most use cases
Blocking wait support with condition variables

While a true lock-free Michael-Scott queue requires complex memory reclamation (Hazard Pointers or epoch-based reclamation), fine-grained locking offers comparable performance for most practical scenarios.

Naming Clarification

This class is intentionally named "concurrent_queue" (not "lockfree_queue") because:

It uses mutexes for synchronization (fine-grained locking)
"Lock-free" has a specific technical meaning (no locks, uses CAS loops)
"Concurrent" accurately describes thread-safe access without implying lock-free

For a true lock-free queue implementation, see lockfree_job_queue which uses the Michael-Scott algorithm with hazard pointers for memory reclamation.

Key Features

Thread Safety: Safe for concurrent access from multiple threads
Blocking Wait: wait_dequeue provides efficient blocking with timeout
Generic: Works with any move-constructible type
Low Contention: Separate locks for head and tail operations

Performance Characteristics

Enqueue: O(1), acquires tail lock only
Dequeue: O(1), acquires head lock only
No lock contention between enqueue and dequeue operations

Example Usage

detail::concurrent_queue<std::string> queue;
 
// Producer thread
queue.enqueue("message");
 
// Consumer thread (non-blocking)
if (auto value = queue.try_dequeue()) {
    process(*value);
}
 
// Consumer thread (blocking with timeout)
if (auto value = queue.wait_dequeue(std::chrono::milliseconds{100})) {
    process(*value);
}

Template Parameters

T	The element type (must be move-constructible)

Definition at line 156 of file forward.h.

Constructor & Destructor Documentation

◆ concurrent_queue() [1/3]

template<typename T >

kcenon::thread::detail::concurrent_queue< T >::concurrent_queue ( )

inline

Constructs an empty queue.

Initializes the queue with a dummy node to simplify the algorithm.

Definition at line 100 of file concurrent_queue.h.

                       {
        auto* dummy = new node{};
        head_ = dummy;
        tail_ = dummy;
    }

References kcenon::thread::detail::concurrent_queue< T >::head_, and kcenon::thread::detail::concurrent_queue< T >::tail_.

◆ ~concurrent_queue()

template<typename T >

kcenon::thread::detail::concurrent_queue< T >::~concurrent_queue ( )

inline

Destructor - signals shutdown and drains the queue.

Definition at line 109 of file concurrent_queue.h.

                        {
        shutdown();
        // Drain remaining items
        while (try_dequeue()) {
        }
        // Delete the dummy node
        delete head_;
    }

References kcenon::thread::detail::concurrent_queue< T >::head_, kcenon::thread::detail::concurrent_queue< T >::shutdown(), and kcenon::thread::detail::concurrent_queue< T >::try_dequeue().

Here is the call graph for this function:

◆ concurrent_queue() [2/3]

template<typename T >

kcenon::thread::detail::concurrent_queue< T >::concurrent_queue ( const concurrent_queue< T > & )

delete

◆ concurrent_queue() [3/3]

template<typename T >

kcenon::thread::detail::concurrent_queue< T >::concurrent_queue ( concurrent_queue< T > && )

delete

Member Function Documentation

◆ empty()

template<typename T >

auto kcenon::thread::detail::concurrent_queue< T >::empty ( ) const -> bool

inlinenodiscardnoexcept

Checks if the queue appears empty.

Definition at line 231 of file concurrent_queue.h.

                                                      {
        return size_.load(std::memory_order_acquire) == 0;
    }

References kcenon::thread::detail::concurrent_queue< T >::size_.

◆ enqueue()

template<typename T >

void kcenon::thread::detail::concurrent_queue< T >::enqueue ( T value )

inline

Enqueues a value into the queue.

Parameters

value The value to enqueue (moved)

Note: Thread-safe, uses tail lock only

Definition at line 131 of file concurrent_queue.h.

                          {
        if (shutdown_.load(std::memory_order_acquire)) {
            return;
        }
 
        auto* new_node = new node{std::move(value)};
 
        {
            std::lock_guard<std::mutex> lock(tail_mutex_);
            tail_->next = new_node;
            tail_ = new_node;
        }
 
        size_.fetch_add(1, std::memory_order_release);
        notify_one();
    }

References kcenon::thread::detail::concurrent_queue< T >::node::next, kcenon::thread::detail::concurrent_queue< T >::notify_one(), kcenon::thread::detail::concurrent_queue< T >::shutdown_, kcenon::thread::detail::concurrent_queue< T >::size_, kcenon::thread::detail::concurrent_queue< T >::tail_, and kcenon::thread::detail::concurrent_queue< T >::tail_mutex_.

Here is the call graph for this function:

◆ is_shutdown()

template<typename T >

auto kcenon::thread::detail::concurrent_queue< T >::is_shutdown ( ) const -> bool

inlinenodiscardnoexcept

Checks if shutdown has been signaled.

Definition at line 269 of file concurrent_queue.h.

                                                            {
        return shutdown_.load(std::memory_order_acquire);
    }

References kcenon::thread::detail::concurrent_queue< T >::shutdown_.

◆ notify_all()

template<typename T >

void kcenon::thread::detail::concurrent_queue< T >::notify_all ( )

inline

Wakes all waiting consumers.

Definition at line 253 of file concurrent_queue.h.

                      {
        std::lock_guard<std::mutex> lock(cv_mutex_);
        cv_.notify_all();
    }

References kcenon::thread::detail::concurrent_queue< T >::cv_, and kcenon::thread::detail::concurrent_queue< T >::cv_mutex_.

Referenced by kcenon::thread::detail::concurrent_queue< T >::shutdown().

Here is the caller graph for this function:

◆ notify_one()

template<typename T >

void kcenon::thread::detail::concurrent_queue< T >::notify_one ( )

inline

Wakes one waiting consumer.

Definition at line 245 of file concurrent_queue.h.

                      {
        std::lock_guard<std::mutex> lock(cv_mutex_);
        cv_.notify_one();
    }

References kcenon::thread::detail::concurrent_queue< T >::cv_, and kcenon::thread::detail::concurrent_queue< T >::cv_mutex_.

Referenced by kcenon::thread::detail::concurrent_queue< T >::enqueue().

Here is the caller graph for this function:

◆ operator=() [1/2]

template<typename T >

concurrent_queue & kcenon::thread::detail::concurrent_queue< T >::operator= ( concurrent_queue< T > && )

delete

◆ operator=() [2/2]

template<typename T >

concurrent_queue & kcenon::thread::detail::concurrent_queue< T >::operator= ( const concurrent_queue< T > & )

delete

◆ shutdown()

template<typename T >

void kcenon::thread::detail::concurrent_queue< T >::shutdown ( )

inline

Signals shutdown and wakes all waiters.

Definition at line 261 of file concurrent_queue.h.

                    {
        shutdown_.store(true, std::memory_order_release);
        notify_all();
    }

References kcenon::thread::detail::concurrent_queue< T >::notify_all(), and kcenon::thread::detail::concurrent_queue< T >::shutdown_.

Referenced by kcenon::thread::detail::concurrent_queue< T >::~concurrent_queue().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ size()

template<typename T >

auto kcenon::thread::detail::concurrent_queue< T >::size ( ) const -> std::size_t

inlinenodiscardnoexcept

Gets approximate queue size.

Definition at line 238 of file concurrent_queue.h.

                                                          {
        return size_.load(std::memory_order_acquire);
    }

References kcenon::thread::detail::concurrent_queue< T >::size_.

◆ try_dequeue()

template<typename T >

auto kcenon::thread::detail::concurrent_queue< T >::try_dequeue ( ) -> std::optional<T>

inlinenodiscard

Tries to dequeue a value (non-blocking)

Returns: The dequeued value, or nullopt if queue is empty

Note: Thread-safe, uses head lock only

Definition at line 155 of file concurrent_queue.h.

                                                       {
        std::lock_guard<std::mutex> lock(head_mutex_);
 
        node* old_head = head_;
        node* next = old_head->next;
 
        if (next == nullptr) {
            return std::nullopt;  // Queue is empty
        }
 
        // Extract value from next node
        std::optional<T> value;
        if (next->data.has_value()) {
            value = std::move(*next->data);
            next->data.reset();
        }
 
        // Advance head (old_head becomes the new dummy)
        head_ = next;
        delete old_head;
 
        size_.fetch_sub(1, std::memory_order_release);
        return value;
    }

References kcenon::thread::detail::concurrent_queue< T >::node::data, kcenon::thread::detail::concurrent_queue< T >::head_, kcenon::thread::detail::concurrent_queue< T >::head_mutex_, kcenon::thread::detail::concurrent_queue< T >::node::next, and kcenon::thread::detail::concurrent_queue< T >::size_.

Referenced by kcenon::thread::detail::concurrent_queue< T >::wait_dequeue(), and kcenon::thread::detail::concurrent_queue< T >::~concurrent_queue().

Here is the caller graph for this function:

◆ wait_dequeue() [1/2]

template<typename T >

auto kcenon::thread::detail::concurrent_queue< T >::wait_dequeue ( ) -> std::optional<T>

inlinenodiscard

Dequeues a value with indefinite blocking wait.

Returns: The dequeued value, or nullopt if shutdown

Definition at line 224 of file concurrent_queue.h.

                                                        {
        return wait_dequeue(std::chrono::hours{24 * 365});
    }

References kcenon::thread::detail::concurrent_queue< T >::wait_dequeue().

Referenced by kcenon::thread::detail::concurrent_queue< T >::wait_dequeue().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ wait_dequeue() [2/2]

template<typename T >

template<typename Rep , typename Period >

auto kcenon::thread::detail::concurrent_queue< T >::wait_dequeue ( const std::chrono::duration< Rep, Period > & timeout ) -> std::optional<T>

inlinenodiscard

Dequeues a value with blocking wait.

Parameters

timeout Maximum time to wait

Returns: The dequeued value, or nullopt if timeout or shutdown

Definition at line 187 of file concurrent_queue.h.

                          {
        // First try without waiting
        if (auto value = try_dequeue()) {
            return value;
        }
 
        // Wait for notification
        std::unique_lock<std::mutex> lock(cv_mutex_);
        auto deadline = std::chrono::steady_clock::now() + timeout;
 
        while (!shutdown_.load(std::memory_order_acquire)) {
            // Try to dequeue
            lock.unlock();
            if (auto value = try_dequeue()) {
                return value;
            }
            lock.lock();
 
            // Check if we should stop waiting
            if (std::chrono::steady_clock::now() >= deadline) {
                return std::nullopt;
            }
 
            // Wait for notification or timeout
            cv_.wait_until(lock, deadline);
        }
 
        // Final attempt after shutdown
        return try_dequeue();
    }