由无锁容器管理的缓冲区的完整性

（在两个误解答案之后的澄清：如果生产者线程数小于堆栈大小，则代码运行良好。只有 1 个消费者发布槽。我用 32 个制作人与 16 个插槽调整这个演示的方式是快速触发一个糟糕的条件）

在对用于多线程缓冲区管理的无锁堆栈进行压力测试时，我发现缓冲区内容的完整性无法得到保证。我现在非常确定堆栈/后进先出解决方案不是最佳选择;但我仍然想了解这些缓冲区是如何受到损害的。

这个想法是：一个无锁堆栈，其中包含指向"免费"缓冲区的指针。它们可以由许多生产者线程中的一个检索。然后，缓冲区被填充数据并"调度"到单个使用者线程，最终将它们返回到堆栈。

观察结果是：- 两个线程以某种方式获得相同的缓冲区。- 一个线程正在获取一个缓冲区，其内存仍未从刚刚释放它的另一个线程刷新。

以下是我可以放在一起的最简单的示例，用于演示：

更新：我为任何想要使用它的人制作了一个具有更好调试输出的版本，在这里：https://ideone.com/v9VAqU

#include <atomic>
#include <assert.h>
#include <chrono>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>
using namespace std;
#define N_SLOTS 16
#define N_THREADS 32
// The data buffers that are shared among threads
class Buffer { public: int data[N_THREADS] = {0}; } buffers[N_SLOTS];
// The lock-free stack under study
class LockFreeStack
{
    Buffer* stack[N_SLOTS];
    atomic_int free_slots, out_of_slots, retries;
public:
    LockFreeStack() : free_slots(0), out_of_slots(0), retries(0) {
        for (int i=0; i<N_SLOTS; i++)
            release_buffer(&buffers[i]);
    }
    Buffer* get_buffer()
    {
        int slot = --free_slots;
        if (slot < 0) {
            out_of_slots++;
            return nullptr;
        }
/// [EDIT] CAN GET PREEMPTED RIGHT HERE, BREAKING ATOMICITY!
        return stack[slot];
    }
    void release_buffer(Buffer* buf)
    {
        int slot;
        while(true) {
            slot = free_slots;
            if (slot <= 0) {
                stack[0] = buf;
                free_slots = 1;
                break;
            }
            stack[slot] = buf;
            if (free_slots++ == slot)
                break;
            retries++;
        }
    }
    ostream& toStream(ostream& oss) {
        return oss << "LockFreeStack with free_slots=" << free_slots << ", oos=" << out_of_slots << ", retries=" << retries;
    }
} lockFreeStack;
// Utility class to help with test
class PrintQueue {
    queue<Buffer*> q;
    mutex m;
public:
    void add(Buffer* buf) {
        lock_guard<mutex> lock(m);
        q.push(buf);
    }
    Buffer* pop() {
        lock_guard<mutex> lock(m);
        Buffer* buf;
        if (q.empty())
            return nullptr;
        buf = q.front();
        q.pop();
        return buf;
    }
} printQueue;
int main()
{
    vector<thread> workers;
    for (int t = 0; t < N_THREADS; ++t) {
        workers.push_back(thread([&,t] {
            while(true) {
                auto buf = lockFreeStack.get_buffer();
                if (buf) {
                    buf->data[t] = t;
                    this_thread::sleep_for(chrono::milliseconds(10));
                    printQueue.add(buf);
                }
            }
        }));
    }
    while(true) {
        this_thread::sleep_for(chrono::milliseconds(10));
        lockFreeStack.toStream(cout) << endl;
        Buffer *buf;
        while((buf = printQueue.pop())) {
            cout << "Got Buffer " << buf << " #" << (buf-buffers) << " { ";
            int used = 0;
            for(int t=0; t<N_THREADS; t++)
                if (buf->data[t]) {
                    used += 1;
                    cout << 't' << buf->data[t] << ' ';
                    buf->data[t] = 0;
                }
            cout << "}n";
            assert (used == 1);
            lockFreeStack.release_buffer(buf);
        }
    }
    return 0;
}

还有一个错误输出的示例：

> LockFreeStack with free_slots=-2454858, oos=2454836, retries=0
> Got Buffer 0x604a40 #12 { t7 }
> Got Buffer 0x6049c0 #11 { t8 }
> Got Buffer 0x604b40 #14 { t1 }
> Got Buffer 0x604bc0 #15 { }
> test.cpp:111: int main(): Assertion `used == 1' failed.

我试过到处使用std::atomic_thread_fence()，但没有区别。

故障在哪里？

（顺便说一句，在多个版本的GCC上测试，包括5.2和4.6）