Files
ViewDesignEngine/include/vde/core/concurrent_data.h
T

586 lines
19 KiB
C++
Raw Normal View History

#pragma once
/**
* @file concurrent_data.h
* @brief 极致性能 — 无锁并发数据结构
*
* 提供高性能线程安全容器:
* - LockFreeQueue<T> : 无锁队列 (CAS),多生产者多消费者
* - LockFreeStack<T> : 无锁栈 (CAS),单生产者单消费者为主
* - ConcurrentHashMap<K,V>: 分段锁哈希表,读写并发
* - ReadWriteSpinLock : 读写自旋锁(读优先)
*
* @ingroup core
*/
#include <atomic>
#include <vector>
#include <memory>
#include <functional>
#include <mutex>
#include <shared_mutex>
#include <cstddef>
#include <cstdint>
#include <cstring>
#include <new>
#include <utility>
#include <type_traits>
namespace vde::core {
// ═══════════════════════════════════════════════════════════════════════════
// 无锁队列 (Lock-Free Queue)
// ═══════════════════════════════════════════════════════════════════════════
/// 多生产者-多消费者无锁环形队列
///
/// 基于原子 CAS 操作实现,无互斥锁,适用于高吞吐场景。
/// 内部使用 2 的幂次容量保证索引取模为位运算。
///
/// @tparam T 元素类型(必须可平凡拷贝)
///
/// @note 容量必须为 2 的幂,自动向上取整
template<typename T>
class LockFreeQueue {
static_assert(std::is_trivially_copyable_v<T>,
"LockFreeQueue requires trivially copyable types");
public:
/// 构造函数
/// @param capacity 期望容量(自动向上取整到 2 的幂,最小 8)
explicit LockFreeQueue(size_t capacity = 1024)
: mask_(next_power_of_two(std::max(capacity, size_t(8))) - 1)
{
size_t real = mask_ + 1;
buffer_ = static_cast<Element*>(
::operator new(real * sizeof(Element), std::nothrow));
// placement new 初始化
for (size_t i = 0; i < real; ++i) {
new (&buffer_[i]) Element();
}
head_.store(0, std::memory_order_relaxed);
tail_.store(0, std::memory_order_relaxed);
}
~LockFreeQueue() {
if (buffer_) {
size_t real = mask_ + 1;
for (size_t i = 0; i < real; ++i) {
buffer_[i].~Element();
}
::operator delete(buffer_);
}
}
// 禁止拷贝
LockFreeQueue(const LockFreeQueue&) = delete;
LockFreeQueue& operator=(const LockFreeQueue&) = delete;
/// 尝试入队(非阻塞)
/// @param item 要入队的元素
/// @return true 若入队成功
bool try_push(const T& item) {
size_t tail = tail_.load(std::memory_order_relaxed);
for (;;) {
size_t head = head_.load(std::memory_order_acquire);
// 满判断:tail + 1 == head (mod capacity)
if ((tail - head) >= mask_) return false;
if (tail_.compare_exchange_weak(tail, tail + 1,
std::memory_order_release, std::memory_order_relaxed)) {
// 成功分配 slot
size_t idx = tail & mask_;
buffer_[idx].data = item;
buffer_[idx].ready.store(true, std::memory_order_release);
return true;
}
// CAS 失败,tail 已更新,重试
}
}
/// 尝试移动入队
bool try_push(T&& item) {
size_t tail = tail_.load(std::memory_order_relaxed);
for (;;) {
size_t head = head_.load(std::memory_order_acquire);
if ((tail - head) >= mask_) return false;
if (tail_.compare_exchange_weak(tail, tail + 1,
std::memory_order_release, std::memory_order_relaxed)) {
size_t idx = tail & mask_;
buffer_[idx].data = std::move(item);
buffer_[idx].ready.store(true, std::memory_order_release);
return true;
}
}
}
/// 尝试出队(非阻塞)
/// @param item 输出参数,接收出队元素
/// @return true 若出队成功
bool try_pop(T& item) {
size_t head = head_.load(std::memory_order_relaxed);
for (;;) {
size_t tail = tail_.load(std::memory_order_acquire);
if (head == tail) return false; // 空
size_t idx = head & mask_;
if (!buffer_[idx].ready.load(std::memory_order_acquire)) {
// 生产者尚未写入完成,重试
head = head_.load(std::memory_order_relaxed);
continue;
}
if (head_.compare_exchange_weak(head, head + 1,
std::memory_order_release, std::memory_order_relaxed)) {
item = std::move(buffer_[idx].data);
buffer_[idx].ready.store(false, std::memory_order_release);
return true;
}
// CAS 失败,重试
}
}
/// 是否为空
[[nodiscard]] bool empty() const {
return head_.load(std::memory_order_acquire) ==
tail_.load(std::memory_order_acquire);
}
/// 当前元素数(近似值,并发下非精确)
[[nodiscard]] size_t size() const {
size_t h = head_.load(std::memory_order_acquire);
size_t t = tail_.load(std::memory_order_acquire);
return (t >= h) ? (t - h) : 0;
}
/// 容量
[[nodiscard]] size_t capacity() const { return mask_ + 1; }
private:
struct Element {
T data{};
std::atomic<bool> ready{false};
};
Element* buffer_;
const size_t mask_; // capacity - 1 (power of 2)
alignas(64) std::atomic<size_t> head_{0};
alignas(64) std::atomic<size_t> tail_{0};
static size_t next_power_of_two(size_t v) {
v--;
v |= v >> 1; v |= v >> 2;
v |= v >> 4; v |= v >> 8;
v |= v >> 16; v |= v >> 32;
return v + 1;
}
};
// ═══════════════════════════════════════════════════════════════════════════
// 无锁栈 (Lock-Free Stack)
// ═══════════════════════════════════════════════════════════════════════════
/// 无锁栈 — Treiber Stack
///
/// 基于 CAS 的单链表实现,适用于单生产者单消费者或低竞争多生产者场景。
///
/// @tparam T 元素类型
template<typename T>
class LockFreeStack {
struct Node {
T data;
Node* next;
Node(const T& d, Node* n = nullptr) : data(d), next(n) {}
Node(T&& d, Node* n = nullptr) : data(std::move(d)), next(n) {}
};
public:
LockFreeStack() : head_(nullptr) {}
~LockFreeStack() {
Node* node = head_.load(std::memory_order_relaxed);
while (node) {
Node* next = node->next;
delete node;
node = next;
}
}
LockFreeStack(const LockFreeStack&) = delete;
LockFreeStack& operator=(const LockFreeStack&) = delete;
/// 压栈
void push(const T& item) {
Node* node = new Node(item);
node->next = head_.load(std::memory_order_relaxed);
while (!head_.compare_exchange_weak(node->next, node,
std::memory_order_release, std::memory_order_relaxed)) {
// 自旋重试
}
}
/// 移动压栈
void push(T&& item) {
Node* node = new Node(std::move(item));
node->next = head_.load(std::memory_order_relaxed);
while (!head_.compare_exchange_weak(node->next, node,
std::memory_order_release, std::memory_order_relaxed)) {
// 自旋重试
}
}
/// 尝试弹栈
/// @return true 若弹栈成功
bool try_pop(T& item) {
Node* node = head_.load(std::memory_order_acquire);
while (node) {
if (head_.compare_exchange_weak(node, node->next,
std::memory_order_release, std::memory_order_relaxed)) {
item = std::move(node->data);
delete node;
return true;
}
// CAS 失败,node 已更新为新的 head,继续
}
return false;
}
/// 是否为栈空
[[nodiscard]] bool empty() const {
return head_.load(std::memory_order_acquire) == nullptr;
}
/// 清空栈
void clear() {
Node* node = head_.exchange(nullptr, std::memory_order_acq_rel);
while (node) {
Node* next = node->next;
delete node;
node = next;
}
}
private:
std::atomic<Node*> head_;
};
// ═══════════════════════════════════════════════════════════════════════════
// 分段锁哈希表 (Concurrent HashMap)
// ═══════════════════════════════════════════════════════════════════════════
/// 分段锁哈希表
///
/// 将哈希表分成多个段(Segment),每段独立加锁。
/// 不同段的操作完全并行,同一段内部串行。
///
/// @tparam K 键类型
/// @tparam V 值类型
/// @tparam Hash 哈希函数(默认 std::hash<K>
///
/// @note 段数量固定为 64(2 的幂),保证取模为位运算
template<typename K, typename V, typename Hash = std::hash<K>>
class ConcurrentHashMap {
static constexpr size_t NUM_SEGMENTS = 64;
static constexpr size_t SEGMENT_MASK = NUM_SEGMENTS - 1;
struct Entry {
K key;
V value;
Entry* next = nullptr;
Entry(const K& k, const V& v) : key(k), value(v) {}
Entry(K&& k, V&& v) : key(std::move(k)), value(std::move(v)) {}
};
struct alignas(64) Bucket {
Entry* head = nullptr;
};
struct alignas(64) Segment {
std::shared_mutex mutex;
std::vector<Bucket> buckets;
Segment() : buckets(16) {} // 初始 16 桶/段
};
public:
ConcurrentHashMap() {
for (size_t i = 0; i < NUM_SEGMENTS; ++i) {
segments_[i].store(new Segment(), std::memory_order_relaxed);
}
}
~ConcurrentHashMap() {
for (size_t i = 0; i < NUM_SEGMENTS; ++i) {
Segment* seg = segments_[i].load(std::memory_order_relaxed);
if (seg) {
for (auto& bucket : seg->buckets) {
Entry* e = bucket.head;
while (e) {
Entry* next = e->next;
delete e;
e = next;
}
}
delete seg;
}
}
}
ConcurrentHashMap(const ConcurrentHashMap&) = delete;
ConcurrentHashMap& operator=(const ConcurrentHashMap&) = delete;
/// 插入或更新
/// @param key 键
/// @param value 值
void insert(const K& key, const V& value) {
size_t seg_idx = segment_index(key);
Segment* seg = segments_[seg_idx].load(std::memory_order_acquire);
std::unique_lock lock(seg->mutex);
size_t bidx = bucket_index(key, seg->buckets.size());
Entry* e = seg->buckets[bidx].head;
while (e) {
if (e->key == key) { e->value = value; return; }
e = e->next;
}
// 插入链表头
Entry* ne = new Entry(key, value);
ne->next = seg->buckets[bidx].head;
seg->buckets[bidx].head = ne;
}
/// 移动插入
void insert(K&& key, V&& value) {
size_t seg_idx = segment_index(key);
Segment* seg = segments_[seg_idx].load(std::memory_order_acquire);
std::unique_lock lock(seg->mutex);
size_t bidx = bucket_index(key, seg->buckets.size());
Entry* e = seg->buckets[bidx].head;
while (e) {
if (e->key == key) { e->value = std::move(value); return; }
e = e->next;
}
Entry* ne = new Entry(std::move(key), std::move(value));
ne->next = seg->buckets[bidx].head;
seg->buckets[bidx].head = ne;
}
/// 查找
/// @param key 键
/// @param value 输出参数
/// @return true 若找到
[[nodiscard]] bool find(const K& key, V& value) const {
size_t seg_idx = segment_index(key);
Segment* seg = segments_[seg_idx].load(std::memory_order_acquire);
std::shared_lock lock(seg->mutex);
size_t bidx = bucket_index(key, seg->buckets.size());
Entry* e = seg->buckets[bidx].head;
while (e) {
if (e->key == key) { value = e->value; return true; }
e = e->next;
}
return false;
}
/// 检查键是否存在
[[nodiscard]] bool contains(const K& key) const {
V dummy;
return find(key, dummy);
}
/// 删除键
/// @return true 若删除成功
bool erase(const K& key) {
size_t seg_idx = segment_index(key);
Segment* seg = segments_[seg_idx].load(std::memory_order_acquire);
std::unique_lock lock(seg->mutex);
size_t bidx = bucket_index(key, seg->buckets.size());
Entry** prev = &seg->buckets[bidx].head;
Entry* e = *prev;
while (e) {
if (e->key == key) {
*prev = e->next;
delete e;
return true;
}
prev = &e->next;
e = e->next;
}
return false;
}
/// 清空
void clear() {
for (size_t i = 0; i < NUM_SEGMENTS; ++i) {
Segment* seg = segments_[i].load(std::memory_order_acquire);
std::unique_lock lock(seg->mutex);
for (auto& bucket : seg->buckets) {
Entry* e = bucket.head;
while (e) {
Entry* next = e->next;
delete e;
e = next;
}
bucket.head = nullptr;
}
}
}
/// 总元素数(近似值,遍历所有段)
[[nodiscard]] size_t size() const {
size_t total = 0;
for (size_t i = 0; i < NUM_SEGMENTS; ++i) {
Segment* seg = segments_[i].load(std::memory_order_acquire);
std::shared_lock lock(seg->mutex);
for (auto& bucket : seg->buckets) {
Entry* e = bucket.head;
while (e) { ++total; e = e->next; }
}
}
return total;
}
private:
std::atomic<Segment*> segments_[NUM_SEGMENTS];
Hash hasher_;
size_t segment_index(const K& key) const {
return hasher_(key) & SEGMENT_MASK;
}
size_t bucket_index(const K& key, size_t num_buckets) const {
return (hasher_(key) >> 6) % num_buckets;
}
};
// ═══════════════════════════════════════════════════════════════════════════
// 读写自旋锁
// ═══════════════════════════════════════════════════════════════════════════
/// 读优先读写自旋锁
///
/// 多个读者可以同时持有锁;写者独占。
/// 使用原子自旋等待,适合临界区极短(< 100 ns)的场景。
///
/// @note 读者数量限制为 2^30 - 1
class ReadWriteSpinLock {
static constexpr int WRITER_BIT = 0x40000000; // bit 30
static constexpr int READER_MASK = 0x3FFFFFFF; // bits 0-29
public:
ReadWriteSpinLock() : state_(0) {}
// ── 读锁 ──
/// 获取读锁(自旋等待直到无写者)
void lock_read() {
for (int spins = 0; ; ++spins) {
int s = state_.load(std::memory_order_relaxed);
if (!(s & WRITER_BIT)) {
if (state_.compare_exchange_weak(s, s + 1,
std::memory_order_acquire, std::memory_order_relaxed)) {
return;
}
}
// 自适应退避
if (spins > 100) {
// PAUSE / YIELD
#if defined(__x86_64__) || defined(_M_X64)
__builtin_ia32_pause();
#elif defined(__aarch64__)
__asm__ volatile("yield");
#endif
}
}
}
/// 释放读锁
void unlock_read() {
state_.fetch_sub(1, std::memory_order_release);
}
// ── 写锁 ──
/// 获取写锁(自旋等待直到无读者且无写者)
void lock_write() {
for (int spins = 0; ; ++spins) {
int expected = 0;
if (state_.compare_exchange_weak(expected, WRITER_BIT,
std::memory_order_acquire, std::memory_order_relaxed)) {
return;
}
if (spins > 100) {
#if defined(__x86_64__) || defined(_M_X64)
__builtin_ia32_pause();
#elif defined(__aarch64__)
__asm__ volatile("yield");
#endif
}
}
}
/// 释放写锁
void unlock_write() {
state_.store(0, std::memory_order_release);
}
/// 尝试获取读锁(非阻塞)
[[nodiscard]] bool try_lock_read() {
int s = state_.load(std::memory_order_relaxed);
if (s & WRITER_BIT) return false;
return state_.compare_exchange_strong(s, s + 1,
std::memory_order_acquire, std::memory_order_relaxed);
}
/// 尝试获取写锁(非阻塞)
[[nodiscard]] bool try_lock_write() {
int expected = 0;
return state_.compare_exchange_strong(expected, WRITER_BIT,
std::memory_order_acquire, std::memory_order_relaxed);
}
/// 是否有写者
[[nodiscard]] bool is_write_locked() const {
return state_.load(std::memory_order_acquire) & WRITER_BIT;
}
private:
std::atomic<int> state_;
};
/// RAII 读锁守卫
class ReadLockGuard {
public:
explicit ReadLockGuard(ReadWriteSpinLock& lock) : lock_(&lock) { lock_->lock_read(); }
~ReadLockGuard() { if (lock_) lock_->unlock_read(); }
ReadLockGuard(const ReadLockGuard&) = delete;
ReadLockGuard& operator=(const ReadLockGuard&) = delete;
ReadLockGuard(ReadLockGuard&& o) noexcept : lock_(o.lock_) { o.lock_ = nullptr; }
ReadLockGuard& operator=(ReadLockGuard&& o) noexcept {
if (this != &o) { lock_ = o.lock_; o.lock_ = nullptr; }
return *this;
}
private:
ReadWriteSpinLock* lock_;
};
/// RAII 写锁守卫
class WriteLockGuard {
public:
explicit WriteLockGuard(ReadWriteSpinLock& lock) : lock_(&lock) { lock_->lock_write(); }
~WriteLockGuard() { if (lock_) lock_->unlock_write(); }
WriteLockGuard(const WriteLockGuard&) = delete;
WriteLockGuard& operator=(const WriteLockGuard&) = delete;
WriteLockGuard(WriteLockGuard&& o) noexcept : lock_(o.lock_) { o.lock_ = nullptr; }
WriteLockGuard& operator=(WriteLockGuard&& o) noexcept {
if (this != &o) { lock_ = o.lock_; o.lock_ = nullptr; }
return *this;
}
private:
ReadWriteSpinLock* lock_;
};
} // namespace vde::core