如何用两个互斥体同步生产者消费者问题？

Question

我目前正在为我的大学做家庭作业。 我们应该使循环缓冲区线程安全，给定一个带有两个互斥体和两个信号的 struct rbuf。这两个互斥体可以追溯到两个写入器线程和两个读取器线程始终处于活动状态的事实。到目前为止，读取和写入功能运行得很好，因为所有未事先测试线程安全的测试都是成功的。它们基本上只负责写入缓冲区和从缓冲区读取。我们还应该使用

pthread_cond_timedwait

。我当前的实现总是存在数据竞争（我知道这一点要感谢 ThreadSanitizer），不幸的是我完全不知道如何更好地同步它。

这是我到目前为止的代码，我所做的对我来说非常有意义：

/*
typedef struct {
    uint8_t* read;
    uint8_t* write;
    uint8_t* begin;
    uint8_t* end;
    pthread_mutex_t mutex_read;
    pthread_mutex_t mutex_write;
    pthread_cond_t signal_read;
    pthread_cond_t signal_write;
} rbuf;
*/

int producer(rbuf *buffer, void *message, size_t message_len) {
    int val;
    struct timespec to;
    while (1) {
        pthread_mutex_lock(&buffer->mutex_write);
        clock_gettime(CLOCK_REALTIME, &to);
        to.tv_sec += 1;
        retval = write(buffer, message, message_len);
        if (val == 1 //SUCCESS) {
            pthread_cond_signal(&buffer->signal_read);
            pthread_mutex_unlock(&buffer->mutex_write);
            return retval;
        }
        if (pthread_cond_timedwait(&buffer->signal_write, &buffer->mutex_write, &to) == ETIMEDOUT) {
            pthread_mutex_unlock(&buffer->mutex_write);
            return val;
        }
    }
}

int consumer(rbuf *buffer, void *buffer_read, size_t *buffer_len) {
    int val;
    struct timespec to;
    while (1) {
        pthread_mutex_lock(&buffer->mutex_read);
        clock_gettime(CLOCK_REALTIME, &to);
        to.tv_sec += 1;
        retval = read(buffer, buffer_read, buffer_len);
        if (val == 1 //SUCCESS) {
            pthread_cond_signal(&buffer->signal_write);
            pthread_mutex_unlock(&buffer->mutex_read);
            return retval;
        }
        if (pthread_cond_timedwait(&buffer->signal_read, &buffer->mutex_read, &to) == ETIMEDOUT)  {
            pthread_mutex_unlock(&buffer->mutex_read);
            return val;
        }
        pthread_mutex_unlock(&buffer->mutex_read);
    }
}

我尝试了几种实现，但这个是迄今为止最好的。生产者写入并在写入成功时发送信号。消费者也这样做。这两个函数处于 while 循环中，因为这对我来说有意义，但我不知道这是否正确。我会非常高兴收到改进建议或如何更好地解决这个问题的建议。预先感谢您的回答。

亲切的问候！

Answer 1

目前，您正在经历数据竞争，因为生产者和消费者正在锁定不同的互斥体。我假设 read() 和 write() 函数都查看读取和写入指针。

写入缓冲区时，write()函数会检查写入指针和读取指针之间的距离，以确定是否有空间可写入，然后写入数据。然而，当这种情况发生时，读取函数可以读取一些数据并增加读取指针的值，这就是 ThreadSanitizer 所抱怨的。

您当前的代码可能会通过以下方式陷入死锁：

生产者线程尝试写入缓冲区，但发现缓冲区已满。但在它有机会等待条件变量之前，它会被抢占并休眠一段时间。当它处于睡眠状态时，读取器线程进入，读取缓冲区中的所有数据并向条件变量发出信号。条件变量不具有“粘性”。换句话说，如果没有人等待，信号就会丢失。

生产者线程随后醒来并在条件变量上休眠。由于缓冲区是空的，因此读取器线程不会醒来并发出信号表明有更多可用空间。然后生产者线程将等待条件超时。

我不确定你的教授试图让你用两个互斥体完成什么任务。使用带有两个条件变量的单个互斥体对我来说更有意义。在这种情况下，如果生产者和消费者要锁定相同的互斥体，我预计不会出现竞争条件。

教授可能试图让您进行读取器/写入器锁定，以减少某些工作负载下的争用。在这种情况下，生产者会首先锁定写锁，然后锁定读锁，而消费者只会锁定读锁，但我正在抓住这个解释的救命稻草。