BS::thread_pool Class Reference

A fast, lightweight, and easy-to-use C++17 thread pool class. More...

#include <BS_thread_pool.hpp>

Collaboration diagram for BS::thread_pool:

Classes
class	blocks
	A helper class to divide a range into blocks. Used by detach_blocks(), submit_blocks(), detach_loop(), and submit_loop(). More...

Public Member Functions
	thread_pool ()
	Construct a new thread pool. The number of threads will be the total number of hardware threads available, as reported by the implementation. This is usually determined by the number of cores in the CPU. If a core is hyperthreaded, it will count as two threads.
	thread_pool (const concurrency_t num_threads)
	Construct a new thread pool with the specified number of threads.
	thread_pool (const std::function< void()> &init_task)
	Construct a new thread pool with the specified initialization function.
	thread_pool (const concurrency_t num_threads, const std::function< void()> &init_task)
	Construct a new thread pool with the specified number of threads and initialization function.
	thread_pool (const thread_pool &)=delete
	thread_pool (thread_pool &&)=delete
thread_pool &	operator= (const thread_pool &)=delete
thread_pool &	operator= (thread_pool &&)=delete
	~thread_pool ()
	Destruct the thread pool. Waits for all tasks to complete, then destroys all threads. Note that if the pool is paused, then any tasks still in the queue will never be executed.
size_t	get_tasks_queued () const
	Get the number of tasks currently waiting in the queue to be executed by the threads.
size_t	get_tasks_running () const
	Get the number of tasks currently being executed by the threads.
size_t	get_tasks_total () const
	Get the total number of unfinished tasks: either still waiting in the queue, or running in a thread. Note that get_tasks_total() == get_tasks_queued() + get_tasks_running().
concurrency_t	get_thread_count () const
	Get the number of threads in the pool.
std::vector< std::thread::id >	get_thread_ids () const
	Get a vector containing the unique identifiers for each of the pool's threads, as obtained by std::thread::get_id().
void	purge ()
	Purge all the tasks waiting in the queue. Tasks that are currently running will not be affected, but any tasks still waiting in the queue will be discarded, and will never be executed by the threads. Please note that there is no way to restore the purged tasks.
template<typename F >
void	detach_task (F &&task BS_THREAD_POOL_PRIORITY_INPUT)
	Submit a function with no arguments and no return value into the task queue, with the specified priority. To push a function with arguments, enclose it in a lambda expression. Does not return a future, so the user must use wait() or some other method to ensure that the task finishes executing, otherwise bad things will happen.
template<typename T , typename F >
void	detach_blocks (const T first_index, const T index_after_last, F &&block, const size_t num_blocks=0 BS_THREAD_POOL_PRIORITY_INPUT)
	Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue, with the specified priority. The block function takes two arguments, the start and end of the block, so that it is only called only once per block, but it is up to the user make sure the block function correctly deals with all the indices in each block. Does not return a multi_future, so the user must use wait() or some other method to ensure that the loop finishes executing, otherwise bad things will happen.
template<typename T , typename F >
void	detach_loop (const T first_index, const T index_after_last, F &&loop, const size_t num_blocks=0 BS_THREAD_POOL_PRIORITY_INPUT)
	Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue, with the specified priority. The loop function takes one argument, the loop index, so that it is called many times per block. Does not return a multi_future, so the user must use wait() or some other method to ensure that the loop finishes executing, otherwise bad things will happen.
template<typename T , typename F >
void	detach_sequence (const T first_index, const T index_after_last, F &&sequence BS_THREAD_POOL_PRIORITY_INPUT)
	Submit a sequence of tasks enumerated by indices to the queue, with the specified priority. Does not return a multi_future, so the user must use wait() or some other method to ensure that the sequence finishes executing, otherwise bad things will happen.
void	reset ()
	Reset the pool with the total number of hardware threads available, as reported by the implementation. Waits for all currently running tasks to be completed, then destroys all threads in the pool and creates a new thread pool with the new number of threads. Any tasks that were waiting in the queue before the pool was reset will then be executed by the new threads. If the pool was paused before resetting it, the new pool will be paused as well.
void	reset (const concurrency_t num_threads)
	Reset the pool with a new number of threads. Waits for all currently running tasks to be completed, then destroys all threads in the pool and creates a new thread pool with the new number of threads. Any tasks that were waiting in the queue before the pool was reset will then be executed by the new threads. If the pool was paused before resetting it, the new pool will be paused as well.
void	reset (const std::function< void()> &init_task)
	Reset the pool with the total number of hardware threads available, as reported by the implementation, and a new initialization function. Waits for all currently running tasks to be completed, then destroys all threads in the pool and creates a new thread pool with the new number of threads and initialization function. Any tasks that were waiting in the queue before the pool was reset will then be executed by the new threads. If the pool was paused before resetting it, the new pool will be paused as well.
