ctpl::thread_pool Class Reference

#include <ctpl.h>

Public Member Functions
	thread_pool ()
	thread_pool (int nThreads, int queueSize=_ctplThreadPoolLength_)
	~thread_pool ()
int	size ()
int	n_idle ()
std::thread &	get_thread (int i)
void	resize (int nThreads)
void	clear_queue ()
std::function< void(int)>	pop ()
void	stop (bool isWait=false)
template<typename F , typename... Rest>
auto	push (F &&f, Rest &&... rest) -> std::future< decltype(f(0, rest...))>
template<typename F >
auto	push (F &&f) -> std::future< decltype(f(0))>
	thread_pool ()
	thread_pool (int nThreads)
	~thread_pool ()
int	size ()
int	n_idle ()
std::thread &	get_thread (int i)
void	resize (int nThreads)
void	clear_queue ()
std::function< void(int)>	pop ()
void	stop (bool isWait=false)
template<typename F , typename... Rest>
auto	push (F &&f, Rest &&... rest) -> std::future< decltype(f(0, rest...))>
template<typename F >
auto	push (F &&f) -> std::future< decltype(f(0))>

Detailed Description

Definition at line 48 of file ctpl.h.

Constructor & Destructor Documentation

thread_pool() [1/4]

ctpl::thread_pool::thread_pool ( )

inline

Definition at line 52 of file ctpl.h.

: q(_ctplThreadPoolLength_) { this->init(); }

thread_pool() [2/4]

ctpl::thread_pool::thread_pool ( int  nThreads, int  queueSize = _ctplThreadPoolLength_  )

inline

Definition at line 53 of file ctpl.h.

: q(queueSize) { this->init(); this->resize(nThreads); }

~thread_pool() [1/2]

ctpl::thread_pool::~thread_pool ( )

inline

Definition at line 56 of file ctpl.h.

                       {
            this->stop(true);
        }

thread_pool() [3/4]

ctpl::thread_pool::thread_pool ( )

inline

Definition at line 75 of file ctpl_stl.h.

{ this->init(); }

thread_pool() [4/4]

ctpl::thread_pool::thread_pool ( int  nThreads )

inline

Definition at line 76 of file ctpl_stl.h.

{ this->init(); this->resize(nThreads); }

~thread_pool() [2/2]

ctpl::thread_pool::~thread_pool ( )

inline

Definition at line 79 of file ctpl_stl.h.

                       {
            this->stop(true);
        }

Member Function Documentation

clear_queue() [1/2]

void ctpl::thread_pool::clear_queue ( )

inline

Definition at line 99 of file ctpl.h.

                           {
            std::function<void(int id)> * _f;
            while (this->q.pop(_f))
                delete _f;  // empty the queue
        }

clear_queue() [2/2]

void ctpl::thread_pool::clear_queue ( )

inline

Definition at line 122 of file ctpl_stl.h.

                           {
            std::function<void(int id)> * _f;
            while (this->q.pop(_f))
                delete _f; // empty the queue
        }

get_thread() [1/2]

std::thread& ctpl::thread_pool::get_thread ( int  i )

inline

Definition at line 65 of file ctpl.h.

{ return *this->threads[i]; }

get_thread() [2/2]

std::thread& ctpl::thread_pool::get_thread ( int  i )

inline

Definition at line 88 of file ctpl_stl.h.

{ return *this->threads[i]; }

n_idle() [1/2]

int ctpl::thread_pool::n_idle ( )

inline

Definition at line 64 of file ctpl.h.

{ return this->nWaiting; }

n_idle() [2/2]

int ctpl::thread_pool::n_idle ( )

inline

Definition at line 87 of file ctpl_stl.h.

{ return this->nWaiting; }

pop() [1/2]

std::function<void(int)> ctpl::thread_pool::pop ( )

inline

Definition at line 106 of file ctpl.h.

                                     {
            std::function<void(int id)> * _f = nullptr;
            this->q.pop(_f);
            std::unique_ptr<std::function<void(int id)>> func(_f);  // at return, delete the function even if an exception occurred
            
            std::function<void(int)> f;
            if (_f)
                f = *_f;
            return f;
        }

