单一生产者和多个消费者

Question

我正在尝试使用一个场景,其中生产者在缓冲区(state.value)中生成一个值,并且多个消费者正在读取缓冲区并在数组中更新它.下面是代码.

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


    pthread_mutex_t mutex;
    pthread_cond_t prod, cons;

    static int count = 0;
    struct shared_state{
    int done;
    int value;
    int value_available;
    int *array;
    int j;
    }state;

   void * producer(void *arg){
      int a[] = {12,11,10,9,8,7};
      int size = sizeof(a)/sizeof(a[0]);
      int i = 0;
      while(i<size){
      pthread_mutex_lock(&mutex);
      while(state.value_available)
        pthread_cond_wait(&prod, &mutex);

      state.value = a[i++];
      printf("In producer: %d\n",state.value);
      state.value_available = 1;
      pthread_cond_signal(&cons);
      pthread_mutex_unlock(&mutex);
    }
      state.done = 1;
      count++;
      printf("Producer count: %d\n",count);
      pthread_exit(NULL);
    }


    void * consumer(void *arg){

       while(!(state.done)){
       pthread_mutex_lock(&mutex);
       while(!state.value_available)
       pthread_cond_wait(&cons,&mutex);
       state.array[(state.j)] = state.value;
       printf("In consumer: %d\t%d\n",state.array[state.j], state.j);
       (state.j)++;
       state.value_available = 0;
       pthread_cond_signal(&prod);
       pthread_mutex_unlock(&mutex);
     }
     int i; 
     for(i=0;i<6;i++)
     printf("%d-->",state.array[i]);
     printf("\n");  
     count++;
     printf("Consumer count: %d\n",count);
   }

   int main(void){

     state.done = 0;
     pthread_t pro,con,con2;
     state.value_available = 0;
     state.j = 0;
     state.array = (int *)malloc(sizeof(int)*6);
     pthread_create(&pro, NULL, producer, (void *)NULL);
     pthread_create(&con, NULL, consumer, (void *)NULL);
     pthread_create(&con2, NULL, consumer, (void *)NULL);
     pthread_join(pro,NULL);
     pthread_join(con,NULL);
     pthread_join(con2,NULL);
     pthread_exit(NULL);
     printf("\n");
     return 0;
   }  

以下是我收到的输出.但是,第二个使用者线程不会退出并进入无限循环.如果有人可以帮助我识别错误,那将会很有帮助.谢谢.

     In producer: 12
     In consumer: 12    
     In producer: 11
     In consumer: 11
     In producer: 10
     In consumer: 10
     In producer: 9
     In consumer: 9 
     In producer: 8
     In consumer: 8
     In producer: 7
     Producer count: 1
     In consumer: 7
     Consumer array: 12-->11-->10-->9-->8-->7-->
     Consumer count: 2

Answer 1

首先,您无法初始化互斥锁和条件变量.因为它们是全局的,它们的初始状态不是因为这个原因不确定,但它也不一定是有效的.您必须使用适当的初始化函数或使用for-purpose初始化宏进行初始化.例如,

pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t prod = PTHREAD_COND_INITIALIZER;
pthread_cond_t cons = PTHREAD_COND_INITIALIZER;

你似乎对这个问题感到幸运,但这并不意味着你不应该修复它.

在第二位,您不检查函数调用的返回值是否有错误代码.你必须这样做才能让你的代码变得健壮.由于代码损坏以及由于不受控制的运行时问题而导致错误,如果您认为函数调用将始终成功,那么您迟早会感到悲伤.

然而,你有一个更大的问题:pthread_cond_signal()唤醒一个线程等待给定的条件变量,如果确实有任何等待.当生产者最后一次发出CV信号时,两个消费者线程很可能会被阻塞.在这种情况下,一个人醒来并执行其处理,但另一个仍然被阻止.因为消费者在等待唤醒后对条件谓词进行了适当的检查,所以您可以通过使用pthread_cond_broadcast()来解决该问题.

但这只是解决方案的一半.因为您确实执行了正确的谓词检查,并且因为消费者将在释放互斥锁之前更新共享状态以使其自己的谓词为假,所以如果第二个消费者确实从等待中醒来,它将继续等待.此外,如果它没有恢复等待会怎么办？它没有可供使用的价值,也没有其他可供选择的途径.

底线:

• 您的制作人必须向消费者的简历广播,而不是向其发出信号,至少在制作完最后一个项目之后.
• 当消费者在等待CV时醒来时,它不仅要检查一个值是否可用,还要检查生产者是否完成.在任何一种情况下都不能恢复等待,但只有在实际可用的情况下才必须消耗一个值.