void	reset (const concurrency_t num_threads, const std::function< void()> &init_task)
	Reset the pool with a new number of threads and a new initialization function. Waits for all currently running tasks to be completed, then destroys all threads in the pool and creates a new thread pool with the new number of threads and initialization function. Any tasks that were waiting in the queue before the pool was reset will then be executed by the new threads. If the pool was paused before resetting it, the new pool will be paused as well.
template<typename F , typename R = std::invoke_result_t<std::decay_t<F>>>
std::future< R >	submit_task (F &&task BS_THREAD_POOL_PRIORITY_INPUT)
	Submit a function with no arguments into the task queue, with the specified priority. To submit a function with arguments, enclose it in a lambda expression. If the function has a return value, get a future for the eventual returned value. If the function has no return value, get an std::future<void> which can be used to wait until the task finishes.
template<typename T , typename F , typename R = std::invoke_result_t<std::decay_t<F>, T, T>>
multi_future< R >	submit_blocks (const T first_index, const T index_after_last, F &&block, const size_t num_blocks=0 BS_THREAD_POOL_PRIORITY_INPUT)
	Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue, with the specified priority. The block function takes two arguments, the start and end of the block, so that it is only called only once per block, but it is up to the user make sure the block function correctly deals with all the indices in each block. Returns a multi_future that contains the futures for all of the blocks.
template<typename T , typename F >
multi_future< void >	submit_loop (const T first_index, const T index_after_last, F &&loop, const size_t num_blocks=0 BS_THREAD_POOL_PRIORITY_INPUT)
	Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue, with the specified priority. The loop function takes one argument, the loop index, so that it is called many times per block. It must have no return value. Returns a multi_future that contains the futures for all of the blocks.
template<typename T , typename F , typename R = std::invoke_result_t<std::decay_t<F>, T>>
multi_future< R >	submit_sequence (const T first_index, const T index_after_last, F &&sequence BS_THREAD_POOL_PRIORITY_INPUT)
	Submit a sequence of tasks enumerated by indices to the queue, with the specified priority. Returns a multi_future that contains the futures for all of the tasks.
void	wait ()
	Wait for tasks to be completed. Normally, this function waits for all tasks, both those that are currently running in the threads and those that are still waiting in the queue. However, if the pool is paused, this function only waits for the currently running tasks (otherwise it would wait forever). Note: To wait for just one specific task, use submit_task() instead, and call the wait() member function of the generated future.
template<typename R , typename P >
bool	wait_for (const std::chrono::duration< R, P > &duration)
	Wait for tasks to be completed, but stop waiting after the specified duration has passed.
template<typename C , typename D >
bool	wait_until (const std::chrono::time_point< C, D > &timeout_time)
	Wait for tasks to be completed, but stop waiting after the specified time point has been reached.

Private Member Functions
void	create_threads (const std::function< void()> &init_task)
	Create the threads in the pool and assign a worker to each thread.
void	destroy_threads ()
	Destroy the threads in the pool.
void	worker (const concurrency_t idx, const std::function< void()> &init_task)
	A worker function to be assigned to each thread in the pool. Waits until it is notified by detach_task() that a task is available, and then retrieves the task from the queue and executes it. Once the task finishes, the worker notifies wait() in case it is waiting.

Static Private Member Functions
static concurrency_t	determine_thread_count (const concurrency_t num_threads)
	Determine how many threads the pool should have, based on the parameter passed to the constructor or reset().

Private Attributes
std::condition_variable	task_available_cv = {}
	A condition variable to notify worker() that a new task has become available.
std::condition_variable	tasks_done_cv = {}
	A condition variable to notify wait() that the tasks are done.
std::queue< std::function< void()> >	tasks = {}
	A queue of tasks to be executed by the threads.
size_t	tasks_running = 0
	A counter for the total number of currently running tasks.
std::mutex	tasks_mutex = {}
	A mutex to synchronize access to the task queue by different threads.
concurrency_t	thread_count = 0
	The number of threads in the pool.
std::unique_ptr< std::thread[]>	threads = nullptr
	A smart pointer to manage the memory allocated for the threads.
bool	waiting = false
	A flag indicating that wait() is active and expects to be notified whenever a task is done.
bool	workers_running = false
	A flag indicating to the workers to keep running. When set to false, the workers terminate permanently.

Detailed Description

A fast, lightweight, and easy-to-use C++17 thread pool class.

Constructor & Destructor Documentation

thread_pool() [1/4]

BS::thread_pool::thread_pool ( )

inline

Construct a new thread pool. The number of threads will be the total number of hardware threads available, as reported by the implementation. This is usually determined by the number of cores in the CPU. If a core is hyperthreaded, it will count as two threads.

: thread_pool(0, [] {}) {}

