You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

基于CAS与tagged pointer的无锁栈是否线程安全?相关疑问咨询

无锁栈节点销毁的线程安全问题

我使用std::atomic结合CAS实现了一个无锁栈,为解决ABA问题采用了tagged pointer,相关代码如下:

template <typename T>
union tagged_ptr {
    struct
    {
        std::uint64_t tag : 12, ptr : 52;
    };
    std::uint64_t full;

    tagged_ptr(const std::uint64_t &full)
    {
        this->full = full;
    }
    tagged_ptr(T *ptr = nullptr, std::uint16_t cnt = 0)
    {
        tag = cnt;
        this->ptr = reinterpret_cast<std::uint64_t>(ptr);
    }
    T *get()
    {
        return reinterpret_cast<T *>(ptr);
    }
};

template <typename T>
class lfstack
{
    struct Node
    {
        T data;
        Node *next;
        Node(const T &data, Node *next = nullptr)
        {
            this->data = data;
            this->next = next;
        }
    };

    std::atomic_uint64_t m_top;
    std::atomic_size_t m_size;

  public:
    lfstack()
    {
        m_top = 0;
        m_size = 0;
    }

    ~lfstack()
    {
        Node *ptr = reinterpret_cast<Node *>(m_top.load(std::memory_order_relaxed)), *next;
        while (ptr)
        {
            next = ptr->next;
            delete ptr;
            ptr = next;
        }
    }

    size_t size()
    {
        return m_size.load();
    }

    bool empty()
    {
        return !size();
    }

    const T &top()
    {
        return tagged_ptr<Node>(m_top.load()).get()->data;
    }

    std::optional<T> pop()
    {
        tagged_ptr<Node> local_ptr(m_top.load(std::memory_order_relaxed));
        while (true)
        {
            if (!local_ptr.get())
                return std::nullopt;
            tagged_ptr<Node> local_next(local_ptr.get()->next, local_ptr.tag); 
            if (m_top.compare_exchange_weak(local_ptr.full, local_next.full))
            {
                T ret_val = std::move(local_ptr.get()->data);
                delete local_ptr.get();
                m_size.fetch_sub(1, std::memory_order_relaxed);
                return ret_val;
            }
        }
    }

    void push(const T &data)
    {
        tagged_ptr<Node> local_ptr(m_top.load(std::memory_order_relaxed)), new_ptr(new Node(data));
        while (true)
        {
            new_ptr.get()->next = local_ptr.get();
            new_ptr.tag = local_ptr.tag + 1;
            if (m_top.compare_exchange_weak(local_ptr.full, new_ptr.full))
            {
                m_size.fetch_add(1, std::memory_order_relaxed);
                break;
            }
        }
    }
};

我已在MSVC x64编译器上进行多线程测试,结果看似正常,但怀疑pop函数中的tagged_ptr<Node> local_next(local_ptr.get()->next, local_ptr.tag);代码行并非线程安全。若其他线程执行delete local_ptr.get();,此时访问local_ptr.get()->next会导致未定义行为,请问是否需要延迟对象销毁?


你的怀疑完全正确,这行代码确实存在严重的线程安全问题,必须采用延迟对象销毁的策略。

问题根源

当你从m_top加载得到local_ptr后,其他线程可能已经完成了对该节点的pop操作并执行了delete,此时你访问local_ptr.get()->next本质是访问已被释放的内存,属于未定义行为。测试结果看似正常只是因为内存尚未被系统重新分配或覆盖,这种巧合在高并发场景下必然会引发崩溃或数据错乱。

可行的延迟销毁方案

  1. 危险指针(Hazard Pointers)

    • 每个线程在访问节点前,将该节点指针注册为"危险指针",标记当前线程正在使用这个节点;
    • 当需要销毁节点时,先检查所有线程的危险指针列表,确认没有线程引用该节点后再执行销毁;
    • 这是无锁数据结构中回收内存的标准方案之一,能可靠避免悬空指针访问。
  2. 基于纪元的回收(Epoch-Based Reclamation)

    • 维护全局的"纪元"计数器,每个线程在操作数据结构时记录当前纪元;
    • 待销毁节点被放入对应纪元的回收队列,当全局纪元推进到足够大(确保所有旧纪元的线程都已完成操作),再批量销毁该纪元队列中的节点。
  3. 线程本地回收池(简化版)

    • 给每个线程分配本地的待销毁节点列表,pop成功后不直接delete,而是将节点加入本地列表;
    • 当列表达到预设阈值(如100个节点)时,再批量销毁。但这种方案仅能降低风险,无法彻底避免问题,因为其他线程仍可能持有对这些节点的引用,适合对性能要求极高且并发强度可控的场景。

额外的代码问题提示

  • 内存顺序:pop和push中m_size.fetch_sub/fetch_add应使用std::memory_order_release,配合compare_exchange_weak的内存顺序,确保其他线程能及时看到栈大小的变化;
  • top函数:直接加载m_top后访问节点数据同样存在悬空指针风险,需要结合危险指针或类似机制保证访问时节点的有效性。

内容的提问来源于stack exchange,提问作者tongstar

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.23 09:22:31