kcenon::monitoring::lockfree_queue< T > Class Template Reference

Thread-safe lock-free MPMC (Multiple Producer Multiple Consumer) queue. More...

#include <lockfree_queue.h>

Collaboration diagram for kcenon::monitoring::lockfree_queue< T >:

Classes
struct	slot

Public Member Functions
	lockfree_queue ()
	Default constructor with default configuration.
	lockfree_queue (const lockfree_queue_config &config)
	Construct with configuration.
	lockfree_queue (const lockfree_queue &)=delete
lockfree_queue &	operator= (const lockfree_queue &)=delete
	lockfree_queue (lockfree_queue &&other) noexcept
lockfree_queue &	operator= (lockfree_queue &&other) noexcept
common::Result< bool >	push (const T &value)
	Push an element to the queue.
common::Result< bool >	push (T &&value)
	Push an element to the queue (move version)
common::Result< T >	pop ()
	Pop an element from the queue.
bool	empty () const
	Check if the queue is empty.
size_t	size () const
	Get current queue size.
size_t	capacity () const
	Get queue capacity.
const lockfree_queue_statistics &	get_statistics () const
	Get queue statistics.
void	reset_statistics ()
	Reset statistics.

Private Member Functions
template<typename U >
common::Result< bool >	push_impl (U &&value)

Private Attributes
lockfree_queue_config	config_
size_t	capacity_
std::vector< slot >	buffer_
std::atomic< size_t >	head_
std::atomic< size_t >	tail_
std::atomic< size_t >	size_
lockfree_queue_statistics	stats_

Detailed Description

template<typename T>
class kcenon::monitoring::lockfree_queue< T >

Thread-safe lock-free MPMC (Multiple Producer Multiple Consumer) queue.

This implementation uses a bounded ring buffer with atomic operations for thread-safe access without locks.

Template Parameters

T	The type of elements stored in the queue

Definition at line 120 of file lockfree_queue.h.

Constructor & Destructor Documentation

lockfree_queue() [1/4]

template<typename T >

kcenon::monitoring::lockfree_queue< T >::lockfree_queue ( )

inline

Default constructor with default configuration.

Definition at line 125 of file lockfree_queue.h.

: lockfree_queue(lockfree_queue_config{}) {}

lockfree_queue() [2/4]

template<typename T >

kcenon::monitoring::lockfree_queue< T >::lockfree_queue ( const lockfree_queue_config & config )

inlineexplicit

Construct with configuration.

Parameters

config

Queue configuration

Definition at line 131 of file lockfree_queue.h.

        : config_(config)
        , capacity_(config.initial_capacity)
        , buffer_(config.initial_capacity)
        , head_(0)
        , tail_(0)
        , size_(0) {
        // Initialize each slot's sequence to its index
        for (size_t i = 0; i < capacity_; ++i) {
            buffer_[i].sequence.store(i, std::memory_order_relaxed);
        }
    }

References kcenon::monitoring::lockfree_queue< T >::buffer_, and kcenon::monitoring::lockfree_queue< T >::capacity_.

lockfree_queue() [3/4]

template<typename T >

kcenon::monitoring::lockfree_queue< T >::lockfree_queue ( const lockfree_queue< T > & )

delete

lockfree_queue() [4/4]

template<typename T >

kcenon::monitoring::lockfree_queue< T >::lockfree_queue ( lockfree_queue< T > && other )

inlinenoexcept

Definition at line 149 of file lockfree_queue.h.

        : config_(std::move(other.config_))
        , capacity_(other.capacity_)
        , buffer_(std::move(other.buffer_))
        , head_(other.head_.load())
        , tail_(other.tail_.load())
        , size_(other.size_.load())
        , stats_(std::move(other.stats_)) {}

Member Function Documentation

capacity()

template<typename T >

size_t kcenon::monitoring::lockfree_queue< T >::capacity ( ) const

inline

Get queue capacity.

Returns