thread_pool() [2/4]

BS::thread_pool::thread_pool ( const concurrency_t  num_threads )

inlineexplicit

Construct a new thread pool with the specified number of threads.

Parameters

num_threads

The number of threads to use.

: thread_pool(num_threads, [] {}) {}

thread_pool() [3/4]

BS::thread_pool::thread_pool ( const std::function< void()> &  init_task )

inlineexplicit

Construct a new thread pool with the specified initialization function.

Parameters

init_task

An initialization function to run in each thread before it starts to execute any submitted tasks. The function must take no arguments and have no return value. It will only be executed exactly once, when the thread is first constructed.

: thread_pool(0, init_task) {}

thread_pool() [4/4]

BS::thread_pool::thread_pool ( const concurrency_t  num_threads, const std::function< void()> &  init_task  )

inline

Construct a new thread pool with the specified number of threads and initialization function.

Parameters

num_threads	The number of threads to use.
init_task	An initialization function to run in each thread before it starts to execute any submitted tasks. The function must take no arguments and have no return value. It will only be executed exactly once, when the thread is first constructed.

                                                                                       : thread_count(determine_thread_count(num_threads)), threads(std::make_unique<std::thread[]>(determine_thread_count(num_threads)))
    {
        create_threads(init_task);
    }

~thread_pool()

BS::thread_pool::~thread_pool ( )

inline

Destruct the thread pool. Waits for all tasks to complete, then destroys all threads. Note that if the pool is paused, then any tasks still in the queue will never be executed.

    {
        wait();
        destroy_threads();
    }

Member Function Documentation

get_tasks_queued()

size_t BS::thread_pool::get_tasks_queued ( ) const

inline

Get the number of tasks currently waiting in the queue to be executed by the threads.

Returns

The number of queued tasks.

    {
        const std::scoped_lock tasks_lock(tasks_mutex);
        return tasks.size();
    }

Here is the caller graph for this function:

get_tasks_running()

size_t BS::thread_pool::get_tasks_running ( ) const

inline

Get the number of tasks currently being executed by the threads.

Returns

The number of running tasks.

    {
        const std::scoped_lock tasks_lock(tasks_mutex);
        return tasks_running;
    }

Here is the caller graph for this function:

get_tasks_total()

size_t BS::thread_pool::get_tasks_total ( ) const

inline

Get the total number of unfinished tasks: either still waiting in the queue, or running in a thread. Note that get_tasks_total() == get_tasks_queued() + get_tasks_running().

Returns

The total number of tasks.

    {
        const std::scoped_lock tasks_lock(tasks_mutex);
        return tasks_running + tasks.size();
    }

Here is the caller graph for this function:

get_thread_count()

concurrency_t BS::thread_pool::get_thread_count ( ) const

inline

Get the number of threads in the pool.

Returns

The number of threads.

    {
        return thread_count;
    }

Here is the caller graph for this function:

get_thread_ids()

std::vector< std::thread::id > BS::thread_pool::get_thread_ids ( ) const

inline

Get a vector containing the unique identifiers for each of the pool's threads, as obtained by std::thread::get_id().

Returns

The unique thread identifiers.

    {
        std::vector<std::thread::id> thread_ids(thread_count);
        for (concurrency_t i = 0; i < thread_count; ++i)
        {
            thread_ids[i] = threads[i].get_id();
        }
        return thread_ids;
    }

Here is the caller graph for this function:

purge()

void BS::thread_pool::purge ( )

inline

Purge all the tasks waiting in the queue. Tasks that are currently running will not be affected, but any tasks still waiting in the queue will be discarded, and will never be executed by the threads. Please note that there is no way to restore the purged tasks.

    {
        const std::scoped_lock tasks_lock(tasks_mutex);
        while (!tasks.empty())
            tasks.pop();
    }

Here is the caller graph for this function:

detach_task()

template<typename F >

void BS::thread_pool::detach_task ( F &&task  BS_THREAD_POOL_PRIORITY_INPUT )

inline

Submit a function with no arguments and no return value into the task queue, with the specified priority. To push a function with arguments, enclose it in a lambda expression. Does not return a future, so the user must use wait() or some other method to ensure that the task finishes executing, otherwise bad things will happen.

Template Parameters

F	The type of the function.

Parameters

task	The function to push.
priority	The priority of the task. Should be between -32,768 and 32,767 (a signed 16-bit integer). The default is 0. Only enabled if BS_THREAD_POOL_ENABLE_PRIORITY is defined.

    {
        {
            const std::scoped_lock tasks_lock(tasks_mutex);
            tasks.emplace(std::forward<F>(task) BS_THREAD_POOL_PRIORITY_OUTPUT);
        }
        task_available_cv.notify_one();
    }

Here is the caller graph for this function:

detach_blocks()

template<typename T , typename F >

