Lock-free ring buffer with atomic operations. More...

#include <ring_buffer.h>

Inheritance diagram for kcenon::monitoring::ring_buffer< T >:

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

Public Member Functions
	ring_buffer (const ring_buffer_config &config={})
	Constructor with configuration.

	ring_buffer (const ring_buffer &)=delete

ring_buffer &	operator= (const ring_buffer &)=delete

	ring_buffer (ring_buffer &&)=default

ring_buffer &	operator= (ring_buffer &&)=default

common::VoidResult	write (T &&item)
	Write a single element to the buffer.

size_t	write_batch (std::vector< T > &&items)
	Write multiple elements in batch.

common::VoidResult	read (T &item)
	Read a single element from the buffer.

size_t	read_batch (std::vector< T > &items, size_t max_count=SIZE_MAX)
	Read multiple elements in batch.

common::VoidResult	peek (T &item) const
	Peek at the next item without removing it.

size_t	size () const noexcept
	Get current number of elements in buffer.

bool	empty () const noexcept
	Check if buffer is empty.

bool	full () const noexcept
	Check if buffer is full.

size_t	capacity () const noexcept
	Get buffer capacity.

void	clear () noexcept
	Clear all elements in the buffer.

const ring_buffer_config &	get_config () const noexcept
	Get buffer configuration.

const ring_buffer_stats &	get_stats () const noexcept
	Get buffer statistics.

void	reset_stats () noexcept
	Reset statistics.

bool	is_overflow_rate_high () const noexcept
	Check if buffer is experiencing high overflow.

double	get_overflow_rate () const noexcept
	Get overflow rate percentage.

Private Member Functions
size_t	get_mask () const noexcept
	Get the mask for efficient modulo operation.

bool	is_full_unsafe (size_t write_idx, size_t read_idx) const noexcept
	Check if buffer is full.

bool	is_empty_unsafe (size_t write_idx, size_t read_idx) const noexcept
	Check if buffer is empty.

Private Attributes
std::atomic< size_t >	write_index_ {0}

std::atomic< size_t >	read_index_ {0}

std::unique_ptr< T[]>	buffer_

ring_buffer_config	config_

ring_buffer_stats	stats_

Detailed Description

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

Lock-free ring buffer with atomic operations.

Template Parameters

T	The type of elements to store

This implementation uses atomic operations for thread-safety and provides efficient circular buffer semantics with configurable overflow behavior.

Definition at line 141 of file ring_buffer.h.

Constructor & Destructor Documentation

◆ ring_buffer() [1/3]

template<typename T >

kcenon::monitoring::ring_buffer< T >::ring_buffer ( const ring_buffer_config & config = {} )

inlineexplicit

Constructor with configuration.

Parameters

config Ring buffer configuration options

Exceptions

std::invalid_argument if configuration validation fails

Definition at line 180 of file ring_buffer.h.

                                                            {})
        : buffer_(std::make_unique<T[]>(config.capacity))
        , config_(config) {
 
        // Validate configuration
        auto validation = config_.validate();
        if (validation.is_err()) {
            throw std::invalid_argument("Invalid ring buffer configuration: " +
                                      validation.error().message);
        }
    }

◆ ring_buffer() [2/3]

template<typename T >

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

delete

◆ ring_buffer() [3/3]

template<typename T >

kcenon::monitoring::ring_buffer< T >::ring_buffer ( ring_buffer< T > && )

default

Member Function Documentation

◆ capacity()

template<typename T >

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

inlinenoexcept

Get buffer capacity.

Definition at line 416 of file ring_buffer.h.

                                     {
        return config_.capacity;
    }

References kcenon::monitoring::ring_buffer_config::capacity, and kcenon::monitoring::ring_buffer< T >::config_.

Referenced by TEST_F().

Here is the caller graph for this function:

◆ clear()

template<typename T >

void kcenon::monitoring::ring_buffer< T >::clear ( )

inlinenoexcept

Clear all elements in the buffer.

Definition at line 423 of file ring_buffer.h.

                          {
        write_index_.store(0, std::memory_order_release);
        read_index_.store(0, std::memory_order_release);
    }

References kcenon::monitoring::ring_buffer< T >::read_index_, and kcenon::monitoring::ring_buffer< T >::write_index_.

◆ empty()

template<typename T >

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

inlinenoexcept

Check if buffer is empty.

Definition at line 400 of file ring_buffer.h.

                                {
        return size() == 0;
    }

References kcenon::monitoring::ring_buffer< T >::size().

