C语言线程：生产者消费者模型死锁问题求助

问题描述 #

实现C语言生产者消费者模型，程序接收生产者和消费者数量作为输入，需完成8192个元素的生产与消费操作。测试发现：

• 生产者与消费者数量相等时程序运行正常
• 数量不等时出现死锁
• 使用helgrind工具排查，提示Possible data race Locks held: none，但无法定位问题根源

原代码 #

#include <pthread.h>
#include <semaphore.h>
#include <stdio.h>
#include <stdlib.h>

pthread_mutex_t mutex;
sem_t empty;
sem_t full;
int prod=0;
int cons=0;
int buffer[8];
int in=0;
int out=0;

void producer(void* x)
{
  int z=1;
  while(z)
  {
    int item=rand();
    sem_wait(&empty);
    pthread_mutex_lock(&mutex);
    if (prod!=8192){
        buffer[in]=item;
        prod+=1;
        printf("Prod %d:item ins %d at %d\n",*((int *)x),buffer[in],in);
        in=(in+1)%8;
    }
    
    if(prod==8192){z=0;}
    pthread_mutex_unlock(&mutex);
    sem_post(&full); 
    for (int i=0; i<10000; i++);
  }
}

void consumer(void*x)
{
int z=1;
 while(z)
 {
   sem_wait(&full);
   pthread_mutex_lock(&mutex);
    if (cons!=8192){
        int item = buffer[out];
        cons+=1;
        printf("Cons %d:item rm %d at %d\n",*((int *)x),item,out);
        out =(out+1)%8;
    }
   if(cons==8192){z=0;}

   pthread_mutex_unlock(&mutex);
   sem_post(&empty);
   for (int i=0; i<10000; i++);
 }
}

int main(int argc, char const *argv[])
{
    int nbProd=atoi(argv[1]);
    int nbCons=atoi(argv[2]);
    pthread_t produc[nbProd],conso[nbCons];
    pthread_mutex_init(&mutex,NULL);
    sem_init(&empty,0,8);
    sem_init(&full,0,0);
    int idP[nbProd];
    for (int i=0; i< nbProd;i++){
        idP[i]=i;
    }
    int idC[nbCons];
    for (int i=0; i< nbCons;i++){
        idC[i]=i;
    }
    for(int i=0;i<nbProd;i++){
        pthread_create(&produc[i],NULL,(void *)producer,(void*)&idP[i]);
    }
    for(int i=0;i<nbCons;i++){
        pthread_create(&conso[i],NULL,(void *)consumer,(void*)&idC[i]);
    }
    for(int i=0;i<nbProd;i++){
        pthread_join(produc[i],NULL);
    }
    for(int i=0;i<nbCons;i++){
        pthread_join(conso[i],NULL);
    }
    pthread_mutex_destroy(&mutex);
    sem_destroy(&empty);
    sem_destroy(&full);
    printf("prod:%d,-cons:%d",prod,cons);
    return 0;
}

问题根源分析 #

1. 死锁核心原因：

   • 生产者线程在prod达到8192后，仍会执行sem_wait(&empty)占用空槽信号量，进入临界区后不执行生产操作，但依然调用sem_post(&full)错误增加满槽信号量。
   • 消费者线程同理，cons达到8192后仍会执行sem_wait(&full)占用满槽信号量，进入临界区后不执行消费操作，却调用sem_post(&empty)错误增加空槽信号量。
   • 这种行为会导致信号量状态混乱，部分线程永久阻塞在信号量等待操作上，引发死锁。

2. 数据竞争隐患：
   线程在进入信号量等待前，未通过互斥锁保护读取prod/cons的完成状态，可能出现多个线程同时进入信号量等待的情况，进一步加剧信号量状态的混乱。

修复后的代码 #

#include <pthread.h>
#include <semaphore.h>
#include <stdio.h>
#include <stdlib.h>

pthread_mutex_t mutex;
sem_t empty;
sem_t full;
int prod = 0;
int cons = 0;
int buffer[8];
int in = 0;
int out = 0;
#define TOTAL_ITEMS 8192

