33
\$\begingroup\$

I have been working on this thread pool for awhile to make it as simple to use as possible. I need some tips on improving performance, and some good ways to test its performance. I was wondering if anyone had any opinions/suggestions!

Here is the class:

#pragma once

#include<thread>
#include<vector>
#include<queue>
#include<mutex>
#include<condition_variable>
#include<functional>
#include<future>

#define MAX_THREADS std::thread::hardware_concurrency() - 1;

//portable way to null the copy and assignment operators
#define NULL_COPY_AND_ASSIGN(T) \
    T(const T& other) {(void)other;} \
    void operator=(const T& other) { (void)other; }

/* ThreadPool class
It is a singleton. To prevent spawning
tons of threads, I made it a singleton */
class ThreadPool{
public:

    //getInstance to allow the second constructor to be called
    static ThreadPool& getInstance(int numThreads){
        static ThreadPool instance(numThreads);

        return instance;
    }

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

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

        //package the task
        std::packaged_task<retType()> task(std::move(std::bind(f, args...)));

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

        //get the future from the task before the task is moved into the jobqueue
        std::future<retType> future = task.get_future();

        //place the job into the queue
        JobQueue.emplace( std::make_shared<AnyJob<retType> > (std::move(task)) );

        //notify a thread that there is a new job
        thread.notify_one();

        //return the future for the function so the user can get the return value
        return future;
    }

    /* utility functions will go here*/
    inline void resize(int newTCount){

        int tmp = MAX_THREADS;
        if(newTCount > tmp || newTCount < 1){
            tmp = numThreads;
            numThreads = MAX_THREADS;
            Pool.resize(newTCount);
            for (int i = tmp; i != numThreads; ++i) {
                Pool.emplace_back(std::thread(&ThreadPool::threadManager, this));
                Pool.back().detach();
            }
        }
        else if (newTCount > numThreads) {
            uint8_t tmp = numThreads;
            numThreads = newTCount;
            Pool.resize(numThreads);
            for (int i = tmp; i != numThreads; ++i) {
                Pool.emplace_back(std::thread(&ThreadPool::threadManager, this));
                Pool.back().detach();
            }
        }
        else {
            numThreads = (uint8_t)newTCount;
            Pool.resize(newTCount);
        }


    }

    inline uint8_t getThreadCount(){
        return numThreads;
    }

private:

    //used polymorphism to store any type of function in the job queue
    class Job {
    private:
        std::packaged_task<void()> func;
    public:
        virtual ~Job() {}
        virtual void execute() = 0;
    };

    template <typename RetType>
    class AnyJob : public Job {
    private:
        std::packaged_task<RetType()> func;
    public:
        AnyJob(std::packaged_task<RetType()> func) : func(std::move(func)) {}
        void execute() {
            func();
        }
    }; 
    // end member classes

    //member variables
    uint8_t numThreads; // number of threads in the pool
    std::vector<std::thread> Pool; //the actual thread pool
    std::queue<std::shared_ptr<Job>> JobQueue;
    std::condition_variable thread;// used to notify threads about available jobs
    std::mutex JobMutex; // used to push/pop jobs to/from the queue
    //end member variables

    /* infinite loop function */
    inline void threadManager() {
        while (true) {

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

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

            (*JobQueue.front()).execute();

            JobQueue.pop();
        }
    }

    /*  Constructors */
    ThreadPool(); //prevent default constructor from being called

    //real constructor that is used
    inline ThreadPool(uint8_t numThreads) : numThreads(numThreads) {
        int tmp = MAX_THREADS;
        if(numThreads > tmp){
            numThreads = tmp;
        }
        Pool.reserve(numThreads);
        for(int i = 0; i != numThreads; ++i){
            Pool.emplace_back(std::thread(&ThreadPool::threadManager, this));
            Pool.back().detach();
        }
    }
    /* end constructors */


NULL_COPY_AND_ASSIGN(ThreadPool);
}; /* end ThreadPool Class */

Here is example usage:

#include "ThreadPool.h"
#include <iostream>

int main(){

    ThreadPool& pool = ThreadPool::getInstance(4); //create pool with 4 threads

    auto testFunc = [](int x){ return x*x; };

    auto returnValue = pool.push(testFunc, 5);

    std::cout << returnValue.get() << std::endl;

    return 0;
}
\$\endgroup\$
36
\$\begingroup\$

