Thread Pool in C++

Question

I know thread pools have been reviewed dozens of time and there is plenty of libraries around that implements them, but for fun and for the sake of learning more about multi threading and C++ I wanted to create my own thread pool. Functionality-wise it is currently doing what I expect it to do and is quite performant as well (could be improved probably though). I would like to receive some feedback on my code, in particular about readability but any feedback is a treasure. If you have nice ideas on how to improve it under any point of view I am super happy to hear from you guys.

The code is structured in two classes: ThreadQueued and ThreadPoolQueued, where the latter uses the first one to implement a ThreadPool.

below is the code (it is currently split in .hpp and .cpp all under ThreadQueued and ThreadPoolQueued:

ThreadQueued.hpp

#include <functional>
#include <deque>
#include <thread>
#include <condition_variable>
#include <vector>

namespace ThreadQueuedUtil
{
    const unsigned int MAX_QUEUE_SIZE = 10000; ///< How many functions the thread can hold
    const int PERFORMANCE_RATIO = 2; ///< At MAX_QUEUE_SIZE/PERFORMANCE_RATIO functions will be executed in batches
}

class ThreadQueued
{
public:
    ThreadQueued(unsigned int nMaxQueueSize = ThreadQueuedUtil::MAX_QUEUE_SIZE);
    
    void Start();
    
    void Wait();
    
    void push_back(std::function<void()> function);
    
    size_t getQueueLength() const;
    
private:
    
    // Main thread related variables //
    std::thread m_mainThread; ///< main thread
    void internalMainThread(); ///< Internal thread method
    bool m_bWaitCalled = false; ///< Track wether the Wait has been called or not
    std::mutex m_mutexWaitCalled; ///< Mutex for the above boolean
    std::deque<std::function<void()>> m_liFunction; ///< The functions that will be called at each cycle
    std::mutex m_mutexFunctions; ///< Mutex for the deque above
    std::condition_variable_any m_cvSleepCheck; ///< Condition variable to avoid wasting processing time
    ///////////////////////////////////
    
    ///\brief The internal queue cannot go over 10k elements,
    /// if the user continue to push_back then the additional
    /// elements will be dropped
    unsigned int m_nMaximumQueueSize = ThreadQueuedUtil::MAX_QUEUE_SIZE;
    size_t m_nInternalQueueLength = 0; ///< Helper variable to avoid locking for just checking the size

    ///\brief Internal performance related variables
    std::vector<std::function<void ()>> m_liPerformanceExecutionFunctions;
    
    
    unsigned int m_nDroppedFunctions = 0;
    
};

ThreadQueued.cpp

#include "MultipleLocks.hpp"
#include <iostream>

ThreadQueued::ThreadQueued(unsigned int nMaxQueueSize)
:   m_nMaximumQueueSize(nMaxQueueSize)
{
    
}

void ThreadQueued::Start()
{
    m_mainThread = std::thread(&ThreadQueued::internalMainThread, this);
}

void ThreadQueued::Wait()
{
    m_mutexWaitCalled.lock();
    m_bWaitCalled = true;
    m_mutexWaitCalled.unlock();
    
    m_cvSleepCheck.notify_all();
    
    if (m_mainThread.joinable()) {
        m_mainThread.join();
    }
}

void ThreadQueued::push_back(std::function<void ()> function)
{
    /// Warning: This method can be accessed from different threads
    if (m_nInternalQueueLength >= m_nMaximumQueueSize) {
        ++m_nDroppedFunctions;
        std::cout << "Dropping function! Single thread!" << std::endl;
        return;
    }
    
    { // lock function's mutex
        std::lock_guard<std::mutex> lockFunctions(m_mutexFunctions);
        m_liFunction.push_back(function);
        m_nInternalQueueLength = m_liFunction.size();
    }
    
    m_cvSleepCheck.notify_all();
}

size_t ThreadQueued::getQueueLength() const
{
    return m_nInternalQueueLength;
}

void ThreadQueued::internalMainThread()
{
    while(true)
    {
        std::unique_lock<std::mutex> lockFunctions(m_mutexFunctions, std::defer_lock);
        lockFunctions.lock();
        if(!m_liFunction.empty())
        {
            std::function<void ()> function = m_liFunction.front();
            m_liFunction.pop_front();
            m_nInternalQueueLength = m_liFunction.size();
            lockFunctions.unlock();
            
            // Execute what has been requested
            if (function) {
                // Callable
                function();
            }
            
            // If we have many functions ready to be processed,
            // we can take a bunch of them and process them, without
            // the need to gain a lock everytime.
            if (m_nInternalQueueLength > m_nMaximumQueueSize/ThreadQueuedUtil::PERFORMANCE_RATIO) {
                lockFunctions.lock();
                unsigned long nNumberOfFunctionsToExtract = m_nInternalQueueLength;                m_liPerformanceExecutionFunctions.reserve(nNumberOfFunctionsToExtract);
                for (std::deque<std::function<void ()>>::iterator iter = m_liFunction.begin(); iter != m_liFunction.end(); ++iter) {
                    m_liPerformanceExecutionFunctions.push_back(*iter);
                }
                m_liFunction.clear();
                m_nInternalQueueLength = 0;
                lockFunctions.unlock();
            }
            
            // Now we can process with all the time of the world
            for (std::vector<std::function<void ()>>::iterator iter = m_liPerformanceExecutionFunctions.begin(); iter != m_liPerformanceExecutionFunctions.end(); ++iter) {
                if ((*iter)) {
                    // Callable
                    (*iter)();
                }
            }
            m_liPerformanceExecutionFunctions.clear();
        }
        else
        {
            lockFunctions.unlock();
            std::unique_lock<std::mutex> lockWaitCalled(m_mutexWaitCalled, std::defer_lock);
            lockWaitCalled.lock();
            if (m_bWaitCalled) {
                // Game over
                return;
            }
            else
            {
                lockWaitCalled.unlock();
                
                MultipleLocks locks(m_mutexFunctions, m_mutexWaitCalled);
                
                m_cvSleepCheck.wait(locks, [this]() -> bool {return (!m_liFunction.empty() || m_bWaitCalled);});
                
                continue;
            }
        }
    }
}

ThreadPoolQueued.hpp

// Libraries
#include <vector>
#include <functional>
#include <thread>
#include <deque>
#include <condition_variable>

// Locals
#include "ThreadQueued.hpp"

namespace ThreadPoolUtil
{
    const unsigned int MAX_QUEUE_SIZE = 40000;
}

class ThreadPoolQueued
{
public:
    
    ThreadPoolQueued(unsigned int nMaximumNumberOfThreads = std::thread::hardware_concurrency());
    ~ThreadPoolQueued();
    
    void start();
    void wait();
    
    void push_back(std::function<void(void)> function);
    
private:
    
    //\brief internal thread related variables //
    std::thread m_mainThread; ///< Main execution thread
    
    void internalMainThread(); ///< Internal processing cycle
    
    void balancedLoad(std::function<void ()> & function); ///< Call this method if you want to assign the function in a balanced way to the workers
    
    void unbalancedLoad(std::function<void ()> & function); ///< Call this method if you want to maximize the speed of the main thread
    unsigned int m_nNextThreadToLoad = 0; ///< Helper variable to increase efficiency in the unbalancedLoad
    
    bool m_bWaitCalled = false; ///< Track wether the Wait has been called or not
    
    std::mutex m_mutexWaitCalled; ///< Mutex for the above boolean
    
    std::deque<std::function<void()>> m_liFunction; ///< The functions that will be called at each cycle
    
    std::mutex m_mutexFunctions; ///< Mutex for the deque above
    
    std::condition_variable_any m_cvSleepCheck; ///< Condition variable to avoid wasting processing time

    std::vector<ThreadQueued> m_liThreads; ///< Internal list of threads
    
    unsigned int m_nMaximumNumberOfThreads;
    ////////////////////////////////////////////////
};

ThreadPoolQueued.cpp

#include "ThreadPoolQueued.hpp"   // <-- Loki Added this line
#include "MultipleLocks.hpp"
#include <iostream>

ThreadPoolQueued::ThreadPoolQueued(unsigned int nMaximumNumberOfThreads)
:   m_nMaximumNumberOfThreads(nMaximumNumberOfThreads),
m_liThreads(nMaximumNumberOfThreads)
{
    
}

ThreadPoolQueued::~ThreadPoolQueued()
{
    if (m_mainThread.joinable()) {
        m_mainThread.join();
    }
}

void ThreadPoolQueued::start()
{
    m_mainThread = std::thread(&ThreadPoolQueued::internalMainThread, this);
    
    for (std::vector<ThreadQueued>::iterator iter = m_liThreads.begin(); iter != m_liThreads.end(); ++iter) {
        (*iter).Start();
    }
}

void ThreadPoolQueued::wait()
{
    m_mutexWaitCalled.lock();
    m_bWaitCalled = true;
    m_mutexWaitCalled.unlock();
    
    m_cvSleepCheck.notify_all();
}

void ThreadPoolQueued::push_back(std::function<void ()> function)
{
    balancedLoad(function);
    m_cvSleepCheck.notify_all();
}

void ThreadPoolQueued::balancedLoad(std::function<void ()> & function)
{
    // Pass the function to the less loaded thread
    unsigned int nLessLoadedThreadIndex = 0;
    unsigned int nCounter = 0;
    size_t nLowestLoad = ThreadQueuedUtil::MAX_QUEUE_SIZE;
    for(std::vector<ThreadQueued>::iterator iter = m_liThreads.begin(); iter != m_liThreads.end(); ++iter, ++nCounter)
    {
        size_t nCurrentThreadQueueSize = (*iter).getQueueLength();
        if (nCurrentThreadQueueSize < nLowestLoad)
        {
            nLowestLoad = nCurrentThreadQueueSize;
            nLessLoadedThreadIndex = nCounter;
        }
    }
    m_liThreads.at(nLessLoadedThreadIndex).push_back(function);
}

void ThreadPoolQueued::unbalancedLoad(std::function<void ()> & function)
{
    m_liThreads.at(m_nNextThreadToLoad).push_back(function);
    ++m_nNextThreadToLoad;
    if (m_nNextThreadToLoad == m_nMaximumNumberOfThreads) {
        m_nNextThreadToLoad = 0;
    }
    return;
}

void ThreadPoolQueued::internalMainThread()
{
    while (true) {
        m_mutexFunctions.lock();
        m_mutexWaitCalled.lock();
        if (m_liFunction.empty() && m_bWaitCalled) {
            m_mutexWaitCalled.unlock();
            m_mutexFunctions.unlock();
            // Game over! Wait for all threads to finish their job
            for (std::vector<ThreadQueued>::iterator iter = m_liThreads.begin(); iter != m_liThreads.end(); ++iter)
            {
                (*iter).Wait();
            }
            return;
        }
        else
        {
            m_mutexWaitCalled.unlock();
            if (m_liFunction.empty()) {
                m_mutexFunctions.unlock();
                
                MultipleLocks locks(m_mutexFunctions, m_mutexWaitCalled);
                
                m_cvSleepCheck.wait(locks, [this]() -> bool {return !m_liFunction.empty() || m_bWaitCalled;});
                continue;
            }
            else
            {
                std::function<void ()> function = m_liFunction.front();
                m_liFunction.pop_front();
                m_mutexFunctions.unlock();
                
                // balance or unbalanced load?
                balancedLoad(function);
                
            }
        }
    }
}

Answer 1

Design

This is not a "Thread Pool" as I understand the term (but is something close). So lets define the terms.

To me. A thread pool is a set of threads that pull units of work from a shared queue. When there is no work in the queue a thread will suspend (i.e. not consume resources).

Thus it is accesses to the shared queue that becomes the reason for the pool and to protect accesses to this queue in a threaded environment.

Note sure what to call your thing. But what you have is:

1. A thread with its own fixed-sized queue of work (ThreadQueued).
2. A workload manager thread that distributes incoming work to ThreadQueued objects.

The problem I see is that the workload manager tries to balance the jobs by job count with no context on how expensive the job actually is. Sure this may work in an environment were all jobs are micro-transactions and all take very little time. But there is no indication of this in the documentation or the interface.

If any of the work packages is large then your system could force job starvation in multiple threads while one thread still has thousands of jobs still to do (because the current one will take a long time to complete).

State checking

You don't use locks consistently. You can NOT read a value then create a lock to update it. If you are in a multi-threaded environment then the read needs to be consistent with the write.

// If you have one spot left in your queue for a piece of work.
if (m_nInternalQueueLength >= m_nMaximumQueueSize) {
    ++m_nDroppedFunctions;
    return;
}
// Then multiple threads can get past this check
// To this point. You can then have multiple threads
// waiting to acquire the lock and add the function to `m_liFunction`.
// Even while the below code is executing (inside the lock) more
// threads can still get past the above check and pile up against this
// lock.
{
    std::lock_guard<std::mutex> lockFunctions(m_mutexFunctions);
    m_liFunction.push_back(function);
    m_nInternalQueueLength = m_liFunction.size();
}

// This code is not thread safe.

In this case you are manually calling lock/unlock.

m_mutexWaitCalled.lock();
m_bWaitCalled = true;
m_mutexWaitCalled.unlock();

This is just not idiomatic C++ where we use RAII to perform this type of operation.

Exception Safety

When using threads you need to take into account exceptions. If a thread terminates because of an exception then the application will quit (after calling std::terminate).

So any random user code needs to be handled inside a try catch block.

Overall

I think the code execution is very clunky and overcomplicated, as you try and make ThreadQueue handle both a single job and a list of jobs within the same lock region.

I think you need to look at a couple more examples of thread pools before trying this again.

Several people have had their Thread Pools reviewed on this site. That is a good place to start reading.

Here is one I published: A simple Thread Pool

Code Review

Funny Namespace

Why are only the constants in the namespace?

namespace ThreadQueuedUtil
{
    const unsigned int MAX_QUEUE_SIZE = 10000;
    const int PERFORMANCE_RATIO = 2;
}

I would expect all your information to be inside the namespace!

class ThreadQueued
{

That includes the classes.

Two Phase Startup.

Two phase startup is considered a bad design. This is where you construct an object but then also need to call a method to finish the initialization.

class ThreadQueued
{
public:
    ThreadQueued(unsigned int nMaxQueueSize = ThreadQueuedUtil::MAX_QUEUE_SIZE);

    void Start();

A constructor and Start() method sort of indicate two phase startup.

When the constructor has finished your object should be fully formed and already operational. I should not need to prime it (that was the job of the constructor).

The problem is that somebody will forget to call Start() and that will be an error for them that they need to track down. By using a single initialization function (the constructor) that is automatically called you reduce the potential for incorrect usage.

Naming conventions

C++ does not have any standard naming conventions. BUT a very common one is that User defined Types start with an initial capital letter. Functions/Methods/Variables all start with an initial lower case letter and all standard types are prefixed with std::.

Now I can't say this is wrong:

    void Start();    
    void Wait();

When you look at just these two functions. But if we clump them together with your other methods we see an inconsistent naming convention on your functions.

    void Start();    
    void Wait();
    void push_back(std::function<void()> function);    
    size_t getQueueLength() const;

Even if you don't use the one I mention above you should have a convention that is consistent and easy to spot. I prefer Snake case on method names but other prefer the underscore (the community is still very divided on that one). Pick one or the other and be consistent (you are using both).

Bad Comments

Don't write useless comments.

    // Main thread related variables //

They not only make the code untidy and harder to read - by providing no extra information you are adding an extra cognitive load to the developer that gives them no new information.

Additionally it is a real measurable maintenance effect to have bad comments. Over time comments can stray from the code (as the code is changed). Then when a developer comes across an inconsistency does he assume the code is correct and the comment should be deleted/fixed - or is the comment correct and thus the code has a bug and needs to be fixed?

When you write comments they should be about WHY. Why do you have these values? Why are you using this specific algorithm (provide a link to a wiki about the algorithm)?

The code should describe how (by using self documenting code).

More useless comments:

    std::thread m_mainThread; ///< main thread    // is that why its name is mainThread.
    void internalMainThread(); ///< Internal thread method   // is that why its name is Internal 
    bool m_bWaitCalled = false; ///< Track wether the Wait has been called or not  // Comment says the same as the variable name

I could go on. All these comments are terrible and should be stripped.

Nearly a good comment.

    ///\brief The internal queue cannot go over 10k elements,
    /// if the user continue to push_back then the additional
    /// elements will be dropped
    unsigned int m_nMaximumQueueSize = ThreadQueuedUtil::MAX_QUEUE_SIZE;

First the comment is 10K but the variable has a size of MAX_QUEUE_SIZE. What happens if these are not consistent? I don't mind you telling my what the use of this value is but it would be nice to put that comment with the code and maybe explain WHY the limit is 10K (or MAX_QUEUE_SIZE). I can probably read the code and deduce it is dropping elements easily. But to help maintainers it would be nice to understand WHY.

Not Thread safe

void ThreadQueued::push_back(std::function<void ()> function)
{
    // We covered Why up above.
}

Handle one then many

You seem to handle one then many. Why not just many?

void ThreadQueued::internalMainThread()
{

Break this function into several smaller functions.

void ThreadQueued::internalMainThread()
{

Auto is your friend

unsigned long nNumberOfFunctionsToExtract = m_nInternalQueueLength;                
m_liPerformanceExecutionFunctions.reserve(nNumberOfFunctionsToExtract);
for (std::deque<std::function<void ()>>::iterator iter = m_liFunction.begin(); iter != m_liFunction.end(); ++iter) {
    m_liPerformanceExecutionFunctions.push_back(*iter);
}
m_liFunction.clear();

The above loop can be simplified with auto.
It can be further simplified with a range-based for.