我正在尝试实现一个使用线性循环缓冲区来存储数据的无锁队列。与通用无锁队列相比，我有以下放宽条件:

• 我知道将存储在队列中的最坏情况下元素的数量。队列是对一组固定元素进行操作的系统的一部分。代码永远不会尝试在队列中存储更多元素，因为此固定集合中有元素。
• 没有多生产者/多消费者。队列将用于多生产者/单消费者或单生产者/多消费者设置。

概念上，队列实现如下

• 标准二次幂环形缓冲区。底层数据结构是一个使用power-of-two trick的标准环形缓冲区。 .读写索引只会递增。当使用简单的位掩码对数组进行索引时，它们被限制在底层数组的大小。读指针在 pop() 中以原子方式递增，写指针在 push() 中以原子方式递增。
• 大小变量控制对 pop() 的访问。一个额外的“大小”变量跟踪队列中元素的数量。这消除了对读取和写入索引执行算术的需要。在整个写入操作发生后，大小变量自动递增，即数据已写入后备存储并且写游标已递增。我正在使用 compare-and-swap (CAS)在 pop() 中以原子方式递减大小的操作，并且仅在大小不为零时才继续。这样pop()应该保证返回有效数据。

我的队列实现如下。请注意，每当 pop() 尝试读取之前由 push() 写入的内存时，调试代码就会停止执行。这永远不应该发生，因为 ‒ 至少在概念上是这样的 ‒ pop() 可能只有在队列中有元素(不应该有下溢)时才会继续。

#include <atomic>
#include <cstdint>
#include <csignal> // XXX for debugging

template <typename T>
class Queue {
private:
    uint32_t m_data_size;   // Number of elements allocated
    std::atomic<T> *m_data; // Queue data, size is power of two
    uint32_t m_mask;        // Bitwise AND mask for m_rd_ptr and m_wr_ptr
    std::atomic<uint32_t> m_rd_ptr; // Circular buffer read pointer
    std::atomic<uint32_t> m_wr_ptr; // Circular buffer write pointer
    std::atomic<uint32_t> m_size;   // Number of elements in the queue

    static uint32_t upper_power_of_two(uint32_t v) {
        v--; // https://graphics.stanford.edu/~seander/bithacks.html
        v |= v >> 1; v |= v >> 2; v |= v >> 4; v |= v >> 8; v |= v >> 16;
        v++;
        return v;
    }

public:
    struct Optional { // Minimal replacement for std::optional
        bool good;
        T value;
        Optional() : good(false) {}
        Optional(T value) : good(true), value(std::move(value)) {}
        explicit operator bool() const { return good; }
    };

    Queue(uint32_t max_size)
        : // XXX Allocate 1 MiB of additional memory for debugging purposes
          m_data_size(upper_power_of_two(1024 * 1024 + max_size)),
          m_data(new std::atomic<T>[m_data_size]),
          m_mask(m_data_size - 1),
          m_rd_ptr(0),
          m_wr_ptr(0),
          m_size(0) {
        // XXX Debug code begin
        // Fill the memory with a marker so we can detect invalid reads
        for (uint32_t i = 0; i < m_data_size; i++) {
            m_data[i] = 0xDEADBEAF;
        }
        // XXX Debug code end
    }

    ~Queue() { delete[] m_data; }

    Optional pop() {
        // Atomically decrement the size variable
        uint32_t size = m_size.load();
        while (size != 0 && !m_size.compare_exchange_weak(size, size - 1)) {
        }

        // The queue is empty, abort
        if (size <= 0) {
            return Optional();
        }

        // Read the actual element, atomically increase the read pointer
        T res = m_data[(m_rd_ptr++) & m_mask].load();

        // XXX Debug code begin
        if (res == T(0xDEADBEAF)) {
            std::raise(SIGTRAP);
        }
        // XXX Debug code end
        return res;
    }

    void push(T t) {
        m_data[(m_wr_ptr++) & m_mask].store(t);
        m_size++;
    }

    bool empty() const { return m_size == 0; }
};

但是，下溢确实会发生，并且很容易在多线程压力测试中触发。在这个特定的测试中，我维护了两个队列 q1 和 q2。在主线程中，我将固定数量的元素提供给 q1。两个工作线程从 q1 读取并在紧密循环中推送到 q2。主线程从q2读取数据，反馈给q1。

如果只有一个工作线程(单生产者/单消费者)，或者只要所有工作线程与主线程在同一个 CPU 上，这就可以正常工作。但是，一旦有两个工作线程被显式调度到与主线程不同的 CPU 上，它就会失败。

下面的代码实现了这个测试

#include <pthread.h>
#include <thread>
#include <vector>