Funny how the universe works - I just finished my own implementation of a thread pool (albeit in C++17), and it looks a lot like yours. I found this question on the front page when I went to post my own - here's hoping we're both on the right track!

Mark the copy c'tor and operator= as deleted

Instead of actually implementing something that you don't want to be used, in C++11 and newer you can explicitly disallow invocations of the copy constructor and assignment operator:

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

Or, if you would rather stick with the macro:

#define DISALLOW_COPY_AND_ASSIGN(T) \
    T(const T&) = delete; \
    T &operator=(const T&) = delete;

When a function is marked as deleted, attempting to invoke it results in a compile-time error.

Do not declare a default constructor

ThreadPool(); //prevent default constructor from being called

//real constructor that is used
inline ThreadPool(uint8_t numThreads);

The declaration of the default constructor above does nothing, other than obscure the real reason the program won't build if you try to invoke the default ctor. A default constructor is compiler-defined if, and only if, there are no constructors explicitly declared.

By declaring a default constructor but not implementing it, you make it legal for a compiler to build code that attempts to invoke it - only to find yourself in a position where the linker will instead fail, which is not the expected outcome of building code that is (supposed to be) ill-formed.

Drop the inline specifier

According to 9.1.6 [dcl.inline]:

  1. A function defined within a class definition is an inline function.

Your inline specifiers do nothing!

Get rid of Job::func

What is the point of declaring Job::func, only to have it shadowed by AnyJob<T>::func? The base class member variable never even gets touched - it's just adding complexity to the code without any purpose. The only point of Job is to serve as a common base to different types of polymorphic functors (your AnyJob<T>s). Make it as simple as possible.

Use good names for your variables, and get rid of redundant ones

//member variables
uint8_t numThreads; // number of threads in the pool
std::vector<std::thread> Pool; //the actual thread pool
std::queue<std::shared_ptr<Job>> JobQueue;
std::condition_variable thread;// used to notify threads about available jobs
std::mutex JobMutex; // used to push/pop jobs to/from the queue
//end member variables

Why do you have numThreads in the first place, if your vector of threads already keeps track of that information? Why is thread the name of a condition_variable? I'd expect that to be a thread object or container. Pool is also not a great name - something such as workers or threads would be better.

Further, what's with the naming convention? You have camel case (numThreads) and Pascal case (JobQueue) mixed together - that's pretty weird. If you want to take it a step further and make it more C++ey, the C++ Core Guidelines advise you to prefer underscore case.

Make threadManager private - or get rid of it altogether

ThreadPool::threadManager is clearly not meant to be called by the thread that owns the ThreadPool object. Why make it available? It's always better to make it difficult or impossible for mistakes to be done (if within the realm of reason).

In light of the fact that it its only purpose is transitioning the worker threads from working to idle and vice versa, the name is a bit odd. Further, why not simply pass a lambda to the std::thread constructor? This would make your implementation more succinct.

And other things

I would also take into consideration things such as the constructor silently capping the number of threads to a set number - is that Funny Behavior™, that the programmer should instead be warned about? Maybe we want the program to be able to have more threads than the processor? It could be possible that one of the tasks given to you waits on I/O a significant amount of time - context switching can be your friend!

but, overall...

From what I can tell, this is good code. I was fairly nitpicky with it - but with the intention that in doing so, you can make your code better and hopefully learn a few tricks along the way. Good job!

\$\endgroup\$
  • 1
    \$\begingroup\$ Yep that naming convention just went to hell. Mostly because I was just trying to get everything to work. Ill get rid of the numThreads as im not sure why I put it there. As far as inline goes, the functions are inside a class definition, so I believe it applies here. Also my threadManager() function is already private. Thank you for the tips! Ill be making the changes you suggest! \$\endgroup\$ – Paul Jun 4 at 6:31
  • 1
    \$\begingroup\$ @Paul my mistake about threadManager - it is a public member of a privately declared class, hence my confusion. My point about the inline declarations is that since all of your functions are defined within the class definition, they are all already implicitly declared inline. But regardless, I'm glad I could help! You've got some good code going on there :) \$\endgroup\$ – osuka_ Jun 4 at 6:43
14
\$\begingroup\$

@osuka_ and @anderas gave some very good advice. I just have a couple things I want to add:

Macros

#define MAX_THREADS std::thread::hardware_concurrency() - 1;

While it doesn't really make sense to do arithmetic on this, macros like this should be surrounded by parentheses so the order of operations works correctly. Without it you get weird stuff like (assume std::thread::hardware_concurrency() is 4): MAX_THREADS * 5 => 4 - 1 * 5 => -1 instead of MAX_THREAD * 5 => (4 - 1) * 5 => 15. Also, macros should not end in a semicolon. The user of the macro should add the semicolon (like you did - int tmp = MAX_THREADS;).

Alternatively, avoid macros altogether (this is C++ after all) and use const auto MAX_THREADS = std::thread::hardware_concurrency() - 1;

In push():

std::unique_lock<std::mutex> lock(JobMutex);

This is a minor point, but seeing this I would expect something to unlock lock at some point (like the condition variable in threadManager(), which by the way has the confusing name thread). If the lock should be held until the end of its scope like in this case use a std::lock_guard<std::mutex> instead.

\$\endgroup\$
  • 1
    \$\begingroup\$ hardware_concurrency is not constexpr (compile time constant), so your MAX_THREADS variable should be const. When discussing the macro, you should mention that the macro definition should not end in a semicolon. \$\endgroup\$ – 1201ProgramAlarm Jun 4 at 14:18
  • \$\begingroup\$ @1201ProgramAlarm Thanks, edited. I didn't even notice the semicolon! \$\endgroup\$ – Kevin Jun 4 at 15:05
11
\$\begingroup\$

You've received good style advice so far.

But your pool doesn't work. So let's try to address that.

1. 'resize' is both wrong and not needed.

As others mentioned, it's not thread safe. Making it so is exceedingly difficult. And in practice, you don't need to change the thread pool's size after after start.

2. You only use one thread at a time.

In threadManager(), you execute jobs with the mutex taken. That means a single job a time can be executed, negating the very reason you created the pool in the first place.

Fix it by copying the job to a local variable, pop it from the queue, unlock the mutex and only then execute it.

3. shared_ptr is slower than unique_ptr

and not needed for the jobs queue. But best to get rid of them both, as suggested at #6.

4. detach() is lazy and dangerous

There are only few good uses of it in practice, because it will kill threads at program exit, rather than gracefully wait for jobs to be completed.

