Easy to use Thread Pool Implementation

Question

awhile ago I asked about a thread pool implementation I had made. I took the advice down and modified it a lot over time to fit my needs. It has been serving me extremely well in all of my projects but I was wondering if there was anything else I could improve on?

Here is the class:

class ThreadPool{
public:
    /*  Constructor */
    ThreadPool(uint8_t numThreads) {
        assert(numThreads > 0);
        createThreads(numThreads);
    }

    //Destructor
    ~ThreadPool(){
        shutdown = true;
        notifier.notify_all();
        for(int i = 0; i < threads.size(); ++i){
            threads[i].join();
        }
    }

    //add any arg # function to queue
    template <typename Func, typename... Args >
    auto push(Func&& f, Args&&... args){

    //get return type of the function
    typedef decltype(f(args...)) retType;

    std::packaged_task<retType()> task(std::bind(f, args...));

    std::future<retType> future = task.get_future();

    {
        // lock jobQueue mutex, add job to the job queue 
        std::unique_lock<std::mutex> lock(JobMutex);

        jobQueue.emplace(std::packaged_task<void()>(std::move(task)));
    }
    notifier.notify_one();

    return future;
    }

private:

    std::vector<std::thread> threads;
    std::queue<std::packaged_task<void()>> jobQueue;
    std::condition_variable notifier;
    std::mutex JobMutex;
    std::atomic<bool> shutdown = false;

    void createThreads(uint8_t numThreads) {
        auto threadFunc = [this]() {
            while (true) {
                std::packaged_task<void()> job;

                {
                    std::unique_lock<std::mutex> lock(JobMutex);
                    notifier.wait(lock, [this] {return !jobQueue.empty(); });

                    if(shutdown){
                        break;
                    }

                    //strange bug where it will continue even if the job queue is empty
                    if (jobQueue.size() < 1)
                        continue;

                    job = std::move(jobQueue.front());

                    jobQueue.pop();
                }
                job();
            }

        };
        threads.reserve(numThreads);
        for (int i = 0; i != numThreads; ++i) {
            threads.emplace_back(std::thread(threadFunc));
        }
    }


    ThreadPool(const ThreadPool& other) = delete; 
    void operator=(const ThreadPool& other) = delete;

}; /* end ThreadPool Class */

Answer 1

You have a logical bug in the code. Imagine all jobs are finished and then shutdown is called. You'll never leave the notifier.wait(...).

Also, if a job exists but shutdown is called you quit instantly without doing it - I do not think that this is a good idea. I think you should finish all jobs prior to quitting. Make a special hard reset routine for the immediate exit. Though, it probably requires usage of exceptions.

Also, consider what will happen if an exception is called inside the function f. In the thread pool one of the threads will die and kill the whole program - most likely. It is not good. It should somehow forward the error to the caller. To this end I have thread pool functionality divided - made a supplementary class for user that wraps whatever additional functionality is needed.

Technical issues:

1. std::bind(f, args...) should be std::bind(f, std::forward<Args>(args)...) to ensure perfect forwarding as otherwise you might end up with unnecessary copying of data.

2. numThreads should be defaulted to std::thread::hardware_concurrency and I believe and uint8_t is getting too small for some of the latest processors. Additionally the ThreadPool should have the functionality to tell to its user how many threads it has.

3. I am not too fond of std::future - it is somewhat incomplete as Executors TS is still incomplete. Normally I'd expect that user would want to call the push task several times to complete a single operation. In which, case you generate multiple std::future<void>. It would be better to somehow to make it into a single wait instead of several. But I think I might be nitpicking - honestly I'd wait for Executor TS near completion and try to implement or get from somewhere something with the same interface as long as Executor TS is unavailable.

4. The declarations

   ThreadPool(const ThreadPool& other) = delete; 
   void operator=(const ThreadPool& other) = delete;
   

are redundant as the class already has a std::mutex that makes the whole class non-copyable and non-movable. Yeah, it isn't important.

Answer 2

Code Review:

Please put your code in a namespace.

---

ThreadPool(uint8_t numThreads) {
    assert(numThreads > 0);

This is dangerous and will never assert!

The input type here uint8_t which means it is not signed and thus can only be positive (so the only number that it will assert on is if numThreads is zero.

The problem is that C++ integer type conversion will "automatically" convert a negative signed number into a unsigned value (which will usually result in a very large positive value).

ThreadPool  pool(-10);   // Will compile.
                         // Will not assert as -10 is a signed value and will
                         // be auto converted to unsigned at which point it
                         // will be positive and pass the assert test.

---

Here is a good place to use the range based for:

    for(int i = 0; i < threads.size(); ++i){
        threads[i].join();
    }

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

---

The use of typedef has been replaced by the using declaration:

//get return type of the function
typedef decltype(f(args...)) retType;

// Rather use this:
using retType = decltype(f(args...));

---

Not in favor of using decltype() here as you are getting the type of the function (not the return type). Then getting the return type by effectively inferring the calling.

typedef decltype(f(args...)) retType;
std::packaged_task<retType()> task(std::bind(f, args...));
                   ^^^^^^^^^.  Inferred function call

There are actually templates to extract this value directly from the type:

using RetType = invoke_result_t<Func, Args....>;

---

Still on the same statement. It is sort of standard practice for user defined types retType in the case to have an initial uppercase letter. This allows you to easily see types over objects.

---

If you set the shutdown flag this still never leaves the wait() unless there are also objects in the jobQueue.

                notifier.wait(lock, [this] {return !jobQueue.empty(); });

                if(shutdown){
                    break;
                }

You need to test for the shutdown inside thewait()` test method.

                notifier.wait(lock, [this] {return !jobQueue.empty() || shutdown; });

                if(shutdown){
                    break;
                }

---

This is worrying. Though I can't spot the issue:

                //strange bug where it will continue even if the job queue is empty
                if (jobQueue.size() < 1)
                    continue;

---

This seems to be wrong advice.
I though function return values were already R-Values.
Where is my thinking going wrong?

You don't need the std::move() here.

                job = std::move(jobQueue.front());

---

When you are using emplace() you are building the object the container holds by using its constructor.

So here you are creating a temporary object then call the move constructor to move the thread into the container. But you don't need to constructor the temporary thread object as the threadFunc will be forwarded to the thread constructor in the object.

        threads.emplace_back(std::thread(threadFunc));

        // Remove the thread
        threads.emplace_back(threadFunc);

---