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

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 >:

Classes
struct	validated_tag

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.

Static Public Member Functions
static common::Result< std::unique_ptr< ring_buffer > >	create (const ring_buffer_config &config={})
	Create a ring buffer with validated configuration.

Private Member Functions
	ring_buffer (const ring_buffer_config &config, validated_tag)
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/4]

template<typename T >

kcenon::monitoring::ring_buffer< T >::ring_buffer ( const ring_buffer_config & config, validated_tag  )

inlineprivate

Definition at line 156 of file ring_buffer.h.

        : buffer_(std::make_unique<T[]>(config.capacity))
        , config_(config) {}

ring_buffer() [2/4]

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

Deprecated

Use create() for Result-based error handling

Definition at line 207 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() [3/4]

template<typename T >

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

delete

ring_buffer() [4/4]

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 443 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 450 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_.

create()

template<typename T >

static common::Result< std::unique_ptr< ring_buffer > > kcenon::monitoring::ring_buffer< T >::create ( const ring_buffer_config & config = {} )

inlinestatic

Create a ring buffer with validated configuration.

Parameters

config

Ring buffer configuration options

Returns

Result containing the ring buffer or error

Definition at line 187 of file ring_buffer.h.

                                           {}) {
        auto validation = config.validate();
        if (validation.is_err()) {
            return common::Result<std::unique_ptr<ring_buffer>>::err(
                error_info(monitoring_error_code::invalid_configuration,
                           "Invalid ring buffer configuration: " +
                               validation.error().message)
                    .to_common_error());
        }
        return common::ok(std::unique_ptr<ring_buffer>(
            new ring_buffer(config, validated_tag{})));
    }

Referenced by TEST_F(), and TEST_F().

Here is the caller graph for this function:

empty()

template<typename T >

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

inlinenoexcept

Check if buffer is empty.

Definition at line 427 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 434 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 458 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 163 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 493 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 465 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 177 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 170 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 486 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 395 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 338 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 366 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 472 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 411 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 230 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();
    }

References kcenon::monitoring::ring_buffer< T >::buffer_, kcenon::monitoring::ring_buffer_config::capacity, kcenon::monitoring::collection_failed, kcenon::monitoring::ring_buffer< T >::config_, kcenon::monitoring::ring_buffer_stats::failed_writes, kcenon::monitoring::ring_buffer< T >::get_mask(), kcenon::monitoring::ring_buffer< T >::is_full_unsafe(), kcenon::monitoring::ring_buffer_config::overwrite_old, kcenon::monitoring::ring_buffer_stats::overwrites, kcenon::monitoring::ring_buffer< T >::read_index_, kcenon::monitoring::ring_buffer< T >::size(), kcenon::monitoring::ring_buffer< T >::stats_, kcenon::monitoring::storage_full, kcenon::monitoring::error_info::to_common_error(), kcenon::monitoring::ring_buffer_stats::total_writes, and kcenon::monitoring::ring_buffer< T >::write_index_.

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 309 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 151 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 152 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 149 of file ring_buffer.h.

{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 153 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 148 of file ring_buffer.h.

{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