void BS::thread_pool::detach_blocks ( const T  first_index, const T  index_after_last, F &&  block, const size_t  num_blocks = 0 BS_THREAD_POOL_PRIORITY_INPUT  )

inline

Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue, with the specified priority. The block function takes two arguments, the start and end of the block, so that it is only called only once per block, but it is up to the user make sure the block function correctly deals with all the indices in each block. Does not return a multi_future, so the user must use wait() or some other method to ensure that the loop finishes executing, otherwise bad things will happen.

Template Parameters

T	The type of the indices. Should be a signed or unsigned integer.
F	The type of the function to loop through.

Parameters

first_index	The first index in the loop.
index_after_last	The index after the last index in the loop. The loop will iterate from first_index to (index_after_last - 1) inclusive. In other words, it will be equivalent to for (T i = first_index; i < index_after_last; ++i). Note that if index_after_last <= first_index, no blocks will be submitted.
block	A function that will be called once per block. Should take exactly two arguments: the first index in the block and the index after the last index in the block. block(start, end) should typically involve a loop of the form for (T i = start; i < end; ++i).
num_blocks	The maximum number of blocks to split the loop into. The default is 0, which means the number of blocks will be equal to the number of threads in the pool.
priority	The priority of the tasks. Should be between -32,768 and 32,767 (a signed 16-bit integer). The default is 0. Only enabled if BS_THREAD_POOL_ENABLE_PRIORITY is defined.

    {
        if (index_after_last > first_index)
        {
            const blocks blks(first_index, index_after_last, num_blocks ? num_blocks : thread_count);
            for (size_t blk = 0; blk < blks.get_num_blocks(); ++blk)
                detach_task(
                    [block = std::forward<F>(block), start = blks.start(blk), end = blks.end(blk)]
                    {
                        block(start, end);
                    } BS_THREAD_POOL_PRIORITY_OUTPUT);
        }
    }

Here is the call graph for this function:

Here is the caller graph for this function:

detach_loop()

template<typename T , typename F >

void BS::thread_pool::detach_loop ( const T  first_index, const T  index_after_last, F &&  loop, const size_t  num_blocks = 0 BS_THREAD_POOL_PRIORITY_INPUT  )

inline

Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue, with the specified priority. The loop function takes one argument, the loop index, so that it is called many times per block. Does not return a multi_future, so the user must use wait() or some other method to ensure that the loop finishes executing, otherwise bad things will happen.

Template Parameters

T	The type of the indices. Should be a signed or unsigned integer.
F	The type of the function to loop through.

Parameters

first_index	The first index in the loop.
index_after_last	The index after the last index in the loop. The loop will iterate from first_index to (index_after_last - 1) inclusive. In other words, it will be equivalent to for (T i = first_index; i < index_after_last; ++i). Note that if index_after_last <= first_index, no blocks will be submitted.
loop	The function to loop through. Will be called once per index, many times per block. Should take exactly one argument: the loop index.
num_blocks	The maximum number of blocks to split the loop into. The default is 0, which means the number of blocks will be equal to the number of threads in the pool.
priority	The priority of the tasks. Should be between -32,768 and 32,767 (a signed 16-bit integer). The default is 0. Only enabled if BS_THREAD_POOL_ENABLE_PRIORITY is defined.

    {
        if (index_after_last > first_index)
        {
            const blocks blks(first_index, index_after_last, num_blocks ? num_blocks : thread_count);
            for (size_t blk = 0; blk < blks.get_num_blocks(); ++blk)
                detach_task(
                    [loop = std::forward<F>(loop), start = blks.start(blk), end = blks.end(blk)]
                    {
                        for (T i = start; i < end; ++i)
                            loop(i);
                    } BS_THREAD_POOL_PRIORITY_OUTPUT);
        }
    }

Here is the call graph for this function:

Here is the caller graph for this function:

detach_sequence()

template<typename T , typename F >

void BS::thread_pool::detach_sequence ( const T  first_index, const T  index_after_last, F &&sequence  BS_THREAD_POOL_PRIORITY_INPUT  )

inline

Submit a sequence of tasks enumerated by indices to the queue, with the specified priority. Does not return a multi_future, so the user must use wait() or some other method to ensure that the sequence finishes executing, otherwise bad things will happen.

Template Parameters

T	The type of the indices. Should be a signed or unsigned integer.
F	The type of the function used to define the sequence.

Parameters

