cmvr-es/include/utils/base/ring_buffer.h

//
// Created by xtkuang on 2025/5/30.
//

#ifndef CMVR_ES_RING_BUFFER_H
#define CMVR_ES_RING_BUFFER_H

#pragma once
#include <deque>
#include <mutex>
#include <vector>
#include <atomic>
#include <optional>

template<typename T>
class RingBuffer {
public:
    explicit RingBuffer(size_t capacity) : capacity_(capacity) {}

    void push(const T& item) {
        std::lock_guard<std::mutex> lock(mutex_);
        if (buffer_.size() >= capacity_) {
            buffer_.pop_front();
        }
        buffer_.push_back(item);
    }

    std::vector<T> getAll() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return std::vector<T>(buffer_.begin(), buffer_.end());
    }

    void clear() {
        std::lock_guard<std::mutex> lock(mutex_);
        buffer_.clear();
    }

    size_t size() const {
        std::lock_guard<std::mutex> lock(mutex_);
        return buffer_.size();
    }

private:
    size_t capacity_;
    std::deque<T> buffer_;
    mutable std::mutex mutex_;
};

template<typename T>
class SPMCRingBuffer {
public:
    explicit SPMCRingBuffer(size_t capacity)
        : buffer_(capacity), capacity_(capacity),
          head_(0), tail_(0) {}

    // 写入操作（仅支持单个生产者）
    void push(const T& item) {
        size_t head = head_.load(std::memory_order_relaxed);
        size_t tail = tail_.load(std::memory_order_acquire);
        buffer_[head % capacity_] = item;
        head = head + 1;
        head_.store(head, std::memory_order_release);
        if (head - tail >= capacity_) {
            // 队列满，覆盖最旧的数据
            tail_.store(tail + 1, std::memory_order_release);
        }
    }

    // 单消费者使用（内部 tail_）
    std::optional<T> pop() {
        size_t tail = tail_.load(std::memory_order_relaxed);
        size_t head = head_.load(std::memory_order_acquire);
        if (tail >= head) return std::nullopt;
        T value = buffer_[tail % capacity_];
        tail_.store(tail + 1, std::memory_order_release);
        return value;
    }

    std::optional<T> getLast() {
        size_t tail = tail_.load(std::memory_order_relaxed);
        size_t head = head_.load(std::memory_order_acquire);
        if (tail >= head) return std::nullopt;
        T value = buffer_[head_ % capacity_];
        return value;
    }

    // 多消费者使用（每个读者独立维护 reader_tail）
    std::optional<T> pop(size_t& reader_tail) const {
        size_t head = head_.load(std::memory_order_acquire);
        if (reader_tail >= head) return std::nullopt;
        if (head > reader_tail + capacity_) {
            // 数据已被覆盖，跳过无效读取区间
            reader_tail = head - capacity_;
            return std::nullopt;
        }
        T value = buffer_[reader_tail % capacity_];
        reader_tail++;
        return value;
    }

    size_t size() const {
        return head_.load(std::memory_order_acquire) - tail_.load(std::memory_order_acquire);
    }

    size_t getHead() const {
        return head_.load(std::memory_order_acquire);
    }

    size_t getTail() const {
        return tail_.load(std::memory_order_acquire);
    }

    bool empty() const {
        return size() == 0;
    }

    bool full() const {
        return size() >= capacity_;
    }

    void clear() {
        head_.store(0, std::memory_order_release);
        tail_.store(0, std::memory_order_release);
    }

private:
    std::vector<T> buffer_;
    const size_t capacity_;

    std::atomic<size_t> head_;  // 共享写指针
    std::atomic<size_t> tail_;  // 共享读指针（仅用于 SPSC 模式）
};



#endif //CMVR_ES_RING_BUFFER_H