2
\$\begingroup\$

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.

\$\endgroup\$
1
  • 3
    \$\begingroup\$ I have rolled back your last edit. Please do not update the code in your question to incorporate feedback from answers, doing so goes against the Question + Answer style of Code Review. This is not a forum where you should keep the most updated version in your question. Please see what you may and may not do after receiving answers. \$\endgroup\$
    – Heslacher
    Jul 26 '19 at 4:42
2
\$\begingroup\$

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
\$\endgroup\$
9
  • \$\begingroup\$ I appreciate for taking time to reply 1. Don't use references to shared_ptr"- Isn't reference counting expensive? 2. Use lock_guard instead of unique_lock in threadFund - condition_variable::wait requires a unique_lock and it has advantages like being able to lock and unlock when needed. \$\endgroup\$
    – user82039
    Jul 24 '19 at 1:05
  • \$\begingroup\$ 3. Don't create objects to replace them anyway - If you look closely I am not actually replacing them. My thread data struct has a deleted copy and assignment operators and therefore I have to create them inside my container. Inside the loop I'm just initialising a member of the struct. \$\endgroup\$
    – user82039
    Jul 24 '19 at 1:05
  • \$\begingroup\$ 4. duplication of packaged_task - In this particular implementation I am already managing ITasks in the main thread and the thread pool must update the state of them regularly. Then the main thread will use accessors to retrieve the state of ITask whenever it needs. Also, I do not want to return a future which adds an extra managing load in the main thread. \$\endgroup\$
    – user82039
    Jul 24 '19 at 1:10
  • \$\begingroup\$ 1. Not it isn't, your locks and vtable lockup probably take more time, besides, you are copying the shared_ptr in the insert in submit anyway. Which is good, otherwise it would defeat the purpose of shared_ptr all along and you could use raw pointers or references. 2. You lock and unlock the the mutex in the inner scope, which is exactly what lock_guard would do. Your manual usage + the wrong scope defeats the RAII purpose of unique_lock. 3. Then I suggest you write a constructor for that, and do that in emplace. When you are concerned about performace, as 1. suggests, here is wh to start. \$\endgroup\$ Jul 24 '19 at 7:43
  • \$\begingroup\$ 4. I don't see from your description how the usage of future would not make the main code easier. \$\endgroup\$ Jul 24 '19 at 7:44

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.