Referenced by TEST_F().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ full()

template<typename T >

bool kcenon::monitoring::ring_buffer< T >::full ( ) const

inlinenoexcept

Check if buffer is full.

Definition at line 407 of file ring_buffer.h.

                               {
        size_t write_idx = write_index_.load(std::memory_order_acquire);
        size_t read_idx = read_index_.load(std::memory_order_acquire);
        return is_full_unsafe(write_idx, read_idx);
    }

References kcenon::monitoring::ring_buffer< T >::is_full_unsafe(), kcenon::monitoring::ring_buffer< T >::read_index_, and kcenon::monitoring::ring_buffer< T >::write_index_.

Referenced by TEST_F().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ get_config()

template<typename T >

const ring_buffer_config & kcenon::monitoring::ring_buffer< T >::get_config ( ) const

inlinenoexcept

Get buffer configuration.

Definition at line 431 of file ring_buffer.h.

                                                          {
        return config_;
    }

References kcenon::monitoring::ring_buffer< T >::config_.

◆ get_mask()

template<typename T >

size_t kcenon::monitoring::ring_buffer< T >::get_mask ( ) const

inlineprivatenoexcept

Get the mask for efficient modulo operation.

Definition at line 156 of file ring_buffer.h.

                                     {
        return config_.capacity - 1;
    }

References kcenon::monitoring::ring_buffer_config::capacity, and kcenon::monitoring::ring_buffer< T >::config_.

Referenced by kcenon::monitoring::ring_buffer< T >::is_full_unsafe(), kcenon::monitoring::ring_buffer< T >::read(), and kcenon::monitoring::ring_buffer< T >::write().

Here is the caller graph for this function:

◆ get_overflow_rate()

template<typename T >

double kcenon::monitoring::ring_buffer< T >::get_overflow_rate ( ) const

inlinenoexcept

Get overflow rate percentage.

Definition at line 466 of file ring_buffer.h.

                                              {
        return stats_.get_overflow_rate();
    }

References kcenon::monitoring::ring_buffer_stats::get_overflow_rate(), and kcenon::monitoring::ring_buffer< T >::stats_.

Here is the call graph for this function:

◆ get_stats()

template<typename T >

const ring_buffer_stats & kcenon::monitoring::ring_buffer< T >::get_stats ( ) const

inlinenoexcept

Get buffer statistics.

Definition at line 438 of file ring_buffer.h.

                                                        {
        return stats_;
    }

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

◆ is_empty_unsafe()

template<typename T >

bool kcenon::monitoring::ring_buffer< T >::is_empty_unsafe	(	size_t	write_idx,
		size_t	read_idx ) const

inlineprivatenoexcept

Check if buffer is empty.

Definition at line 170 of file ring_buffer.h.

                                                                           {
        return write_idx == read_idx;
    }

Referenced by kcenon::monitoring::ring_buffer< T >::peek(), and kcenon::monitoring::ring_buffer< T >::read().

Here is the caller graph for this function:

◆ is_full_unsafe()

template<typename T >

bool kcenon::monitoring::ring_buffer< T >::is_full_unsafe	(	size_t	write_idx,
		size_t	read_idx ) const

inlineprivatenoexcept

Check if buffer is full.

Definition at line 163 of file ring_buffer.h.

                                                                          {
        return ((write_idx + 1) & get_mask()) == read_idx;
    }

References kcenon::monitoring::ring_buffer< T >::get_mask().

Referenced by kcenon::monitoring::ring_buffer< T >::full(), and kcenon::monitoring::ring_buffer< T >::write().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ is_overflow_rate_high()

template<typename T >

bool kcenon::monitoring::ring_buffer< T >::is_overflow_rate_high ( ) const

inlinenoexcept

Check if buffer is experiencing high overflow.

Returns: true if overflow rate exceeds threshold

Definition at line 459 of file ring_buffer.h.

                                                {
        return stats_.is_overflow_rate_high();
    }

References kcenon::monitoring::ring_buffer_stats::is_overflow_rate_high(), and kcenon::monitoring::ring_buffer< T >::stats_.

Here is the call graph for this function:

◆ operator=() [1/2]

template<typename T >

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

delete

◆ operator=() [2/2]

template<typename T >

ring_buffer & kcenon::monitoring::ring_buffer< T >::operator= ( ring_buffer< T > && )

default

◆ peek()

template<typename T >

common::VoidResult kcenon::monitoring::ring_buffer< T >::peek ( T & item ) const

inline

Peek at the next item without removing it.

