ViewDesignEngine/include/vde/core/concurrent_data.h

#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