2
\$\begingroup\$

I just finished writing a Thread Pool in C++11 and wanted to get some feedback since I'm not particularly experienced in multithreaded programming, I used to have a thread pool written in Boost::Asio mostly using their work dispatcher and io_service. This thread pool implementation seems to block less and has a slightly more reliable framerate in my game when used for particle effects.

Any feedback on performance improvements would be greatly appreciated! Thank you.

https://github.com/M2tM/MutedThreadPool

#ifndef _MV_THREADPOOL_H_
#define _MV_THREADPOOL_H_

#include <thread>
#include <condition_variable>
#include <deque>
#include <vector>
#include <atomic>
#include <list>
#include <iostream>

namespace MV {
    class ThreadPool {
        class Worker;
    public:
        class Job {
            friend Worker;
        public:
            Job(const std::function<void()> &a_action) :
                action(a_action) {
            }
            Job(const std::function<void()> &a_action, const std::function<void()> &a_onFinish) :
                action(a_action),
                onFinish(a_onFinish) {
            }
            Job(const std::function<void()> &a_action, const std::function<void()> &a_onFinish, const std::shared_ptr<std::atomic<size_t>> &a_groupCounter, const std::shared_ptr<std::function<void()>> &a_onGroupFinish) :
                action(a_action),
                onFinish(a_onFinish),
                groupCounter(a_groupCounter),
                onGroupFinish(a_onGroupFinish) {
            }
            Job(const std::function<void()> &a_action, const std::shared_ptr<std::atomic<size_t>> &a_groupCounter, const std::shared_ptr<std::function<void()>> &a_onGroupFinish) :
                action(a_action),
                groupCounter(a_groupCounter),
                onGroupFinish(a_onGroupFinish) {
            }
            Job(Job&& a_rhs) = default;
            Job(const Job& a_rhs) = default;

            void group(const std::shared_ptr<std::atomic<size_t>> &a_groupCounter, const std::shared_ptr<std::function<void()>> &a_onGroupFinish) {
                groupCounter = a_groupCounter;
                onGroupFinish = a_onGroupFinish;
            }

            void operator()() noexcept {
                try { action(); } catch (std::exception &e) { parent->exception(e); }
                bool groupFinished = groupCounter && --(*groupCounter) == 0 && onGroupFinish && *onGroupFinish;
                if (onFinish && !groupFinished) {
                    parent->task(onFinish);
                }
                else if (onFinish && groupFinished) {
                    parent->task([=] {
                        try { onFinish(); } catch (std::exception &e) { parent->exception(e); }
                        try { (*onGroupFinish)(); } catch (std::exception &e) { parent->exception(e); }
                    });
                }
                else if (groupFinished) {
                    parent->task((*onGroupFinish));
                }
            }
        private:
            ThreadPool* parent;

            std::function<void()> action;
            std::function<void()> onFinish;
            std::shared_ptr<std::atomic<size_t>> groupCounter;
            std::shared_ptr<std::function<void()>> onGroupFinish;
        };
        friend Job;
    private:
        class Worker {
        public:
            Worker(ThreadPool* a_parent) :
                parent(a_parent) {
                thread = std::make_unique<std::thread>([=]() { work(); });
            }
            ~Worker() { if (thread && thread->joinable()) { thread->join(); } }

            void cleanup() {
                while (!finished) {}
            }
        private:
            void work() {
                while (!parent->stopped) {
                    std::unique_lock<std::mutex> guard(parent->lock);
                    parent->notify.wait(guard, [=] {return parent->jobs.size() > 0 || parent->stopped; });
                    if (parent->stopped) { break; }

                    auto job = std::move(parent->jobs.front());
                    parent->jobs.pop_front();
                    guard.unlock();

                    job.parent = parent;
                    job();
                }
                finished = true;
            }
            ThreadPool* parent;
            bool finished = false;
            std::unique_ptr<std::thread> thread;
        };
        friend Worker;

    public:
        ThreadPool() :
            ThreadPool(std::thread::hardware_concurrency() > 1 ? std::thread::hardware_concurrency() - 1 : 1) {
        }

        ThreadPool(size_t a_threads) {
            for (size_t i = 0; i < a_threads; ++i) {
                workers.push_back(std::make_unique<Worker>(this));
            }
        }

        ~ThreadPool() {
            {
                std::lock_guard<std::mutex> guard(lock);
                jobs.clear();
                stopped = true;
            }
            notify.notify_all();
            for (auto&& worker : workers) {
                worker->cleanup();
            }
        }

