3
\$\begingroup\$

This is my first time writing a custom thread pool in C++. I would love some feedback on how my code looks and if I am missing best practices or have glaring/subtle issues. Thanks!

ThreadPool.hpp:

#pragma once 

#include <atomic> 
#include <vector> 
#include <iostream> 
#include <thread>
#include <future>
#include <mutex>
#include <queue>
#include <functional>
#include <ThreadSafeQueue.hpp>

class ThreadPool{
    public: 
        ThreadPool(std::atomic_bool& result); 
        void waitForCompletion();
        void addJob(std::function<bool()> newJob);
        void setComplete();
    private: 
        void workLoop(std::atomic_bool& result); 
        int m_numThreads; 
        std::vector<std::thread> m_threads; 
        std::atomic_bool m_workComplete; 
        std::mutex m_mutex; 
        std::condition_variable m_jobWaitCondition; 
        ThreadSafeQueue<std::function<bool()>> m_JobQueue;
};

ThreadPool.cpp:

#include <ThreadPool.hpp> 

ThreadPool::ThreadPool(std::atomic_bool& result){ 
    m_numThreads = std::thread::hardware_concurrency();
    m_workComplete = false;
    for (int i = 0; i < m_numThreads; i++)
    {
        m_threads.push_back(std::thread(&ThreadPool::workLoop, this, std::ref(result)));
    }
}

// each thread executes this loop 
void ThreadPool::workLoop(std::atomic_bool& result){ 
    while(!m_workComplete){
        std::function<bool()> currentJob;
        bool popped;
        {
            std::unique_lock<std::mutex> lock(m_mutex); 
            m_jobWaitCondition.wait(lock, [this](){
                return !m_JobQueue.empty() || m_workComplete.load();
            });
            
            popped = m_JobQueue.pop(currentJob);
        }
        if(popped){
            result = currentJob() && result;
        }
    }
}

void ThreadPool::addJob(std::function<bool()> newJob){ 
    m_JobQueue.push(newJob);
    m_jobWaitCondition.notify_one();
}

void ThreadPool::setComplete(){
    m_workComplete = true; 
}

void ThreadPool::waitForCompletion(){
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        m_workComplete.store(true);
    }
    
    m_jobWaitCondition.notify_all();

    for(auto& thread : m_threads){ 
        thread.join();
    }
    
    m_threads.clear();
}

ThreadSafeQueue.hpp:

#pragma once

#include <mutex>
#include <queue>

template <class T>
class ThreadSafeQueue {
   public:
    ThreadSafeQueue(){};
    void push(T element) {
        std::unique_lock<std::mutex> lock(m_mutex);
        m_queue.push(element);
    }
    bool pop(T& retElement) {
        std::unique_lock<std::mutex> lock(m_mutex);
        if (m_queue.empty()) {
            return false;
        }
        retElement = m_queue.front();
        m_queue.pop();
        return true;
    }
    bool empty(){ 
        std::unique_lock<std::mutex> lock(m_mutex); 
        return m_queue.empty();
    }

