doubleringbuffer.h - master - krust/DoubleRingBuffer

doubleringbuffer.h
394 строки13 KB
krust
initial commit
07 авг 2025, 20:44
07 авг 2025, 20:44031d23b
#ifndef DOUBLERINGBUFFER_H
#define DOUBLERINGBUFFER_H

#include <array>
#include <atomic>
#include <cstdint>
#include <cstring>
#include <span>

/**
 * @brief A lock-free double-buffered ring buffer for high-performance concurrent I/O
 *
 * This class implements a ring buffer that provides contiguous memory access for both
 * reading and writing operations. It uses a double-buffering technique where data is
 * stored in two mirrored regions, allowing readers to access data linearly without
 * dealing with wrap-around logic.
 *
 * @tparam BufSize The capacity of the buffer (must be a power of 2)
 *
 * @section Thread Safety
 * The buffer supports one concurrent reader and one concurrent writer without locks.
 * It uses atomic flags to prevent multiple simultaneous read or write operations.
 *
 * @section Usage Examples
 * @code
 * DoubleRingBuffer<1024> buffer;
 *
 * // Simple write/read
 * const char* data = "Hello World";
 * buffer.write(data, strlen(data));
 *
 * char read_data[100];
 * size_t read_count = buffer.read(read_data, sizeof(read_data));
 *
 * // Advanced usage with manual control
 * uint8_t* write_ptr = buffer.begin_write();
 * if (write_ptr) {
 *     // Write data directly to buffer
 *     memcpy(write_ptr, my_data, data_size);
 *     buffer.end_write(data_size);
 * }
 *
 * const uint8_t* read_ptr = buffer.begin_read();
 * if (read_ptr) {
 *     // Process data directly from buffer
 *     process_data(read_ptr, buffer.size());
 *     buffer.end_read(processed_size);
 * }
 * @endcode
 */
template<size_t BufSize = 1024>
class DoubleRingBuffer
{
private:
    // Ensure BufSize is a power of 2 for efficient modulo operations
    static_assert(BufSize > 0 && (BufSize & (BufSize - 1)) == 0,
                  "Buffer capacity must be a power of 2 for efficient wrap-around");

    /**
     * @brief Double-sized buffer providing contiguous access
     *
     * The buffer is organized as two mirrored regions:
     * - [0, BufSize): Active region for read/write operations
     * - [BufSize, 2*BufSize): Shadow region for maintaining data contiguity
     */
    std::array<uint8_t, 2 * BufSize> buffer_{};

    size_t head_;                  ///< Read position in range [0, BufSize)
    size_t tail_;                  ///< Write position in range [0, BufSize)
    size_t size_;                  ///< Current number of elements in buffer
    std::atomic<bool> readerBusy_; ///< Flag preventing concurrent read operations
    std::atomic<bool> writerBusy_; ///< Flag preventing concurrent write operations

    /**
     * @brief Synchronize data between active and shadow regions
     *
     * This function ensures data consistency between the active buffer region
     * [0, BufSize) and the shadow region [BufSize, 2*BufSize). When write
     * operations span across the buffer boundary, this function mirrors the
     * data to maintain contiguous access for readers.
     *
     * @param write_start Starting position of the write operation
     * @param write_count Number of bytes written
     */
    void mirror_new_data(size_t write_start, size_t write_count) noexcept
    {
        const size_t mirror_offset = BufSize;
        const size_t start_idx = write_start & (BufSize - 1);
        const size_t end_idx = start_idx + write_count;

        if (end_idx > BufSize) {
            // Wrapping occurs - synchronize in two parts
            const size_t first_part_size = BufSize - start_idx;
            const size_t second_part_size = end_idx - BufSize;

            // Copy first part from active to shadow region for future reads
            std::memcpy(&buffer_[start_idx + mirror_offset], &buffer_[start_idx], first_part_size);

            // Copy second part (written to shadow) back to active region
            std::memcpy(&buffer_[0], &buffer_[mirror_offset], second_part_size);

        } else {
            // No wrapping - simple copy from active to shadow
            std::memcpy(&buffer_[start_idx + mirror_offset], &buffer_[start_idx], write_count);
        }
    }

public:
    /**
     * @brief Construct a new Double Ring Buffer
     *
     * Initializes the buffer with empty state and clears busy flags.
     * All positions are set to zero and size is initialized to zero.
     */
    DoubleRingBuffer() noexcept
        : head_(0)
        , tail_(0)
        , size_(0)
        , readerBusy_(false)
        , writerBusy_(false)
    {}

