15
\$\begingroup\$

A very simple Thread pool:

Any work added to the pool will be executed. The destructor will wait for all work to be finished before letting the threads stop. We then join all threads before letting the destructor exit.

#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <vector>
#include <list>
#include <iostream>

class SimpleWorkQueue
{
    bool                                stopping;
    bool                                finished;
    std::mutex                          lock;
    std::condition_variable             cond;
    std::condition_variable             threadBarrier;
    std::list<std::function<void()>>    work;
    std::vector<std::thread>            workers;

    std::function<void()> getNextWorkItem();
    void workerAction();
    void tellAllThreadsToStop();
    public:
        SimpleWorkQueue(int initialThreadCount);
        ~SimpleWorkQueue();
        void addWorkItem(std::function<void()> action);
};

SimpleWorkQueue::SimpleWorkQueue(int initialThreadCount)
    : stopping(false)
    , finished(false)
{
    std::condition_variable     constructBarrier;
    int count = 0;

    workers.reserve(initialThreadCount);
    for(int loop=0;loop < initialThreadCount;++loop)
    {
        /*
         * Create all the threads we want in this loop.
         * For each thread we also add a job so that we
         * make sure it starts up correctly.
         *
         * To make sure that each thread only adds one to
         * the count the thread is held in the action
         * until all the threads have started up
         */
        work.push_back([&constructBarrier, &count, this]()
        {
            std::unique_lock<std::mutex> locker(this->lock);
            ++count;                            // Add one to the count
            constructBarrier.notify_one();      // And notify the Work-Queue.

            this->threadBarrier.wait(locker);   // Wait until all threads
                                                // have started.
        });
        // Create a thread to deal with the job we just started.
        workers.emplace_back(&SimpleWorkQueue::workerAction, this);
    }
    /*
     * We have created all the threads.
     * But we must now wait for all threads to
     * enter the while loop inside `workerAction()`
     *
     * We want to make sure that all threads have entered
     * the while loop before the destructor is entered
     * because if the destructor is entered we could
     * set finished to false before the thread enters the
     * function which makes reasoning about exit conditions
     * very hard.
     */
    std::unique_lock<std::mutex> locker(lock);
    constructBarrier.wait(locker, [&count, this](){
        return count == this->workers.size();
    });

    /*
     * All the thread have entered the check above and
     * incremented the counter. This released this thread
     * We can now release all the threads to start accepting
     * work items.
     */
    threadBarrier.notify_all();
}

SimpleWorkQueue::~SimpleWorkQueue()
{
    tellAllThreadsToStop();

    cond.notify_all();
    for(std::thread& thread: workers)
    {
        thread.join();
    }
}

void SimpleWorkQueue::tellAllThreadsToStop()
{
    std::unique_lock<std::mutex> locker(lock);
    stopping = true;
    /*
     * Push one job for each thread.
     * After this job is executed it will exit the while loop
     * in `workerAction()` (because finished is true)
     * thus causing the thread to exit.
     *
     * Thus each thread will execute one of these jobs then exit.
     * Thus all threads will eventually die.
     *
     * We do it this way to make sure that all work that has
     * been added will be completed before the threads start
     * exiting.
     */
    for(std::thread& thread: workers)
    {
        work.push_back([this](){
            std::unique_lock<std::mutex> locker(this->lock);
            this->finished = true;
        });
    }
}

void SimpleWorkQueue::addWorkItem(std::function<void()> action)
{
    std::unique_lock<std::mutex> locker(lock);
    if (stopping) {
        throw std::runtime_error("Can't add work after destructor entered");
    }
    work.push_back(action);
    cond.notify_one();
}

void SimpleWorkQueue::workerAction()
{
    while(!finished)
    {
        std::function<void()> item = getNextWorkItem();
        try
        {
            item();
        }
        catch(...)
        {
            /*
             * Must catch all exceptions generate by work.
             * If you don't then the user code may cause the
             * thread to accidentally exit.
             *
             * In old threading systems (like pthreads)
             * it was undefined behavior if the thread exited
             * with an exception. That's probably not the case here
             *
             * But you definitely don't want the thread to end
             * because of an exception in user space. So
             * catch and throw away (log if you must).
             */
        }
    }
}

std::function<void()> SimpleWorkQueue::getNextWorkItem()
{
    std::unique_lock<std::mutex> locker(lock);
    cond.wait(locker, [this](){return !this->work.empty();});

    std::function<void()> result = work.front();
    work.pop_front();
    return result;
}

Example usage

int main()
{
    SimpleWorkQueue queue(10);

    queue.addWorkItem([](){std::cerr << "Hi" << time(NULL) << "\n";});
}
\$\endgroup\$
0

1 Answer 1

10
\$\begingroup\$

On startup a lot of the code is there to deal with making sure that threads don't start work until the SimpleWorkQueue object is correctly initialized.

This work is not related to the work queue and should be factored out into its own class.

The concept of a barrier was mentioned in the question comments. So here it is. A barrier is a thread concept that blocks all threads that check-in until the specified number of threads have all arrived then they are all released simultaneously. This is often used for synchronization.