Parameters

item	Reference to store the peeked item

Returns: Result indicating success or failure

Definition at line 368 of file ring_buffer.h.

                                         {
        size_t current_read = read_index_.load(std::memory_order_acquire);
        size_t current_write = write_index_.load(std::memory_order_acquire);
 
        if (is_empty_unsafe(current_write, current_read)) {
            return common::VoidResult::err(error_info(monitoring_error_code::collection_failed,
                             "Ring buffer is empty").to_common_error());
        }
 
        item = buffer_[current_read]; // Copy, don't move
        return common::ok();
    }

References kcenon::monitoring::ring_buffer< T >::buffer_, kcenon::monitoring::collection_failed, kcenon::monitoring::ring_buffer< T >::is_empty_unsafe(), kcenon::monitoring::ring_buffer< T >::read_index_, and kcenon::monitoring::ring_buffer< T >::write_index_.

Referenced by TEST_F().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ read()

template<typename T >

common::VoidResult kcenon::monitoring::ring_buffer< T >::read ( T & item )

inline

Read a single element from the buffer.

Parameters

item	Reference to store the read item

Returns: Result indicating success or failure

Definition at line 311 of file ring_buffer.h.

                                   {
        stats_.total_reads.fetch_add(1, std::memory_order_relaxed);
 
        size_t current_read = read_index_.load(std::memory_order_acquire);
        size_t current_write = write_index_.load(std::memory_order_acquire);
 
        if (is_empty_unsafe(current_write, current_read)) {
            stats_.failed_reads.fetch_add(1, std::memory_order_relaxed);
            return common::VoidResult::err(error_info(monitoring_error_code::collection_failed,
                             "Ring buffer is empty").to_common_error());
        }
 
        // Read the item
        item = std::move(buffer_[current_read]);
 
        // Update read index
        size_t new_read = (current_read + 1) & get_mask();
        read_index_.store(new_read, std::memory_order_release);
 
        return common::ok();
    }

References kcenon::monitoring::ring_buffer< T >::buffer_, kcenon::monitoring::collection_failed, kcenon::monitoring::ring_buffer_stats::failed_reads, kcenon::monitoring::ring_buffer< T >::get_mask(), kcenon::monitoring::ring_buffer< T >::is_empty_unsafe(), kcenon::monitoring::ring_buffer< T >::read_index_, kcenon::monitoring::ring_buffer< T >::stats_, kcenon::monitoring::ring_buffer_stats::total_reads, and kcenon::monitoring::ring_buffer< T >::write_index_.

Referenced by kcenon::monitoring::ring_buffer< T >::read_batch(), TEST_F(), TEST_F(), and TEST_F().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ read_batch()

template<typename T >

size_t kcenon::monitoring::ring_buffer< T >::read_batch	(	std::vector< T > &	items,
		size_t	max_count = SIZE_MAX )

inline

Read multiple elements in batch.

Parameters

items	Vector to store read items
max_count	Maximum number of items to read

Returns: Number of items actually read

Definition at line 339 of file ring_buffer.h.

                                                                        {
        if (max_count == 0) {
            return 0;
        }
 
        size_t batch_size = std::min(max_count, config_.batch_size);
        items.reserve(items.size() + batch_size);
 
        size_t read_count = 0;
        T temp_item;
 
        while (read_count < batch_size) {
            auto result = read(temp_item);
            if (result.is_err()) {
                break; // No more items to read
            }
 
            items.emplace_back(std::move(temp_item));
            ++read_count;
        }
 
        return read_count;
    }

References kcenon::monitoring::ring_buffer_config::batch_size, kcenon::monitoring::ring_buffer< T >::config_, and kcenon::monitoring::ring_buffer< T >::read().

Referenced by TEST_F().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ reset_stats()

template<typename T >

void kcenon::monitoring::ring_buffer< T >::reset_stats ( )

inlinenoexcept

Reset statistics.

Definition at line 445 of file ring_buffer.h.

                                {
        stats_.total_writes.store(0);
        stats_.total_reads.store(0);
        stats_.overwrites.store(0);
        stats_.failed_writes.store(0);
        stats_.failed_reads.store(0);
        stats_.contention_retries.store(0);
        stats_.creation_time = std::chrono::system_clock::now();
    }

