C++如何使无锁堆栈推送原子

Question

我需要写一个 void push(const T& val)无锁堆栈的实现。问题是 compare_exchange_weak 期望非原子 node*但我必须使用 std::atomic<node*> next字段而不是常规 node* next .我试图通过这样做来解决这个问题

void push(const T& val) {
    node* new_node = new node(val);
    node* local_next = new_node->next.load();
    while (!head.compare_exchange_weak(local_next, new_node));
}

但是创建 if local_next让事情变得更糟。我测试了 2 种代码变体。第一个有非原子场node* next我在下面的测试代码中丢失了大约 20-30 个元素。使用第二个变体我陷入了僵局。

测试代码:

#include <iostream>
#include <thread>
#include <atomic>
#include "lock_free_stack.h"

using namespace std;

void test(lock_free_stack<int>& st, atomic<int>& sum) {
    st.push(1);
    shared_ptr<int> val(st.pop());
    while (val == nullptr) { }
    sum.store(sum.load() + *val);
}

int main(int argc, const char * argv[]) {
    atomic<int> sum;
    sum.store(0);
    for (int i = 0; i < 100; ++i) {
        lock_free_stack<int> st;
        thread t1(test, ref(st), ref(sum));
        thread t2(test, ref(st), ref(sum));
        thread t3(test, ref(st), ref(sum));
        thread t4(test, ref(st), ref(sum));
        thread t5(test, ref(st), ref(sum));
        thread t6(test, ref(st), ref(sum));
        thread t7(test, ref(st), ref(sum));
        thread t8(test, ref(st), ref(sum));
        t1.join();
        t2.join();
        t3.join();
        t4.join();
        t5.join();
        t6.join();
        t7.join();
        t8.join();
    }
    if (sum.load() == 800) {
        cout << "CORRECT" << endl;
    } else {
        cout << "TIME TO REWRITE STACK " << sum << endl;
    }
    return 0;
}

还有我的无锁堆栈代码(第一个变体):

#ifndef lock_free_stack_hard_lock_free_stack_h
#define lock_free_stack_hard_lock_free_stack_h

template <typename T>
class lock_free_stack {
private:
    struct node {
        node* next;
        std::shared_ptr<T> value;
        node (const T& val) : value(std::make_shared<T>(val)) { }
    };
    std::atomic<node*> head;
    std::shared_ptr<T> default_value;

public:
    lock_free_stack() : head(nullptr), default_value(std::make_shared<T>()) { }

    void push(const T& val) {
        node* new_node = new node(val);
        new_node->next = head.load();
        while (!head.compare_exchange_weak(new_node->next, new_node));
    }

    std::shared_ptr<T> pop() {
        node* old_head = head.load();
        while (old_head && !head.compare_exchange_weak(old_head, old_head->next));
        if (old_head) {
            return old_head->value;
        } else {
            return std::shared_ptr<T>();
        }
    }
};

#endif

第二个变体:

#ifndef lock_free_stack_hard_lock_free_stack_h
    #define lock_free_stack_hard_lock_free_stack_h

    template <typename T>
    class lock_free_stack {
    private:
        struct node {
            std::atomic<node*> next;
            std::shared_ptr<T> value;
            node (const T& val) : value(std::make_shared<T>(val)) { }
        };
        std::atomic<node*> head;
        std::shared_ptr<T> default_value;

    public:
        lock_free_stack() : head(nullptr), default_value(std::make_shared<T>()) { }

        void push(const T& val) {
            node* new_node = new node(val);
            new_node->next = head.load();
            node* local_next = new_node->next.load();
            while (!head.compare_exchange_weak(local_next, new_node)); 
        }

        std::shared_ptr<T> pop() {
            node* old_head = head.load();
            while (old_head && !head.compare_exchange_weak(old_head, old_head->next));
            if (old_head) {
                return old_head->value;
            } else {
                return std::shared_ptr<T>();
            }
        }
    };

    #endif

所以最后一个问题是如何创建 local_next正确吗？

谢谢。

Answer 1

测试的一个问题是行 sum.store(sum.load() + *val);

使用原子操作，例如 sum += *val;