在 std::thread 中等待条件 A 或条件 B

Question

我有一个看似相当简单的需求，但我是使用 std::thread 的新手，我不确定我是否理解正确。

我的线程的工作是运行一个循环：等到一个对象需要处理，然后处理它，然后等待，...

我正要使用condition_variable 来实现这一点，但我意识到当线程空闲等待新对象时，它不会注意到stopThread 标志已设置。

我实际上想要一种方法来做wait_for_either(new_data,exit_thread)，但我不确定如何优雅地实现它。类似队列函数的旧代码使用 Windows API WaitForMultipleObjects 但我想借此机会学习 C++11 方式。

Answer 1

当您等待数据处理时，您实际上是在等待条件变量发出信号。所以当你想退出线程的时候只要给条件变量发信号，就好像stopThread标志是一个特殊的要处理的数据一样。

代码可能如下所示：

void thread_func()
{
    std::unique_lock<std::mutex> lock(mutex);
    for (;;)
    {
        cond.wait(lock, [] () { return stopThread || !dataContainer.empty(); });
        if (stopThread)
            return; //exit thread
        //process data from dataContainer
    }
}

插入数据：

{
    std::unique_lock<std::mutex> lock(mutex);
    dataContainer.push_back(new_data);
    cond.notify_all();
}

然后，当你想停止线程时：

{
    std::unique_lock<std::mutex> lock(mutex);
    stopThread = true;
    cond.notify_all();
}
thread.join(); //not necessary but probably a good idea

Answer 2

这是一个带有中止选项的强大数据消耗循环：

while(true) {
  decltype(dataContainer) data;
  {
    std::unique_lock<std::mutex> lock(mutex);
    cond.wait(lock, [] () { return stopThread || !dataContainer.empty(); });
    if (stopThread)
      return; //exit thread
    data = std::move(dataContainer);
  }
  for (auto&& d:data) {
    if (stopThread) return; // abort
    //process data from d
  }
}

stopThread 应该是atomic 或者底部的for(:) 循环中的访问需要用mutex 保护。

for(:) 循环中对stopThread 的访问是可选的；没有它，它在完成它拾取的工作包之前不会中止。

dataContainer 是以std::vector 或类似的形式完成的一系列工作。线程唤醒，抓取所有待办事项，然后继续工作。

您也可以从dataContainer 弹出一项任务，而不是全部执行。生成的代码稍微简单一些。

要将数据排队到dataContainer，你必须锁定mutex，放入数据，然后通知：

{
  std::unique_lock<std::mutex> lock(mutex);
  dataContainer.push_back(new_data);
}
cond.notify_one();

关闭：

{
  std::unique_lock<std::mutex> lock(mutex);
  stopThread = true;
}
cond.notify_all();

请注意，即使stopThread 是原子的，您也需要获取互斥锁。否则存在竞争条件。

Answer 3

半 - 伪代码。

std::atomic_bool programRunning;
std::condition_variable cv;
std::mutex mtx;

std::thread loopThread([&]{
   while(programRunning.load()){
       std::unique_lock<std::mutex> lock(mtx);
        if (newDataAvailable){
            //process new data
        } else{
            cv.wait(lock,[&]{ return dataAvailable || !progamRunning.load(); });
        }
   }

});

{
   std::lock_guard<std::mutex> lock(mtx);
   queueMoreData();
   cv.notify_one();
}

//on exit:
programRunning.store(false);
cv.notify_one();