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