ThreadContext, C++ (gcc)

ThreadContext

#ifndef THREADCONTEXT_HPP_INCLUDED
#define THREADCONTEXT_HPP_INCLUDED

#include <condition_variable>
#include <iostream>
#include <vector>
#include <future>
#include <thread>
#include <mutex>
#include "StaticQueue.hpp"
#include "FunctionWrapper.hpp"

namespace GeneticAlgorithm
{
//=========================================================================================
template <typename ReturnT>
class ThreadContext
{
    //###########################################################################

static size_t hwConcurrency_;

std::vector<std::thread> threads_;
    std::condition_variable condition_;
    std::mutex internalMutex_;

// queue of tasks for threads
    StaticQueue<std::packaged_task<ReturnT()>, 8> queue_;

bool exitflag_;

//###########################################################################

void Runnable()
    {
        while(exitflag_ == false)
        {
            std::unique_lock<std::mutex> locker{internalMutex_};

// wakeup and handles passed task
            condition_.wait(locker, [this]{ return queue_.Empty(); });

std::packaged_task<ReturnT()> task;
            queue_.Pop(task);

// unlock exactly here
            // in order to only one thread got task
            locker.unlock();

task();
        }
    }

// the signal for exit from thread
    static ReturnT KillThread(ThreadContext& r)
    {
        r.exitflag_ = true;
        return ReturnT{};
    }

public:
    ThreadContext() : threads_{hwConcurrency_}, exitflag_{false}
    {
        // Run threads and suspend their
        for(auto& th : threads_)
        {
            th = std::thread{&ThreadContext::Runnable, this};
        }
    }

/********************************************//**
     * \brief Add someone task to queue for execute
     *
     * \param func Some callable object (static function or function-member)
     * \param pack Parameter pack for callable object
     *
     * \return Returns the shared future object
     *           associated with the return-type of passed callable object
     *
     ***********************************************/

template <typename F, typename... Args>
    std::shared_future<ReturnT> AddAsyncTask(F func, Args&&... pack)
    {
        // TODO:
        // 1) If at the moment of call this method all threads are busy
        //      then new task don't be execute (it is an undefined behavior)
        //      need to wait  when someone thread suspend on the condition variable
        //
        // 2) To add the ability of transfer a function-member

auto callable = std::bind(func, std::forward<Args>(pack)...);

// OK: this object copies the callable object inside
        std::packaged_task<ReturnT()> newTask{callable};
        std::shared_future<ReturnT> futureObject = newTask.get_future();

// packaged_task<T> is only movable (move-assignable, move-constructible)
        queue_.Push(std::move(newTask));
        condition_.notify_one();

return futureObject;
    }

~ThreadContext()
    {
        // send to all threads exit signal
        for(size_t i = 0; i < hwConcurrency_; ++i)
        {
            AddAsyncTask(KillThread, std::ref(*this));
        }

// waits all threads for correct exit
        for(auto& thread : threads_)
        {
            if(thread.joinable())
            {
                thread.join();
            }
        }
    }
};
//=========================================================================================

template <typename ReturnT>
size_t ThreadContext<ReturnT>::hwConcurrency_ =
    std::thread::hardware_concurrency() != 0 ?
        std::thread::hardware_concurrency() - 1 : 2;

} // end of namespace GeneticAlgorithm

#endif // THREADCONTEXT_HPP_INCLUDED


run \| edit \| history \| help	0