multithreading - 使用互斥锁和条件变量的线程同步

Question

我正在尝试实现一个多线程作业，一个生产者和一个消费者，基本上我想做的是，当消费者完成数据时，它通知生产者，以便生产者提供新数据。

棘手的部分是，在我目前的 impl 中，生产者和消费者都相互通知并相互等待，我不知道如何正确实现这部分。

例如，看下面的代码，

mutex m;
condition_variable cv;

vector<int> Q;  // this is the queue the consumer will consume
vector<int> Q_buf;  // this is a buffer Q into which producer will fill new data directly

// consumer
void consume() {
  while (1) {
    if (Q.size() == 0) {  // when consumer finishes data
      unique_lock<mutex> lk(m);
      // how to notify producer to fill up the Q?
      ...
      cv.wait(lk);
    }

    // for-loop to process the elems in Q
    ...
  }
}

// producer
void produce() {
  while (1) {
    // for-loop to fill up Q_buf
    ...

    // once Q_buf is fully filled, wait until consumer asks to give it a full Q
    unique_lock<mutex> lk(m);
    cv.wait(lk);
    Q.swap(Q_buf);  // replace the empty Q with the full Q_buf
    cv.notify_one();
  }
}

我不确定这是上面使用 mutex 的代码和 condition_variable是实现我的想法的正确方法，
请给我一些建议!

Answer 1

该代码错误地假定 vector<int>::size()和 vector<int>::swap()是原子的。他们不是。

另外，spurious wakeups必须由 while 处理循环(或另一个 cv::wait 过载)。

修复:

mutex m;
condition_variable cv;
vector<int> Q;

// consumer
void consume() {
    while(1) {
        // Get the new elements.
        vector<int> new_elements;
        {
            unique_lock<mutex> lk(m);
            while(Q.empty())
                cv.wait(lk);
            new_elements.swap(Q);
        }
        // for-loop to process the elems in new_elements
    }
}

// producer
void produce() {
    while(1) {
        vector<int> new_elements;
        // for-loop to fill up new_elements

        // publish new_elements
        {
            unique_lock<mutex> lk(m);
            Q.insert(Q.end(), new_elements.begin(), new_elements.end());
            cv.notify_one();
        }
    }
}