324 lines
11 KiB
C++
324 lines
11 KiB
C++
#pragma once
|
|
#include "../system/export.h"
|
|
#include "Core/Base/global_include.h"
|
|
|
|
#include <algorithm>
|
|
#include <cstdint>
|
|
#include <cstring>
|
|
#include <mutex>
|
|
#include <string>
|
|
#include <string_view>
|
|
#include <vector>
|
|
|
|
namespace Psc {
|
|
|
|
|
|
|
|
// -------------------- utils --------------------
|
|
template <typename Size_T>
|
|
bool is_pow2(Size_T x) { return x && ((x & (x - 1)) == 0); }
|
|
|
|
template <typename Size_T>
|
|
Size_T ceil_to_pow2(Size_T x) {
|
|
if (x <= 1) return 1;
|
|
--x;
|
|
for (Size_T i = 1; i < sizeof(Size_T) * 8; i <<= 1) x |= x >> i;
|
|
return x + 1;
|
|
}
|
|
|
|
// -------------------- Base RingBuffer --------------------
|
|
// - capacity 必须为 2^n
|
|
// - mask = capacity - 1
|
|
// - used = (tail - head) & mask
|
|
// - free = (capacity - 1) - used (浪费 1 字节)
|
|
template <typename MutexT, typename Size_T>
|
|
class Base_RingBuffer {
|
|
static_assert(std::is_unsigned_v<Size_T>, "Size_T must be unsigned");
|
|
public:
|
|
Base_RingBuffer() = default;
|
|
void init(Size_T min_capacity) {
|
|
// 防御：min_capacity + 1 溢出
|
|
if (min_capacity >= std::numeric_limits<Size_T>::max()) {
|
|
fail_fast("Ring buffer size is too large" + Psc::to_string(min_capacity));
|
|
}
|
|
// +1 维持“浪费一个字节”的策略
|
|
const Size_T cap = ceil_to_pow2(min_capacity + 1);
|
|
capacity = cap;
|
|
head = 0;
|
|
tail = 0;
|
|
mask_ = cap - 1;
|
|
buf_.assign(cap, 0);
|
|
|
|
if (!is_pow2(capacity)) {
|
|
std::terminate();
|
|
}
|
|
}
|
|
|
|
[[nodiscard]] bool empty() const { return head == tail; }
|
|
|
|
[[nodiscard]] bool full() const { return ((tail + 1) & mask_) == head; }
|
|
|
|
[[nodiscard]] std::string state_str() {
|
|
std::lock_guard<MutexT> lock(mutex_);
|
|
auto free_space = free_space_unsafe();
|
|
auto used_space = used_space_unsafe();
|
|
return VAR_STR_5(free_space, used_space, head, tail, capacity);
|
|
}
|
|
|
|
protected:
|
|
// used_space / free_space 与共享内存版一致
|
|
[[nodiscard]] Size_T used_space_unsafe() const {
|
|
return (tail - head) & mask_;
|
|
}
|
|
[[nodiscard]] Size_T free_space_unsafe() const {
|
|
return (capacity - 1) - used_space_unsafe();
|
|
}
|
|
|
|
void rollback_unsafe(Size_T n) { head = (head - n) & mask_; }
|
|
|
|
void write_bytes_unsafe(const void* data, Size_T len) {
|
|
const Size_T end = std::min(len, capacity - tail);
|
|
std::memcpy(buf_.data() + tail, data, end);
|
|
std::memcpy(buf_.data(), static_cast<const std::uint8_t*>(data) + end, len - end);
|
|
tail = (tail + len) & mask_;
|
|
}
|
|
|
|
void read_bytes_unsafe(void* out, Size_T len) {
|
|
const Size_T end = std::min(len, capacity - head);
|
|
std::memcpy(out, buf_.data() + head, end);
|
|
std::memcpy(static_cast<std::uint8_t*>(out) + end, buf_.data(), len - end);
|
|
head = (head + len) & mask_;
|
|
}
|
|
|
|
void peek_bytes_unsafe(void* out, Size_T len) const {
|
|
const Size_T end = std::min(len, capacity - head);
|
|
std::memcpy(out, buf_.data() + head, end);
|
|
std::memcpy(static_cast<std::uint8_t*>(out) + end, buf_.data(), len - end);
|
|
}
|
|
|
|
protected:
|
|
MutexT mutex_{};
|
|
Size_T capacity = 0;
|
|
Size_T head = 0;
|
|
Size_T tail = 0;
|
|
std::vector<uint8_t> buf_{};
|
|
Size_T mask_ = 0;
|
|
};
|
|
|
|
// -------------------- Packet mode (length-prefixed) --------------------
|
|
template <class MutexT, typename Size_T>
|
|
class RingBuffer : public Base_RingBuffer<MutexT, Size_T> {
|
|
|
|
public:
|
|
static constexpr Size_T HEADER_SIZE = sizeof(Size_T);
|
|
|
|
// 包模式：空间不够则整体失败（不拆包）
|
|
bool write(const void* src, Size_T len) {
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
if (len > (this->capacity - 1 - HEADER_SIZE)) return false;
|
|
const Size_T need = HEADER_SIZE + len;
|
|
if (this->free_space_unsafe() < need) return false;
|
|
this->write_bytes_unsafe(reinterpret_cast<const void*>(&len), HEADER_SIZE);
|
|
if (len) this->write_bytes_unsafe(src, len);
|
|
return true;
|
|
}
|
|
|
|
// out_len: in 代表 out 缓冲最大容量；out_len: out 代表实际读到的长度
|
|
bool read(void* out, Size_T& out_len) {
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
if (this->empty()) return false;
|
|
|
|
Size_T msg_len = 0;
|
|
this->peek_bytes_unsafe(reinterpret_cast<void*>(&msg_len), HEADER_SIZE);
|
|
|
|
const Size_T need = HEADER_SIZE + static_cast<Size_T>(msg_len);
|
|
if (this->used_space_unsafe() < need) return false; // 半包
|
|
|
|
if (out_len < msg_len) return false; // 调用者提供的 out 不够大
|
|
|
|
Size_T dummy = 0;
|
|
this->read_bytes_unsafe(reinterpret_cast<void*>(&dummy), HEADER_SIZE);
|
|
if (msg_len) this->read_bytes_unsafe(out, msg_len);
|
|
|
|
out_len = msg_len;
|
|
return true;
|
|
}
|
|
|
|
// 只 peek 包头，不消费；成功返回 true，并写出 msg_len
|
|
bool peek_len(Size_T& msg_len) {
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
if (this->empty()) return false;
|
|
this->peek_bytes_unsafe(reinterpret_cast<void*>(&msg_len), HEADER_SIZE);
|
|
|
|
// 防御：坏包头（避免队列被毒化）
|
|
if (msg_len > (this->capacity - 1 - HEADER_SIZE)) return false;
|
|
|
|
const Size_T need = HEADER_SIZE + msg_len;
|
|
if (this->used_space_unsafe() < need) return false; // 半包
|
|
return true;
|
|
}
|
|
|
|
// peek 下一条消息内容（不消费）
|
|
// out_len: in 为 out 缓冲容量；out_len: out 为消息实际长度
|
|
bool peek(void* out, Size_T& out_len) {
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
if (this->empty()) return false;
|
|
|
|
Size_T msg_len = 0;
|
|
this->peek_bytes_unsafe(reinterpret_cast<void*>(&msg_len), HEADER_SIZE);
|
|
|
|
if (msg_len > (this->capacity - 1 - HEADER_SIZE)) return false;
|
|
const Size_T need = HEADER_SIZE + msg_len;
|
|
if (this->used_space_unsafe() < need) return false; // 半包
|
|
if (out_len < msg_len) return false;
|
|
|
|
// 先临时把 head 向前偏移 HEADER_SIZE，再 peek msg_len 字节
|
|
// 不修改 head：用局部 off 来计算
|
|
Size_T off = (this->head + HEADER_SIZE) & this->mask_;
|
|
|
|
// 复用 copy 逻辑：这里写一个局部的“从任意 off 读”版本（不改 head）
|
|
// 由于 Base 里没有该函数，这里直接展开一份（与 peek_bytes_unsafe 类似）
|
|
const Size_T cap = this->capacity;
|
|
const Size_T end = std::min(msg_len, cap - off);
|
|
std::memcpy(out, this->buf_.data() + off, end);
|
|
std::memcpy(static_cast<std::uint8_t*>(out) + end, this->buf_.data(), msg_len - end);
|
|
|
|
out_len = msg_len;
|
|
return true;
|
|
}
|
|
|
|
// 仅丢弃/消费下一条消息（不拷贝内容）
|
|
// 成功返回 true；若空/半包/坏包头则 false
|
|
bool skip_one() {
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
if (this->empty()) return false;
|
|
|
|
Size_T msg_len = 0;
|
|
this->peek_bytes_unsafe(reinterpret_cast<void*>(&msg_len), HEADER_SIZE);
|
|
|
|
// 防御：坏包头，避免队列被毒化
|
|
if (msg_len > (this->capacity - 1 - HEADER_SIZE)) return false;
|
|
|
|
const Size_T need = HEADER_SIZE + msg_len;
|
|
if (this->used_space_unsafe() < need) return false; // 半包
|
|
|
|
// 直接前移 head（不实际读数据）
|
|
this->head = (this->head + need) & this->mask_;
|
|
return true;
|
|
}
|
|
|
|
// 仅丢弃/消费下一条消息的 payload 的前 n 字节（不拷贝）
|
|
// 用于“读一条大消息但分段处理”的场景：你先 peek_len/peek 拿到长度，再分段 consume
|
|
// 注意：它不会跳过 header；它假设你已经消费过 header 或者你自定义协议
|
|
// 如果你只需要 skip 整包，用 skip_one()
|
|
bool skip_bytes(Size_T n) {
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
if (n == 0) return true;
|
|
if (this->used_space_unsafe() < n) return false;
|
|
this->head = (this->head + n) & this->mask_;
|
|
return true;
|
|
}
|
|
|
|
};
|
|
|
|
// -------------------- Stream mode --------------------
|
|
template <class MutexT, typename Size_T>
|
|
class Stream_RingBuffer : public Base_RingBuffer<MutexT, Size_T> {
|
|
public:
|
|
bool write(const void* src, Size_T len) {
|
|
if (len == 0) return true;
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
if (this->free_space_unsafe() < len) return false;
|
|
this->write_bytes_unsafe(src, len);
|
|
return true;
|
|
}
|
|
|
|
// out_len 代表要读多少；成功则读满 out_len
|
|
bool read(void* out, Size_T& out_len) {
|
|
if (out_len == 0) return true;
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
if (this->used_space_unsafe() < out_len) {
|
|
out_len = 0;
|
|
return false;
|
|
}
|
|
this->read_bytes_unsafe(out, out_len);
|
|
return true;
|
|
}
|
|
|
|
// peek 最多 out_len 字节（不消费）；返回实际 peek 到的字节数
|
|
Size_T peek_best_effort(void* out, Size_T max_len) {
|
|
if (max_len == 0) return 0;
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
|
|
const Size_T can_read = this->used_space_unsafe();
|
|
const Size_T actual = std::min(max_len, can_read);
|
|
if (actual == 0) return 0;
|
|
|
|
// 等价于从 head 开始 peek actual 字节
|
|
const Size_T end = std::min(actual, this->capacity - this->head);
|
|
std::memcpy(out, this->buf_.data() + this->head, end);
|
|
std::memcpy(static_cast<std::uint8_t*>(out) + end, this->buf_.data(), actual - end);
|
|
return actual;
|
|
}
|
|
|
|
// peek 指定 out_len 字节（不消费）；成功则返回 true 并保持 out_len 不变
|
|
bool peek(void* out, Size_T out_len) {
|
|
if (out_len == 0) return true;
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
if (this->used_space_unsafe() < out_len) return false;
|
|
|
|
const Size_T end = std::min(out_len, this->capacity - this->head);
|
|
std::memcpy(out, this->buf_.data() + this->head, end);
|
|
std::memcpy(static_cast<std::uint8_t*>(out) + end, this->buf_.data(), out_len - end);
|
|
return true;
|
|
}
|
|
|
|
|
|
// best effort：尽可能写入，返回实际写入字节数
|
|
Size_T write_best_effort(const void* src, Size_T len) {
|
|
if (len == 0) return 0;
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
const Size_T can_write = this->free_space_unsafe();
|
|
const Size_T actual = std::min(len, can_write);
|
|
if (actual) this->write_bytes_unsafe(src, actual);
|
|
return actual;
|
|
}
|
|
|
|
// best effort：尽可能读取，返回实际读取字节数
|
|
Size_T read_best_effort(void* out, Size_T max_len) {
|
|
if (max_len == 0) return 0;
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
const Size_T can_read = this->used_space_unsafe();
|
|
const Size_T actual = std::min(max_len, can_read);
|
|
if (actual) this->read_bytes_unsafe(out, actual);
|
|
return actual;
|
|
}
|
|
// 丢弃/消费 n 字节（不拷贝）；成功则返回 true
|
|
bool skip(Size_T n) {
|
|
if (n == 0) return true;
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
if (this->used_space_unsafe() < n) return false;
|
|
this->head = (this->head + n) & this->mask_;
|
|
return true;
|
|
}
|
|
|
|
// best effort：尽可能丢弃，返回实际丢弃字节数
|
|
Size_T skip_best_effort(Size_T n) {
|
|
if (n == 0) return 0;
|
|
std::lock_guard<MutexT> lock(this->mutex_);
|
|
const Size_T can = this->used_space_unsafe();
|
|
const Size_T actual = std::min(n, can);
|
|
if (actual) this->head = (this->head + actual) & this->mask_;
|
|
return actual;
|
|
}
|
|
};
|
|
|
|
// -------------------- Convenient aliases --------------------
|
|
using RingBuffer_ST = RingBuffer<Empty_Lock, std::size_t>;
|
|
using StreamRingBuffer_ST = Stream_RingBuffer<Empty_Lock, std::size_t>;
|
|
|
|
using RingBuffer_MT = RingBuffer<std::mutex, std::size_t>;
|
|
using StreamRingBuffer_MT = Stream_RingBuffer<std::mutex, std::size_t>;
|
|
|
|
} // namespace Psc
|