C++ 线程安全和 notify_all()

Posted 2023-02-22

【中文标题】C++ 线程安全和 notify_all()

【英文标题】：C++ Thread safety and notify_all()

【发布时间】：2016-04-22 20:50:45

【问题描述】：

真正的代码要复杂得多，但我想我设法制作了一个 mcve。

我正在尝试执行以下操作：

让一些线程工作 将它们全部置于暂停状态 唤醒他们中的第一个，等待它完成，然后唤醒第二个，等待它完成，唤醒第三个......等等。

我正在使用的代码如下，它似乎可以工作

std::atomic_int which_thread_to_wake_up;
std::atomic_int threads_asleep;
threads_asleep.store(0);
std::atomic_bool ALL_THREADS_READY;
ALL_THREADS_READY.store(false);
int threads_num = .. // Number of threads
bool thread_has_finished = false;

std::mutex mtx;
std::condition_variable cv;

std::mutex mtx2;
std::condition_variable cv2;


auto threadFunction = [](int my_index) 

  // some heavy workload here..
  ....

  
        std::unique_lock<std::mutex> lck(mtx);
        ++threads_asleep;
        cv.notify_all(); // Wake up any other thread that might be waiting
   

  std::unique_lock<std::mutex> lck(mtx);
  bool all_ready = ALL_THREADS_READY.load();
  size_t index = which_thread_to_wake_up.load();
  cv.wait(lck, [&]() 
      all_ready = ALL_THREADS_READY.load();
      index = which_thread_to_wake_up.load();
      return all_ready && my_index == index;
  );

  // This thread was awaken for work!
  .. do some more work that requires synchronization..

  std::unique_lock<std::mutex> lck2(mtx2);
  thread_has_finished = true;
  cv2.notify_one(); // Signal to the main thread that I'm done
;

// launch all the threads..
std::vector<std::thread> ALL_THREADS;
for (int i = 0; i < threads_num; ++i)
  ALL_THREADS.emplace_back(threadFunction, i);      


// Now the main thread needs to wait for ALL the threads to finish their first phase and go to sleep    

std::unique_lock<std::mutex> lck(mtx);
size_t how_many_threads_are_asleep = threads_asleep.load();
    while (how_many_threads_are_asleep < threads_num) 
      cv.wait(lck, [&]() 
        how_many_threads_are_asleep = threads_asleep.load();
        return how_many_threads_are_asleep == numThreads;
      );
    

// At this point I'm sure ALL THREADS ARE ASLEEP!

// Wake them up one by one (there should only be ONE awake at any time before it finishes his computation)

for (int i = 0; i < threads_num; i++) 

  which_thread_to_wake_up.store(i);
  cv.notify_all(); // (*) Wake them all up to check if they're the chosen one
  std::unique_lock<std::mutex> lck2(mtx2);
  cv2.wait(lck, [&]()  return thread_has_finished; ); // Wait for the chosen one to finish
  thread_has_finished = false;

我担心最后一次notify_all()调用（我用（*）标记的那个）可能会导致以下情况：

所有线程都处于休眠状态 通过调用notify_all()从主线程唤醒所有线程 具有正确索引的线程完成最后一次计算并释放锁 所有其他线程已被唤醒，但尚未检查原子变量 主线程发出第二个 notify_all() 并且 THIS GETS LOST（因为线程都被唤醒了，他们还没有简单地检查原子）

这会发生吗？如果 notify_all() 的调用以某种方式 缓冲 或与实际检查条件变量的函数的同步顺序，我找不到任何措辞。

【问题讨论】：

我没有仔细研究过你的代码，但是你描述的一般情况是可以的。条件变量不存储信号。只有那些在发出信号时等待条件变量的线程才能看到该信号。被信号唤醒的线程不再等待，即使它们还没有从wait()函数返回。

【参考方案1】：

根据 (notify_all) 上的文档

notify_all 只是继续一个线程的要求的一半。条件语句也必须为真。所以必须有一个交通警察设计来唤醒第一个，唤醒第二个，唤醒第三个。 notify 函数告诉线程检查该条件。

我的回答比具体代码更高级，但我希望能有所帮助。

【参考方案2】：

您认为的情况可能会发生。如果您的工作线程（从属）在调用notify_all() 时被唤醒，那么它们可能会错过该信号。

防止这种情况的一种方法是在cv.notify_all() 之前锁定mtx，然后再解锁。正如wait() 的文档中所建议的，锁被用作pred() 访问的保护。如果主线程获取mtx，则没有其他线程同时检查条件。虽然他们当时可能正在做其他工作，但在你的代码中他们不太可能再次输入wait。