read/writes 到共享变量 b/w pthread 未同步

Question

我正在尝试使用 pthreads 实现一个简单的 producer/consumer 代码。生产者和消费者线程之间唯一共同的共享数据是 count 变量，用于计算共享数组中可用元素的数量。发生的事情是在一个线程中更新的 count 没有反映在其他线程中。如何确保一个线程中对 count 的写入也出现在其他线程中？我错过了什么吗？

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

#define ARRAY_SIZE 100

int array[ARRAY_SIZE];
volatile int count;
int head;
int tail;

pthread_cond_t full = PTHREAD_COND_INITIALIZER;
pthread_cond_t empty = PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

void *producer(void *args)
{
        int res = 0;
        while (1) {
                pthread_mutex_lock(&mutex);
                if (count == ARRAY_SIZE) {
                        printf("\nNo space for new items waiting for consumer to consume");
                        pthread_cond_wait(&empty, &mutex);

                        // Sometimes, why is count variable still ARRAY_SIZE.
                        // How do I make sure writes to 'count' variable in
                        // consumer thread is visible immediately in producer
                        // thread?

                        if (count == ARRAY_SIZE) {
                                printf("\ncount is still ARRAY_SIZE");
                                exit(0);
                        }
                }

                head %= ARRAY_SIZE;
                count++;
                array[head] = head;
                printf("\nproduced %d/%d", head, count);
                head++;
                pthread_mutex_unlock(&mutex);
                pthread_cond_signal(&full);
        }
}

void *consumer(void *args)
{
        int res = 0;
        while (1) {
                pthread_mutex_lock(&mutex);
                if (count == 0) {
                        printf("\nNo items available waiting for producer to produce");
                        pthread_cond_wait(&full, &mutex);

                        // Sometimes, why is count variable still zero. How do I
                        // make sure writes to 'count' variable in producer
                        // thread is visible immediately in consumer thread?

                        if (count == 0) {
                                printf("\ncount is still zero");
                                exit(0);
                        }
                }

                tail %= ARRAY_SIZE;
                int ele = array[tail];
                count--;
                printf("\nconsumed %d/%d", tail, count);
                tail++;
                pthread_mutex_unlock(&mutex);
                pthread_cond_signal(&empty);
        }
}

int main()
{
        pthread_t producer_thread;
        pthread_t consumer_thread;
        int ret = 0;

        setbuf(stdout, NULL);

        ret = pthread_create(&producer_thread, NULL, producer, NULL);
        if (ret != 0) {
                printf("\nUnable to create producer thread %d", ret);
                goto exit;
        }

        ret = pthread_create(&consumer_thread, NULL, consumer, NULL);
        if (ret != 0) {
                printf("\nUnable to create consumer thread %d", ret);
                goto exit;
        }

        pthread_join(producer_thread, NULL);
        pthread_join(consumer_thread, NULL);

exit:
        return ret;
}

produced 72/99
produced 73/100
No space for new items waiting for consumer to consume
consumed 74/99
consumed 75/98
consumed 76/97
consumed 77/96
produced 74/97
produced 75/98
produced 76/99
produced 77/100
No space for new items waiting for consumer to consume
count is still ARRAY_SIZE   <------ incorrect

consumed 21/2
consumed 22/1
consumed 23/0
No items available waiting for producer to produce
produced 24/1
consumed 24/0
No items available waiting for producer to produce
produced 25/1
produced 26/2
produced 27/3
consumed 25/2
consumed 26/1
consumed 27/0
No items available waiting for producer to produce
count is still zero   <------ incorrect

Zan Lynx 修复后有效的解决方案

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

#define ARRAY_SIZE 100

int array[ARRAY_SIZE];
volatile int count;
int head;
int tail;

pthread_cond_t full = PTHREAD_COND_INITIALIZER;
pthread_cond_t empty = PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

void *producer(void *args)
{
        int res = 0;
        while (1) {
                pthread_mutex_lock(&mutex);
                if (count == ARRAY_SIZE) {
                        printf("\nNo space for new items waiting for consumer to consume");

                        // Spurious wakeups from the pthread_cond_timedwait() or
                        // pthread_cond_wait() functions may occur. Since the
                        // return from pthread_cond_timedwait() or
                        // pthread_cond_wait() does not imply anything about the
                        // value of this predicate, the predicate should be
                        // re-evaluated upon such return.

                        while (count == ARRAY_SIZE)
                                pthread_cond_wait(&empty, &mutex);
                }

                head %= ARRAY_SIZE;
                count++;
                array[head] = head;
                printf("\nproduced %d/%d", head, count);
                head++;
                pthread_mutex_unlock(&mutex);
                pthread_cond_signal(&full);
        }

        return NULL;
}

void *consumer(void *args)
{
        int res = 0;
        while (1) {
                pthread_mutex_lock(&mutex);
                if (count == 0) {
                        printf("\nNo items available waiting for producer to produce");

                        // Spurious wakeups from the pthread_cond_timedwait() or
                        // pthread_cond_wait() functions may occur. Since the
                        // return from pthread_cond_timedwait() or
                        // pthread_cond_wait() does not imply anything about the
                        // value of this predicate, the predicate should be
                        // re-evaluated upon such return.

                        while (count == 0)
                                pthread_cond_wait(&full, &mutex);
                }

                tail %= ARRAY_SIZE;
                int ele = array[tail];
                count--;
                printf("\nconsumed %d/%d", tail, count);
                tail++;
                pthread_mutex_unlock(&mutex);
                pthread_cond_signal(&empty);
        }

        return NULL;
}

int main()
{
        pthread_t producer_thread;
        pthread_t consumer_thread;
        int ret = 0;

        setbuf(stdout, NULL);

        ret = pthread_create(&producer_thread, NULL, producer, NULL);
        if (ret != 0) {
                printf("\nUnable to create producer thread %d", ret);
                goto exit;
        }

        ret = pthread_create(&consumer_thread, NULL, consumer, NULL);
        if (ret != 0) {
                printf("\nUnable to create consumer thread %d", ret);
                goto exit;
        }

        pthread_join(producer_thread, NULL);
        pthread_join(consumer_thread, NULL);

exit:
        return ret;
}

Answer 1

我相信您错过了条件等待必须始终在等待后再次检查谓词的事实 returns。必有循环。

除 signal/notify 调用外，等待可能因各种原因而结束。

read/writes 到共享变量 b/w pthread 未同步

read/writes to shared variable b/w pthread not synchronized

multithreading

pthreads