CODEWITHSUNDEEP
c++multithreadingthread-safety
Dean
Dean

Reputation: 6958

C++ Thread safety and notify_all()

The real code is way more complex but I think I managed to make a mcve.

I'm trying to do the following:

  1. Have some threads do work
  2. Put them ALL into a pause state
  3. Wake up the first of them, wait for it to finish, then wake up the second one, wait for it to finish, wake up the third one.. etc..

The code I'm using is the following and it seems to work

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

I'm afraid that the last notify_all() call (the one I marked with (*)) might cause the following situation:

Could this ever happen? I couldn't find any wording for notify_all() if its calls are somehow buffered or the order of synchronization with the functions that actually check the condition variables.

Upvotes: 2

Views: 816

Answers (2)

gdlmx
gdlmx

Reputation: 6789

The situation you consider can happen. If your working threads (slaves) are awaken when the notify_all() is invoked, then they will probably miss that signal.

One way to prevent this situation is to lock mtx before cv.notify_all() and unlock it afterward. As suggested in the documentation of wait(), lock is used as a guard to pred() access. If the master thread aquires mtx, no other thread are checking the conditions at the same moment. Although they may be doing other jobs at that time, but in your code they are not likely to enter wait again.

Upvotes: 0

ioMatrix
ioMatrix

Reputation: 633

As per the docs on (notify_all)

notify_all is only one half of the requirements to continue a thread. The condition statement has to be true as well. So there has to be a traffic cop designed to wake up the first, wake up the second, wake up the third. The notify function tells the thread to check that condition.

My answer is more high level than code specific but I hope that helps.

Upvotes: 0

Related Questions