first_index	The first index in the sequence.
index_after_last	The index after the last index in the sequence. The sequence will iterate from first_index to (index_after_last - 1) inclusive. In other words, it will be equivalent to for (T i = first_index; i < index_after_last; ++i). Note that if index_after_last <= first_index, no tasks will be submitted.
sequence	The function used to define the sequence. Will be called once per index. Should take exactly one argument, the index.
priority	The priority of the tasks. Should be between -32,768 and 32,767 (a signed 16-bit integer). The default is 0. Only enabled if BS_THREAD_POOL_ENABLE_PRIORITY is defined.

    {
        for (T i = first_index; i < index_after_last; ++i)
            detach_task(
                [sequence = std::forward<F>(sequence), i]
                {
                    sequence(i);
                } BS_THREAD_POOL_PRIORITY_OUTPUT);
    }

Here is the caller graph for this function:

reset() [1/4]

void BS::thread_pool::reset ( )

inline

Reset the pool with the total number of hardware threads available, as reported by the implementation. Waits for all currently running tasks to be completed, then destroys all threads in the pool and creates a new thread pool with the new number of threads. Any tasks that were waiting in the queue before the pool was reset will then be executed by the new threads. If the pool was paused before resetting it, the new pool will be paused as well.

    {
        reset(0, [] {});
    }

Here is the caller graph for this function:

reset() [2/4]

void BS::thread_pool::reset ( const concurrency_t  num_threads )

inline

Reset the pool with a new number of threads. Waits for all currently running tasks to be completed, then destroys all threads in the pool and creates a new thread pool with the new number of threads. Any tasks that were waiting in the queue before the pool was reset will then be executed by the new threads. If the pool was paused before resetting it, the new pool will be paused as well.

Parameters

num_threads

The number of threads to use.

    {
        reset(num_threads, [] {});
    }

reset() [3/4]

void BS::thread_pool::reset ( const std::function< void()> &  init_task )

inline

Reset the pool with the total number of hardware threads available, as reported by the implementation, and a new initialization function. Waits for all currently running tasks to be completed, then destroys all threads in the pool and creates a new thread pool with the new number of threads and initialization function. Any tasks that were waiting in the queue before the pool was reset will then be executed by the new threads. If the pool was paused before resetting it, the new pool will be paused as well.

Parameters

init_task

An initialization function to run in each thread before it starts to execute any submitted tasks. The function must take no arguments and have no return value. It will only be executed exactly once, when the thread is first constructed.

    {
        reset(0, init_task);
    }

reset() [4/4]

void BS::thread_pool::reset ( const concurrency_t  num_threads, const std::function< void()> &  init_task  )

inline

Reset the pool with a new number of threads and a new initialization function. Waits for all currently running tasks to be completed, then destroys all threads in the pool and creates a new thread pool with the new number of threads and initialization function. Any tasks that were waiting in the queue before the pool was reset will then be executed by the new threads. If the pool was paused before resetting it, the new pool will be paused as well.

Parameters

num_threads	The number of threads to use.
init_task	An initialization function to run in each thread before it starts to execute any submitted tasks. The function must take no arguments and have no return value. It will only be executed exactly once, when the thread is first constructed.

    {
#ifdef BS_THREAD_POOL_ENABLE_PAUSE
        std::unique_lock tasks_lock(tasks_mutex);
        const bool was_paused = paused;
        paused = true;
        tasks_lock.unlock();
#endif
        wait();
        destroy_threads();
        thread_count = determine_thread_count(num_threads);
        threads = std::make_unique<std::thread[]>(thread_count);
        create_threads(init_task);
#ifdef BS_THREAD_POOL_ENABLE_PAUSE
        tasks_lock.lock();
        paused = was_paused;
#endif
    }

submit_task()

template<typename F , typename R = std::invoke_result_t<std::decay_t<F>>>

std::future< R > BS::thread_pool::submit_task ( F &&task  BS_THREAD_POOL_PRIORITY_INPUT )

inline

Submit a function with no arguments into the task queue, with the specified priority. To submit a function with arguments, enclose it in a lambda expression. If the function has a return value, get a future for the eventual returned value. If the function has no return value, get an std::future<void> which can be used to wait until the task finishes.

Template Parameters

F	The type of the function.
R	The return type of the function (can be void).

Parameters

task	The function to submit.
priority	The priority of the task. Should be between -32,768 and 32,767 (a signed 16-bit integer). The default is 0. Only enabled if BS_THREAD_POOL_ENABLE_PRIORITY is defined.

Returns

A future to be used later to wait for the function to finish executing and/or obtain its returned value if it has one.

    {
        const std::shared_ptr<std::promise<R>> task_promise = std::make_shared<std::promise<R>>();
        detach_task(
            [task = std::forward<F>(task), task_promise]
            {
#ifndef BS_THREAD_POOL_DISABLE_EXCEPTION_HANDLING
                try
                {
#endif
                    if constexpr (std::is_void_v<R>)
                    {
                        task();
                        task_promise->set_value();
                    }
                    else
                    {
                        task_promise->set_value(task());
                    }
#ifndef BS_THREAD_POOL_DISABLE_EXCEPTION_HANDLING
                }
                catch (...)
                {
                    try
                    {
                        task_promise->set_exception(std::current_exception());
                    }
                    catch (...)
                    {
                    }
                }
#endif
            } BS_THREAD_POOL_PRIORITY_OUTPUT);
        return task_promise->get_future();
    }

