它是 C++11 中联锁单链表的正确实现吗?

Is it a correct implementation of interlocked singly linked list in C++11

本文关键字:实现 链表 C++11 单链表 它是      更新时间:2023-10-16

我有以下使用 C++11 原子学的联锁单链表实现:

struct notag {};
template<class T, class Tag=notag>
struct s_list_base
{
};
template<class T, class Tag = notag>
struct s_list : s_list_base<T, Tag>
{
    s_list_base<T, Tag> *next_ptr;
};
template<bool auto_destruct, class T, class Tag = notag>
class atomic_s_list
{
    struct s_head : s_list_base<T, Tag>
    {
        std::atomic<s_list_base<T, Tag > *> next_ptr { this };
    };
    using LinkType = s_list<T, Tag> *;
    s_head head;
public:
    atomic_s_list() = default;
    atomic_s_list(const atomic_s_list &) = delete;
    atomic_s_list &operator =(const atomic_s_list &) = delete;
    ~atomic_s_list()
    {
        clear();
    }
    void clear() noexcept
    {
        if (auto_destruct)
        {
            T *item;
            do
            {
                item = pop();
                delete item;
            } while (item);
        }
        else
            head.next_ptr = &head;
    }
    void push(T *pItem) noexcept
    {
        auto p = static_cast<LinkType>(pItem);
        auto phead = head.next_ptr.load(std::memory_order_relaxed);
        do
        {
            p->next_ptr = phead;
        } while (!head.next_ptr.compare_exchange_weak(phead, p));
    }
    T *pop() noexcept
    {
        auto result = head.next_ptr.load(std::memory_order_relaxed);
        while (!head.next_ptr.compare_exchange_weak(result, static_cast<LinkType>(result)->next_ptr))
            ;
        return result == &head ? nullptr : static_cast<T *>(result);
    }
};

问题是,在实际程序中,我有几个并发运行的线程,这些线程使用 pop 从此列表中获取一个对象,使用它,然后用 push 将其放回原处,似乎我有一场竞赛有时两个线程最终从列表中获取相同的对象。

我试图从这个程序中举一个简单的例子来说明一个种族。在这里:

struct item : s_list<item>
{
    std::atomic<int> use{ 0 };
};
atomic_s_list<true, item> items;
item *allocate()
{
    auto *result = items.pop();
    if (!result)
        result = new item;
    return result;
}
void free(item *p)
{
    items.push(p);
}
int main()
{
    using namespace std::chrono_literals;
    static const int N = 20;
    std::vector<std::thread> threads;
    threads.reserve(N);
    for (int i = 0; i < N; ++i)
    {
        threads.push_back(std::thread([&]
        {
            while (true)
            {
                auto item = allocate();
                if (0 != item->use.fetch_add(1, std::memory_order_relaxed))
                    std::terminate();
                item->use.fetch_sub(1, std::memory_order_relaxed);
                free(item);
            }
        }));
    }
    std::this_thread::sleep_for(20min);
}

所以问题是:这种联锁单链表的实现是否正确?

经过更多的研究,我可以确认我面临ABA问题。

似乎没有人应该在现代硬件(具有大量硬件线程)和高度竞争的联锁列表上信任这种简单的联锁单链表实现。

考虑实现维基百科文章中描述的技巧后,我决定使用提升实现(参见boost::lockfree::stack),因为它似乎在解决ABA问题方面做出了很好的努力。

目前,我的测试代码没有失败,原始程序也没有失败。

不是一个正确的实现!

  1. 线程 A 和线程 B 调用 pop;A 和 B 得到相同的"结果"-> head.next_ptr
  2. 线程 A 修改内存。
  3. 线程 B 读取"结果>next_ptr"获取不正确的数据
  4. 线程 调用推送。 head.next_ptr == "结果" 现在!
  5. 线程 B 调用compare_exchange_weak; 使用不正确的数据head.next_ptr更新