16
\$\begingroup\$

The class supplies a simple thread pool that uses the thread class from C++11.

When I schedule the job, I pass two std::function<void()> objects that the pool will execute: the first function is the real job, the second is executed to send some sort of notification that the job has been done.

I'm not sure if I should merge the two std::function in one and let the user decide if to send the notification or not.

Additionally, not so sure about the naming executeJob vs scheduleJob.

The checking of maxJobsInQueue is not there yet, but when done I cannot decide between throwing exception when the queue is full vs blocking in executeJob until the job can be inserted.

Any other improvement you can suggest?

evr_threadpool.h:

#ifndef EVR_THREADPOOL_H
#define EVR_THREADPOOL_H

#include <thread>
#include <vector>
#include <list>
#include <memory>
#include <functional>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <tuple>

/**
 * @brief The ThreadPool class launches a pre-defined number of threads and
 *        keeps them ready to execute jobs.
 *
 * Each job is composed by two functions (std::function<void()>):
 * the two functions are executed one after another, and the second one
 *  usually sends some sort of notification that the job has been done.
 *
 */
class ThreadPool
{
public:

    /**
     * @brief Initializes the pool and launches the threads.
     *
     * @param numThreads number of threads to launch.
     * @param maxJobsInQueue
     */
    ThreadPool(size_t numThreads, size_t maxJobsInQueue);


    /**
     * @brief Destructors
     *
     * Sends a "terminate" signal to the threads and waits for
     *  their termination.
     *
     * The threads will complete the currently running job
     *  before checking for the "terminate" flag.
     */
    virtual ~ThreadPool();


    /**
     * @brief Schedule a job for execution.
     *
     * The first available thread will pick up the job and run it.
     *
     * @param job             a function that executes the job. It is called
     *                         in the thread that pickec it up
     * @param notificationJob a function that usually sends a notification
     *                         that the job has been executed. it is executed
     *                         immediately after the first function in the same
     *                         thread that ran the first function
     */
    void executeJob(std::function<void()> job, std::function<void()> notificationJob);


private:
    /**
     * @brief Function executed in each worker thread.
     *
     * Runs until the termination flag m_bTerminate is set to true
     *  in the class destructor
     */
    void loop();


    /**
     * @brief Returns the next job scheduled for execution.
     *
     * The function blocks if the list of scheduled jobs is empty until
     *  a new job is scheduled or until m_bTerminate is set to true
     *  by the class destructor, in which case it throws Terminated.
     *
     * When a valid job is found it is removed from the queue and
     *  returned to the caller.
     *
     * @return the next job to execute
     */
    std::pair<std::function<void()>, std::function<void()> > getNextJob();


    /**
     * @brief Contains the running working threads (workers).
     */
    std::vector<std::unique_ptr<std::thread> > m_workers;


    /**
     * @brief Queue of jobs scheduled for execution and not yet executed.
     */
    std::list<std::pair<std::function<void()>, std::function<void()> > > m_jobs;


    /**
     * @brief Mutex used to access the queue of scheduled jobs (m_jobs).
     */
    std::mutex m_lockJobsList;


    /**
     * @brief Condition variable used to notify that a new job has been
     *        inserted in the queue (m_jobs).
     */
    std::condition_variable m_notifyJob;


    /**
     * @brief This flag is set to true by the class destructor to signal
     *         the worker threads that they have to terminate.
     */
    std::atomic<bool> m_bTerminate;


    /**
     * @brief This exception is thrown by getNextJob() when the flag
     *         m_bTerminate has been set.
     */
    class Terminated: public std::runtime_error
    {
    public:
        Terminated(const std::string& what): std::runtime_error(what) {}
    };

};

#endif // EVR_THREADPOOL_H

evr_threadpool.cpp

#include "evr_threadpool.h"


/*
 * Constructor
 *************/
ThreadPool::ThreadPool(size_t numThreads, size_t maxJobsInQueue):
    m_workers(numThreads), m_bTerminate(false)
{
    for(std::unique_ptr<std::thread>& worker: m_workers)
    {
        worker.reset(new std::thread(&ThreadPool::loop, this));
    }
}


/*
 * Destructor
 ************/
ThreadPool::~ThreadPool()
{
    {
        std::unique_lock<std::mutex> lockList(m_lockJobsList);
        m_bTerminate = true;
        m_notifyJob.notify_all();
    }

    for(std::unique_ptr<std::thread>& worker: m_workers)
    {
        worker->join();
    }
}


/*
 * Schedule a job
 ****************/
void ThreadPool::executeJob(std::function<void()> job, std::function<void()> notificationJob)
{
    std::unique_lock<std::mutex> lockList(m_lockJobsList);
    m_jobs.push_back(std::pair<std::function<void()>, std::function<void()> >(job, notificationJob));
    m_notifyJob.notify_one();
}


/*
 * Retrieve the next job to execute
 **********************************/