Replace it with join() in the class destructor. (One more reason to stop using that singleton that a few others explain it's bad).

You will need to add extra code to control the exit:

  • add an atomic bool isStopping.
  • initialize it to false in the constructor's initializer list.
  • On destructor, set it to true.
  • Then call notify_all() on the condition variable. This way all threads are woken up and can test for the value of isStopping.
  • in threadManager(), before and after executing the job, check if isStopping is set to true, and return if needed, exiting the thread.

You will also need to adjust the condition variable lambda to return if isStopping is true.

  • Finally, back in the destructor, call join() on all threads.

You can play with two different exit strategies: execute all pending jobs first or discard them. Discarding is a good default, because otherwise the exit will be delayed for an unspecified amount of time untill the queue is processed.

5. That singleton

It not only prevents you from closing the thread pool gracefully (because singleton destructors are called very late in the exit process, but can prevent genuine use-cases of your pool. Let's say you want to process two kinds of tasks - one whick is very fast and one which is very slow. Imagine you queue many slow tasks in the pool, which will make the fast tasks wait for them all to be executed.

Having two pools, one for the fast, one for the slow ones allows you to separate resources and offer better performance for the fast ones.

6. You can replace both Job and AnyJob with std::function

and a proper initializer with a lambda which captures the packaged_task.

7. There is no good default for number of threads

A purely computational load - like scientific simulations - running on a dedicated server will best work with a thread per core (actually even this basic assumption is wrong in the face of hyperthreading). But this is a tiny minority of cases. If you do I/O, you can use effectively many more threads than cores. If you use multiple pools for different parts of your app (one for I/O, one for processing), you'll need to choose wisely the resource distribution between them. And in a server shared by more than one app, you need to keep other tenants in mind.

I suggest removing the use of hardware_concurrency altogether. It invites the user to take lazy and poor decisions.

\$\endgroup\$
  • \$\begingroup\$ Those are actual compelling arguments for not using a singleton, Ill take that into consideration. I am about to post revised code ( I actually found that bug where one thread is being used at a time because of the mutex). I would like to see an example of what you mean in #6. I found no way of storing any type of packaged task, as a std::function. I require packaged_task because they allow the use of futures, where std::function, does not. I was not too sure how to get .join() to work. It seemed that if I didnt join in the constructor, I got compile errors. \$\endgroup\$ – Paul Jun 4 at 19:41
  • \$\begingroup\$ Suggestion for the atomic bool: initialize it to false in the constructor. Don't construct it and then do an atomic store with mo_seq_cst by assigning to it. \$\endgroup\$ – Peter Cordes Jun 6 at 4:05
  • \$\begingroup\$ Two serious problems with the suggestions here: first, discarding tasks is going to result in exceptions thrown (since you’re getting rid of a future that was never fulfilled). Secondly, std::function cannot be used here as a result of std::packaged_task not being copiable (only movable). The standard requires an std::function be constructed from a Callable and CopyAssignable object; the tasks, in this case, are only one of those. \$\endgroup\$ – osuka_ Jun 8 at 17:40
9
\$\begingroup\$

std::thread::hardware_concurrency() can return 0,you should handle this case.

\$\endgroup\$
8
\$\begingroup\$

@osuka_ has already provided a thorough review, but I want to show an important point that is missing from his review: The choice of making your class a singleton and the way that you implemented it.

I suppose that you have a really good reason to make this class a singleton. But sometimes, the singleton pattern is considered an anti-pattern because it often makes testing harder (among other downsides). Alternatives would be to simply make the ThreadPool a normal class and use dependency injection/inversion techniques to locate a shared object.

A possible source for problems can be found in your getInstance method:

    //getInstance to allow the second constructor to be called
    static ThreadPool& getInstance(int numThreads){
        static ThreadPool instance(numThreads);

        return instance;
    }

This constructs a single instance with a given value of numThreads on the first usage. The problem is: you have to take care that either

  • this method is ALWAYS called with the same numThreads
  • or the very first usage is ALWAYS at a point where you can be completely sure that the value is correct.

Both result in maintainability issues. Consider for example the following function:

void doWork() {
    auto& pool = ThreadPool::getInstance(4);
    // ... use the pool
}

This would later get changed to

void doWork() {
    prepareWork();
    auto& pool = ThreadPool::getInstance(4);
    // ... use the pool
}

Here, you would have to check whether prepareWork() also uses the ThreadPool and, if so, whether it passes the correct number of threads. In larger codebases, this can easily lead to avoidable bugs.

Conclusion: Please reconsider whether making this class a singleton is really the best choice, and maybe select a better way of initializing the number of threads.

\$\endgroup\$
  • \$\begingroup\$ I know a lot of people consider singletons to be a bad practice. The reason I did this was because of hardware limitations. I did not want the user to be able to create many different instances. 100 instances of a thread pool all using the max threads possible seemed like a bad idea to me, but I have no experience in what would happen in that case. Would it just slow to a crawl until no memory is left or crash? \$\endgroup\$ – Paul Jun 4 at 19:13
  • \$\begingroup\$ Implementation-defined, but making the thread pool itself a singleton is still the wrong way to handle this. \$\endgroup\$ – Useless Jun 5 at 14:59
  • \$\begingroup\$ @Paul: If that many threads are just waiting for I/O (e.g. listening on network sockets), then no big deal. If they're all doing CPU-bound work that doesn't touch much memory, everything will be slow but not unusable on e.g. an 8-core system. Your OS's scheduling timeslice is short enough that interactive programs will get CPU time occasionally. (And with a good OS with a heuristic that boosts the priority of tasks that "look interactive", e.g. don't use up their full timeslice, it might not be very noticeable). If they use much RAM, you get swap space thrashing or at least L3 cache. \$\endgroup\$ – Peter Cordes Jun 6 at 4:15
5
\$\begingroup\$

And another one for posterity ...

Macros - just say No.

#define MAX_THREADS std::thread::hardware_concurrency() - 1;
  • don't put a semicolon at the end of a macro, it means you can't write resize(MAX_THREADS)
  • do parenthesize your macros, like

    #define MAX_THREADS (std::thread::hardware_concurrency() - 1)
    

    so you can write resize(MAX_THREADS/2)

  • don't use macros in the first place, we're not writing K&R C any more

    unsigned int max_threads() {
        return std::thread::hardware_concurrency() - 1;
    }
    
  • don't use thread::hardware_concurrency() either - it doesn't guarantee what you expect anyway:

    ... The value should be considered only a hint.

Constructors

//portable way to null the copy and assignment operators

... it isn't and it doesn't. Just write = delete explicitly, as osuka_ says.

ThreadPool(); //prevent default constructor from being called

... you mean because it's private? Just use = delete here too. That expresses your intent so clearly you don't even need a comment, which is why it was added the language.

inline ThreadPool(uint8_t numThreads)

always mark single-argument constructors explicit unless you want implicit conversion from uint8_t (I'm pretty sure you don't). And inline does nothing here.

For some reason you're limiting thread pools to std::numeric_limits<uint8_t>::max() threads at construction time, but allow them to be later resized up to std::numeric_limits<int>::max(). If you really wanted, for some reason, a pool of 260 threads, that's a bit awkward.

Singletons

/* ... To prevent spawning tons of threads, I made it a singleton */
  • Avoid singletons anyway where you can
  • If you need a singleton (and you really don't), write a Singleton<T> wrapper instead of baking it into the class. Singleton-ness is not a core concern of a thread pool, and a sane user might quite reasonably want two thread pools, each with a small number of threads, for separating different types of task.
  • You already allow your pool to be up to hardware_concurrency()-1 in size, and this could legally be enormous, so this doesn't avoid the problem anyway. At some point, you need to just trust that your users aren't going to start a million thread pools with INT_MAX threads each.
  • Your getInstance method constructs the single instance with numThreads threads (implicitly truncated to uint8_t as mentioned above) on the first call, but subsequent calls will ignore the numThreads parameter entirely. This is confusing and error-prone, which strongly suggests the instance management and configuration shouldn't be coupled at all.

Threads

  • when you resize the pool smaller, you detach the surplus worker threads, but you never tell them to die. Any thread pool should have a way to tell worker threads to exit. A virtual bool shutdown() method on the Job would be sufficient, but you also need some way to tell the resize method which threads exited, so it can clean up the pool correctly.
  • exactly the same cleanup problem on destruction. If you remove the resize() method entirely, as suggested in another answer, you can use a simple bool shuttingDown flag - otherwise, you can resize to zero and use the same shutdown notification mechanism.
  • your worker thread keeps holding the mutex while executing each task, so you'll have virtually no actual concurrency.

    Move the current task ptr from the queue while holding the lock, and then use a scoped unlocker (like a unique_lock, but exactly backwards) to release the mutex while executing it.

\$\endgroup\$
3
\$\begingroup\$
/* ThreadPool class
It is a singleton. To prevent spawning
tons of threads, I made it a singleton */

ThreadPools can be useful as both singletons and not singletons.

There is zero need to mix the ThreadPool implementation with the "this is a singleton" implementation. There is a lot of need to not; there are some nasty things you have to look out for with singletons in non-trivial applications.

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

While std::thread and std::async supports passing arguments to a task, that is in my experience a needless complication.

Just take a nullary function. The caller can bundle up their arguemnts into a lambda really easily if they need to.

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

        //package the task
        std::packaged_task<retType()> task(std::move(std::bind(f, args...)));

An example of why what you did is a bad idea. You used std::bind. If f was already the result of a std::bind, this doesn't do the same thing as calling f with the arguments args....

Instead it does the insane thing std::bind does.

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

I'd advise, using the single responsibility principle, to split your job queue off from your thread pool.

template<class T>
struct threadsafe_queue;

About half of the complexity of your ThreadPool is the queue, the other half is managing threads. By splitting the two, you have two piece of code each half as complex.

And a threadsafe_queue can be reused elsewhere.

    /* utility functions will go here*/
    inline void resize(int newTCount){

Nix inline.

resize is a horrible name. You aren't a container.

        int tmp = MAX_THREADS;

Having more threads that hardware concurrency size -1 is a perfectly sane thing to do if you know you have lots of blocking operations.

This kind of logic does not belong in a class named ThreadPool.

Having easy access to "spawn max concurrency threads" or "max concurrency -1 threads" is good. Setting a hard limit is bad.

                Pool.back().detach();

Detaching threads is the wrong thing to do 99.999% of the time. Don't do it. Threads running after main ends is extremely dangerous and toxic.

        else {
            numThreads = (uint8_t)newTCount;
            Pool.resize(newTCount);
        }

You need to be really clear about what shrinking the number of threads means.

    //used polymorphism to store any type of function in the job queue
    class Job {

While std::function<void()> isn't sufficient, as packaged_task<R()> is move only, a type like it is pretty sane and useful.

The simplest to find type that can store a packaged_task<R()> is a packaged_task<void()>. Try it.

Replace Job with std::packaged_task<void()>. Stop messing with pointers.

But really, find a move-only std::function and use that. When in doubt, use value semantics.

    //member variables
    uint8_t numThreads; // number of threads in the pool
    std::vector<std::thread> Pool; //the actual thread pool

Belongs in ThreadPool. But really, numThreads is redundant; Pool.size() has that information.

Maybe have a second vector of "parked threads".

Also, always =0 or whatever data.

    std::queue<std::packaged_task<void()>> JobQueue;
    std::condition_variable thread;// used to notify threads about available jobs
    std::mutex JobMutex; // used to push/pop jobs to/from the queue
    //end member variables

Belongs in threadsafe_queue.

    /* infinite loop function */
    inline void threadManager() {
        while (true) {

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

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

            (*JobQueue.front()).execute();

            JobQueue.pop();
        }

don't do the work while the mutex is engaged.

This is an example of where mixing the thread safe queue with the thread pool has messed you up.

threadsafe_queue< std::packaged_task<void()> > jobs;

template<class F>
auto push_task( F&& f ) {
  using dF = std::decay_t<F>;
  using R = std::result_of_t< dF&() >;
  std::packaged_task< R() > task = std::forward<F>(f);
  auto retval = task.get_future();
  jobs.push_back( std::move(task) ); // may require an explicit cast
  return retval;
}

wow, that is a simpler push.

struct killable_thread {
  std::thread t;
  std::shared_ptr<std::atomic<bool>> kill;
  killable_thread( std::thread tin, std::shared_ptr<std::atomic<bool>> kin ):
    t(std::move(tin)),
    kill(std::move(kin))
  {}
};
std::vector<killable_thread> threads;

void add_thread(std::size_t n = 1) {
  while (n > 0 ) {
    auto kill = std::make_shared<std::atomic<bool>>(false);
    std::thread t( [this, kill]{
      while (!*kill) {
        auto job = jobs.pop_back();
        if (!job)
          return; // done
        (*job)();
      }
    });
    threads.emplace_back( std::move(t), std::move(kill) );
    --n;
  }
}

remove_threads sets *kill and moves the threads somewhere else to clean up later.

Possibly we augment the thread code to have them move themselves to a "to be cleaned up" queue, and other threads can even clean up that queue, leaving at most one "waiting to be joined" task in that queue.

Note that we want threadsafe_queue to support .abort() -- that means pop_back returns an optional<T> (boost or not) instead of a T, so it can return "pop failed". An alternative is that it could throw.

If the queue is killed, all pops abort.

    /*  Constructors */
    ThreadPool(); //prevent default constructor from being called

=delete.

//real constructor that is used
inline ThreadPool(uint8_t numThreads)  {
        int tmp = MAX_THREADS;
        if(numThreads > tmp){
            numThreads = tmp;
        }

again, anti-pattern.

Entire body should read:

ThreadPool(uint8_t numThreads)  {
  add_thread(numThreads);
}

DRY -- don't repeat yourself. There should be one function for adding threads, uesd both here and elsewhere

NULL_COPY_AND_ASSIGN(ThreadPool);

really?

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

that is so taxing you'll write a macro to avoid typing it?

Here is a threadsafe queue:

template<class T>
struct threaded_queue {
  using lock = std::unique_lock<std::mutex>;
  void push_back( T t ) {
    lock l(m);
    data.push_back(std::move(t));
    cv.notify_one();
  }
  boost::optional<T> pop_front() {
    lock l(m);
    cv.wait(l, [this]{ return abort || !data.empty(); } );
    if (abort) return {};
    auto r = std::move(data.front());
    data.pop_front();
    return std::move(r);
  }
  void terminate() {
    lock l(m);
    abort = true;
    data.clear();
    cv.notify_all();
  }
  ~threaded_queue()
  {
    terminate();
  }
private:
  std::mutex m;
  std::deque<T> data;
  std::condition_variable cv;
  bool abort = false;
};

another operation I find useful is the ability to "abandon all queued tasks", without aborting the queue.

You'll notice how much smaller and more clean this is when it isn't mixed in with the thread pool code. The thread pool code also gets cleaner.

\$\endgroup\$
  • 1
    \$\begingroup\$ Could be just me, but it sounds like wrapping a queue in another class entirely is a significant overkill. Wrapping a thread is also not anything you can’t solve by adding a line to the waiting task implementation. More importantly, it’s awkward (impossible?) to use a queue with T = std::packaged_task<void()> if you want to return a future with any other T. The point of Job and AnyJob is to add some type erasure to this structure, so that you can simultaneously return a meaningful value through your future, and store tasks that return different types. \$\endgroup\$ – osuka_ Jun 8 at 18:07
  • 1
    \$\begingroup\$ And on second thought: mimicking std::thread and std::bind is a good idea for many reasons: the programmer is probably familiar with it, it’s easier to use, and you avoid making them #include <functional> for a bind call. There are many very desirable advantages to his approach, at little cost (since it doesn’t add much complexity) \$\endgroup\$ – osuka_ Jun 8 at 18:12
  • \$\begingroup\$ @osuk nobody should be using std bind. Just use lambdas. Packaged task void is a type erasure class; it can store packaged task R for any type R. Using the thread call bindings is much more difficult than using a lambda, because the r/l value transformations of arguments passed are sufficiently obscure that it took msvc multiple versions to get their implementation correct. Most C++ programmers won't get it right; the OP doesn't get it right. You probably didn't notice they got it wrong. Users probably don't know the right magic behaviour, so both wrong and right surprise users. \$\endgroup\$ – Yakk Jun 8 at 21:00
  • \$\begingroup\$ Why should nobody use bind? It does take a decent understanding of the language to use, but I can't see why it should be avoided at all costs as you seem to suggest. Lambdas have a few oddities (such as the fact that move-capturing a non-copiable object makes the lambda non-copiable). If you understand move semantics and reference binding, implementing something correctly isn't that difficult. The problem with using packaged_task<void()> is that you cannot, then, accept a callable that returns any type (unless some type erasure happens before it is added to the queue). \$\endgroup\$ – osuka_ Jun 8 at 21:39
  • 1
    \$\begingroup\$ @Yakk, std::packaged_task<int()> foo; std::packaged_task<void()> bar = std::move(foo); does not compile, but std::packaged_task<int()> foo; std::packaged_task<void()> bar(std::move(foo)); does. \$\endgroup\$ – Evg Jun 28 at 21:24
2
\$\begingroup\$

There are a lot of good comments here already, overall good work ...

The resize function is not threadsafe, especially since you are using a singleton you don't know where this will be called from, not making this threadsafe leaves you open to race conditions.

You're not unlocking JobQueue before the call to notify, that seems strange, the thread that gets woken up would at least have to wait for the push call to release the lock.

While the number of cores is definitely a factor that should be taken under consideration, it's really the runtime behavior of the threads that determines the best size of the threadpool, by creating the artificial limit you're limiting the use of your pool. I'd use the std::thread::hardware_concurrency() as a default value for the constructor but not to limit the size of the pool. Anybody playing around with the pool size can now determine the optimal size for their use case.

I have to put in my two cents with regards to singletons, I use singletons in some places in my code but I have not found a good reason to limit construction of singleton objects, by allowing singleton access or new construction of the object you create opportunities for varied use, it makes testing easier, and sometimes lets you get rid of the singleton access. Just make the constructor public anybody looking at your class can see ah i can use the singleton, or i can create one and pass it around.

\$\endgroup\$
  • 1
    \$\begingroup\$ Good advice overall - but a singleton that can be instantiated more than once is, by definition, not a singleton. If you can "use the singleton or create one and pass it around", it is not a singleton - it's just a global. \$\endgroup\$ – osuka_ Jun 4 at 17:57
  • \$\begingroup\$ Thank you! I solved the issue with the locking and unlocking. Profiling showed that it was actually slowing performance because the threadManager (since been deleted in place of a lambda) was holding onto the lock WHILE executing the job, preventing every other thread from accessing the queue. Ill edit with the updated code \$\endgroup\$ – Paul Jun 4 at 19:34

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.