        void task(const Job &a_newWork) {
            {
                std::lock_guard<std::mutex> guard(lock);
                jobs.emplace_back(std::move(a_newWork));
            }
            notify.notify_one();
        }
        void task(const std::function<void()> &a_task) {
            {
                std::lock_guard<std::mutex> guard(lock);
                jobs.emplace_back(a_task);
            }
            notify.notify_one();
        }
        void task(const std::function<void()> &a_task, const std::function<void()> &a_onComplete) {
            {
                std::lock_guard<std::mutex> guard(lock);
                jobs.emplace_back(a_task, a_onComplete);
            }
            notify.notify_one();
        }

        void tasks(const std::vector<Job> &a_tasks, const std::function<void()> &a_onGroupComplete) {
            std::shared_ptr<std::atomic<size_t>> counter = std::make_shared<std::atomic<size_t>>(a_tasks.size());
            std::shared_ptr<std::function<void()>> sharedGroupComplete = std::make_shared<std::function<void()>>(std::move(a_onGroupComplete));
            for (auto&& job : a_tasks)
            {
                {
                    std::lock_guard<std::mutex> guard(lock);
                    jobs.emplace_back(std::move(job));
                    jobs.back().group(counter, sharedGroupComplete);
                }
                notify.notify_one();
            }
        }

        void schedule(const std::function<void()> &a_method) {
            std::lock_guard<std::mutex> guard(scheduleLock);
            scheduled.push_back(a_method);
        }

        void run() {
            for (auto i = scheduled.begin(); i != scheduled.end();) {
                try {
                    (*i)();
                } catch (std::exception &e) {
                    exception(e);
                }

                std::lock_guard<std::mutex> guard(scheduleLock);
                scheduled.erase(i++);
            }
        }

        void exceptionHandler(std::function<void(std::exception &e)> a_onException) {
            std::lock_guard<std::mutex> guard(exceptionLock);
            onException = a_onException;
        }

        size_t threads() const {
            return workers.size();
        }
    private:
        void exception(std::exception &e) {
            std::lock_guard<std::mutex> guard(exceptionLock);
            if (onException) {
                onException(e);
            } else {
                std::cerr << "Uncaught Exception in Thread Pool: " << e.what() << '\n';
            }
        }

        std::condition_variable notify;
        bool stopped = false;
        std::mutex lock;
        std::mutex scheduleLock;
        std::mutex exceptionLock;
        std::vector<std::unique_ptr<Worker>> workers;
        std::deque<Job> jobs;
        std::list<std::function<void()>> scheduled;
        std::deque<std::string> exceptionMessages;
        std::function<void(std::exception &e)> onException;
    };
}

#endif
\$\endgroup\$
6
  • 2
    \$\begingroup\$ Any specific reason (other than lifetime management) why groupCounter has to be std::shared_ptr<std::atomic<size_t>> instead of std::atomic<size_t>&? Same question for onGroupFinish. \$\endgroup\$
    – hoffmale
    Commented Jul 24, 2017 at 17:29
  • 1
    \$\begingroup\$ Much too large and complicated. Thread pools have been reviewed several times before. Check out some of the existing thread pools and see if you can steal from them to simplify your implementation. PS modern C++ should not need such low level primitives as they are the same functionality is provided by the standard. You should be using higher level abstractions like std::promise std::future and std::async \$\endgroup\$ Commented Jul 24, 2017 at 17:38
  • \$\begingroup\$ @hoffmale Lifetime management is the reason they are std::shared_ptr, they are not references because they are not necessary unless they are part of a group, and being part of a group is optional. So not every task has a groupCounter or onGroupFinish. They are actually shared pointers because they are shared state between the group. \$\endgroup\$
    – M2tM
    Commented Jul 24, 2017 at 17:51
  • \$\begingroup\$ @LokiAstari Yeah I had considered making use of the std::future interface, but unfortunately the lack of a std::when_all and std::future::then makes the standard a little immature for my most common use case which is chaining together continuations. Maybe in a few years when std::future is more composable I will re-visit this. std::async is cool, but is an awkward place for use in general code because I can't control how many threads I get, and that matters on iOS (really don't want too many), and I don't think it's much simpler than worker dispatching with that constraint. I'm open to it. \$\endgroup\$
    – M2tM
    Commented Jul 24, 2017 at 17:59
  • \$\begingroup\$ @LokiAstari But I would honestly need you to point me at an example of std::async in a thread pool, I haven't seen any yet. It feels like it's a kind of abstraction meant for use in a higher level than a thread pool, but it is too low level to be useful generally without guarantees about its own thread creation policies. IE: If I use std::async directly in my particle emitters, and each emitter doesn't know about others (they are components in a scene graph) each emitter may create x threads and that would be .5 MB of ram per job basically. With a thread pool I have more explicit control. \$\endgroup\$
    – M2tM
    Commented Jul 24, 2017 at 18:03