    /**
     * @brief Get the maximum capacity of the buffer
     * @return constexpr size_t Maximum number of elements the buffer can hold
     */
    [[nodiscard]] static constexpr size_t capacity() noexcept { return BufSize; }

    /**
     * @brief Check if the buffer is empty
     * @return true if buffer contains no elements
     * @return false if buffer contains elements
     */
    [[nodiscard]] bool is_empty() const noexcept { return size_ == 0; }

    /**
     * @brief Check if the buffer is full
     * @return true if buffer cannot accept more elements
     * @return false if buffer has available space
     */
    [[nodiscard]] bool is_full() const noexcept { return size_ == BufSize; }

    /**
     * @brief Get the current number of elements in the buffer
     * @return size_t Current element count
     */
    [[nodiscard]] size_t size() const noexcept { return size_; }

    /**
     * @brief Get the number of available spaces in the buffer
     * @return size_t Number of elements that can be written
     */
    [[nodiscard]] size_t available() const noexcept { return BufSize - size_; }

    /**
     * @brief Check if a read operation is currently in progress
     * @return true if a reader has started but not finished an operation
     * @return false if no read operation is active
     */
    [[nodiscard]] bool reader_is_busy() const noexcept
    {
        return readerBusy_.load(std::memory_order_relaxed);
    }

    /**
     * @brief Check if a write operation is currently in progress
     * @return true if a writer has started but not finished an operation
     * @return false if no write operation is active
     */
    [[nodiscard]] bool writer_is_busy() const noexcept
    {
        return writerBusy_.load(std::memory_order_relaxed);
    }

    /**
     * @brief Reset the buffer to empty state
     *
     * @warning This function does NOT check if read/write operations are in progress.
     *          Use with caution in concurrent environments.
     *
     * @post Buffer is empty with head=tail=0 and busy flags cleared
     */
    void clear() noexcept
    {
        head_ = 0;
        tail_ = 0;
        size_ = 0;
        readerBusy_.store(false, std::memory_order_release);
        writerBusy_.store(false, std::memory_order_release);
    }

    /**
     * @brief Begin a write operation and obtain a contiguous write buffer
     *
     * This function acquires exclusive write access and returns a pointer to
     * a contiguous memory region where data can be written. The region is
     * guaranteed to be at least `available()` bytes long.
     *
     * @return uint8_t* Pointer to writable memory, or nullptr if writer is busy
     *
     * @see end_write()
     *
     * @note Must be paired with end_write() to complete the operation
     */
    [[nodiscard]] uint8_t *begin_write() noexcept
    {
        bool expected = false;
        if (!writerBusy_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) {
            return nullptr; // Writer is already busy
        }
        return &buffer_[tail_];
    }

    /**
     * @brief Complete a write operation and update buffer state
     *
     * This function finalizes a write operation started with begin_write().
     * It synchronizes the written data between active and shadow regions and
     * updates the buffer's write position and size.
     *
     * @param count Number of bytes actually written
     * @return size_t Number of bytes successfully added to buffer (may be less than count)
     *
     * @see begin_write()
     *
     * @pre begin_write() must have been called successfully
     */
    size_t end_write(size_t count) noexcept
    {
        if (!writerBusy_.load(std::memory_order_acquire)) {
            return 0; // Writer wasn't started
        }

        count = std::min(count, available());

        if (count > 0) {
            // Mirror the written data to maintain consistency
            mirror_new_data(tail_, count);
            tail_ = (tail_ + count) & (BufSize - 1);
        }

        size_ += count;
        writerBusy_.store(false, std::memory_order_release);
        return count;
    }

