c - 如何让两个 pthread 线程响应彼此的等待和信号条件？

Question

我在基本的双线程安排工作方面遇到了一些麻烦。

我正在一个“生产者”线程中从 stdin 读取一大块字节到内存中，并在第二个“消费者”线程中处理这些字节(一旦这些字节可用)。一旦字节被消耗，消费者线程返回休眠状态，生产者线程再次运行。

我使用 pthread_cond_wait() 和 pthread_cond_signal() 让两个线程相互通信生成或消耗数据。

这是两个线程的代码:

void * produce_bytes(void *t_data)
{ 
    pthread_data_t *d = (pthread_data_t *)t_data;

    do {
        pthread_mutex_lock(&d->input_lock);
        d->n_bytes = fread(d->in_buf, sizeof(unsigned char), BUF_LENGTH_VALUE, stdin);
        if (d->n_bytes > 0) { 
            fprintf(stdout, "PRODUCER ...signaling consumer...\n");
            pthread_cond_signal(&d->input_cond);
            fprintf(stdout, "PRODUCER ...consumer signaled...\n");
        }
        pthread_mutex_unlock(&d->input_lock);
    } while (d->n_bytes > 0);

    return NULL;
}

void * consume_bytes(void *t_data) 
{
    pthread_data_t *d = (pthread_data_t *)t_data;

    pthread_mutex_lock(&d->input_lock);
    while (d->n_bytes == 0)
        pthread_cond_wait(&d->input_cond, &d->input_lock);
    fprintf(stdout, "CONSUMER ...consuming chunk...\n");
    d->n_bytes = 0;
    fprintf(stdout, "CONSUMER ...chunk consumed...\n");
    pthread_mutex_unlock(&d->input_lock);
}

pthread_data_t 是我用来跟踪状态的结构:

typedef struct {
    pthread_mutex_t input_lock;
    pthread_cond_t input_cond;
    unsigned char in_buf[BUF_LENGTH_VALUE];
    size_t n_bytes;
} pthread_data_t;

我在我的 main() 函数中配置变量；以下是相关摘录:

pthread_t producer_thread = NULL;
pthread_t consumer_thread = NULL;
pthread_data_t *thread_data = NULL;

thread_data = malloc(sizeof(pthread_data_t));
thread_data->n_bytes = 0;
pthread_mutex_init(&(thread_data->input_lock), NULL);
pthread_cond_init(&(thread_data->input_cond), NULL);

pthread_create(&producer_thread, NULL, produce_bytes, (void *) thread_data);
pthread_create(&consumer_thread, NULL, consume_bytes, (void *) thread_data);

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

当我运行此命令时，product_bytes() 在第一次迭代中成功发出 consume_bytes() 信号，但在第二次及后续迭代中，信号被发送到 consume_bytes() 并且永远不会被听到，因此消费者函数永远不会再次运行:

PRODUCER ...signaling consumer...
PRODUCER ...consumer signaled...
CONSUMER ...consuming chunk...
CONSUMER ...chunk consumed...
PRODUCER ...signaling consumer...
PRODUCER ...consumer signaled...
PRODUCER ...signaling consumer...
PRODUCER ...consumer signaled...
PRODUCER ...signaling consumer...
PRODUCER ...consumer signaled...
...

我正在使用教程here作为我想做的事情的基础。我做错了什么？

Answer 1

该代码存在一些问题:

1. product_bytes 在阻塞调用 fread 期间锁定互斥锁。响应式应用程序的一般经验法则是锁定互斥锁的时间尽可能短。您可能希望首先将输入读入临时缓冲区，然后锁定互斥体并将数据复制到线程之间共享的缓冲区。同样适用于 consume_bytes，它在调用可以阻塞的 fprintf 时保存互斥锁。
while(d->n_bytes > 0) 中的
2. product_bytes 不保存互斥体，这是一个竞争条件，因为 consume_bytes 分配了一个d->n_bytes 的新值。假设您想在 fread 返回 0 (EOF) 时退出该循环，则需要将 fread 的返回值复制到不在线程之间共享的局部变量中并使用该变量作为 while(read_bytes > 0)
中的条件
3. consume_bytes 周围没有任何循环，因此它会在第一个条件变量通知后返回。您可能希望将其包装到 while 循环中，并仅在读取 EOF(0 字节)时退出。