1人のプロデューサーと複数のコンシューマー

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; 
    } 

以下は、私が受け取っている出力です。ただし、2番目のコンシューマスレッドは終了せず、無限ループに入ります。誰かがエラーを特定する際に助けてくれると助かります。ありがとう。

 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

最初に、ミューテックスと条件変数の初期化に失敗しました。それらはグローバルであるため、初期状態はその理由で不確定ではありませんが、どちらも必ずしも有効ではありません。適切な初期化関数を使用するか、目的の初期化マクロを使用して初期化する必要があります。例えば、

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

あなたはその問題に出luckingように見えるが、それは、あなたがそれを修正するべきではないという意味ではありません。

第2に、エラーコードの関数呼び出しの戻り値をチェックしません。あなたは本当にあなたのコードが堅牢であるようにする必要があります。エラーが発生するのは、コードが壊れていることと、実行時に問題が発生していることが原因です。関数呼び出しが常に成功すると仮定すると、遅かれ早かれ悲しいことになります。

しかし、より大きな問題があります。pthread_cond_signal() wakes スレッドが実際に待機している場合は、指定された条件変数で待機しています。プロデューサが最後にCVに信号を送るときに、両方の消費者スレッドがブロックされる可能性があります。その場合、処理を開始して処理を実行しますが、もう一方はブロックされたままです。コンシューマは、待機状態から復帰した後に条件述語を適切にチェックするため、代わりにpthread_cond_broadcast()を使用してその問題を解決できます。

しかし、それは半分の解決策です。実際には適切な述語チェックを実行し、消費者はミューテックスを解放する前に自身の述語が偽であるように共有状態を更新するので、2番目のコンシューマは待機から復帰した場合にも待機を再開します。また、それが待って再開しなければ何をするだろうか？それが消費するための利用可能な価値はなく、それに代わる経路はありません。

ボトムライン：

• あなたのプロデューサーは、少なくともその最後のアイテムを製造した後代わりに、それへのシグナリングの消費者のCVに放送しなければなりません。
• 消費者がCVを待って目を覚ますとき、値が利用可能かどうかだけでなく、プロデューサが完了したかどうかをチェックする必要があります。いずれの状況でも待機を再開してはなりませんが、実際に利用可能な場合にのみ値を消費する必要があります。