PndLmdThreadPool.h

Go to the documentation of this file.

/*
 * @author Sehe (https://stackoverflow.com/users/85371/sehe)
 *
 * Thread Pool implementation using boost::threads.
 *
 * taken from this StackOverflow answer:
 * https://stackoverflow.com/questions/22685176/boost-group-threads-maximal-number-of-parallel-thread
 *
 */

#include <boost/thread.hpp>
#include <boost/phoenix.hpp>
#include <boost/optional.hpp>

//using namespace boost;
using namespace boost::phoenix::arg_names;

class PndLmdThreadPool {
private:
        boost::mutex mx;
        boost::condition_variable cv;

        typedef boost::function<void()> job_t;
        std::deque<job_t> _queue;

        boost::thread_group pool;

        boost::atomic_bool shutdown;
        static void worker_thread(PndLmdThreadPool& q) {
                while (boost::optional<job_t> job = q.dequeue())
                        (*job)();
        }

public:
        //create thread pool with maximum possible threads
        PndLmdThreadPool() :
                        shutdown(false) {
                for (unsigned i = 0; i < boost::thread::hardware_concurrency(); ++i)
                        pool.create_thread(boost::bind(worker_thread, boost::ref(*this)));
        }

        //create thread pool with maxThreads threads
        PndLmdThreadPool(unsigned int maxThreads) :
                        shutdown(false) {
                if (maxThreads > boost::thread::hardware_concurrency() || maxThreads == 0) maxThreads =
                    boost::thread::hardware_concurrency();
                for (unsigned i = 0; i < maxThreads; ++i)
                        pool.create_thread(boost::bind(worker_thread, boost::ref(*this)));
        }

        void enqueue(job_t job) {
                boost::lock_guard<boost::mutex> lk(mx);
                _queue.push_back(job);
                cv.notify_one();
        }

        boost::optional<job_t> dequeue() {
                boost::unique_lock<boost::mutex> lk(mx);
                namespace phx = boost::phoenix;
                cv.wait(lk, phx::ref(shutdown) || !phx::empty(phx::ref(_queue)));
                if (_queue.empty()) return boost::none;
                job_t job = _queue.front();
                _queue.pop_front();
                return job;
        }

        // wait for all threads to complete
        void wait() {
                shutdown = true;
                {
                        boost::lock_guard<boost::mutex> lk(mx);
                        cv.notify_all();
                }
                pool.join_all();
        }

        ~PndLmdThreadPool() {
                shutdown = true;
                {
                        boost::lock_guard<boost::mutex> lk(mx);
                        cv.notify_all();
                }
                pool.join_all();
        }
};