#pragma once /** * @file concurrent_data.h * @brief 极致性能 — 无锁并发数据结构 * * 提供高性能线程安全容器: * - LockFreeQueue : 无锁队列 (CAS),多生产者多消费者 * - LockFreeStack : 无锁栈 (CAS),单生产者单消费者为主 * - ConcurrentHashMap: 分段锁哈希表,读写并发 * - ReadWriteSpinLock : 读写自旋锁(读优先) * * @ingroup core */ #include #include #include #include #include #include #include #include #include #include #include #include namespace vde::core { // ═══════════════════════════════════════════════════════════════════════════ // 无锁队列 (Lock-Free Queue) // ═══════════════════════════════════════════════════════════════════════════ /// 多生产者-多消费者无锁环形队列 /// /// 基于原子 CAS 操作实现,无互斥锁,适用于高吞吐场景。 /// 内部使用 2 的幂次容量保证索引取模为位运算。 /// /// @tparam T 元素类型(必须可平凡拷贝) /// /// @note 容量必须为 2 的幂,自动向上取整 template class LockFreeQueue { static_assert(std::is_trivially_copyable_v, "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( ::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 ready{false}; }; Element* buffer_; const size_t mask_; // capacity - 1 (power of 2) alignas(64) std::atomic head_{0}; alignas(64) std::atomic 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 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 head_; }; // ═══════════════════════════════════════════════════════════════════════════ // 分段锁哈希表 (Concurrent HashMap) // ═══════════════════════════════════════════════════════════════════════════ /// 分段锁哈希表 /// /// 将哈希表分成多个段(Segment),每段独立加锁。 /// 不同段的操作完全并行,同一段内部串行。 /// /// @tparam K 键类型 /// @tparam V 值类型 /// @tparam Hash 哈希函数(默认 std::hash) /// /// @note 段数量固定为 64(2 的幂),保证取模为位运算 template> 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 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 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 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