多生产者多消费者无锁队列崩溃：std::bad_alloc异常排查求助

我实现了一个多生产者多消费者（MPMC）无锁队列，但运行时在dequeue函数中抛出std::bad_alloc异常并崩溃。我无法理解该异常的成因，因为我既没有手动分配内存，也未使用会触发内存分配的STL容器。以下是我的代码及错误输出，请帮我排查问题所在：

class QuoteQueue {
    public:

        QuoteQueue() {
            // Make sure that atomic which we use is lock free:
            assert(numOfElements_.is_lock_free());
        }

        // Blocks if the queue is full
        void enqueue(Quote&& quote) {
            try
            {
                
                while(numOfElements_.fetch_add(1) >= MAX_CAPACITY) {
                    numOfElements_.fetch_sub(1);
                }
                
                int curTail = 0;
                int newTail = 0;

                do {
                    curTail = tail_.load();
                    newTail = (curTail + 1) % MAX_CAPACITY;
                } while(!tail_.compare_exchange_weak(curTail, newTail));

                quotes_[curTail] = quote;
            }
            catch(std::exception& e)
            {
                std::cout << "exception during enqueue: " << e.what() << std::endl;
            }
        }

        // Blocks if the queue is empty
        Quote dequeue() {
            
            try
            {
                
                while(numOfElements_.fetch_sub(1) <= 0) {
                    numOfElements_.fetch_add(1);
                }
                
                int curHead = 0;
                int newHead = 0;

                do {
                    curHead = head_.load();
                    newHead = (curHead + 1) % MAX_CAPACITY;
                } while(!head_.compare_exchange_weak(curHead, newHead));

                return quotes_[curHead];
            }
            catch(std::exception& e)
            {
                std::cout << "exception during dequeue: " << e.what() << std::endl;
            }
            
            return {};
        }

    private:
        static const int MAX_CAPACITY = 1000;

        std::array<Quote, MAX_CAPACITY> quotes_;

        std::atomic<int> head_ = 0;
        std::atomic<int> tail_ = 0;

        std::atomic<int> numOfElements_ = 0;
};

int main() {

    std::vector<std::thread> enqThreads;
    std::vector<std::thread> deqThreads;

    QuoteQueue queue;

    for (int i = 0; i < 10; i++) {
        enqThreads.emplace_back([&queue] {
                int count = 0;
                while (1) {
                    Quote quote({"2024-02-12 13:12:13:567.788", 100. + count, 50. + count * 0.5, true});
                    queue.enqueue(std::move(quote));
                    count = (count + 1) % 100;
                    //std::cout << quote << std::flush;
                }
            });
    }

    for (int i = 0; i < 20; i++) {
        deqThreads.emplace_back([&queue]
        {
            while (1) {
                auto quote = queue.dequeue(); // dequeue should block if queue is empty
                quote.produc = false;
                //std::cout << quote << std::flush; // [Leonid] I added flush for testing
            }
        });
    }

    // Your code goes here.
    // Wait for enqueue, dequeue threads to finisih

    for (int i = 0; i < 10; i++) {
        enqThreads[i].join();
        deqThreads[i].join();
    }
    
    return 0;
}

错误输出：

exception during dequeue: std::bad_alloc
Segmentation fault

问题根源 #

1. numOfElements_的竞态条件触发非法内存访问 #

dequeue中numOfElements_.fetch_sub(1) <= 0的逻辑存在严重竞态：

• 队列为空时，多个消费者线程同时执行fetch_sub(1)，会把numOfElements_降到负数。
• 后续线程进入循环执行fetch_add(1)恢复数值，但已有线程跳过循环继续执行head_的CAS操作。
• 当head_超过tail_（甚至绕回数组边界后），访问quotes_[curHead]会指向数组越界的内存区域，破坏堆结构，最终触发std::bad_alloc（堆损坏后，任何内存分配操作都会失败）或直接段错误。

2. enqueue的队列满判断逻辑同样有竞态 #

多个生产者同时执行fetch_add(1)时，可能导致numOfElements_超过MAX_CAPACITY，后续的fetch_sub(1)无法原子性恢复数值，导致队列实际元素数量超过容量，覆盖未被消费的元素。

3. 返回Quote{}可能触发内存分配 #

如果Quote类的默认构造函数包含内存分配逻辑（比如std::string成员），当dequeue的try块抛出异常后，返回{}会调用默认构造函数，此时堆已损坏，就会抛出std::bad_alloc。

修复方案 #

修正numOfElements_的原子操作逻辑 #

必须保证修改head_/tail_之前，numOfElements_的增减严格对应队列实际元素变化，且循环判断是原子性的：

修正enqueue：

void enqueue(Quote&& quote) {
    try {
        int expected;
        do {
            expected = numOfElements_.load();
            if (expected >= MAX_CAPACITY) {
                std::this_thread::yield(); // 让出CPU减少忙等
                continue;
            }
        } while (!numOfElements_.compare_exchange_weak(expected, expected + 1));

        int curTail = 0;
        int newTail = 0;
        do {
            curTail = tail_.load();
            newTail = (curTail + 1) % MAX_CAPACITY;
        } while (!tail_.compare_exchange_weak(curTail, newTail));

        quotes_[curTail] = std::move(quote);
    } catch(std::exception& e) {
        std::cout << "exception during enqueue: " << e.what() << std::endl;
    }
}

修正dequeue：

Quote dequeue() {
    try {
        int expected;
        do {
            expected = numOfElements_.load();
            if (expected <= 0) {
                std::this_thread::yield();
                continue;
            }
        } while (!numOfElements_.compare_exchange_weak(expected, expected - 1));

        int curHead = 0;
        int newHead = 0;
        do {
            curHead = head_.load();
            newHead = (curHead + 1) % MAX_CAPACITY;
        } while (!head_.compare_exchange_weak(curHead, newHead));

        return std::move(quotes_[curHead]);
    } catch(std::exception& e) {
        std::cout << "exception during dequeue: " << e.what() << std::endl;
        throw; // 抛出异常避免返回默认构造的Quote
    }
}

额外建议 #

• 使用std::optional<Quote>作为dequeue的返回值，避免默认构造函数的潜在问题。
• 忙等时使用std::this_thread::yield()或C++20的std::atomic_wait，降低CPU占用。
• 确保Quote类的移动/拷贝操作线程安全且无异常，避免破坏队列状态。

内容的提问来源于stack exchange，提问作者Andrey Rubliov