c++ - std::condition_variable 多次调用 notify_all

Question

首先，让我向您介绍一下我的问题。

我的代码是这样的:

#include <iostream>
#include <thread>
#include <condition_variable>

std::mutex mtx;
std::mutex cvMtx;
std::mutex mtx2;
bool ready{false};
std::condition_variable cv;
int threadsFinishedCurrentLevel{0};

void tfunc() {
  for(int i = 0; i < 5; i++) {
    //do something
    for (int j = 0; j < 10000; j++) {
      std::cout << j << std::endl;
    }
    //this is i-th level
    mtx2.lock();
    threadsFinishedCurrentLevel++;
    if (threadsFinishedCurrentLevel == 2) {
        //this is last thread in current level
        threadsFinishedCurrentLevel = 0;
        cvMtx.unlock();
    }
    mtx2.unlock();
    {
      //wait for notify
      unique_lock<mutex> lck(mtx);
      while (!ready) cv_.wait(lck);
    }
  }
}

int main() {
  cvMtx.lock(); //init

  std::thread t1(tfunc);
  std::thread t2(tfunc);

  for (int i = 0; i < 5; i++) {
    cvMtx.lock();
    {
      unique_lock<mutex> lck(mtx);
      ready = true;
      cv.notify_all();
    }
  }
  t1.join();
  t2.join();
  return 0;
}

我有 2 个线程。我的计算由级别组成(对于这个例子，假设我们有 5 个级别)。在同一层次上，计算可以划分为线程。然后每个线程计算问题的一部分。当我想进入下一个(更高)级别时，必须先完成较低级别。所以我的想法是这样的。当当前级别的最后一个线程完成时，它会解锁主线程，因此它可以通知所有线程继续到下一个级别。但是必须多次调用此通知。因为有很多这样的级别。这个 condition_variable 可以重新启动吗？或者我是否需要为每个级别设置一个条件变量？因此，例如，当我有 1000 个级别时，我需要动态分配 1000x condition_variable？

Answer 1

是我一个人还是您正在尝试使用互斥锁阻塞主线程(这是您尝试在所有线程完成时通知它的方式？)，我的意思是这不是互斥锁的任务。这就是应该使用条件变量的地方。

// New condition_variable, to nofity main thread when child is done with level
std::condition_variable cv2;

// When a child is done, it will update this counter
int counter = 0; // This is already protected by cvMtx, otherwise it could be atomic.

// This is to sync cout
std::mutex cout_mutex;

void tfunc()
{
    for (int i = 0; i < 5; i++)
    {
        {
            std::lock_guard<std::mutex> l(cout_mutex);
            std::cout << "Level " << i + 1 << " " << std::this_thread::get_id() << std::endl;
        }

        {
            std::lock_guard<std::mutex> l(cvMtx);
            counter++; // update counter &
        }

        cv2.notify_all(); // notify main thread we are done.

        {
            //wait for notify
            unique_lock<mutex> lck(mtx);
            cv.wait(lck);
            // Note that I've removed the "ready" flag here
            // That's because u would need multiple ready flags to make that work
        }
    }
}

int main()
{
    std::thread t1(tfunc);
    std::thread t2(tfunc);

    for (int i = 0; i < 5; i++)
    {
        {
            unique_lock<mutex> lck(cvMtx);

            // Wait takes a predicate which u can take advantage of
            cv2.wait(lck, [] { return (counter == 2); });
            counter = 0;

            // This thread will get notified multiple times
            // But it only will wake up when counter matches 2
            // Which equals to how many threads we've created.              
        }

        // Sleeping a bit to know the code is working
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));

        // Wake up all threds and continue to next level.
        unique_lock<mutex> lck(mtx);
        cv.notify_all();
    }

    t1.join();
    t2.join();
    return 0;
}