threadpool.hpp

Go to the documentation of this file.

 
#ifndef __JOIN_THREADPOOL_HPP__
#define __JOIN_THREADPOOL_HPP__
 
// libjoin.
#include <join/condition.hpp>
#include <join/thread.hpp>
 
// C++.
#include <functional>
#include <thread>
#include <memory>
#include <atomic>
#include <vector>
#include <deque>
 
namespace join
{
    class ThreadPool;
 
    class WorkerThread
    {
    public:
        WorkerThread (ThreadPool& pool);
 
        ~WorkerThread ();
 
    private:
        void work ();
 
        ThreadPool* _pool = nullptr;
 
        Thread _thread;
    };
 
    class ThreadPool
    {
    public:
        ThreadPool (int workers = std::max (int (std::thread::hardware_concurrency ()), 1));
 
        ~ThreadPool ();
 
        template <class Function, class... Args>
        void push (Function&& func, Args&&... args)
        {
            ScopedLock <Mutex> lock (_mutex);
            _jobs.emplace_back (std::bind (std::forward <Function> (func), std::forward <Args> (args)...));
            _condition.signal ();
        }
 
        size_t size ();
 
    private:
        std::vector <std::unique_ptr <WorkerThread>> _workers;
 
        Condition _condition;
 
        Mutex _mutex;
 
        std::atomic <bool> _stop;
 
        std::deque <std::function <void ()>> _jobs;
 
        friend class WorkerThread;
    };
 
    template <class InputIt, class Func>
    void distribute (InputIt first, InputIt last, Func function)
    {
        // check if we have tasks to perform.
        int count = std::distance (first, last);
        if (count)
        {
            // determine number of threads and tasks per thread to run.
            int concurrency = std::max (int (std::thread::hardware_concurrency ()), 1);
            // no need to create more threads than tasks.
            concurrency     = std::min (concurrency, count);
            int elements    = count / concurrency;
            int rest        = count % concurrency;
            // distribute tasks to threads.
            std::vector <int> tasks (concurrency, elements);
            for (int i = 0; i < rest; ++i)
            {
                ++tasks[i];
            }
 
            // determine the real thread pool size (concurrency minus 1 as we are a thread).
            std::vector <Thread> pool;
            int nth = concurrency - 1;
            pool.reserve (nth);
 
            // create threads.
            auto beg = first, end = first;
            for (int i = 0; i < nth; ++i)
            {
                std::advance (end, tasks[i]);
                pool.emplace_back (function, beg, end);
                beg = end;
            }
 
            // we are a thread so we can help.
            function (beg, last);
 
            // wait for threads to terminate.
            for (auto& thread : pool)
            {
                thread.join ();
            }
        }
    }
 
    template <class InputIt, class Func>
    void parallelForEach (InputIt first, InputIt last, Func function)
    {
        distribute (first, last, [&function] (InputIt beg, InputIt end)
        {
            for (; beg != end; ++beg)
            {
                function (*beg);
            }
        });
    }
}
 
#endif