pop() [2/2]

std::function<void(int)> ctpl::thread_pool::pop ( )

inline

Definition at line 129 of file ctpl_stl.h.

                                     {
            std::function<void(int id)> * _f = nullptr;
            this->q.pop(_f);
            std::unique_ptr<std::function<void(int id)>> func(_f); // at return, delete the function even if an exception occurred
            std::function<void(int)> f;
            if (_f)
                f = *_f;
            return f;
        }

push() [1/4]

template<typename F , typename... Rest>

auto ctpl::thread_pool::push ( F &&  f, Rest &&...  rest  ) -> std::future<decltype(f(0, rest...))>

inline

Definition at line 152 of file ctpl.h.

                                                                               {
            auto pck = std::make_shared<std::packaged_task<decltype(f(0, rest...))(int)>>(
                std::bind(std::forward<F>(f), std::placeholders::_1, std::forward<Rest>(rest)...)
            );

            auto _f = new std::function<void(int id)>([pck](int id) {
                (*pck)(id);
            });
            this->q.push(_f);

            std::unique_lock<std::mutex> lock(this->mutex);
            this->cv.notify_one();

            return pck->get_future();
        }

push() [2/4]

template<typename F >

auto ctpl::thread_pool::push ( F &&  f ) -> std::future<decltype(f(0))>

inline

Definition at line 171 of file ctpl.h.

                                                      {
            auto pck = std::make_shared<std::packaged_task<decltype(f(0))(int)>>(std::forward<F>(f));

            auto _f = new std::function<void(int id)>([pck](int id) {
                (*pck)(id);
            });
            this->q.push(_f);

            std::unique_lock<std::mutex> lock(this->mutex);
            this->cv.notify_one();

            return pck->get_future();
        }

push() [3/4]

template<typename F , typename... Rest>

auto ctpl::thread_pool::push ( F &&  f, Rest &&...  rest  ) -> std::future<decltype(f(0, rest...))>

inline

Definition at line 173 of file ctpl_stl.h.

                                                                               {
            auto pck = std::make_shared<std::packaged_task<decltype(f(0, rest...))(int)>>(
                std::bind(std::forward<F>(f), std::placeholders::_1, std::forward<Rest>(rest)...)
                );
            auto _f = new std::function<void(int id)>([pck](int id) {
                (*pck)(id);
            });
            this->q.push(_f);
            std::unique_lock<std::mutex> lock(this->mutex);
            this->cv.notify_one();
            return pck->get_future();
        }

push() [4/4]

template<typename F >

auto ctpl::thread_pool::push ( F &&  f ) -> std::future<decltype(f(0))>

inline

Definition at line 189 of file ctpl_stl.h.

                                                      {
            auto pck = std::make_shared<std::packaged_task<decltype(f(0))(int)>>(std::forward<F>(f));
            auto _f = new std::function<void(int id)>([pck](int id) {
                (*pck)(id);
            });
            this->q.push(_f);
            std::unique_lock<std::mutex> lock(this->mutex);
            this->cv.notify_one();
            return pck->get_future();
        }

resize() [1/2]

void ctpl::thread_pool::resize ( int  nThreads )

inline

Definition at line 70 of file ctpl.h.

                                  {
            if (!this->isStop && !this->isDone) {
                int oldNThreads = static_cast<int>(this->threads.size());
                if (oldNThreads <= nThreads) {  // if the number of threads is increased
                    this->threads.resize(nThreads);
                    this->flags.resize(nThreads);

                    for (int i = oldNThreads; i < nThreads; ++i) {
                        this->flags[i] = std::make_shared<std::atomic<bool>>(false);
                        this->set_thread(i);
                    }
                }
                else {  // the number of threads is decreased
                    for (int i = oldNThreads - 1; i >= nThreads; --i) {
                        *this->flags[i] = true;  // this thread will finish
                        this->threads[i]->detach();
                    }
                    {
                        // stop the detached threads that were waiting
                        std::unique_lock<std::mutex> lock(this->mutex);
                        this->cv.notify_all();
                    }
                    this->threads.resize(nThreads);  // safe to delete because the threads are detached
                    this->flags.resize(nThreads);  // safe to delete because the threads have copies of shared_ptr of the flags, not originals
                }
            }
        }