References kcenon::monitoring::ring_buffer_stats::contention_retries, kcenon::monitoring::ring_buffer_stats::creation_time, kcenon::monitoring::ring_buffer_stats::failed_reads, kcenon::monitoring::ring_buffer_stats::failed_writes, kcenon::monitoring::ring_buffer_stats::overwrites, kcenon::monitoring::ring_buffer< T >::stats_, kcenon::monitoring::ring_buffer_stats::total_reads, and kcenon::monitoring::ring_buffer_stats::total_writes.

◆ size()

template<typename T >

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

inlinenoexcept

Get current number of elements in buffer.

Definition at line 384 of file ring_buffer.h.

                                 {
        size_t write_idx = write_index_.load(std::memory_order_acquire);
        size_t read_idx = read_index_.load(std::memory_order_acquire);
 
        // Handle wraparound correctly
        if (write_idx >= read_idx) {
            return write_idx - read_idx;
        } else {
            // Wraparound case: write has wrapped around but read hasn't
            return config_.capacity - read_idx + write_idx;
        }
    }

References kcenon::monitoring::ring_buffer_config::capacity, kcenon::monitoring::ring_buffer< T >::config_, kcenon::monitoring::ring_buffer< T >::read_index_, and kcenon::monitoring::ring_buffer< T >::write_index_.

Referenced by kcenon::monitoring::ring_buffer< T >::empty(), TEST_F(), TEST_F(), and kcenon::monitoring::ring_buffer< T >::write().

Here is the caller graph for this function:

◆ write()

template<typename T >

common::VoidResult kcenon::monitoring::ring_buffer< T >::write ( T && item )

inline

Write a single element to the buffer.

Parameters

item	Item to write

Returns: Result indicating success or failure

Definition at line 203 of file ring_buffer.h.

                                     {
        stats_.total_writes.fetch_add(1, std::memory_order_relaxed);
 
        // Atomically claim a write slot using CAS loop to avoid ABA problem
        size_t current_write;
        size_t new_write;
        bool overflow_handled = false;
        size_t retry_count = 0;
        constexpr size_t max_retries = 100;
 
        do {
            current_write = write_index_.load(std::memory_order_acquire);
            size_t current_read = read_index_.load(std::memory_order_acquire);
 
            // Check for buffer full condition
            if (is_full_unsafe(current_write, current_read)) {
                if (config_.overwrite_old) {
                    // Advance read index to overwrite oldest data
                    // Use strong CAS in a loop to ensure it succeeds
                    size_t expected_read = current_read;
                    size_t new_read = (current_read + 1) & get_mask();
 
                    // Try to advance read index with strong CAS
                    // If it fails, another thread already advanced it, which is fine
                    if (read_index_.compare_exchange_strong(expected_read, new_read,
                                                           std::memory_order_acq_rel,
                                                           std::memory_order_acquire)) {
                        // Successfully advanced read index
                        if (!overflow_handled) {
                            stats_.overwrites.fetch_add(1, std::memory_order_relaxed);
                            overflow_handled = true;
                        }
                    }
                    // Continue to claim write slot even if CAS failed
                    // (another thread may have already made space)
                } else {
                    // Buffer is full and overwrite is not allowed
                    stats_.failed_writes.fetch_add(1, std::memory_order_relaxed);
 
                    // Provide more detailed error information
                    size_t current_size = size();
                    return common::VoidResult::err(error_info(monitoring_error_code::storage_full,
                                     "Ring buffer is full (size: " +
                                     std::to_string(current_size) +
                                     "/" + std::to_string(config_.capacity) +
                                     ", overwrites: " +
                                     std::to_string(stats_.overwrites.load()) + ")").to_common_error());
                }
            }
 
            new_write = (current_write + 1) & get_mask();
 
            // Prevent infinite loop in case of extreme contention
            if (++retry_count > max_retries) {
                stats_.failed_writes.fetch_add(1, std::memory_order_relaxed);
                return common::VoidResult::err(error_info(monitoring_error_code::collection_failed,
                                 "Failed to write to ring buffer after " +
                                 std::to_string(max_retries) + " retries (high contention)").to_common_error());
            }
 
            // Atomically claim the write slot
        } while (!write_index_.compare_exchange_weak(current_write, new_write,
                                                     std::memory_order_acq_rel,
                                                     std::memory_order_acquire));
 
        // Write the item to the claimed slot
        buffer_[current_write] = std::move(item);
 
        // Memory fence ensures data write completes before index update is visible
        std::atomic_thread_fence(std::memory_order_release);
 
        return common::ok();
    }

Referenced by TEST_F(), TEST_F(), TEST_F(), and kcenon::monitoring::ring_buffer< T >::write_batch().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ write_batch()

template<typename T >

