Implementing a thread pool for task execution

Question

I wanted to experiment with a reusable thread pool that allows to add tasks and execute them repetitively. After going through,

https://github.com/progschj/ThreadPool and another Stack Overflow post I came up with the implementation in https://github.com/dragsu/CPPThread.

The main bit of the code is,

ThreadPool.h

#pragma once
#include <condition_variable>
#include <future>
#include <list>
#include <memory>
#include <mutex>
#include <thread>
#include <map>

#include "ITask.h"

/**
 * A simple imlpementation of a thread pool that can excute
 * {@link ITask}s.
 */
class ThreadPool
{

public:
    //----------------------------------------------------------
    //                      CONSTRUCTORS
    //----------------------------------------------------------
    ThreadPool( size_t threadCount );
    ~ThreadPool();

    //----------------------------------------------------------
    //                    INSTANCE METHODS
    //----------------------------------------------------------

    /**
     * Adds a {@link ITask} to a thread.
     * <p/>
     * <b>Note:</b> The task will be added to the thread with a minimum
     * task load.
     *
     * @param task The {@link ITask} to be added to the task queue of a thread
     */
    void submit( std::shared_ptr<ITask> &task );

    /**
     * Removes a {@link ITask} from a thread.
     *
     * @param task The {@link ITask} to be removed from the queue
     * @return <code>true</code> if the task been removed successfully,
     *         otherwise false
     */
    bool remove( std::shared_ptr<ITask> &task );

    /**
     * Sets a wait time for threads before starting another execution
     * round.
     *
     * @param milliseconds Wait time of individual thread after
     *        completing a round of its job queue.
     */
    void setWaitTime( unsigned int milliseconds );

private:
    /*
     * Holds an individual thread and related resources of it
     */
    struct ThreadData
    {
        ThreadData() : threadObj(), jobs(), stop( false )
        {

        }
        ThreadData( const ThreadData& ) = delete;
        ThreadData& operator=( const ThreadData& ) = delete;

        std::thread threadObj;
        std::map<int, std::shared_ptr<ITask>> jobs;
        std::mutex mutexGuard;
        std::condition_variable condition;
        bool stop;
    };

private:
    void threadFunc( ThreadData* data );

private:
    // To keep track of threads so we can join them at the destruction
    std::list< ThreadData > workers;
    unsigned int waitTIme;
};

ThreadPool.cpp

#include "ThreadPool.h"

using namespace std;
ThreadPool::ThreadPool( size_t threadCount ) : waitTIme(1)
{
    for( size_t i = 0; i < threadCount; ++i )
    {
        workers.emplace_back(); // create threadData object inside the list 
    }

    for( auto &thredData : workers )
    {
        thredData.threadObj = thread( &ThreadPool::threadFunc, this, &thredData );
    }
}

ThreadPool::~ThreadPool()
{
    for( auto &worker :workers )
    {
        worker.stop = true;
        worker.condition.notify_one();
    }

    // Join all the threads
    for( auto &worker : workers )
    {
        worker.threadObj.join();
    }
}

void ThreadPool::threadFunc( ThreadData* data )
{
    unique_lock<mutex> lock( data->mutexGuard, defer_lock );

    while( true )
    {
        lock.lock();
        data->condition.wait( lock, [data, &lock]{
            bool shouldContinue = false;
            shouldContinue = data->stop || !data->jobs.empty();
            return shouldContinue;
        });

        // Stop was signaled, let's exit the thread
        if( data->stop ) return;

        for( auto &task : data->jobs )
        {
            task.second->run();
        }
        lock.unlock();
        this_thread::sleep_for( chrono::milliseconds(waitTIme) );
    }
}

void ThreadPool::submit( shared_ptr<ITask> &task )
{
    ThreadData *candidate = nullptr;
    int currentMin = INT_MAX;
    for( auto &thredData : workers )
    {
        int jobCount = thredData.jobs.size();
        if( jobCount < currentMin )
        {
            currentMin = jobCount;
            candidate = &thredData;
        }
    }
    if( candidate )
    {
        unique_lock<mutex> l( candidate->mutexGuard );
        candidate->jobs.insert( pair<int, shared_ptr<ITask>>(task->getId(), task) );
        candidate->condition.notify_one();
    }
}

bool ThreadPool::remove( shared_ptr<ITask> &task )
{
    bool success = false;
    for( auto &thredData : workers )
    {
        unique_lock<mutex> l( thredData.mutexGuard );
        auto it = thredData.jobs.find( task->getId() );
        if( it != thredData.jobs.end() )
        {
            thredData.jobs.erase( it );
            success = true;
            break;
        }
    }
    return success;
}

void ThreadPool::setWaitTime( unsigned int milliseconds )
{
    this->waitTIme = milliseconds;
}

I would be glad if I can get some feedback on this.

Answer 1

First of all, the good things:

The use of Doxygen documentation of classes and functions is very good style, which even pros too often "forget", and I like it. Also the use of a setter for the wait time. Also your try to use your locks in a RAII way is good thinking. Also to define return values at start and have only one return statement, instead of return statements in branches, is very good style and even has sometimes performance benefits. And the best part is that you also checked for spurious wake ups, something most people, including professors for concurrency programming, tend to forget.

However your code has a very high count of issues which make it unusable, and in general I find in overall a bad design too.

---

First your code is very hard to read since you do things twice or do unnecessary stuff. Then you reinvent the wheel often, which could be solved with classes and stuff from std. You use things wrong or don`t seem to know how they are intended to be used. You don't care about exception safety at all, lower the usability for the caller by restricting the usage of an interface for its tasks.

It seems to me you were overwhelmed by the good solution your first link represents, since it uses the best but advanced stuff, but knew run time polymorphism and some basic stuff already. Unfortunately you still need to improve your understandings of C++ basics first and also a bit in concurrency, although already respectable, like the use of atomics and the uselessness of the sleep. Keep on learning, you are on a good path, including by appreciating code reviews and putting yourself out in the open for critique! C++ is hard, but returns that with the fastest code you can get, while being safer and easier to use then C.

Here are some of the problems I found (far from complete):

1. Don't use references to std::shared_ptr, you always want your own copy, thats the reason whz you create a new one when you insert them to, but reduce caller usage
2. submit will not tell the caller if the task was submitted
3. Use of an own Interface Class ITask, why the overhead of run-time polymorphism and duplication of std::packaged_task
4. Missing header #include <climits> int currentMin = INT_MAX;
5. Use lock_guard instead of unique_lock in threadFund
6. Consider the precision of your data types, for instance you assign the size_t of jobs.size() to an int, which is very bad, since the int could overflow and lead to undefined behavior
7. Don't create objects to replace them anyway as in ThreadPool::ThreadPool, you could have done the things you want already in the emplace_back and also replace the for loop with std::generate_n