resize() [2/2]

void ctpl::thread_pool::resize ( int  nThreads )

inline

Definition at line 93 of file ctpl_stl.h.

                                  {
            if (!this->isStop && !this->isDone) {
                int oldNThreads = static_cast<int>(this->threads.size());
                if (oldNThreads <= nThreads) {  // if the number of threads is increased
                    this->threads.resize(nThreads);
                    this->flags.resize(nThreads);

                    for (int i = oldNThreads; i < nThreads; ++i) {
                        this->flags[i] = std::make_shared<std::atomic<bool>>(false);
                        this->set_thread(i);
                    }
                }
                else {  // the number of threads is decreased
                    for (int i = oldNThreads - 1; i >= nThreads; --i) {
                        *this->flags[i] = true;  // this thread will finish
                        this->threads[i]->detach();
                    }
                    {
                        // stop the detached threads that were waiting
                        std::unique_lock<std::mutex> lock(this->mutex);
                        this->cv.notify_all();
                    }
                    this->threads.resize(nThreads);  // safe to delete because the threads are detached
                    this->flags.resize(nThreads);  // safe to delete because the threads have copies of shared_ptr of the flags, not originals
                }
            }
        }

size() [1/2]

int ctpl::thread_pool::size ( )

inline

Definition at line 61 of file ctpl.h.

{ return static_cast<int>(this->threads.size()); }

size() [2/2]

int ctpl::thread_pool::size ( )

inline

Definition at line 84 of file ctpl_stl.h.

{ return static_cast<int>(this->threads.size()); }

stop() [1/2]

void ctpl::thread_pool::stop ( bool  isWait = false )

inline

Definition at line 121 of file ctpl.h.

                                       {
            if (!isWait) {
                if (this->isStop)
                    return;
                this->isStop = true;
                for (int i = 0, n = this->size(); i < n; ++i) {
                    *this->flags[i] = true;  // command the threads to stop
                }
                this->clear_queue();  // empty the queue
            }
            else {
                if (this->isDone || this->isStop)
                    return;
                this->isDone = true;  // give the waiting threads a command to finish
            }
            {
                std::unique_lock<std::mutex> lock(this->mutex);
                this->cv.notify_all();  // stop all waiting threads
            }
            for (int i = 0; i < static_cast<int>(this->threads.size()); ++i) {  // wait for the computing threads to finish
                if (this->threads[i]->joinable())
                    this->threads[i]->join();
            }
            // if there were no threads in the pool but some functors in the queue, the functors are not deleted by the threads
            // therefore delete them here
            this->clear_queue();
            this->threads.clear();
            this->flags.clear();
        }

stop() [2/2]

void ctpl::thread_pool::stop ( bool  isWait = false )

inline

Definition at line 142 of file ctpl_stl.h.

                                       {
            if (!isWait) {
                if (this->isStop)
                    return;
                this->isStop = true;
                for (int i = 0, n = this->size(); i < n; ++i) {
                    *this->flags[i] = true;  // command the threads to stop
                }
                this->clear_queue();  // empty the queue
            }
            else {
                if (this->isDone || this->isStop)
                    return;
                this->isDone = true;  // give the waiting threads a command to finish
            }
            {
                std::unique_lock<std::mutex> lock(this->mutex);
                this->cv.notify_all();  // stop all waiting threads
            }
            for (int i = 0; i < static_cast<int>(this->threads.size()); ++i) {  // wait for the computing threads to finish
                    if (this->threads[i]->joinable())
                        this->threads[i]->join();
            }
            // if there were no threads in the pool but some functors in the queue, the functors are not deleted by the threads
            // therefore delete them here
            this->clear_queue();
            this->threads.clear();
            this->flags.clear();
        }

---

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

• CTPL/ctpl.h
• CTPL/ctpl_stl.h