c - 多线程程序在完成之前卡住

Question

我编写了我的第一个多线程程序，并且它的大部分工作正常。共享缓冲区数组最初填充为 -1，指示生产者它是空的并且需要填充。然后，生产者用随机值 1 - 10 填充共享缓冲区，然后生产者轮流填充缓冲区。然后，生产者向消费者发出信号，告知其已填充缓冲区的一个元素并开始消费它。生产者需要填写 120 个元素，消费者需要填写每个条目。程序运行得非常好，直到到达第 110 项。然后它就卡住了，我不明白为什么。我该如何解决这个问题？

这是输出的片段。

 Item: 85, Consuming value 8, my thread id is: 1216
 Item: 86, Consuming value 7, my thread id is: 298320
Signal
Producer thread 231296 and value: 0
Producer thread 297552 and value: 2
Producer thread 298576 and value: 0
 Item: 87, Consuming value 9, my thread id is: 297808
Signal
Producer thread 960 and value: 3
Producer thread 298064 and value: 2
 Item: 88, Consuming value 3, my thread id is: 231744
 Item: 89, Consuming value 7, my thread id is: 298320
 Item: 90, Consuming value 3, my thread id is: 1216
 Item: 91, Consuming value 7, my thread id is: 298832
Signal
Producer thread 231296 and value: 3
Producer thread 297552 and value: 8
Producer thread 298576 and value: 6
 Item: 92, Consuming value 2, my thread id is: 297808
Signal
Producer thread 960 and value: 9
Producer thread 298064 and value: 7
 Item: 93, Consuming value 5, my thread id is: 298320
 Item: 94, Consuming value 2, my thread id is: 298832
 Item: 95, Consuming value 0, my thread id is: 1216
 Item: 96, Consuming value 2, my thread id is: 231744

这是我的代码

#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>
#include <stdint.h>


#define THREADS 5
#define ELEMENTS 120


pthread_t tid_producer[THREADS], tid_consumer[THREADS];
int value = 0;
int saveValue = 0;
void *produce(void *arg);
void *consume(void *arg);
int producerCount =0;
int consumerCount = ELEMENTS;

struct {
  pthread_mutex_t mutex;
  int index;
  int value; 
  int MyShBuff[ELEMENTS];
} add = {PTHREAD_MUTEX_INITIALIZER, 0, 0}; 


  struct{
   pthread_mutex_t    mutex;
   pthread_cond_t     cond;
   int nready;
   int value;
   int empty;
   int counter;

   /* number ready for consumer */
} nready = {PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER,0, -2, ELEMENTS};

int main()
{

    int i, j, k;

    //Ready buffer for producers
    for (i =0; i < ELEMENTS; i++)
    {
        add.MyShBuff[i]=-1;
    }



    for(j = 0; j < THREADS; j++) {

   pthread_create(&tid_producer[j], NULL, &produce, NULL);
   pthread_create(&tid_consumer[j], NULL, &consume, NULL);
}



 /* wait for all producers and the consumer*/

    for(k = 0; k < THREADS; k++) {
        pthread_join(tid_producer[k], NULL);
        pthread_join(tid_consumer[k], NULL);    
    }

     /* Clean up and exit */

  pthread_mutex_destroy(&nready.mutex);
  pthread_mutex_destroy(&add.mutex);
   pthread_cond_destroy(&nready.cond);
   pthread_exit(NULL);

    exit(0);    
    return 0;
}

 void *produce(void *arg)
    { 
    int i = 0;


    for ( ; ; ) 
    {
        pthread_mutex_lock(&add.mutex);
        if(add.index  >= ELEMENTS)
        {

            pthread_mutex_unlock(&add.mutex);
            return NULL;
        }
        if(add.MyShBuff[add.index] == -1)
        {
         add.value = rand() % 10 + 0;   
         add.MyShBuff[add.index] = add.value;
         printf("Producer thread %d and value: %d\n" ,pthread_self(), add.MyShBuff[add.index]);
         add.index++;
        }
         pthread_mutex_unlock(&add.mutex);
         pthread_mutex_lock(&nready.mutex);

         if(nready.nready == 0)
         {
         pthread_cond_broadcast(&nready.cond);
         printf("Signal\n");
         }

        nready.nready++;
        pthread_mutex_unlock(&nready.mutex);

        }


    }


