基于互斥锁和信号量实现的类MPSC队列出队时触发空队列错误的问题排查

嘿，我来帮你拆解这个MPSC队列的问题——你说消费者最终会触发空队列的断言失败，咱们一步步找根源：

问题背景 #

你想要用互斥锁和信号量实现一个类MPSC（多生产者单消费者）的队列，但运行后发现消费者会尝试从空队列中出队，触发assert(!is_queue_empty(queue))的错误。

最小复现场景 #

你准备了一个复现示例：5个生产者，每个把自己的ID入队4次，队列容量设为4，消费者循环20次执行出队操作。代码如下：

#include <assert.h>
#include <pthread.h>
#include <semaphore.h>
#include <stdbool.h>
#include <stdlib.h>

typedef struct {
        int capacity;
        int* data;
        int length;
        int offset;
} Queue;

typedef struct {
        int id;
        Queue* queue;
        pthread_mutex_t* queue_lock;
        sem_t* queue_sem;
} ProducerArgs;

int dequeue(Queue*);
void enqueue(Queue*, int);
void init_queue(Queue*, int);
bool is_queue_empty(Queue*);
bool is_queue_full(Queue*);
void* producer(void*);

int main(int argc, char* argv[]) {
        pthread_mutex_t queue_lock;

        sem_t queue_sem;
        sem_init(&queue_sem, 0, 0);

        Queue queue;
        init_queue(&queue, 4);

        ProducerArgs args[5];
        pthread_t producers[5];
        for (int i = 0; i < 5; i++) {
                args[i].id = i;
                args[i].queue = &queue;
                args[i].queue_lock = &queue_lock;
                args[i].queue_sem = &queue_sem;
                pthread_create(&producers[i], NULL, producer, (void*)&args[i]);
        }

        for (int j = 0; j < 20; j++) {
                sem_wait(&queue_sem);
                pthread_mutex_lock(&queue_lock);
                int value = dequeue(&queue);
                pthread_mutex_unlock(&queue_lock);
                (void)value;
        }

        return 0;
}

int dequeue(Queue* queue) {
        assert(!is_queue_empty(queue));
        int value = queue->data[queue->offset];
        queue->offset = (queue->offset + 1) % queue->capacity;
        queue->length--;
        return value;
}

void enqueue(Queue* queue, int value) {
        assert(!is_queue_full(queue));
        queue->data[(queue->offset + queue->length) % queue->capacity] = value;
        queue->length++;
}

bool is_queue_empty(Queue* queue) {
        return queue->length == 0;
}

bool is_queue_full(Queue* queue) {
        return queue->length == queue->capacity;
}

void init_queue(Queue* queue, int capacity) {
        queue->capacity = capacity;
        queue->data = malloc(sizeof(int) * capacity);
        queue->length = 0;
        queue->offset = 0;
}

void* producer(void* ptr) {
        ProducerArgs* args = (ProducerArgs*)ptr;

        int completed = 0;
        while (completed < 4) {
                pthread_mutex_lock(args->queue_lock);
                if (is_queue_full(args->queue)) {
                        pthread_mutex_unlock(args->queue_lock);
                } else {
                        enqueue(args->queue, args->id);
                        pthread_mutex_unlock(args->queue_lock);
                        sem_post(args->queue_sem);
                        completed++;
                }
        }
        return NULL;
}

你提到信号量初始值设为0，认为这样能保证消费者必须等生产者调用sem_post后才能出队，但实际还是出了问题——你的核心理解是对的，但代码里有两个致命的问题：

问题根源分析 #

1. 互斥锁未初始化，导致临界区完全失效 #

你在main函数里只声明了pthread_mutex_t queue_lock;，但没有调用pthread_mutex_init(&queue_lock, NULL);初始化互斥锁！

未初始化的互斥锁状态是未定义的，这意味着多个生产者可能同时进入临界区（也就是pthread_mutex_lock之后的代码块），同时修改队列的length、offset等状态。比如：