Current capacity of the queue

Definition at line 245 of file lockfree_queue.h.

                            {
        return capacity_;
    }

References kcenon::monitoring::lockfree_queue< T >::capacity_.

empty()

template<typename T >

bool kcenon::monitoring::lockfree_queue< T >::empty ( ) const

inline

Check if the queue is empty.

Returns

true if empty

Definition at line 229 of file lockfree_queue.h.

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

References kcenon::monitoring::lockfree_queue< T >::size_.

Referenced by TEST_F(), TEST_F(), TEST_F(), and TEST_F().

Here is the caller graph for this function:

get_statistics()

template<typename T >

const lockfree_queue_statistics & kcenon::monitoring::lockfree_queue< T >::get_statistics ( ) const

inline

Get queue statistics.

Returns

Reference to statistics

Definition at line 253 of file lockfree_queue.h.

                                                            {
        return stats_;
    }

References kcenon::monitoring::lockfree_queue< T >::stats_.

Referenced by TEST_F(), TEST_F(), and TEST_F().

Here is the caller graph for this function:

operator=() [1/2]

template<typename T >

lockfree_queue & kcenon::monitoring::lockfree_queue< T >::operator= ( const lockfree_queue< T > & )

delete

operator=() [2/2]

template<typename T >

lockfree_queue & kcenon::monitoring::lockfree_queue< T >::operator= ( lockfree_queue< T > && other )

inlinenoexcept

Definition at line 158 of file lockfree_queue.h.

                                                               {
        if (this != &other) {
            config_ = std::move(other.config_);
            capacity_ = other.capacity_;
            buffer_ = std::move(other.buffer_);
            head_.store(other.head_.load());
            tail_.store(other.tail_.load());
            size_.store(other.size_.load());
            stats_ = std::move(other.stats_);
        }
        return *this;
    }

References kcenon::monitoring::lockfree_queue< T >::buffer_, kcenon::monitoring::lockfree_queue< T >::capacity_, kcenon::monitoring::lockfree_queue< T >::config_, kcenon::monitoring::lockfree_queue< T >::head_, kcenon::monitoring::other, kcenon::monitoring::lockfree_queue< T >::size_, kcenon::monitoring::lockfree_queue< T >::stats_, and kcenon::monitoring::lockfree_queue< T >::tail_.

pop()

template<typename T >

common::Result< T > kcenon::monitoring::lockfree_queue< T >::pop ( )

inline

Pop an element from the queue.

Returns

common::Result<T> containing the value on success, error if queue is empty

Definition at line 193 of file lockfree_queue.h.

                          {
        stats_.pop_attempts++;
 
        size_t current_head = head_.load(std::memory_order_relaxed);
 
        while (true) {
            size_t index = current_head % capacity_;
            auto& slot = buffer_[index];
            size_t seq = slot.sequence.load(std::memory_order_acquire);
            intptr_t diff = static_cast<intptr_t>(seq) - static_cast<intptr_t>(current_head + 1);
 
            if (diff == 0) {
                // Slot is ready to be read
                if (head_.compare_exchange_weak(current_head, current_head + 1,
                                                 std::memory_order_relaxed)) {
                    T value = std::move(slot.data);
                    slot.sequence.store(current_head + capacity_, std::memory_order_release);
                    size_.fetch_sub(1, std::memory_order_relaxed);
                    stats_.pop_successes++;
                    return common::ok(std::move(value));
                }
            } else if (diff < 0) {
                // Queue is empty
                stats_.pop_failures++;
                return common::Result<T>::err(error_info(monitoring_error_code::resource_unavailable, "Queue is empty").to_common_error());
            } else {
                // Another thread is modifying, retry
                current_head = head_.load(std::memory_order_relaxed);
            }
        }
    }