Here is the caller graph for this function:

submit_blocks()

template<typename T , typename F , typename R = std::invoke_result_t<std::decay_t<F>, T, T>>

multi_future< R > BS::thread_pool::submit_blocks ( const T  first_index, const T  index_after_last, F &&  block, const size_t  num_blocks = 0 BS_THREAD_POOL_PRIORITY_INPUT  )

inline

Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue, with the specified priority. The block function takes two arguments, the start and end of the block, so that it is only called only once per block, but it is up to the user make sure the block function correctly deals with all the indices in each block. Returns a multi_future that contains the futures for all of the blocks.

Template Parameters

T	The type of the indices. Should be a signed or unsigned integer.
F	The type of the function to loop through.
R	The return type of the function to loop through (can be void).

Parameters

first_index	The first index in the loop.
index_after_last	The index after the last index in the loop. The loop will iterate from first_index to (index_after_last - 1) inclusive. In other words, it will be equivalent to for (T i = first_index; i < index_after_last; ++i). Note that if index_after_last <= first_index, no blocks will be submitted, and an empty multi_future will be returned.
block	A function that will be called once per block. Should take exactly two arguments: the first index in the block and the index after the last index in the block. block(start, end) should typically involve a loop of the form for (T i = start; i < end; ++i).
num_blocks	The maximum number of blocks to split the loop into. The default is 0, which means the number of blocks will be equal to the number of threads in the pool.
priority	The priority of the tasks. Should be between -32,768 and 32,767 (a signed 16-bit integer). The default is 0. Only enabled if BS_THREAD_POOL_ENABLE_PRIORITY is defined.

Returns

A multi_future that can be used to wait for all the blocks to finish. If the block function returns a value, the multi_future can also be used to obtain the values returned by each block.

    {
        if (index_after_last > first_index)
        {
            const blocks blks(first_index, index_after_last, num_blocks ? num_blocks : thread_count);
            multi_future<R> future;
            future.reserve(blks.get_num_blocks());
            for (size_t blk = 0; blk < blks.get_num_blocks(); ++blk)
                future.push_back(submit_task(
                    [block = std::forward<F>(block), start = blks.start(blk), end = blks.end(blk)]
                    {
                        return block(start, end);
                    } BS_THREAD_POOL_PRIORITY_OUTPUT));
            return future;
        }
        return {};
    }

Here is the call graph for this function:

Here is the caller graph for this function:

submit_loop()

template<typename T , typename F >

multi_future< void > BS::thread_pool::submit_loop ( const T  first_index, const T  index_after_last, F &&  loop, const size_t  num_blocks = 0 BS_THREAD_POOL_PRIORITY_INPUT  )

inline

Parallelize a loop by automatically splitting it into blocks and submitting each block separately to the queue, with the specified priority. The loop function takes one argument, the loop index, so that it is called many times per block. It must have no return value. Returns a multi_future that contains the futures for all of the blocks.

Template Parameters

T	The type of the indices. Should be a signed or unsigned integer.
F	The type of the function to loop through.

Parameters

first_index	The first index in the loop.
index_after_last	The index after the last index in the loop. The loop will iterate from first_index to (index_after_last - 1) inclusive. In other words, it will be equivalent to for (T i = first_index; i < index_after_last; ++i). Note that if index_after_last <= first_index, no tasks will be submitted, and an empty multi_future will be returned.
loop	The function to loop through. Will be called once per index, many times per block. Should take exactly one argument: the loop index. It cannot have a return value.
num_blocks	The maximum number of blocks to split the loop into. The default is 0, which means the number of blocks will be equal to the number of threads in the pool.
priority	The priority of the tasks. Should be between -32,768 and 32,767 (a signed 16-bit integer). The default is 0. Only enabled if BS_THREAD_POOL_ENABLE_PRIORITY is defined.

Returns

A multi_future that can be used to wait for all the blocks to finish.

    {
        if (index_after_last > first_index)
        {
            const blocks blks(first_index, index_after_last, num_blocks ? num_blocks : thread_count);
            multi_future<void> future;
            future.reserve(blks.get_num_blocks());
            for (size_t blk = 0; blk < blks.get_num_blocks(); ++blk)
                future.push_back(submit_task(
                    [loop = std::forward<F>(loop), start = blks.start(blk), end = blks.end(blk)]
                    {
                        for (T i = start; i < end; ++i)
                            loop(i);
                    } BS_THREAD_POOL_PRIORITY_OUTPUT));
            return future;
        }
        return {};
    }

Here is the call graph for this function:

Here is the caller graph for this function:

submit_sequence()

template<typename T , typename F , typename R = std::invoke_result_t<std::decay_t<F>, T>>

multi_future< R > BS::thread_pool::submit_sequence ( const T  first_index, const T  index_after_last, F &&sequence  BS_THREAD_POOL_PRIORITY_INPUT  )

inline

Submit a sequence of tasks enumerated by indices to the queue, with the specified priority. Returns a multi_future that contains the futures for all of the tasks.

Template Parameters

T	The type of the indices. Should be a signed or unsigned integer.
F	The type of the function used to define the sequence.
R	The return type of the function used to define the sequence (can be void).

Parameters

first_index	The first index in the sequence.
index_after_last	The index after the last index in the sequence. The sequence will iterate from first_index to (index_after_last - 1) inclusive. In other words, it will be equivalent to for (T i = first_index; i < index_after_last; ++i). Note that if index_after_last <= first_index, no tasks will be submitted, and an empty multi_future will be returned.
sequence	The function used to define the sequence. Will be called once per index. Should take exactly one argument, the index.
priority	The priority of the tasks. Should be between -32,768 and 32,767 (a signed 16-bit integer). The default is 0. Only enabled if BS_THREAD_POOL_ENABLE_PRIORITY is defined.

Returns

A multi_future that can be used to wait for all the tasks to finish. If the sequence function returns a value, the multi_future can also be used to obtain the values returned by each task.

    {
        if (index_after_last > first_index)
        {
            multi_future<R> future;
            future.reserve(static_cast<size_t>(index_after_last - first_index));
            for (T i = first_index; i < index_after_last; ++i)
                future.push_back(submit_task(
                    [sequence = std::forward<F>(sequence), i]
                    {
                        return sequence(i);
                    } BS_THREAD_POOL_PRIORITY_OUTPUT));
            return future;
        }
        return {};
    }

Here is the caller graph for this function:

wait()

void BS::thread_pool::wait ( )

inline

Wait for tasks to be completed. Normally, this function waits for all tasks, both those that are currently running in the threads and those that are still waiting in the queue. However, if the pool is paused, this function only waits for the currently running tasks (otherwise it would wait forever). Note: To wait for just one specific task, use submit_task() instead, and call the wait() member function of the generated future.

Exceptions

`wait_deadlock`

if called from within a thread of the same pool, which would result in a deadlock. Only enabled if BS_THREAD_POOL_ENABLE_WAIT_DEADLOCK_CHECK is defined.

    {
#ifdef BS_THREAD_POOL_ENABLE_WAIT_DEADLOCK_CHECK
        if (this_thread::get_pool() == this)
            throw wait_deadlock();
#endif
        std::unique_lock tasks_lock(tasks_mutex);
        waiting = true;
        tasks_done_cv.wait(tasks_lock,
            [this]
            {
                return (tasks_running == 0) && BS_THREAD_POOL_PAUSED_OR_EMPTY;
            });
        waiting = false;
    }

Here is the caller graph for this function:

wait_for()

template<typename R , typename P >

bool BS::thread_pool::wait_for ( const std::chrono::duration< R, P > &  duration )

inline

Wait for tasks to be completed, but stop waiting after the specified duration has passed.

Template Parameters

R	An arithmetic type representing the number of ticks to wait.
P	An std::ratio representing the length of each tick in seconds.

Parameters

duration

The amount of time to wait.

Returns

true if all tasks finished running, false if the duration expired but some tasks are still running.

Exceptions

`wait_deadlock`

if called from within a thread of the same pool, which would result in a deadlock. Only enabled if BS_THREAD_POOL_ENABLE_WAIT_DEADLOCK_CHECK is defined.

    {
#ifdef BS_THREAD_POOL_ENABLE_WAIT_DEADLOCK_CHECK
        if (this_thread::get_pool() == this)
            throw wait_deadlock();
#endif
        std::unique_lock tasks_lock(tasks_mutex);
        waiting = true;
        const bool status = tasks_done_cv.wait_for(tasks_lock, duration,
            [this]
            {
                return (tasks_running == 0) && BS_THREAD_POOL_PAUSED_OR_EMPTY;
            });
        waiting = false;
        return status;
    }

Here is the caller graph for this function:

wait_until()

template<typename C , typename D >

bool BS::thread_pool::wait_until ( const std::chrono::time_point< C, D > &  timeout_time )

inline

Wait for tasks to be completed, but stop waiting after the specified time point has been reached.

Template Parameters

C	The type of the clock used to measure time.
D	An std::chrono::duration type used to indicate the time point.

Parameters

timeout_time

The time point at which to stop waiting.

Returns

true if all tasks finished running, false if the time point was reached but some tasks are still running.

Exceptions

`wait_deadlock`

