C++. std::condition_variable and multiple wait-threads

user_4481855 0
Viewed 349

I have some class, with queue of std::function<void()> member and methods Push and Pop.

I want to implement addition method PushAndWaitUntilExecuted. It is easy when you have one consumer-thread(who call Pop) and one producer-thread(who call Push) - just simple std::condition_variable will be enough.

But my application have dynamic number of threads which can execute the same lines of code with calling PushAndWaitUntilExecuted function in parallel and wait until consumer-thread execute pushed std::function object.

A have the idea with passing std::pair<uint64_t, std::function<void()>> to queue instead of just std::function<void()>, where uint64_t - producer-thread ID(boost::this_thread::get_id()). And then consumer-thread call std::condition_variable::notify_all() and all threads will check if executed std::function have same ID with thread or not.

Is it ok solution or something better can be implemented?

std multithreading boost c++
1 Answers

More than just a condition variable needs to be introduced here, in order to avoid several different race conditions. A mutex and a job completion flag are also required.

At this point, it becomes cleaner to replace your std::function<void()> with a small class that contains this closure, as well as all the additional baggage:

struct job {
   std::function<void()> implementation;
   std::mutex m;
   std::condition_variable flag;
   bool completed=false;
};

Your queue becomes a queue of std::shared_ptr<job>s, instead of a queue of std::functions, with the jobs constructed in dynamic scope (since, of course, mutexes and condition variables are not copyable or movable, and these objects get accessed from both of your threads).

After your worker thread finishes executing the implementation, it:

  1. locks the mutex.
  2. sets completed to true
  3. signals the condition variable.

And your PushAndWaitUntilExecuted, after it executes the push:

  1. locks the mutex
  2. waits on the condition variable, until completed is set

You must thoroughly understand that C++ gives you absolutely no guarantees, whatsoever, that after you push a new closure into your job queue, some thread doesn't immediately grab it, execute it, and finish it, before the original thread (the one that pushed it) gets around to looking at the condition variable. By now, nobody will be signaling the condition variable any more. If all you have to work with is just a condition variable here, you'll be waiting for the condition variable to get signaled until our sun explodes.

Which is why you need more than just a condition variable, a mutex and an explicit flag, and use the above approach, to correctly handle interthread sequencing.

This is a fairly classical, routine approach. You should find examples of many similar implementations in every good C++ textbook on this subject matter.

user_39433120
Related