References kcenon::monitoring::lockfree_queue< T >::buffer_, kcenon::monitoring::lockfree_queue< T >::capacity_, kcenon::monitoring::lockfree_queue< T >::slot::data, kcenon::monitoring::lockfree_queue< T >::head_, kcenon::monitoring::lockfree_queue_statistics::pop_attempts, kcenon::monitoring::lockfree_queue_statistics::pop_failures, kcenon::monitoring::lockfree_queue_statistics::pop_successes, kcenon::monitoring::resource_unavailable, kcenon::monitoring::lockfree_queue< T >::slot::sequence, kcenon::monitoring::lockfree_queue< T >::size_, and kcenon::monitoring::lockfree_queue< T >::stats_.

Referenced by TEST_F(), TEST_F(), TEST_F(), TEST_F(), TEST_F(), TEST_F(), and TEST_F().

Here is the caller graph for this function:

push() [1/2]

template<typename T >

common::Result< bool > kcenon::monitoring::lockfree_queue< T >::push ( const T & value )

inline

Push an element to the queue.

Parameters

value

The value to push

Returns

common::Result<bool> containing true on success, false if queue is full

Definition at line 176 of file lockfree_queue.h.

                                            {
        return push_impl(value);
    }

References kcenon::monitoring::lockfree_queue< T >::push_impl().

Referenced by TEST_F(), TEST_F(), TEST_F(), TEST_F(), TEST_F(), TEST_F(), and TEST_F().

Here is the call graph for this function:

Here is the caller graph for this function:

push() [2/2]

template<typename T >

common::Result< bool > kcenon::monitoring::lockfree_queue< T >::push ( T && value )

inline

Push an element to the queue (move version)

Parameters

value

The value to push

Returns

common::Result<bool> containing true on success, false if queue is full

Definition at line 185 of file lockfree_queue.h.

                                       {
        return push_impl(std::move(value));
    }

References kcenon::monitoring::lockfree_queue< T >::push_impl().

Here is the call graph for this function:

push_impl()

template<typename T >

template<typename U >

common::Result< bool > kcenon::monitoring::lockfree_queue< T >::push_impl ( U && value )

inlineprivate

Definition at line 271 of file lockfree_queue.h.

                                            {
        stats_.push_attempts++;
 
        size_t current_tail = tail_.load(std::memory_order_relaxed);
 
        while (true) {
            size_t index = current_tail % capacity_;
            auto& slot = buffer_[index];
            size_t seq = slot.sequence.load(std::memory_order_acquire);
            intptr_t diff = static_cast<intptr_t>(seq) - static_cast<intptr_t>(current_tail);
 
            if (diff == 0) {
                // Slot is available for writing
                if (tail_.compare_exchange_weak(current_tail, current_tail + 1,
                                                 std::memory_order_relaxed)) {
                    slot.data = std::forward<U>(value);
                    slot.sequence.store(current_tail + 1, std::memory_order_release);
                    size_.fetch_add(1, std::memory_order_relaxed);
                    stats_.push_successes++;
                    return common::ok(true);
                }
            } else if (diff < 0) {
                // Queue is full
                stats_.push_failures++;
                return common::ok(false);
            } else {
                // Another thread is modifying, retry
                current_tail = tail_.load(std::memory_order_relaxed);
            }
        }
    }

References kcenon::monitoring::lockfree_queue< T >::buffer_, kcenon::monitoring::lockfree_queue< T >::capacity_, kcenon::monitoring::lockfree_queue< T >::slot::data, kcenon::monitoring::lockfree_queue_statistics::push_attempts, kcenon::monitoring::lockfree_queue_statistics::push_failures, kcenon::monitoring::lockfree_queue_statistics::push_successes, kcenon::monitoring::lockfree_queue< T >::slot::sequence, kcenon::monitoring::lockfree_queue< T >::size_, kcenon::monitoring::lockfree_queue< T >::stats_, and kcenon::monitoring::lockfree_queue< T >::tail_.

Referenced by kcenon::monitoring::lockfree_queue< T >::push(), and kcenon::monitoring::lockfree_queue< T >::push().

