446 lines
14 KiB
C++
446 lines
14 KiB
C++
//
|
||
// Created by xtkuang on 2025/5/30.
|
||
//
|
||
|
||
#ifndef CMVR_ES_RING_BUFFER_H
|
||
#define CMVR_ES_RING_BUFFER_H
|
||
|
||
#pragma once
|
||
#include <chrono>
|
||
#include <condition_variable>
|
||
#include <cstdint>
|
||
#include <deque>
|
||
#include <memory>
|
||
#include <mutex>
|
||
#include <optional>
|
||
#include <stdexcept>
|
||
#include <utility>
|
||
#include <vector>
|
||
|
||
template<typename T>
|
||
class RingBuffer {
|
||
public:
|
||
explicit RingBuffer(size_t capacity) : capacity_(capacity) {
|
||
if (capacity_ == 0) {
|
||
throw std::invalid_argument("RingBuffer capacity must be greater than zero");
|
||
}
|
||
}
|
||
|
||
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) {
|
||
if (capacity_ == 0) {
|
||
throw std::invalid_argument("SPMCRingBuffer capacity must be greater than zero");
|
||
}
|
||
}
|
||
|
||
// 写入操作（仅支持单个生产者）
|
||
void push(const T& item) {
|
||
{
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
buffer_[head_ % capacity_] = item;
|
||
++head_;
|
||
if (head_ - tail_ > capacity_) {
|
||
// 队列满，覆盖最旧的数据
|
||
tail_ = head_ - capacity_;
|
||
}
|
||
}
|
||
condition_.notify_all();
|
||
}
|
||
|
||
// 单消费者使用（内部 tail_）
|
||
std::optional<T> pop() {
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
if (tail_ >= head_) return std::nullopt;
|
||
T value = buffer_[tail_ % capacity_];
|
||
++tail_;
|
||
return value;
|
||
}
|
||
|
||
std::optional<T> getLast() const {
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
if (tail_ >= head_) return std::nullopt;
|
||
return buffer_[(head_ - 1) % capacity_];
|
||
}
|
||
|
||
// 多消费者使用（每个读者独立维护 reader_tail）。同一个 reader_tail 只能由
|
||
// 一个消费线程拥有，且不要把该游标与无参 pop() 的共享 tail_ 混合作为同一路读取。
|
||
std::optional<T> pop(size_t& reader_tail) const {
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
if (reader_tail >= head_) return std::nullopt;
|
||
if (reader_tail < tail_) {
|
||
// 数据已被覆盖，跳过无效读取区间
|
||
reader_tail = tail_;
|
||
return std::nullopt;
|
||
}
|
||
T value = buffer_[reader_tail % capacity_];
|
||
reader_tail++;
|
||
return value;
|
||
}
|
||
|
||
// 在同一次加锁中把独立读游标跳到当前最新元素并读取，避免先 getHead()
|
||
// 再 pop() 时被高速覆盖造成的检查/读取竞态。
|
||
std::optional<T> getLatest(size_t& reader_tail) const {
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
if (tail_ >= head_) {
|
||
return std::nullopt;
|
||
}
|
||
reader_tail = head_ - 1;
|
||
T value = buffer_[reader_tail % capacity_];
|
||
++reader_tail;
|
||
return value;
|
||
}
|
||
|
||
template<class Rep, class Period>
|
||
std::optional<T> waitPop(
|
||
size_t& reader_tail,
|
||
const std::chrono::duration<Rep, Period>& timeout) const {
|
||
std::unique_lock<std::mutex> lock(mutex_);
|
||
condition_.wait_for(lock, timeout, [&] { return reader_tail < head_; });
|
||
if (reader_tail >= head_) {
|
||
return std::nullopt;
|
||
}
|
||
if (reader_tail < tail_) {
|
||
reader_tail = tail_;
|
||
}
|
||
if (reader_tail >= head_) {
|
||
return std::nullopt;
|
||
}
|
||
T value = buffer_[reader_tail % capacity_];
|
||
++reader_tail;
|
||
return value;
|
||
}
|
||
|
||
size_t size() const {
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
return head_ - tail_;
|
||
}
|
||
|
||
size_t getHead() const {
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
return head_;
|
||
}
|
||
|
||
size_t getTail() const {
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
return tail_;
|
||
}
|
||
|
||
bool empty() const {
|
||
return size() == 0;
|
||
}
|
||
|
||
bool full() const {
|
||
return size() >= capacity_;
|
||
}
|
||
|
||
void clear() {
|
||
{
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
// Keep sequence numbers monotonic so cursors created before clear()
|
||
// cannot alias newly published slots after the reset.
|
||
tail_ = head_;
|
||
}
|
||
condition_.notify_all();
|
||
}
|
||
|
||
private:
|
||
mutable std::vector<T> buffer_;
|
||
const size_t capacity_;
|
||
mutable std::mutex mutex_;
|
||
mutable std::condition_variable condition_;
|
||
size_t head_{0}; // 单调写序号；clear() 仅推进 tail_，避免旧游标 ABA。
|
||
size_t tail_{0}; // 当前仍保留的最旧序号，同时也是 pop() 的共享读指针。
|
||
};
|
||
|
||
// 线程安全的多消费者广播缓冲区。缓冲区只保存不可变共享对象，消费者通过各自
|
||
// 的 Cursor 独立前进；慢消费者被覆盖的数据会累计到 Cursor::dropped_count。
|
||
// Cursor 是单线程所有权对象，不可由多个线程同时读写；每个消费者应创建自己的 Cursor。
|
||
template<typename T>
|
||
class BroadcastFrameRing {
|
||
public:
|
||
using ValuePtr = std::shared_ptr<const T>;
|
||
|
||
enum class StartPosition {
|
||
NEXT_PUBLISHED,
|
||
OLDEST_AVAILABLE,
|
||
LATEST_AVAILABLE
|
||
};
|
||
|
||
struct Cursor {
|
||
uint64_t generation{0};
|
||
uint64_t next_sequence{0};
|
||
uint64_t dropped_count{0};
|
||
StartPosition start_position{StartPosition::NEXT_PUBLISHED};
|
||
|
||
private:
|
||
bool generation_changed{false};
|
||
uint64_t reported_dropped_count{0};
|
||
friend class BroadcastFrameRing<T>;
|
||
};
|
||
|
||
struct ReadResult {
|
||
ValuePtr value;
|
||
uint64_t generation{0};
|
||
uint64_t sequence{0};
|
||
uint64_t dropped_count{0};
|
||
uint64_t dropped_since_last_read{0};
|
||
bool generation_changed{false};
|
||
};
|
||
|
||
struct Stats {
|
||
size_t capacity{0};
|
||
size_t size{0};
|
||
uint64_t generation{0};
|
||
uint64_t next_sequence{0};
|
||
uint64_t dropped_count{0};
|
||
bool closed{false};
|
||
};
|
||
|
||
explicit BroadcastFrameRing(const size_t capacity)
|
||
: capacity_(capacity) {
|
||
if (capacity_ == 0) {
|
||
throw std::invalid_argument("BroadcastFrameRing capacity must be greater than zero");
|
||
}
|
||
}
|
||
|
||
BroadcastFrameRing(const BroadcastFrameRing&) = delete;
|
||
BroadcastFrameRing& operator=(const BroadcastFrameRing&) = delete;
|
||
|
||
Cursor makeCursor(const StartPosition start_position = StartPosition::NEXT_PUBLISHED) const {
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
Cursor cursor;
|
||
cursor.generation = generation_;
|
||
cursor.start_position = start_position;
|
||
cursor.next_sequence = startSequenceLocked_(start_position);
|
||
return cursor;
|
||
}
|
||
|
||
std::optional<uint64_t> publish(ValuePtr value) {
|
||
if (!value) {
|
||
return std::nullopt;
|
||
}
|
||
|
||
std::optional<uint64_t> published_sequence;
|
||
{
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
if (closed_) {
|
||
return std::nullopt;
|
||
}
|
||
|
||
const uint64_t sequence = next_sequence_++;
|
||
if (entries_.size() == capacity_) {
|
||
entries_.pop_front();
|
||
++dropped_count_;
|
||
}
|
||
entries_.push_back(Entry{generation_, sequence, std::move(value)});
|
||
published_sequence = sequence;
|
||
}
|
||
condition_.notify_all();
|
||
return published_sequence;
|
||
}
|
||
|
||
std::optional<ReadResult> tryRead(Cursor& cursor) const {
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
return tryReadLocked_(cursor);
|
||
}
|
||
|
||
// Low-latency consumers can use this before reading to abandon an excessive
|
||
// backlog atomically. When the number of currently readable entries exceeds
|
||
// maximum_pending_frames, every pending entry is discarded and the cursor is
|
||
// advanced to the next sequence that will be published. Frames already
|
||
// overwritten by the ring and frames actively discarded here are both
|
||
// reflected in Cursor::dropped_count; the next successful read reports their
|
||
// sum through ReadResult::dropped_since_last_read.
|
||
uint64_t discardPendingIfExceeds(
|
||
Cursor& cursor,
|
||
const size_t maximum_pending_frames) const {
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
synchronizeCursorGenerationLocked_(cursor);
|
||
|
||
if (!entries_.empty()) {
|
||
const uint64_t oldest_sequence = entries_.front().sequence;
|
||
if (cursor.next_sequence < oldest_sequence) {
|
||
cursor.dropped_count += oldest_sequence - cursor.next_sequence;
|
||
cursor.next_sequence = oldest_sequence;
|
||
}
|
||
}
|
||
|
||
const uint64_t pending =
|
||
cursor.next_sequence < next_sequence_
|
||
? next_sequence_ - cursor.next_sequence
|
||
: 0;
|
||
if (pending <= static_cast<uint64_t>(maximum_pending_frames)) {
|
||
return 0;
|
||
}
|
||
|
||
cursor.next_sequence = next_sequence_;
|
||
cursor.dropped_count += pending;
|
||
return pending;
|
||
}
|
||
|
||
template<class Rep, class Period>
|
||
std::optional<ReadResult> waitRead(
|
||
Cursor& cursor,
|
||
const std::chrono::duration<Rep, Period>& timeout) const {
|
||
const auto deadline = std::chrono::steady_clock::now() + timeout;
|
||
std::unique_lock<std::mutex> lock(mutex_);
|
||
while (true) {
|
||
if (auto result = tryReadLocked_(cursor)) {
|
||
return result;
|
||
}
|
||
if (closed_) {
|
||
return std::nullopt;
|
||
}
|
||
if (condition_.wait_until(lock, deadline) == std::cv_status::timeout) {
|
||
return tryReadLocked_(cursor);
|
||
}
|
||
}
|
||
}
|
||
|
||
// 开始一个新的发布代次。旧 Cursor 会在下一次成功读取时收到
|
||
// generation_changed=true，序号从 0 重新开始。
|
||
uint64_t reset() {
|
||
uint64_t generation = 0;
|
||
{
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
entries_.clear();
|
||
++generation_;
|
||
next_sequence_ = 0;
|
||
closed_ = false;
|
||
generation = generation_;
|
||
}
|
||
condition_.notify_all();
|
||
return generation;
|
||
}
|
||
|
||
void close() {
|
||
{
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
closed_ = true;
|
||
}
|
||
condition_.notify_all();
|
||
}
|
||
|
||
bool closed() const {
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
return closed_;
|
||
}
|
||
|
||
Stats stats() const {
|
||
std::lock_guard<std::mutex> lock(mutex_);
|
||
return Stats{capacity_, entries_.size(), generation_, next_sequence_, dropped_count_, closed_};
|
||
}
|
||
|
||
private:
|
||
struct Entry {
|
||
uint64_t generation;
|
||
uint64_t sequence;
|
||
ValuePtr value;
|
||
};
|
||
|
||
uint64_t startSequenceLocked_(const StartPosition start_position) const {
|
||
if (entries_.empty()) {
|
||
return next_sequence_;
|
||
}
|
||
switch (start_position) {
|
||
case StartPosition::OLDEST_AVAILABLE:
|
||
return entries_.front().sequence;
|
||
case StartPosition::LATEST_AVAILABLE:
|
||
return entries_.back().sequence;
|
||
case StartPosition::NEXT_PUBLISHED:
|
||
default:
|
||
return next_sequence_;
|
||
}
|
||
}
|
||
|
||
void synchronizeCursorGenerationLocked_(Cursor& cursor) const {
|
||
if (cursor.generation == generation_) {
|
||
return;
|
||
}
|
||
cursor.generation = generation_;
|
||
cursor.next_sequence = startSequenceLocked_(cursor.start_position);
|
||
cursor.generation_changed = true;
|
||
}
|
||
|
||
std::optional<ReadResult> tryReadLocked_(Cursor& cursor) const {
|
||
synchronizeCursorGenerationLocked_(cursor);
|
||
if (entries_.empty()) {
|
||
return std::nullopt;
|
||
}
|
||
|
||
const uint64_t oldest_sequence = entries_.front().sequence;
|
||
if (cursor.next_sequence < oldest_sequence) {
|
||
cursor.dropped_count += oldest_sequence - cursor.next_sequence;
|
||
cursor.next_sequence = oldest_sequence;
|
||
}
|
||
if (cursor.next_sequence >= next_sequence_) {
|
||
return std::nullopt;
|
||
}
|
||
|
||
const size_t index = static_cast<size_t>(cursor.next_sequence - oldest_sequence);
|
||
if (index >= entries_.size()) {
|
||
return std::nullopt;
|
||
}
|
||
|
||
const Entry& entry = entries_[index];
|
||
++cursor.next_sequence;
|
||
const uint64_t dropped_since_last = cursor.dropped_count - cursor.reported_dropped_count;
|
||
cursor.reported_dropped_count = cursor.dropped_count;
|
||
|
||
ReadResult result;
|
||
result.value = entry.value;
|
||
result.generation = entry.generation;
|
||
result.sequence = entry.sequence;
|
||
result.dropped_count = cursor.dropped_count;
|
||
result.dropped_since_last_read = dropped_since_last;
|
||
result.generation_changed = cursor.generation_changed;
|
||
cursor.generation_changed = false;
|
||
return result;
|
||
}
|
||
|
||
const size_t capacity_;
|
||
mutable std::mutex mutex_;
|
||
mutable std::condition_variable condition_;
|
||
std::deque<Entry> entries_;
|
||
uint64_t generation_{1};
|
||
uint64_t next_sequence_{0};
|
||
uint64_t dropped_count_{0};
|
||
bool closed_{false};
|
||
};
|
||
|
||
|
||
|
||
#endif //CMVR_ES_RING_BUFFER_H
|