threadpool.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_THREADPOOL_HPP
#define JOIN_CORE_THREADPOOL_HPP
 
// libjoin.
#include <join/condition.hpp>
#include <join/thread.hpp>
#include <join/cpu.hpp>
 
// C++.
#include <functional>
#include <memory>
#include <atomic>
#include <vector>
#include <deque>
 
namespace join
{
    class ThreadPool;
 
    class WorkerThread
    {
    private:
        WorkerThread (ThreadPool& pool);
 
    public:
        WorkerThread (const WorkerThread& other) = delete;
 
        WorkerThread& operator= (const WorkerThread& other) = delete;
 
        WorkerThread (WorkerThread&& other) = delete;
 
        WorkerThread& operator= (WorkerThread&& other) = delete;
 
        ~WorkerThread () noexcept;
 
    private:
        void work ();
 
        ThreadPool* _pool = nullptr;
 
        Thread _thread;
 
        friend class ThreadPool;
    };
 
    class ThreadPool
    {
    public:
        ThreadPool (int workers = int (CpuTopology::instance ()->cores ().size ()));
 
        ThreadPool (const ThreadPool& other) = delete;
 
        ThreadPool& operator= (const ThreadPool& other) = delete;
 
        ThreadPool (ThreadPool&& other) = delete;
 
        ThreadPool& operator= (ThreadPool&& other) = delete;
 
        ~ThreadPool () noexcept;
 
        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 () const noexcept;
 
    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 = int (CpuTopology::instance ()->cores ().size ());
            // 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