关于std::atomic操作原子性的两个技术疑问及代码验证

问题1：while(head && !head.compare_exchange_weak(ptr, head.load()->next))是否破坏原子性？ #

在这段代码的while(head && !head.compare_exchange_weak(ptr, head.load()->next))语句中，使用head.load()->next作为std::atomic::compare_exchange_weak的目标值是否会破坏原子性？从输出看似乎没问题，但我认为应该会，这个问题是否成立？

问题2：new_node->next = head.exchange(new_node);是否属于原子操作？ #

std::atomic::exchange本身是原子的，但返回值的赋值操作呢？假设new_node被其他线程访问，是否有线程会在赋值过程中插入执行？

代码示例（C++17标准编译） #

#include <iostream>
#include <thread>
#include <atomic>
#include <iomanip>
#include <array>
#include <future>
#include <mutex>
#include <exception>

static constexpr unsigned num_of_thread{ 100U };

struct Node{
    unsigned data;
    Node* next;

    Node() : data{0}, next{nullptr} {}
    Node(unsigned val) : data{val}, next{nullptr} {}
};

void test1() {
    std::atomic<Node*> head{ nullptr };
    std::cout << "Is lock-free?: " << std::boolalpha << head.is_lock_free() << std::endl;

    auto add_new_node = [&head](std::mutex& mut,
                                std::condition_variable& cond_var,
                                bool& guarantor,
                                unsigned i)
    {
        // Every thread creates its own unique node, so, no race condition here
        Node* new_node{ new Node{i} };
        {
            std::unique_lock<std::mutex> lck{ mut };
            cond_var.wait(lck, [&guarantor](){ return guarantor; });
            // All threads pile up here and wait for notification
            // (basically what std::barrier does)
        }

        // Ideally, all threads hit the below line at the same time.
        // Would there be a race condition here if new_node was a shared variable?
        // (Between the "exchange" atomic operation and the assignment of return value)
        new_node->next = head.exchange(new_node);
    };

    auto pop_a_node = [&head](std::mutex& mut,
                              std::condition_variable& cond_var,
                              bool& guarantor)
        {
        Node* ptr{ head };

        {
            std::unique_lock<std::mutex> lck{ mut };
            cond_var.wait(lck, [&guarantor](){ return guarantor; });
            // All threads pile up here and wait for notification
            // (basically what std::barrier does)
        }

        // Do we break atomicity here by using head.load()->next?
        // Or even, does this question make sense? 
        while(head && !head.compare_exchange_weak(ptr, head.load()->next));

        // Using ptr from this point onwards is safe as it is local to every thread
        if(ptr) {
            unsigned retval{ ptr->data };
            delete ptr;
            
            // Protect concurrent access to std::cout (nothing else)
            // in order not to get a scrambled output
            mut.lock();
            std::cout << retval << ' ';
            mut.unlock();
        }
        else {
            mut.lock();
            std::cout << "* ";
            mut.unlock();
        }
    };

    // Half of the threads add, the other half pops nodes
    [&pop_a_node, &add_new_node]() {
        std::condition_variable cond_var{};
        bool guarantor{ false }; // Needed in order to prevent spurious wakeups
        std::mutex mut;

        std::array<std::thread, num_of_thread> threads{};
        for(unsigned i{}; i<num_of_thread; i++) {
            if(i%2U) {
                threads[i] = std::move(std::thread{add_new_node,
                                                   std::ref(mut),
                                                   std::ref(cond_var),
                                                   std::ref(guarantor),
                                                   i});
            }
            else {
                threads[i] = std::move(std::thread{pop_a_node,
                                                   std::ref(mut),
                                                   std::ref(cond_var),
                                                   std::ref(guarantor)});
            }
        }

        // Wake all threads up at the same time. This should allow
        // more race condition possibilities compared to
        // just starting new threads repeatedly 
        guarantor = true;
        cond_var.notify_all();

        for(unsigned i{}; i<num_of_thread; i++) {
            threads[i].join();
        }
    }();
}

int main() {
    test1();

    return 0;
}

运行输出示例 #

Is lock-free?: true
49 9 71 69 95 * * * 97 3 87 * 27 81 85 51 11 29 53 5 * 31 23 19 99 45 * 55 83 13 * * 75 25 33 77 73 61 93 91 35 57 67 79 63 65 47 41 17

（*表示线程因遇到空列表无法弹出元素，缺失元素仍在列表中）

问题解答 #

针对问题1 #

这段代码确实存在数据竞争和未定义行为，只是测试输出没暴露出来。

head.load()->next的执行分为两步：

1. 原子加载head的当前值，得到一个Node*指针；
2. 解引用该指针访问next成员。

这两步之间没有同步机制，其他线程可能在这期间修改head的值——比如把当前head指向的节点删除，或者替换成新节点。此时第一步加载的指针就变成了悬空指针，解引用它会触发未定义行为。

而compare_exchange_weak本身是原子操作，但它的参数head.load()->next是在原子操作开始前就计算好的，整个while循环的逻辑不是原子的。如果其他线程在head.load()和compare_exchange_weak执行之间修改了head，你传入的next值就已经过时了，这会导致CAS操作的逻辑错误，甚至因为悬空指针访问崩溃。

你没看到问题只是测试场景下的运气好，不代表代码是安全的。正确的写法应该是先把head加载到本地变量，再解引用本地变量的next，然后用这个本地变量作为CAS的预期值：

Node* current = head.load();
while (current && !head.compare_exchange_weak(current, current->next)) {
    // 循环重试，current会被CAS自动更新为当前head的值
}

这样就能避免悬空指针的问题，因为每次循环都会重新加载当前的head到current，再基于最新的current去访问next。

针对问题2 #

new_node->next = head.exchange(new_node);不是一个完整的原子操作，它分为两个独立的步骤：

1. 原子执行head.exchange(new_node)：把head的值替换为new_node，并返回head原来的值；
2. 将返回的旧值赋值给new_node->next。

这两步之间没有原子性保证，其他线程可以在这期间插入执行。但在你的代码里，new_node是每个线程自己创建的局部变量，其他线程根本访问不到它，所以不存在数据竞争。

如果new_node是多个线程共享的变量，那赋值new_node->next的时候就会有风险：比如线程A执行完exchange拿到旧值，还没赋值给new_node->next，线程B就修改了new_node的next，这会导致线程A的赋值覆盖线程B的修改，引发数据竞争。但在你的代码场景里，每个线程的new_node都是独立的，所以这个问题不存在。

内容的提问来源于stack exchange，提问作者Ekrem Kılıç