1 Answer 1

2
\$\begingroup\$

unused variable

As far as I can tell, ThreadPool::exceptionMessages is currently not used.

const shared_ptr<T>&

They are used quite a lot (especially on the Job constructors). You might want to consider exchanging them for simple std::shared_ptr<T> instead - you'll make a copy either way, and this way the caller knows you are taking a copy (a const shared_ptr<T>& says "might take", a shared_ptr<T> enforces it) and the compiler might be able to optimize better.

trying to move a const object

In ThreadPool::tasks there is the following statement: jobs.emplace_back(std::move(job)). This will always make a copy! Why? Because job is a value contained in a_tasks, which is const - so it can't be modified. Note that in case you switched a_task to a non-const vector, it will contain invalid values afterwards (as they will have been moved from).

batch performance

This again concerns ThreadPool::tasks: When enqueueing a batch of N jobs, 2*N+2 shared_ptr objects will be constructed (and once the jobs are finished, they'll have to be destructed again). Since those operations can be costly and can be avoided (with some architectural changes), there might be some performance gains (mostly in case of smaller jobs).

race condition

There is a possible (and very likely, unless scheduled is nearly empty) race condition in ThreadPool::run: If run were to be called concurrently, there might be cases where the object currently pointed to by i will be erased by another thread (as it is already finished processing the function pointed to by i).

data structures

This might be highly situational (call it "premature optimization" if you will), but one could exchange ThreadPool::scheduled for a vector. This might boost performance a bit in case you schedule lots of functions, as you are chasing less pointers around (= less cache misses).

Nearly the same could be said for ThreadPool::jobs (exchange it for a circular buffer implementation), especially if you have a good guess on how many jobs you might have queued up at the same time (Same reason: less pointers to chase = less cache misses).

However, these gains might be very small compared to your usual usage, so always measure!

on another note

With regards to performance, have you tried going lock-free? I know, it makes reasoning about stuff harder, but it might give you another edge of performance if really needed (especially on platforms where locking requires kernel calls).

\$\endgroup\$
11
  • \$\begingroup\$ SUPER good feedback, thank you for your time. I'll reply to a few points, but I just want to say I'm definitely taking some of these things to heart and will be improving the implementation and posting a second version in my question. I think of "run" as being a main thread callback invocation point. So I wasn't so concerned with "run" itself being called on multiple threads, it was not intended to. It is meant to run in a game loop basically and to allow other threads to make calls on the main thread. \$\endgroup\$
    – M2tM
    Commented Jul 24, 2017 at 22:53
  • \$\begingroup\$ Talking more about "run" and the data structure I use for those callbacks. I make use of std::list to avoid invalidating iterators during iteration to allow for more conservative locking (rather than locking for the whole traversal of scheduled I only lock on a per callback basis. This avoids holding the mutex for too long and causing extra contention between threads. \$\endgroup\$
    – M2tM
    Commented Jul 24, 2017 at 22:55
  • \$\begingroup\$ Unused variable: Thank you for pointing it out! I'll remove it. Originally I was playing with a couple different ideas for handling exceptions, and I wrote that and forgot to remove it. \$\endgroup\$
    – M2tM
    Commented Jul 24, 2017 at 22:56
  • 1
    \$\begingroup\$ if you have a hour (or two), you might find this interesting: youtube.com/watch?v=c1gO9aB9nbs (a great talk from Herb Sutter about lock-free programming and its difficulties) \$\endgroup\$
    – hoffmale
    Commented Jul 24, 2017 at 23:15
  • 1
    \$\begingroup\$ That's one way to address the issue (note I've been careful not to give any solutions, as they have different advantages and disadvantages in different use cases, and I don't know yours well enough, especially considering performance). If it works for you, great! If it doesn't (or doesn't work in the general case), adjust (and measure)! \$\endgroup\$
    – hoffmale
    Commented Jul 25, 2017 at 0:13

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.