void *consume(void *arg)
{

pthread_mutex_lock(&nready.mutex);

  while(nready.empty != 0)
  {

    while (nready.nready == 0)
    {


        pthread_cond_wait(&nready.cond,&nready.mutex);


         pthread_mutex_lock(&add.mutex);
         printf(" Item: %d, Consuming value %d, my thread id is: %d\n", nready.counter, add.MyShBuff[nready.counter], pthread_self());
         add.MyShBuff[nready.counter] = -2;
         pthread_mutex_unlock(&add.mutex);

        nready.counter++;   
        nready.empty--;


    }

    nready.nready--;
   pthread_mutex_unlock(&nready.mutex);
  }


  return NULL;


}

Answer 1

正如评论中指出的，您在消费中使用互斥体是错误的(想象一下 while 条件是错误的，线程将在不释放互斥体的情况下离开)此外，我建议限制保存相同信息的变量数量(空，计数器)，这使得流程难以遵循。最后，您的 pthread_wait_cond 中还有一个错误:请参阅Does pthread_cond_wait(&cond_t, &mutex); unlock and then lock the mutex?或解释 pthread_wait_cond 的使用的任何其他问题:它们强调谓词可能在返回时为真，这意味着您必须在开始任何需要谓词为真的更多计算之前再次测试它。

我重写了一些你的代码以使其工作；我希望我没有错过其他东西，但这可能会帮助您走得更远:

#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <stdlib.h>
#include <pthread.h>
#include <stdint.h>


#define THREADS 5
#define ELEMENTS 120


pthread_t tid_producer[THREADS], tid_consumer[THREADS];
int value = 0;
int saveValue = 0;
void *produce (void *arg);
void *consume (void *arg);
int producerCount = 0;
int consumerCount = ELEMENTS;

struct
{
    pthread_mutex_t mutex;
    int index;
    int value;
    int MyShBuff[ELEMENTS];
} add = { PTHREAD_MUTEX_INITIALIZER, 0, 0 };


struct
{
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    int nready;
    int value;
    int empty;
    int counter;

    /* number ready for consumer */
} nready = { PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER, 0, -2, ELEMENTS, 0 };

int main(int argc, char **argv)
{
    //Ready buffer for producers
    for (int i = 0; i < ELEMENTS; i++) {
        add.MyShBuff[i] = -1;
    }


    for (int j = 0; j < THREADS; j++) {
        pthread_create (&tid_producer[j], NULL, &produce, NULL);
        pthread_create (&tid_consumer[j], NULL, &consume, NULL);
    }

    /* wait for all producers and the consumer */
    for (int k = 0; k < THREADS; k++) {
        pthread_join (tid_producer[k], NULL);
        pthread_join (tid_consumer[k], NULL);
    }

    /* Clean up and exit */
    pthread_mutex_destroy (&nready.mutex);
    pthread_mutex_destroy (&add.mutex);
    pthread_cond_destroy (&nready.cond);

    return 0;
}

void *produce(void *dummy)
{
    pthread_mutex_lock (&add.mutex);

    while (add.index < ELEMENTS) {
        if (add.MyShBuff[add.index] == -1)
        {
            add.value = rand () % 10 + 0;
            add.MyShBuff[add.index] = add.value;
            printf("Producer thread %ld and value: %d\n" ,pthread_self(), add.MyShBuff[add.index]);
            add.index++;
        }
        pthread_mutex_unlock (&add.mutex);

        pthread_mutex_lock (&nready.mutex);
        {
            if (nready.nready == 0)
            {
                pthread_cond_broadcast (&nready.cond);
                printf ("Signal\n");
            }
            nready.nready++;
        }
        pthread_mutex_unlock (&nready.mutex);

        pthread_mutex_lock (&add.mutex);
    }

    pthread_mutex_unlock (&add.mutex);

    return NULL;
}


void *consume(void *dummy)
{
    pthread_mutex_lock (&nready.mutex);

    while (nready.empty != 0)
    {
        // you also need to check it is not time to leave...
        while (nready.nready == 0 && nready.empty != 0) {
            pthread_cond_wait (&nready.cond, &nready.mutex);
        }

        if (nready.empty == 0) {
            break;
        }

        pthread_mutex_lock (&add.mutex);
        printf(" Item: %d, Consuming value %d, my thread id is: %ld\n", nready.counter, add.MyShBuff[nready.counter], pthread_self());
        add.MyShBuff[nready.counter] = -2;

        pthread_mutex_unlock (&add.mutex);

        nready.counter++;
        nready.empty--;

        nready.nready--;
    }

    pthread_mutex_unlock (&nready.mutex);

    return NULL;
}