size_t kcenon::monitoring::ring_buffer< T >::write_batch ( std::vector< T > && items )

inline

Write multiple elements in batch.

Parameters

items Vector of items to write

Returns: Number of items successfully written

Definition at line 282 of file ring_buffer.h.

                                             {
        if (items.empty()) {
            return 0;
        }
 
        size_t written = 0;
        size_t failed = 0;
 
        for (auto& item : items) {
            auto result = write(std::move(item));
            if (result.is_ok()) {
                ++written;
            } else {
                ++failed;
                // Stop on first failure if not overwriting to prevent data loss
                if (!config_.overwrite_old) {
                    break;
                }
            }
        }
 
        return written;
    }

References kcenon::monitoring::ring_buffer< T >::config_, kcenon::monitoring::ring_buffer_config::overwrite_old, and kcenon::monitoring::ring_buffer< T >::write().

Referenced by TEST_F().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

◆ buffer_

template<typename T >

std::unique_ptr<T[]> kcenon::monitoring::ring_buffer< T >::buffer_

private

Definition at line 149 of file ring_buffer.h.

Referenced by kcenon::monitoring::ring_buffer< T >::peek(), kcenon::monitoring::ring_buffer< T >::read(), and kcenon::monitoring::ring_buffer< T >::write().

◆ config_

template<typename T >

ring_buffer_config kcenon::monitoring::ring_buffer< T >::config_

private

Definition at line 150 of file ring_buffer.h.

Referenced by kcenon::monitoring::ring_buffer< T >::capacity(), kcenon::monitoring::ring_buffer< T >::get_config(), kcenon::monitoring::ring_buffer< T >::get_mask(), kcenon::monitoring::ring_buffer< T >::read_batch(), kcenon::monitoring::ring_buffer< T >::size(), kcenon::monitoring::ring_buffer< T >::write(), and kcenon::monitoring::ring_buffer< T >::write_batch().

◆ read_index_

template<typename T >

std::atomic<size_t> kcenon::monitoring::ring_buffer< T >::read_index_ {0}

private

Definition at line 147 of file ring_buffer.h.

147{0}; // Cache line aligned

Referenced by kcenon::monitoring::ring_buffer< T >::clear(), kcenon::monitoring::ring_buffer< T >::full(), kcenon::monitoring::ring_buffer< T >::peek(), kcenon::monitoring::ring_buffer< T >::read(), kcenon::monitoring::ring_buffer< T >::size(), and kcenon::monitoring::ring_buffer< T >::write().

◆ stats_

template<typename T >

ring_buffer_stats kcenon::monitoring::ring_buffer< T >::stats_

mutableprivate

Definition at line 151 of file ring_buffer.h.

Referenced by kcenon::monitoring::ring_buffer< T >::get_overflow_rate(), kcenon::monitoring::ring_buffer< T >::get_stats(), kcenon::monitoring::ring_buffer< T >::is_overflow_rate_high(), kcenon::monitoring::ring_buffer< T >::read(), kcenon::monitoring::ring_buffer< T >::reset_stats(), and kcenon::monitoring::ring_buffer< T >::write().

◆ write_index_

template<typename T >

std::atomic<size_t> kcenon::monitoring::ring_buffer< T >::write_index_ {0}

private

Definition at line 146 of file ring_buffer.h.

146{0}; // Cache line aligned

Referenced by kcenon::monitoring::ring_buffer< T >::clear(), kcenon::monitoring::ring_buffer< T >::full(), kcenon::monitoring::ring_buffer< T >::peek(), kcenon::monitoring::ring_buffer< T >::read(), kcenon::monitoring::ring_buffer< T >::size(), and kcenon::monitoring::ring_buffer< T >::write().

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

include/kcenon/monitoring/utils/ring_buffer.h

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ ring_buffer() [1/3]

◆ ring_buffer() [2/3]

◆ ring_buffer() [3/3]

Member Function Documentation

◆ capacity()

◆ clear()

◆ empty()

◆ full()

◆ get_config()

◆ get_mask()

◆ get_overflow_rate()

◆ get_stats()

◆ is_empty_unsafe()

◆ is_full_unsafe()

◆ is_overflow_rate_high()

◆ operator=() [1/2]

◆ operator=() [2/2]

◆ peek()

◆ read()

◆ read_batch()

◆ reset_stats()

◆ size()

◆ write()

◆ write_batch()

Member Data Documentation

◆ buffer_

◆ config_

◆ read_index_

◆ stats_

◆ write_index_