Here is the caller graph for this function:

reset_statistics()

template<typename T >

void kcenon::monitoring::lockfree_queue< T >::reset_statistics ( )

inline

Reset statistics.

Definition at line 260 of file lockfree_queue.h.

                            {
        stats_.reset();
    }

References kcenon::monitoring::lockfree_queue_statistics::reset(), and kcenon::monitoring::lockfree_queue< T >::stats_.

Referenced by TEST_F().

Here is the call graph for this function:

Here is the caller graph for this function:

size()

template<typename T >

size_t kcenon::monitoring::lockfree_queue< T >::size ( ) const

inline

Get current queue size.

Returns

Number of elements in the queue

Definition at line 237 of file lockfree_queue.h.

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

References kcenon::monitoring::lockfree_queue< T >::size_.

Referenced by TEST_F(), TEST_F(), and TEST_F().

Here is the caller graph for this function:

Member Data Documentation

buffer_

template<typename T >

std::vector<slot> kcenon::monitoring::lockfree_queue< T >::buffer_

private

Definition at line 305 of file lockfree_queue.h.

Referenced by kcenon::monitoring::lockfree_queue< T >::lockfree_queue(), kcenon::monitoring::lockfree_queue< T >::operator=(), kcenon::monitoring::lockfree_queue< T >::pop(), and kcenon::monitoring::lockfree_queue< T >::push_impl().

capacity_

template<typename T >

size_t kcenon::monitoring::lockfree_queue< T >::capacity_

private

Definition at line 304 of file lockfree_queue.h.

Referenced by kcenon::monitoring::lockfree_queue< T >::capacity(), kcenon::monitoring::lockfree_queue< T >::lockfree_queue(), kcenon::monitoring::lockfree_queue< T >::operator=(), kcenon::monitoring::lockfree_queue< T >::pop(), and kcenon::monitoring::lockfree_queue< T >::push_impl().

config_

template<typename T >

lockfree_queue_config kcenon::monitoring::lockfree_queue< T >::config_

private

Definition at line 303 of file lockfree_queue.h.

Referenced by kcenon::monitoring::lockfree_queue< T >::operator=().

head_

template<typename T >

std::atomic<size_t> kcenon::monitoring::lockfree_queue< T >::head_

private

Definition at line 306 of file lockfree_queue.h.

Referenced by kcenon::monitoring::lockfree_queue< T >::operator=(), and kcenon::monitoring::lockfree_queue< T >::pop().

size_

template<typename T >

std::atomic<size_t> kcenon::monitoring::lockfree_queue< T >::size_

private

Definition at line 308 of file lockfree_queue.h.

Referenced by kcenon::monitoring::lockfree_queue< T >::empty(), kcenon::monitoring::lockfree_queue< T >::operator=(), kcenon::monitoring::lockfree_queue< T >::pop(), kcenon::monitoring::lockfree_queue< T >::push_impl(), and kcenon::monitoring::lockfree_queue< T >::size().

stats_

template<typename T >

lockfree_queue_statistics kcenon::monitoring::lockfree_queue< T >::stats_

mutableprivate

Definition at line 309 of file lockfree_queue.h.

Referenced by kcenon::monitoring::lockfree_queue< T >::get_statistics(), kcenon::monitoring::lockfree_queue< T >::operator=(), kcenon::monitoring::lockfree_queue< T >::pop(), kcenon::monitoring::lockfree_queue< T >::push_impl(), and kcenon::monitoring::lockfree_queue< T >::reset_statistics().

tail_

template<typename T >

std::atomic<size_t> kcenon::monitoring::lockfree_queue< T >::tail_

private

Definition at line 307 of file lockfree_queue.h.

Referenced by kcenon::monitoring::lockfree_queue< T >::operator=(), and kcenon::monitoring::lockfree_queue< T >::push_impl().

---

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

• include/kcenon/monitoring/optimization/lockfree_queue.h