A simple Barrier

 // This is a one time use barrier.
 // Often these are written to be re-used. But that becomes harder.
 // On reset you have to guarantee that all the old threads were flushed
 // before you allow new threads to check-in.
 class Barrier
 {
     std::mutex&                lock;
     int                        count;
     std::condition_variable    threadBarrier;

     public:
         Barrier(std::mutex& m, int count) : lock(m), count(count) {}
         void checkin()
         {
             std::unique_lock<std::mutex> locker(lock);
             --count;
             if (count > 0)
             {
                 threadBarrier.wait(locker, [&count, this](){
                     return count <= 0;
                 });
             }
             else
             {
                 threadBarrier.notify_all();
             }
         }
};
         

Now the startup code is much simplified:

SimpleWorkQueue::SimpleWorkQueue(int initialThreadCount)
    : stopping(false)
    , finished(false)
    , threadBarrier(lock, initialThreadCount + 1)
{
    workers.reserve(initialThreadCount);
    for(int loop=0;loop < initialThreadCount;++loop)
    {
        /*
         * Create all the threads we want in this loop.
         * For each thread we also add a job so that we
         * make sure it starts up correctly.
         *
         * Note: Once a thread is in this piece of work it will
         *       not leave until all threads have checked in.
         */
        work.push_back([&threadBarrier]()
        {
            threadBarrier.checkin();
        });
        // Create a thread to deal with the job we just started.
        workers.emplace_back(&SimpleWorkQueue::workerAction, this);
    }
    // Release all the threads.
    // Once they have entered the main loop
    // And checked into the barrier.
    threadBarrier.checkin();
}

It looks like you are locking the same mutex twice in tellAllThreadsToStop(). You can not use a normal mutex and lock it twice for this you need a recursive lock (this will allow the same thread to lock a mutex more than once).

void SimpleWorkQueue::tellAllThreadsToStop()
{
    std::unique_lock<std::mutex> locker(lock);                // <- Lock
    stopping = true;

    for(std::thread& thread: workers)
    {
        work.push_back([this](){
            std::unique_lock<std::mutex> locker(this->lock);  // <- Lock
            this->finished = true;
        });
    }
}

Luckily this is not the case here. The second lock. Is inside a lambda function. Thus not in the same context as the original lock. We are pushing a piece of work into the queue. When this piece of work is executed by a child thread it needs to alter the state of the current object (multiple threads altering state must be done inside a mutal exclusion zone and thus a lock is required).

        work.push_back([this](){
            std::unique_lock<std::mutex> locker(this->lock);
            this->finished = true;
        });

Another thing pointed out in the comments is that all these jobs set finished to true. This seems like a waste as this will be done by each thread (ie initialThreadCount times).

An alternative is to set finished to true in one job and then add initialThreadCount-1 jobs that did nothing. This would have worked just as well and not have been wasteful in attaining the lock for each thread.

Initially this may seem true but this argument means you are thinking serially about the code. Once the job is picked up by a thread it may be unscheduled at the hardware/OS level and a bunch of other threads are then executed (so potentially all the do nothing jobs may finish executing before the job to set finished to true even starts executing (even if it is pulled from the job queue first).

So because we can not guarantee that any particular job will finish before any other job. They must all set finished to true (or be forced to wait at some barrier).

You are accessing mutable state without locking it:

void SimpleWorkQueue::workerAction()
{
    while(!finished)  // <- finished can be mutated by a child.
    {
    }
}

Upps. That is definitely a bug. Because other threads can modify finished it must be accessed after a memory barrier to force synchronization across threads. There are a couple of alternatives.

  • We could lock the mutext lock
  • We could make finished an atomic

But I think an easier way is to alter the shut down code.

The reason finished is set inside a work item is that if we set finished inside tellAllThreadsToStop() then threads will start to exit the main loop in workerAction() as soon as finished is set to true; even if there is still plenty of work to do.

But we see from the discussion above that we don't really want to modify finished in a child thread as that adds a whole set of problems. So an alternative is to alter the termination condition.

void SimpleWorkQueue::tellAllThreadsToStop()
{
    std::unique_lock<std::mutex> locker(lock);
    stopping = true;

    for(std::thread& thread: workers)
    {
        // TerminateThread is a private class
        // so nobody else can throw it.
        work.push_back([](){ throw TerminateThread(); });
    }
}


// Now our main thread loop.
// looks like this.
// Note: We can remove 'finished' from everywhere.
void SimpleWorkQueue::workerAction()
{
    while(true)
    {
        std::function<void()> item = getNextWorkItem();
        try
        {
            item();
        }
        catch(TerminateThread const& e)
        {
            // Break out of the while loop.
            // We threw because we wanted to terminate
            // and only our code can throw objects of this type.
            break;
        }
        catch(...)
        {
            // All other exceptions are user code generated.
            // Must be caught but can be ignored.
            // Though logging them is probably a good idea.
        }
    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.