Controlling parallel jobs with condition_variable

Question

I'm trying to set up a system where I can have an arbitrary number of parallel tasks, all of which wait for a go signal, do their work, and then stop while a manager collects the results, prepares for the next batch of work, and so on.

The easy way I know to do this is to launch a packaged_task for each job, for each iteration. However, I have got the impression that launching threads (and more so with the higher abstractions) have some considerable overhead, which I would want to avoid. My limited testing of those other options confirm that belief.

The best I've come up with so far is to have threads that are kept alive, and loop based on signals from condition_variable entities.

std::mutex mx;
std::condition_variable global;
bool start = false;
bool ready = false;
std::map<int, std::atomic<bool>> status;
using namespace std;
void work(int n) {
    while (true) {
        {
            unique_lock<mutex> lock{mx};
            //std::cout<<"Job #"<<n<<" waiting for ready signal.\n";
            while (!ready) { global.wait(lock); }
        } {
            unique_lock<mutex> lock{mx};
            status[n] = false;
            //std::cout<<"Job #"<<n<<" ready, waiting for start signal.\n";
            while (!start) { global.wait(lock); }
        }
        this_thread::sleep_for(chrono::milliseconds(10)); // simulate work
        status[n] = true;
    }
}
int main(int argc, char ** argv) {
    const auto num_jobs = (argc<=1? 1000:atoi(argv[1]));
    vector<thread> jobs;
    for (auto n=1;n<=num_jobs;n++) { jobs.emplace_back(work, n); }
    while (true) {
        cin.ignore(); {
            unique_lock<mutex> lock{mx};
            start = false;
            ready = true;
            status.clear();
            global.notify_all();
        } cin.ignore(); {
            unique_lock<mutex> lock{mx};
            ready = false;
            start = true;
            global.notify_all();
        }
        while (!all_of(status.begin(), status.end(),   // Wait for all jobs
               [](auto&& t){ return t.second.load(); }))
        { /*this_thread::sleep_for(chrono::seconds(1));*/ }
    }
    for (auto& job : jobs) { job.join(); }
}

I tried to compact the code, but tell me if it's too obfuscated. What the program does is to create some arbitrary number of workers, which wait for the launch signal from cin and then perform the simulated work.

I had to create two stages for the task to possibly be in, in a ready-set-go mechanic where the thread lands in a waiting clause after completing its job. My first attempt was to use just one wait for the task, but I can't figure out a a way to disable the single flag before some tasks are looping back around and then would start a second iteration immediately.

So basically what I'm asking: is this a good approach to the problem discussed in the first paragraph? If yes, can it be made better? Surely this is a well trodden issue for programmers trying to do parallel optimization, what is the conventional way to deal with my scenario?

Answer 1

Overall

Unfortunately your code is broken.

The main thread can potentially call notify_all() twice before any of the threads are released from the first call wait(lock). Say I hit return several time very quickly.

It's even possible for the main thread to call notify_all() before any of the children call wait(lock) the first time (what happens if I pipe a file into standard in).

If you want all threads and the main to wait and sync at a common point you must explicitly create what is called a sync barrier. This is usually done by having a variable that counts the threads that have reached the sync point and only allowing progress after all threads have reached it and decremented the counter to the point where it reaches 0.

To avoid confusion you will also need to use a separate condition_variable for sync and start points.

Finally this code will never finish.

for (auto& job : jobs) { job.join(); }

Waits for all the threads to finish. But none of the jobs ever exit the callback function work().

Note: you can't just remove this line. A thread who's destructor is called before the thread of execution finishes its work will generate an exception that causes the application to quit. So you should call join() on all threads to verify that the thread of execution has completed.

Design.

I understand the need to force all the children thread to wait at an initial sync point. They need to verify that the main thread has successfully initialized all the global state.

But after that each worker thread does not need to wait at a sync point (unless there is an application requirement to do so). It just needs to wait for work to become available grab it then do it. Once the work is done it is usual for a thread to check if there is more work available (as creating threads is rather expensive you usually don't want to create and destroy threads regularly). If no work is available then make it wait on a condition variable until there is work.

All threads should periodically check to see if they are still needed and expire when they are no longer required. Thus allowing the main thread to exit.

Simple Code Review

Global variables a re a bad idea.

std::mutex mx;
std::condition_variable global;
bool start = false;
bool ready = false;
std::map<int, std::atomic<bool>> status;

wrap all this up in a class that you can then use locally inside a function.

Don't do this.

using namespace std;

If your code is more than 10 lines long you are going to cause issues sooner or later. See: Why is “using namespace std;” considered bad practice?

Threaded job Queue.

The following is a generalized job queue.
Now I read in your question that you need to have work done in synchronized bursts. This class does not do that but with some small adjustments that can be done.

void JobQueue::jobWorker()
{
    while(true)
    {
        auto job = getWork();
        if (job.execute())
        {
            // Indicate a thread should exit by returning true from execute.
            break;
        }
    }
}

class JobQueue
{
    private:
       void jobWorker();
    public:

        // Both addWork()/getWork() modify internal state.
        // So calls must be synchronized via locks.
        template<typename F>
        void addWork(F&& work); // Adds a job to an internal queue.           
        Job getWork();          // If no work is available will cause the
                                // thread to wait (on a condition variable)
                                // until there is work available.



         // Constructor initializes all internal state.
         // Then creates `count` threads of execution that call jobWorker.
         JobQueue(int count);

         // The destructor waits for all work (added via addWork())
         // to be finished.
         //
         // This means it must:
         // 1) wait for the work queue to drain.
         // 2) all threads must be given a job that will return true.
         //    This will allow all the worker threads to exit.
         // 3) Wait for all worker threads by calling join() on them.
         ~JobQueue();           
};

Then main is simplified too:

int main()
{
    JobQueue    jobQueue(4);  // not much point it making this number
                              // larger than the number of cores available
                              // on your machine. Actually you can get a default
                              // value that is the number of cores.

    // Call addWork() as many times as required.
    for(int loop = 0;loop < 1'000'000;++loop) {
        jobQueue.addWork([](){std::cout << "Some Work\n";sleep(5);});
    }
}