threadpool.hpp

Go to the documentation of this file.

 
#ifndef PARALLEL_CELLULAR_AUTOMATA_THREADPOOL_HPP
#define PARALLEL_CELLULAR_AUTOMATA_THREADPOOL_HPP
#include <atomic>
#include <future>
#include <memory>
#include <pthread.h>
#include <queues.hpp>
#include <thread>
#include <type_traits>
#include <vector>
class ThreadJoiner
{
  private:
    std::vector<std::thread> &threads;
 
  public:
    ThreadJoiner(std::vector<std::thread> &t);
    ~ThreadJoiner();
};
 
namespace ca
{
class Threadpool
{
 
    using QueuesContentType = std::function<void()>;
 
  public:
    Threadpool(unsigned nw = 0);
 
    ~Threadpool();
 
    size_t get_number_workers() const;
 
    template <class F, class... Args>
    auto submit(F &&f, Args &&...args) -> std::future<typename std::result_of<F(Args...)>::type>
    {
        // type returned by f(args)
        using return_value_type = typename std::result_of<F(Args...)>::type;
 
        /*
         * std::bind creates a function wrapper "g" s.t. g == f(args)
         */
        auto task = std::make_shared<std::packaged_task<return_value_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...));
        // with the packaged task we split the execution from the return value
        std::future<return_value_type> future_result(task->get_future());
 
        if (local_queue)
        {
            // to avoid cache ping-pong and improve performances each thread pushes on its individual queue if it can.
            local_queue->push([task]() {
                // dereference the package
                (*task)();
            });
        }
        else
        {
            threadpool_work_queue.push([task]() { (*task)(); });
        }
 
        return future_result;
    }
 
  private:
    std::atomic<bool> done;
    ThreadSafeQueue<QueuesContentType> threadpool_work_queue;
    std::vector<std::unique_ptr<WorkStealingQueue<QueuesContentType>>> workers_queues;
    std::vector<std::thread> threads;
    ThreadJoiner joiner;
 
    // Try to steal the work from other thread's queues.
    bool try_stealing_work(QueuesContentType &result)
    {
        unsigned qsize = workers_queues.size();
        for (unsigned i = 0; i < qsize; ++i)
        {
            unsigned const index = (thread_index + 1) % qsize;
            if (workers_queues[index]->try_steal(result))
            {
                return true;
            }
        }
        return false;
    }
 
    // thread-local variables containing the thread index and the local queue.
    static thread_local WorkStealingQueue<QueuesContentType> *local_queue;
    static thread_local unsigned thread_index;
 
    // work stealing worker
    void worker_thread(unsigned index);
};
} // namespace ca
 
#endif