gpmp::core::ThreadPool Class Reference

#include <threads.hpp>

Public Member Functions
	ThreadPool ()
	Default constructor that creates a ThreadPool with the number of threads. More...
	ThreadPool (int numThreads)
	Constructs a ThreadPool with a given number of worker threads to dispatch functions. More...
template<class F , class... Args>
auto	enqueue (F &&f, Args &&...args) -> std::future< typename std::invoke_result< F, Args... >::type >
	Enqueues a task to the thread pool. More...
	~ThreadPool ()

Private Attributes
std::vector< std::thread >	workers
std::queue< std::function< void()> >	tasks
std::mutex	queue_mutex
std::condition_variable	condition
bool	stop

Detailed Description

Examples

factors.cpp, and primes.cpp.

Definition at line 60 of file threads.hpp.

Constructor & Destructor Documentation

ThreadPool() [1/2]

gpmp::core::ThreadPool::ThreadPool ( )

inline

Default constructor that creates a ThreadPool with the number of threads.

Definition at line 78 of file threads.hpp.

                 : ThreadPool(std::thread::hardware_concurrency()) {
    }

ThreadPool() [2/2]

gpmp::core::ThreadPool::ThreadPool ( int  numThreads )

inlineexplicit

Constructs a ThreadPool with a given number of worker threads to dispatch functions.

Parameters

numThreads

The number of worker threads to be created.

Constructs a ThreadPool object with the specified number of worker threads.

Definition at line 88 of file threads.hpp.

                                        : stop(false) {
 
        // traverse through the number of threads specified
        for (int i = 0; i < numThreads; ++i) {
            // add a new thread to the vector storing workers using lambda
            // function
            workers.emplace_back([this] {
                for (;;) {
                    // worker thread creates task object that holds next task to
                    // be executed
                    std::function<void()> task_obj;
 
                    // this "symbolizes" the critical section of the TheadPool
                    // class
                    {
                        // worker thread locks queue_mutex
                        std::unique_lock<std::mutex> lock(this->queue_mutex);
                        // wait on conditional_variable (ThreadPool stop OR
                        // queued task), wait() locks/unlocks based on condition
                        // result
                        this->condition.wait(lock, [this] {
                            return this->stop || !this->tasks.empty();
                        });
                        // based on stop OR awaiting tasks, return from the
                        // thread
                        if (this->stop && this->tasks.empty()) {
                            return;
                        }
 
                        // if above isnt met, move first task in TASKS queue to
                        // the task object to transfer ownership
                        task_obj = std::move(this->tasks.front());
 
                        // pop the handed off task to make room for a new one.
                        // only ONE thread should remove a task from the queue
                        // at a time
                        this->tasks.pop();
                    }
 
                    // EXECUTE THE HANDED OFF TASK
                    task_obj();
                }
            });
        }
    }

References workers.

~ThreadPool()

gpmp::core::ThreadPool::~ThreadPool ( )

inline

Definition at line 185 of file threads.hpp.

                  {
        {
            // lock queue_mutex & set stop to true
            std::unique_lock<std::mutex> lock(queue_mutex);
            stop = true;
        }
        // unblock all threads
        condition.notify_all();
        // treaverse threads and join
        for (std::thread &worker : workers) {
            worker.join();
        }
    }

References condition, queue_mutex, stop, and workers.

Member Function Documentation

enqueue()

template<class F , class... Args>

auto gpmp::core::ThreadPool::enqueue ( F &&  f, Args &&...  args  ) -> std::future<typename std::invoke_result<F, Args...>::type>

inline

Enqueues a task to the thread pool.

Template Parameters

F	Type of the function to be enqueued.
Args	Variadic template parameter pack of the arguments passed to the function.

Parameters

f	Function to be enqueued.
args	Arguments passed to the function

Returns

std::future<typename std::result_of<F(Args...)>::type> A future object that will contain the result of the function execution.

Exceptions

std::runtime_error

If the ThreadPool has already been stopped.

Examples

primes.cpp.

Definition at line 148 of file threads.hpp.

                                                                {
 
        // this is the return type of the passed in function
        // using return_type = typename std::result_of<F(Args...)>::type;
        using return_type = typename std::invoke_result<F, Args...>::type;
 
        // * SHARED POINTER to PACKAGED TASK used to store the passed in i
        //      function + its arguments
        // * std::bind used to create function object binded to the
        //      function `f` + its args to the packaged tasks
        // * std::forward used for forwarding an argument to another
        //      function
        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...));
 
        // the FUTURE obj retrieves the return value of the function passed in
        std::future<return_type> res = task->get_future();
        {
            // aquire lock on queue_mutex for synchronization
            std::unique_lock<std::mutex> lock(queue_mutex);
            // check if threadpool stop is initiated
            if (stop) {
                throw std::runtime_error("enqueue on stopped ThreadPool");
            }
            // add a task using emplace to the queue as a lambda that calls the
            // packaged task
            tasks.emplace([task]() { (*task)(); });
        } // once this is hit, unique_lock is out of scope & mutex is
          // automatically unlocked
        // notify one waiting thread of one new task added to the queue
        condition.notify_one();
        // the return is the future object
        return res;
    }

References condition, queue_mutex, python.linalg::res, stop, and tasks.

Referenced by testing_miller_thread().

Member Data Documentation

condition

std::condition_variable gpmp::core::ThreadPool::condition

private

Definition at line 69 of file threads.hpp.

Referenced by enqueue(), and ~ThreadPool().

queue_mutex

std::mutex gpmp::core::ThreadPool::queue_mutex

private

Definition at line 67 of file threads.hpp.

Referenced by enqueue(), and ~ThreadPool().

stop

bool gpmp::core::ThreadPool::stop

private

Definition at line 71 of file threads.hpp.

Referenced by enqueue(), and ~ThreadPool().

tasks

std::queue<std::function<void()> > gpmp::core::ThreadPool::tasks

private

Definition at line 65 of file threads.hpp.

Referenced by enqueue().

workers

std::vector<std::thread> gpmp::core::ThreadPool::workers

private

Definition at line 63 of file threads.hpp.

Referenced by ThreadPool(), and ~ThreadPool().

---

The documentation for this class was generated from the following file:

• include/openGPMP/core/threads.hpp