    /**
     * @brief Write data to the buffer in a single operation
     *
     * Copies count bytes from src to the buffer. This is a convenience function
     * that combines begin_write(), memcpy(), and end_write() into one call.
     *
     * @param src Pointer to source data (must not be nullptr if count > 0)
     * @param count Number of bytes to write
     * @return size_t Number of bytes successfully written
     */
    [[nodiscard]] size_t write(const void *src, size_t count) noexcept
    {
        if (src == nullptr || count == 0)
            return 0;

        uint8_t *dst = begin_write();
        if (dst == nullptr)
            return 0; // Writer busy

        count = std::min(count, available());
        std::memcpy(dst, src, count);
        return end_write(count);
    }

    /**
     * @brief Begin a read operation and obtain a contiguous read buffer
     *
     * This function acquires exclusive read access and returns a pointer to
     * a contiguous memory region where data can be read from. The region is
     * guaranteed to be at least `size()` bytes long.
     *
     * @return const uint8_t* Pointer to readable memory, or nullptr if reader is busy
     *
     * @see end_read()
     *
     * @note Must be paired with end_read() to complete the operation
     */
    [[nodiscard]] const uint8_t *begin_read() noexcept
    {
        bool expected = false;
        if (!readerBusy_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) {
            return nullptr; // Reader is already busy
        }
        return &buffer_[head_];
    }

    /**
     * @brief Complete a read operation and update buffer state
     *
     * This function finalizes a read operation by updating the read position
     * and decreasing the buffer size. The consumed data becomes available for
     * future writes.
     *
     * @param count Number of bytes actually read
     * @return size_t Number of bytes successfully consumed from buffer
     *
     * @see begin_read()
     *
     * @pre begin_read() must have been called successfully
     */
    size_t end_read(size_t count) noexcept
    {
        if (!readerBusy_.load(std::memory_order_acquire)) {
            return 0; // Reader wasn't started
        }

        count = std::min(count, size());
        head_ = (head_ + count) & (BufSize - 1);
        size_ -= count;
        readerBusy_.store(false, std::memory_order_release);
        return count;
    }

    /**
     * @brief Read data from the buffer in a single operation
     *
     * Copies up to count bytes from the buffer to dst. This is a convenience function
     * that combines begin_read(), memcpy(), and end_read() into one call.
     *
     * @param dst Pointer to destination buffer (must not be nullptr if count > 0)
     * @param count Maximum number of bytes to read
     * @return size_t Number of bytes successfully read
     */
    [[nodiscard]] size_t read(void *dst, size_t count) noexcept
    {
        if (dst == nullptr || count == 0)
            return 0;

        const uint8_t *src = begin_read();
        if (src == nullptr)
            return 0; // Reader busy

        count = std::min(count, size_);
        std::memcpy(dst, src, count);
        return end_read(count);
    }

    /**
     * @brief Skip (consume) data without copying it
     *
     * Advances the read position by count bytes, effectively consuming the data
     * without copying it to another buffer. Useful for discarding unwanted data.
     *
     * @param count Number of bytes to skip
     * @return size_t Number of bytes actually skipped
     */
    [[nodiscard]] size_t skip(size_t count) noexcept
    {
        bool expected = false;
        if (!readerBusy_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) {
            return 0; // Reader is busy
        }

        count = std::min(count, size_);
        head_ = (head_ + count) & (BufSize - 1);
        size_ -= count;
        readerBusy_.store(false, std::memory_order_release);
        return count;
    }

    /**
     * @brief Consume data and return it as a span
     *
     * This function atomically consumes up to count bytes from the buffer and
     * returns them as a std::span. The data is immediately removed from the buffer.
     *
     * @param count Maximum number of bytes to consume
     * @return std::span<const uint8_t> Span containing the consumed data
     *
     * @warning The returned span is only valid until the next buffer operation
     */
    [[nodiscard]] std::span<const uint8_t> consume_span(size_t count) noexcept
    {
        bool expected = false;
        if (!readerBusy_.compare_exchange_strong(expected, true, std::memory_order_acq_rel)) {
            return {}; // Reader is busy
        }

        const size_t consume_count = std::min(count, size_);
        const std::span<const uint8_t> result{&buffer_[head_], consume_count};
        head_ = (head_ + consume_count) & (BufSize - 1);
        size_ -= consume_count;
        readerBusy_.store(false, std::memory_order_release);
        return result;
    }
};

#endif // DOUBLERINGBUFFER_H