static void queue_stress_test_main(std::atomic<uint32_t> &done_count,
                                   Queue<int> &queue_rd, Queue<int> &queue_wr) {
    for (size_t i = 0; i < (1UL << 24); i++) {
        auto res = queue_rd.pop();
        if (res) {
            queue_wr.push(res.value);
        }
    }
    done_count++;
}

static void set_thread_affinity(pthread_t thread, int cpu) {
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(cpu, &cpuset);
    if (pthread_setaffinity_np(thread, sizeof(cpu_set_t),
                               &cpuset) != 0) {
        throw "Error while calling pthread_setaffinity_np";
    }
}

int main() {
    static constexpr uint32_t n_threads{2U}; // Number of worker threads
    //static constexpr uint32_t n_threads{1U}; // < Works fine
    static constexpr uint32_t max_size{16U}; // Elements in the queue
    std::atomic<uint32_t> done_count{0};     // Number of finished threads
    Queue<int> queue1(max_size), queue2(max_size);

    // Launch n_threads threads, make sure the main thread and the two worker
    // threads are on different CPUs.
    std::vector<std::thread> threads;
    for (uint32_t i = 0; i < n_threads; i++) {
        threads.emplace_back(queue_stress_test_main, std::ref(done_count),
                             std::ref(queue1), std::ref(queue2));
        set_thread_affinity(threads.back().native_handle(), 0);
    }
    set_thread_affinity(pthread_self(), 1);
    //set_thread_affinity(pthread_self(), 0); // < Works fine

    // Pump data from queue2 into queue1
    uint32_t elems_written = 0;
    while (done_count < n_threads || !queue2.empty()) {
        // Initially fill queue1 with all values from 0..max_size-1
        if (elems_written < max_size) {
            queue1.push(elems_written++);
        }

        // Read elements from queue2 and put them into queue1
        auto res = queue2.pop();
        if (res) {
            queue1.push(res.value);
        }
    }

    // Wait for all threads to finish
    for (uint32_t i = 0; i < n_threads; i++) {
        threads[i].join();
    }
}

大多数时候这个程序会触发队列代码中的陷阱，这意味着 pop() 试图读取从未被 push() 触及的内存‒ 尽管 pop() 应该 只有在 push() 至少与 pop() 一样频繁地被调用时才会成功>.

您可以在 Linux 上使用 GCC/clang 编译并运行上述程序

c++ -std=c++11 queue.cpp -o queue -lpthread && ./queue

要么只是连接上面的两个代码块，要么下载完整的程序 here .

请注意，在谈到无锁数据结构时，我完全是个新手。我非常清楚有大量经过实战检验的 C++ 无锁队列实现。但是，我就是想不通为什么上面的代码不能按预期工作。

最佳答案

您有两个错误，其中一个可能导致您观察到的失败。

让我们看看您的推送代码，除了我们将只允许每个语句执行一个操作:

void push(T t)
{
    auto const claimed_index = m_wr_ptr++;               /* 1 */
    auto const claimed_offset = claimed_index & m_mask; /* 2 */
    auto& claimed_data = m_data[claimed_offset];         /* 3 */
    claimed_data.store(t);                               /* 4 */
    m_size++;                                            /* 5 */
}

现在，对于有两个生产者的队列，在操作 1 和 4 之间存在竞争条件的漏洞窗口:

之前:

m_rd_ptr == 1
m_wr_ptr == 1
m_size == 0

制作人 A:

/* 1 */ claimed_index = 1; m_wr_ptr = 2;
/* 2 */ claimed_offset = 1;

• Scheduler 让 Producer A 在这里 sleep

制作人 B:

/* 1 */ claimed_index = 2; m_wr_ptr = 3;
/* 2 */ claimed_offset = 2;
/* 3 */ claimed_data = m_data[2];
/* 4 */ claimed_data.store(t);
/* 5 */ m_size = 1;

之后:

m_size == 1
m_rd_ptr == 1
m_wr_ptr == 3
m_data[1] == 0xDEADBEAF
m_data[2] == value_produced_by_B

消费者现在运行，看到 m_size > 0，并从 m_data[1] 读取，同时将 m_rd_ptr 从 1 增加到 2。但是m_data[1]还没有被Producer A写入，Producer B写入了m_data[2]。

第二个错误是 pop() 中的补充情况，当消费者线程在 m_rd_ptr++ 操作和 .load() 之间中断时> 打电话。它可能导致读取值乱序，可能乱序到队列完全环绕并覆盖原始值。

仅仅因为单个源语句中的两个操作是原子的并不能使整个语句成为原子。

关于c++ - C++ 无锁队列实现中的虚假下溢，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/50094792/