void producer(void* x)
{
    while (1)
    {
        pthread_mutex_lock(&mutex);
        // 先判断是否已完成全部生产，避免占用信号量
        if (prod >= TOTAL_ITEMS)
        {
            pthread_mutex_unlock(&mutex);
            break;
        }
        pthread_mutex_unlock(&mutex);

        int item = rand();
        sem_wait(&empty);
        pthread_mutex_lock(&mutex);

        // 再次判断，防止其他线程已经完成生产
        if (prod >= TOTAL_ITEMS)
        {
            pthread_mutex_unlock(&mutex);
            sem_post(&empty); // 释放已占用的信号量
            break;
        }

        buffer[in] = item;
        prod += 1;
        printf("Prod %d: item ins %d at %d\n", *((int *)x), buffer[in], in);
        in = (in + 1) % 8;

        pthread_mutex_unlock(&mutex);
        sem_post(&full);

        // 模拟耗时操作
        for (int i = 0; i < 10000; i++);
    }
}

void consumer(void* x)
{
    while (1)
    {
        pthread_mutex_lock(&mutex);
        // 先判断是否已完成全部消费，避免占用信号量
        if (cons >= TOTAL_ITEMS)
        {
            pthread_mutex_unlock(&mutex);
            break;
        }
        pthread_mutex_unlock(&mutex);

        sem_wait(&full);
        pthread_mutex_lock(&mutex);

        // 再次判断，防止其他线程已经完成消费
        if (cons >= TOTAL_ITEMS)
        {
            pthread_mutex_unlock(&mutex);
            sem_post(&full); // 释放已占用的信号量
            break;
        }

        int item = buffer[out];
        cons += 1;
        printf("Cons %d: item rm %d at %d\n", *((int *)x), item, out);
        out = (out + 1) % 8;

        pthread_mutex_unlock(&mutex);
        sem_post(&empty);

        // 模拟耗时操作
        for (int i = 0; i < 10000; i++);
    }
}

int main(int argc, char const *argv[])
{
    if (argc != 3)
    {
        printf("Usage: %s <num_producers> <num_consumers>\n", argv[0]);
        return 1;
    }

    int nbProd = atoi(argv[1]);
    int nbCons = atoi(argv[2]);
    pthread_t produc[nbProd], conso[nbCons];

    pthread_mutex_init(&mutex, NULL);
    sem_init(&empty, 0, 8);
    sem_init(&full, 0, 0);

    int idP[nbProd];
    for (int i = 0; i < nbProd; i++)
    {
        idP[i] = i;
    }
    int idC[nbCons];
    for (int i = 0; i < nbCons; i++)
    {
        idC[i] = i;
    }

    for (int i = 0; i < nbProd; i++)
    {
        pthread_create(&produc[i], NULL, (void *)producer, (void *)&idP[i]);
    }
    for (int i = 0; i < nbCons; i++)
    {
        pthread_create(&conso[i], NULL, (void *)consumer, (void *)&idC[i]);
    }

    for (int i = 0; i < nbProd; i++)
    {
        pthread_join(produc[i], NULL);
    }
    for (int i = 0; i < nbCons; i++)
    {
        pthread_join(conso[i], NULL);
    }

    pthread_mutex_destroy(&mutex);
    sem_destroy(&empty);
    sem_destroy(&full);
    printf("prod: %d, cons: %d\n", prod, cons);
    return 0;
}

修复说明 #

1. 提前判断任务完成状态：在进入信号量等待前，先通过互斥锁保护读取prod/cons，判断是否已完成全部任务，避免无效的信号量占用。
2. 双重检查机制：在获取信号量后再次判断任务状态，防止多个线程同时进入信号量等待后，其中一个线程完成了全部任务，其他线程需要释放已占用的信号量并退出。
3. 信号量释放修正：当判断任务已完成时，及时释放已占用的信号量，避免信号量计数异常。
4. 增加参数校验：main函数中增加命令行参数数量检查，提升程序健壮性。

经过上述修改后，无论生产者和消费者数量是否相等，程序都能正常完成8192个元素的生产与消费操作，不会出现死锁。

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