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;
}

在上面的代码中，如果有两个生产者，line 1 将在单个消费者的两个生产者线程中为woken up（当队列未满时）。所以两个生产者都可以添加到队列中导致队列溢出。

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

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

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

Answer 1

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

notify(cv) 唤醒 所有 当前在该 cv 上等待的线程

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

另外documentation of pthread_cond_signal规定

pthread_cond_signal() 调用解除阻塞至少一个线程 在指定的条件变量 cond 上被阻塞（如果有 线程在条件下被阻塞）。

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

如作者所示，以上代码的完整性并没有由多个生产者维护。

编辑：

pseudocode of a condition variablewait 可能看起来像

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);