c++ - pthread_mutex_wait 多个生产者和消费者

Question

我正在浏览 this和 this关联。基本上，他们在讨论为什么经典的单线程生产者消费者设计(使用 signal 和 wait 不适用于多生产者场景)。我有一个疑问一直困扰着我 -

作者提出的论点
考虑引用代码

char queue[MAX];  //global
int head = 0, tail = 0; //global
struct cv not_full, not_empty;
struct lock qlock;

void produce(char data) {
  acquire(&qlock);
  if ((head + 1) % MAX  ==  tail) {
    wait(&not_full, &qlock);   //line 1
  }
  queue[head] = data; //line 2
  head = (head + 1) % MAX;
  notify(&not_full);
  release(&qlock);
}

char consume(void) {
  acquire(&qlock);
  if (tail == head) {
    wait(&not_empty, &qlock);
  }
  e = queue[tail];
  tail = (tail + 1) % MAX;
  notify(&not_empty);
  release(&qlock);
  return e;
}

在上面的代码中，如果有两个生产者，第 1 行 将被单个消费者在两个生产者线程中唤醒(当队列未满时)。所以两个生产者都可以添加到队列导致队列溢出。

我的疑问
A。我们用互斥锁保护队列。因此，即使 wait 在多个生产者线程上被唤醒，只有一个生产者仍然拥有互斥锁——因此从逻辑上讲，只有一个生产者“拥有”添加到队列的权利。因为当我们从 wait 出来时，我们将获得互斥量。

告诫
我使用 POSIX mutex,cond var 作为引用。但是，我没有看到使用 POSIX 编写的文章作为引用标准。

问题
我对 wait 特别是 pthread_cond_wait 的理解是否正确？并且在多个生产者的情况下，代码的完整性仍然得到保持。

Answer 1

作者在其文章末尾 wait() 和 notify() 的语义 下提到了

notify(cv) wakes up all threads that are currently waiting on that cv

所以你对wait的理解是不正确的，notify在posix中应该是pthread_cond_broadcast。

此外 documentation of pthread_cond_signal规定

The pthread_cond_signal() call unblocks at least one of the threads that are blocked on the specified condition variable cond (if any threads are blocked on cond).

这仍然不同于您“只有一个生产者”的假设。

如作者所示，上述代码的完整性无法与多个生产者保持一致。

编辑:

pseudocode of a condition variable wait 可能看起来像

void wait (condition *cv, mutex *mx) 
{
    mutex_acquire(&c->listLock);  /* protect the queue */
    enqueue (&c->next, &c->prev, thr_self()); /* enqueue */
    mutex_release (&c->listLock); /* we're done with the list */

    /* The suspend and release_mutex() operation should be atomic */
    release_mutex (mx));
    thr_suspend (self);  /* Sleep 'til someone wakes us */

    <-------- notify executes somewhere else

    mutex_acquire (mx); /* Woke up -- our turn, get resource lock */
    return;
}

在 signal 期间，队列中至少有一个处于 suspend 状态的线程被唤醒。但是 pthread 不保证它只会是一个。现在它们可以运行了。但是他们仍然需要获取锁来保证彼此之间的互斥。

所以当第一个释放互斥锁时，第二个释放它，依此类推。

这意味着被唤醒的生产者将一个接一个地执行

queue[head] = data; //line 2
head = (head + 1) % MAX;
notify(&not_full);
release(&qlock);