   private:
    std::queue<T> m_queue;
    std::mutex m_mutex;
};
```
\$\endgroup\$

2 Answers 2

2
\$\begingroup\$

#include <filename> vs #include "filename"

#include <ThreadPool.hpp> is typically how you would include third-party headers. You would do #include "ThreadPool.hpp" for your own headers.

hardware_concurrency error-checking

hardware_concurrency could return zero. Better handle that error. You might create a constant number of threads if that happens, or you might want to throw an exception:

m_numThreads = std::max(std::thread::hardware_concurrency(), 2u);

Consider emplace_back over push_back:

m_threads.push_back(std::thread(&ThreadPool::workLoop, this, std::ref(result)));

// replace with

m_threads.emplace_back(&ThreadPool::workLoop, this, std::ref(result));

addJob takes an unnecessary copy:

m_JobQueue.push(newJob);

// replace with 

m_JobQueue.push(std::move(newJob));

Destructor

It's better to check and call waitForCompletion in your destructor.

Empty constructor

ThreadSafeQueue(){};

// replace with 

ThreadSafeQueue() = default;

Questionable API

bool pop(T &retElement) is a really bad way of going about doing that. You force the object to have a default constructor and a copy constructor, basically. A much better way would be: std::optional<T> pop();. You would also move the element, instead of copying it.

void push(T element) is not a bad idea, but you should move element into the queue, instead of copying it.

Finally

Not exactly the same, but do check-out my similar post from some time ago: Asynchronous dispatch queue

\$\endgroup\$
4
  • \$\begingroup\$ Just curious, why should emplace_back be considered over push_back? \$\endgroup\$ Mar 28 at 21:26
  • 1
    \$\begingroup\$ @MartinSand push_back, in this specific case, constructs a temporary, moves out of it, and destroys the temporary. emplace_back though constructs the new object inside the queue in one go and no temporary is involved. \$\endgroup\$ Mar 28 at 22:33
  • \$\begingroup\$ Thanks for the code review Ayxan! If I changed pop's signature to be std::optional then I wouldn't be able to get the return status of the function, would there be way to do both? I agree that the signature is a bit strange right now. And can you expand on moving it instead of copying for pop? \$\endgroup\$ Mar 29 at 19:44
  • \$\begingroup\$ I guess this sample answers both your questions: gcc.godbolt.org/z/a7qM5hn6b \$\endgroup\$ Mar 29 at 20:18
2
\$\begingroup\$

Ayxan Haqverdili already gave a great answer, I'd just like to add this:

Remove empty()

Asking a thread-safe queue whether it is empty is almost always useless. By the time empty() returns the answer, another thread could have either popped the remaining element or have pushed a new element. The return value is therefore rather meaningless.

Make m_workComplete a regular bool

Making m_workComplete a std::atomic<bool> sounds great, but the problem is that the atomic access scope is different from that of other variables that are guarded by m_mutex. Also consider calling setComplete(), while it will set m_workComplete, it doesn't signal any threads. If the queue was already empty, that means none of the workers will notice that they should terminate.

The best thing to do here is to make m_workComplete a regular bool, and have it guarded by m_mutex. This way it is in the same atomic scope as the job queue, and setComplete() should be written like so:

void ThreadPool::setComplete(){
    std::unique_lock<std::mutex> lock(m_mutex); 
    m_workComplete = true;
    m_jobWaitCondition.notify_all();
}

And also call this function in waitForCompletion() instead of duplicating it.

\$\endgroup\$
5
  • \$\begingroup\$ Hmm, how would I be able to check if the queue is empty then? This is also my first time doing low level multithreading, so please bear with me, but I thought mutexes would protect the queue from other thread accesses? And can you expand on the atomic access scope? \$\endgroup\$ Mar 29 at 19:46
  • 1
    \$\begingroup\$ @user2793618 you wouldn't check if the queue is empty. You would go ahead and try to do whatever you're trying to do. If the queue has appropriate number of elements, it would succeed, otherwise it'd do nothing. Checking if the queue is empty is useless, because this fact can change between you checking it and trying to use the queue. \$\endgroup\$ Mar 29 at 20:20
  • 1
    \$\begingroup\$ As for scope, an std::atomic variable only provides atomic access to that variable, so the scope is just that variable. A std::mutex only ensures atomic access to those variables that are only ever accessed while the mutex is locked. In your code. In your code, m_workComplete and m_queue have different scopes, but they should be in the same scope for questions like "I want to read from the queue until the work is complete" to be safe to ask. \$\endgroup\$
    – G. Sliepen
    Mar 29 at 20:26
  • \$\begingroup\$ Huh I see, but if I lock the mutex before checking for if the queue is empty should it not block the other threads from accessing it until it's unlocked? \$\endgroup\$ Apr 1 at 18:11
  • \$\begingroup\$ And I see, so if you want some sort of joint statement to be completed, you want the same atomic scope such that the mutex is protecting all of the resources you need? Which in this case would be m_workComplete and the queue since I want to check for the emptiness of the queue and if the work is complete? \$\endgroup\$ Apr 1 at 18:13

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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