if called from within a thread of the same pool, which would result in a deadlock. Only enabled if BS_THREAD_POOL_ENABLE_WAIT_DEADLOCK_CHECK is defined.

    {
#ifdef BS_THREAD_POOL_ENABLE_WAIT_DEADLOCK_CHECK
        if (this_thread::get_pool() == this)
            throw wait_deadlock();
#endif
        std::unique_lock tasks_lock(tasks_mutex);
        waiting = true;
        const bool status = tasks_done_cv.wait_until(tasks_lock, timeout_time,
            [this]
            {
                return (tasks_running == 0) && BS_THREAD_POOL_PAUSED_OR_EMPTY;
            });
        waiting = false;
        return status;
    }

Here is the caller graph for this function:

create_threads()

void BS::thread_pool::create_threads ( const std::function< void()> &  init_task )

inlineprivate

Create the threads in the pool and assign a worker to each thread.

Parameters

init_task

An initialization function to run in each thread before it starts to execute any submitted tasks.

    {
        {
            const std::scoped_lock tasks_lock(tasks_mutex);
            tasks_running = thread_count;
            workers_running = true;
        }
        for (concurrency_t i = 0; i < thread_count; ++i)
        {
            threads[i] = std::thread(&thread_pool::worker, this, i, init_task);
        }
    }

destroy_threads()

void BS::thread_pool::destroy_threads ( )

inlineprivate

Destroy the threads in the pool.

    {
        {
            const std::scoped_lock tasks_lock(tasks_mutex);
            workers_running = false;
        }
        task_available_cv.notify_all();
        for (concurrency_t i = 0; i < thread_count; ++i)
        {
            threads[i].join();
        }
    }

determine_thread_count()

static concurrency_t BS::thread_pool::determine_thread_count ( const concurrency_t  num_threads )

inlinestaticprivate

Determine how many threads the pool should have, based on the parameter passed to the constructor or reset().

Parameters

num_threads

The parameter passed to the constructor or reset(). If the parameter is a positive number, then the pool will be created with this number of threads. If the parameter is non-positive, or a parameter was not supplied (in which case it will have the default value of 0), then the pool will be created with the total number of hardware threads available, as obtained from std::thread::hardware_concurrency(). If the latter returns zero for some reason, then the pool will be created with just one thread.

Returns

The number of threads to use for constructing the pool.

    {
        if (num_threads > 0)
            return num_threads;
        if (std::thread::hardware_concurrency() > 0)
            return std::thread::hardware_concurrency();
        return 1;
    }

worker()

void BS::thread_pool::worker ( const concurrency_t  idx, const std::function< void()> &  init_task  )

inlineprivate

A worker function to be assigned to each thread in the pool. Waits until it is notified by detach_task() that a task is available, and then retrieves the task from the queue and executes it. Once the task finishes, the worker notifies wait() in case it is waiting.

Parameters

idx	The index of this thread.
init_task	An initialization function to run in this thread before it starts to execute any submitted tasks.

    {
        this_thread::get_index.index = idx;
        this_thread::get_pool.pool = this;
        init_task();
        std::unique_lock tasks_lock(tasks_mutex);
        while (true)
        {
            --tasks_running;
            tasks_lock.unlock();
            if (waiting && (tasks_running == 0) && BS_THREAD_POOL_PAUSED_OR_EMPTY)
                tasks_done_cv.notify_all();
            tasks_lock.lock();
            task_available_cv.wait(tasks_lock,
                [this]
                {
                    return !BS_THREAD_POOL_PAUSED_OR_EMPTY || !workers_running;
                });
            if (!workers_running)
                break;
            {
#ifdef BS_THREAD_POOL_ENABLE_PRIORITY
                const std::function<void()> task = std::move(std::remove_const_t<pr_task&>(tasks.top()).task);
                tasks.pop();
#else
                const std::function<void()> task = std::move(tasks.front());
                tasks.pop();
#endif
                ++tasks_running;
                tasks_lock.unlock();
                task();
            }
            tasks_lock.lock();
        }
        this_thread::get_index.index = std::nullopt;
        this_thread::get_pool.pool = std::nullopt;
    }

Member Data Documentation

task_available_cv

std::condition_variable BS::thread_pool::task_available_cv = {}

private

A condition variable to notify worker() that a new task has become available.

{};

tasks_done_cv

std::condition_variable BS::thread_pool::tasks_done_cv = {}

private

A condition variable to notify wait() that the tasks are done.

{};

tasks

std::queue<std::function<void()> > BS::thread_pool::tasks = {}

private

A queue of tasks to be executed by the threads.

{};

tasks_mutex

std::mutex BS::thread_pool::tasks_mutex = {}

mutableprivate

A mutex to synchronize access to the task queue by different threads.

{};

---

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

• external/threadpool/include/BS_thread_pool.hpp