137 lines
3.6 KiB
C++
137 lines
3.6 KiB
C++
//
|
||
// 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
|