c++ - 如何将第二个消费者添加到基于 pthread 的生产者-消费者设置中？

Question

我目前有两个线程用于生产者-消费者设置，它使用 pthread_cond_wait() 和 pthread_cond_signal() 在读取数据和处理数据之间交替。

假设我有一个锁、两个条件和一个 bool 标志，用于说明数据缓冲区中是否有数据:

pthread_mutex_t lock;
pthread_cond_t we_have_data;
pthread_cond_t we_need_data;
bool buffer_is_empty = true;

我有一个 pthread_t，它使用以下函数来生成数据(将数据读入缓冲区):

static void* produce(void* arg) {
    pthread_mutex_lock(&lock);
    for (;;) {
        while (!buffer_is_empty) {
            pthread_cond_wait(&we_need_data, &lock);
        }
        pthread_mutex_unlock(&lock);
        // read some data into our buffer
        pthread_mutex_lock(&lock);
        buffer_is_empty = false;
        pthread_cond_signal(&we_have_data);
    }
}

然后我有第二个 pthread_t，它在收到 we_have_data 信号时使用以下代码来使用该数据:

static void* consume(void* arg) {
    pthread_mutex_lock(&lock);
    for (;;) {
        while (buffer_is_empty) {
            pthread_cond_wait(&we_have_data, &lock);
        }
        pthread_mutex_unlock(&lock);
        // process the data in our buffer
        pthread_mutex_lock(&lock);
        buffer_is_empty = true;
        pthread_cond_signal(&we_need_data);
    }
}

这可以正常工作。

我现在想做的是添加第三个线程，如果缓冲区包含某些数据，它会处理来自 consume() 函数的数据。

我曾尝试添加第三个条件，但我的程序挂起。

我设置了一个条件和 bool 标志:

bool processing_with_second_consumer;
pthread_cond_t we_need_to_process_data_with_another_consumer;

然后我修改消费者:

static void* consume(void* arg) {
    pthread_mutex_lock(&lock);
    for (;;) {
        while (buffer_is_empty && !processing_with_second_consumer) {
            pthread_cond_wait(&we_have_data, &lock);
        }
        pthread_mutex_unlock(&lock);
        // process the data in our buffer
        pthread_mutex_lock(&lock);
        if (data_meets_our_conditions) {
            processing_with_second_consumer = true;
            pthread_cond_signal(&we_need_to_process_data_with_another_consumer);
        }
        buffer_is_empty = true;
        pthread_cond_signal(&we_need_data);
    }
}

然后我修改生产者以等待 bool 值:

static void* produce(void* arg) {
    pthread_mutex_lock(&lock);
    for (;;) {
        while (!buffer_is_empty && !processing_with_second_consumer) {
            pthread_cond_wait(&we_need_data, &lock);
        }
        pthread_mutex_unlock(&lock);
        // read some data into our buffer
        pthread_mutex_lock(&lock);
        buffer_is_empty = false;
        pthread_cond_signal(&we_have_data);
    }
}

并添加第三个线程从消费者消费:

static void* consume_from_the_consumer(void* arg) {
    pthread_mutex_lock(&lock);
    for (;;) {
        while (!buffer_is_empty && processing_with_second_consumer) {
            pthread_cond_wait(&we_need_to_process_data_with_another_consumer, &lock);
        }
        pthread_mutex_unlock(&lock);
        // do more specific processing of the data in our buffer
        pthread_mutex_lock(&lock);
        processing_with_second_consumer = false;
    }
}

我似乎无法让程序正确退出 - 它基本上卡在消费者消费时的无限循环中。

如何使用 pthread 条件正确设置信号，以允许第三个(或第四个、第五个等)线程？

Answer 1

您的生产者仅发出 we_have_data 信号。但由于它将 buffer_is_empty 设置为 false，它可以使 consume_from_the_consumer 线程准备就绪，但它不会解除阻塞，因为它在第二个条件变量上被阻塞。

为了让您的生活更简单，我建议进行两项更改:

1. 始终使用 pthread_cond_broadcast。
2. 只使用一个条件变量。

这可能会稍微降低效率，但它可以避免出现几类完整的细微错误。