• 两个生产者同时判断队列未满，同时执行enqueue操作，导致queue->length被错误地累加（比如两个线程同时执行length++，结果只加了1），或者两个生产者写入同一个队列位置，覆盖了之前的元素。
• 这种竞态条件会导致sem_post的次数和队列中实际存在的元素数量不一致——比如sem_post被调用了20次，但实际队列里只有19个有效元素，那么消费者第20次sem_wait后去出队，队列已经空了，直接触发断言。

2. 生产者队列满时无等待逻辑，自旋浪费资源且加剧竞态 #

当队列满的时候，你的生产者只是解锁后立刻回到循环开头，再次尝试加锁检查队列——这会导致生产者疯狂自旋，占用CPU资源，同时因为频繁的锁竞争，更容易触发上面提到的临界区竞态问题。

修复方案 #

步骤1：初始化互斥锁，并添加资源清理 #

在main函数里，声明互斥锁后立刻初始化，程序结束前销毁锁、信号量，释放队列内存：

int main(int argc, char* argv[]) {
        pthread_mutex_t queue_lock;
        // 初始化互斥锁
        pthread_mutex_init(&queue_lock, NULL);

        sem_t queue_sem;
        sem_init(&queue_sem, 0, 0);
        // 新增：空闲信号量，初始值为队列容量，表示可用的空闲位置
        sem_t empty_sem;
        sem_init(&empty_sem, 0, 4);

        Queue queue;
        init_queue(&queue, 4);

        ProducerArgs args[5];
        pthread_t producers[5];
        for (int i = 0; i < 5; i++) {
                args[i].id = i;
                args[i].queue = &queue;
                args[i].queue_lock = &queue_lock;
                args[i].queue_sem = &queue_sem;
                args[i].empty_sem = &empty_sem; // 传入空闲信号量
                pthread_create(&producers[i], NULL, producer, (void*)&args[i]);
        }

        for (int j = 0; j < 20; j++) {
                sem_wait(&queue_sem);
                pthread_mutex_lock(&queue_lock);
                int value = dequeue(&queue);
                pthread_mutex_unlock(&queue_lock);
                // 释放一个空闲位置
                sem_post(&empty_sem);
                (void)value;
        }

        // 资源清理
        for (int i = 0; i < 5; i++) {
                pthread_join(producers[i], NULL);
        }
        pthread_mutex_destroy(&queue_lock);
        sem_destroy(&queue_sem);
        sem_destroy(&empty_sem);
        free(queue.data);

        return 0;
}

步骤2：修改ProducerArgs结构体，添加空闲信号量指针 #

typedef struct {
        int id;
        Queue* queue;
        pthread_mutex_t* queue_lock;
        sem_t* queue_sem;
        sem_t* empty_sem; // 新增空闲信号量指针
} ProducerArgs;

步骤3：修改生产者逻辑，去掉自旋 #

生产者现在不需要自旋检查队列是否满，直接通过信号量阻塞等待：

void* producer(void* ptr) {
        ProducerArgs* args = (ProducerArgs*)ptr;

        int completed = 0;
        while (completed < 4) {
                // 先等待空闲位置
                sem_wait(args->empty_sem);
                pthread_mutex_lock(args->queue_lock);
                enqueue(args->queue, args->id);
                pthread_mutex_unlock(args->queue_lock);
                // 通知消费者有新元素
                sem_post(args->queue_sem);
                completed++;
        }
        return NULL;
}

修复后的效果 #

这样修改后：

• 互斥锁正确初始化，保证了队列状态修改的原子性，不会再出现竞态条件。
• 生产者在队列满时会阻塞等待，不会自旋浪费CPU。
• 信号量的sem_post和sem_wait次数严格对应队列的元素数量和空闲位置数量，消费者不会再遇到空队列的情况。

备注：内容来源于stack exchange，提问作者Kungfunk