std::pair<std::function<void()>, std::function<void()> > ThreadPool::getNextJob()
{
    std::unique_lock<std::mutex> lockList(m_lockJobsList);

    while(!m_bTerminate)
    {
        if(!m_jobs.empty())
        {
            std::pair<std::function<void()>, std::function<void()> > job = m_jobs.front();
            m_jobs.pop_front();
            return job;
        }

        m_notifyJob.wait(lockList);
    }

    throw Terminated("Thread terminated");
}


/*
 * Function executed by each worker
 **********************************/
void ThreadPool::loop()
{
    try
    {
        for(;;)
        {
            std::pair<std::function<void()>, std::function<void()> > job = getNextJob();
            job.first();
            job.second();
        }
    }
    catch(Terminated& e)
    {
    }
}
\$\endgroup\$
8
\$\begingroup\$

A few thoughts:

  • If I understand this correctly, if one of your job functions throws an exception, it ends processing for your entire thread pool; Threads that are running (but did not throw an exception) will keep running, but the ThreadPool object may be deleted as part of stack unwinding. Consider adding another callback to your executeJob similar to std::function<void(/* err details here */)> on_error.

  • Your executeJob function doesn't actually execute the job - I would rename it to "enqueueJob" or "addJob" or similar.

  • Your Terminated class is a runtime error specialization, but your documentation (or anything in the code really) does not explain why it is a runtime error to have an empty queue.

I assume you use this to stop, not to signal an error state. This may not look like a big thing, but a runtime_error (or specialization) should actually signal an error during the runtime of your applicaition).

For example, the Visual Studio debugger can be set to break automatically when a runtime_error is thrown in the application. Using this implementation would screw up debugging without you meaning to.

Consider either using a second condition variable (or similar) instead of throwing an exception, or adding docs to the source saying "Terminate is a runtime_error specialization because having no jobs to run is an error in this and this case".

\$\endgroup\$
7
\$\begingroup\$

You tagged this as C++11 but the code looks like you want to make it C++03 compatible (most noticably the space in > > and the lack of auto inside for-loops). Most of my suggestions require C++11 and are wrong for C++03.

  • You should add a few std::moves for efficiency, for example inside ThreadPool::executeJob:
    m_jobs.emplace_back(std::move(job), std::move(notificationJob));

  • I do not see the point of maxJobsInQueue. Is there ever a reason to not pass (size_t)-1 or equivalent? I would either remove that entirely and not worry about exceptions either or at least give it that default value.

  • This comment is not easy enough to understand:

    * Each job is composed by two functions (std::function<void()>):
    * the two functions are executed one after another, and the second one
    * usually sends some sort of notification that the job has been done.
    

    What is the point of having two functions? Why not just have one? In a case where I have two functions f and g where f does work and g sends a notification I would just pass []{f(); g();}; and have the same effect with a simpler interface. In case I only have f and don't care about notifications I have to pass an empty lambda as g, which is suboptimal.

  • ThreadPool::m_workers is a std::vector<std::unique_ptr<std::thread>>. What is the unique_ptr for? Just use a std::vector<std::thread> instead. This changes the constructor code to something like

    ThreadPool::ThreadPool(size_t numThreads, size_t maxJobsInQueue = (size_t)-1) :
        m_bTerminate(false)
    {
        m_workers.reserve(numThreads);
        for (size_t i = 0; i < numThreads; i++)
            m_workers.emplace_back(&ThreadPool::loop, this);
    }
    
  • m_jobs is a std::list<std::pair<std::function<void()>, std::function<void()>>>. As a rule of thumb linked lists are useful when you have nothing better to teach in a computer science class and when you require stable iterators. You do not seem to take advantage of either one, so I would suggest std::deque instead.

  • I do not like most of the comments inside evr_threadpool.cpp. I know what a constructor and destructor looks like, so those comments add no information.
    Schedule a job is a contradiction to the function name executeJob. Execute or schedule? Pick one for the function name and remove the comment.
    Retrieve the next job to execute is somewhat confusing. The function is already called getNextJob so the Retrieve the next job-part is redundant. to execute means I can only get the job for executing, so bad things happen if I, say, use getNextJob to clear the queue in a clear function? Not sure what exactly the to execute part is meant to tell me.
    I like the comment Function executed by each worker because it adds useful context information that I cannot see from looking at the loop-function.

  • When testing your threadpool I needed a wait_until_all_jobs_are_done-function, because otherwise ThreadPools destructor sets the terminate flag. Consider adding that function.

  • Consider using std::make_pair instead of manually naming the types in situations like m_jobs.push_back(std::pair<std::function<void()>, std::function<void()> >(job, notificationJob));.
    m_jobs.push_back(std::make_pair(job, notificationJob)); looks nicer.

  • I would consider making a typedef or using declaration for std::pair<std::function<void()>, std::function<void()>>. Not because the type is complicated or long, but because the type is duplicated across the code. If you figure out that the second function should get an enum ErrorType parameter so the notification can tell what happened you have a lot of error prone copy and paste to do which could be done in a single place instead.

\$\endgroup\$

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.