c++ - 为什么我的无锁消息队列段错误 :(?

Question

作为一个纯粹的脑力练习，我试图让它在没有锁或互斥锁的情况下工作。这个想法是，当消费者线程正在读取/执行消息时，它会自动交换生产者线程使用哪个 std::vector 进行写入。这可能吗？我试过玩线程栅栏无济于事。这里某处存在竞争条件，因为它偶尔会出现段错误。我想它在 enqueue 函数的某个地方。有什么想法吗？

// should execute functions on the original thread
class message_queue {
public:
    using fn = std::function<void()>;
    using queue = std::vector<fn>;

    message_queue() : write_index(0) {
    }

    // should only be called from consumer thread
    void run () {
        // atomically gets the current pending queue and switches it with the other one
        // for example if we're writing to queues[0], we grab a reference to queue[0]
        // and tell the producer to write to queues[1]
        queue& active = queues[write_index.fetch_xor(1)];
        // skip if we don't have any messages
        if (active.size() == 0) return;
        // run all messages/callbacks
        for (auto fn : active) {
            fn();
        }
        // clear the active queue so it can be re-used
        active.clear();
        // swap active and pending threads
        write_index.fetch_xor(1);
    }
    void enqueue (fn value) {
        // loads the current pending queue and append some work
        queues[write_index.load()].push_back(value);
    }
private:
    queue queues[2];
    std::atomic<bool> is_empty; // unused for now
    std::atomic<int> write_index;


};
int main(int argc, const char * argv[])
{

    message_queue queue{};
    // flag to stop the message loop
    // doesn't actually need to be atomic because it's only read/wrote on the main thread
    std::atomic<bool> done(false);
    std::thread worker([&queue, &done] {
        int count = 100;
        // send 100 messages
        while (--count) {
            queue.enqueue([count] {
                // should be executed in the main thread
                std::cout << count << "\n";
            });
        }
        // finally tell the main thread we're done
        queue.enqueue([&] {
            std::cout << "done!\n";
            done = true;
        });
    });
    // run messages until the done flag is set
    while(!done) queue.run();
    worker.join();
}

Answer 1

如果我对您的代码的理解正确，则存在数据竞争，例如:

// producer
int r0 = write_index.load(); // r0 == 0

// consumer
int r1 = write_index.fetch_xor(1); // r1 == 0
queue& active = queues[r1];
active.size();

// producer
queue[r0].push_back(...);

现在两个线程同时访问同一个队列。这是一场数据竞赛，这意味着未定义的行为。