A lock-free thread-pool implementation in C++11.
Current version: 1.0.1
Lead Maintainer: Halim Qarroum
This project is an implementation of a lock-free thread-pool in C++. It aims to make it very easy to implement a producer-consumer pattern following C++11 semantics with relatively high performances, although other types of patterns can also be implemented on top of this project.
The implementation uses the lock-free concurrent-queue implementation provided by moodycamel as its underlying thread-safe queuing mechanism for task executions to be spread amongst different worker threads.
To create a thread-pool instance, you call its constructor by providing it with the initial number of threads to provision your thread-pool instance with.
thread::pool::pool_t pool(std::thread::hardware_concurrency() + 1);Note that
std::thread::hardware_concurrencyreturns the number of physical concurrent threads supported by your system. We add 1, since this API can return zero if it was not able to compute the amount of concurrent threads available.
The pool_t class can schedule the execution of callable objects across your worker threads. The simplest way to do so is to invoke the schedule API by providing a callable object, and the arguments to bind to this object. Consequently, an std::future is returned so that you can retrieve the result generated by your worker thread, if any, at call-time.
/**
* \brief Schedules the execution of a lambda function
* taking an integer and returning another integer.
*/
auto worker = [] (int number) {
return (number + 1);
};
// Scheduling the execution of the worker.
auto result = pool.schedule(worker, 42);
// Will block until the result is available.
std::cout << result.get() << std::endl;The schedule method can take any callable object as an argument (static functions, lambda functions, pointers to functions, etc.), and will deduce the appropriate return type of the generated std::future. If the scheduling of the given worker failed, an std::length_error exception will be thrown with a description of the error.
If you do not need to retrieve the result of your work at call-time, you can use the schedule_and_forget method which has a lower overhead in terms of memory usage and performances than the schedule method. This method will never throw exceptions.
auto succeeded = pool.schedule_and_forget(worker, 42);In order to improve performances, it is advised to schedule the execution of callable objects in bulk, by providing an array of callable objects to schedule rather than a single one. The schedule_bulk API is dedicated to bulk insertion of callable objects into the thread-pool.
// Scheduling an array of callables in bulk.
auto result = pool.schedule_bulk(array_of_callables, length));
// Log whether the insertion was successful.
std::cout << "The insertion has " << (result ? "succeeded" : "failed") << std::endl;The schedule_bulk method returns a boolean value indicating whether the bulk insertion has been successful or not. For a complete sample of how to schedule callables in bulk into the thread-pool, have a look at the bulk_insertion example.
In addition to the schedule method, this implementation provides a way to generate functors that can be called like a regular function, but which will instead schedule the execution of your callable on the thread-pool. The functor based syntax provides a more natural way to generate callables and to actually call them.
// Binding a callable function with a thread-pool instance.
auto callable = thread::pool::bind(pool, my_function);
// Scheduling the callable on the thread-pool.
auto result = callable(42);
// Logging the result returned by the callable.
std::cout << result.get() << std::endl;While argument deduction will work seamlessly with other type of callables, in order for the compiler to deduce at compile-time the arguments of a lambda function provided as an argument to thread::pool::bind, you need to provide their types as template parameters.
// Binding using argument types as template parameters.
auto sum = thread::pool::bind<int, int>(pool, [] (int a, int b) {
return (a + b);
});
// Invoking the generated callable.
sum(1, 2);The underlying moodycamel queue used by the thread-pool can optimize the processing of queued elements by using consumer and producer tokens (read more). When dequeuing elements from the queue, the thread-pool already uses an implicit consumer token which is bound to each worker thread.
If you are using multiple producers spread across multiple threads, you might want to create a new consumer token which is unique for the lifetime of that thread. Below is an example of how to create such a producer token.
// Creating a new producer token.
auto token = pool.create_token_of<thread::pool::producer_token_t>();
// Scheduling a callable using the producer token.
auto result = pool.schedule(token, my_function);Every scheduling method (schedule, schedule_and_forget and schedule_bulk) can take an optional producer token as a first argument. If no token is given, scheduling will be done without using tokens.
A thread::pool::parameterized_pool_t class exists in this implementation and allows you to customize the behavior of the thread-pool at compile time. The thread-pool is made such that workers will dequeue callables from the internal queue in bulk in order to improve performances, as such two parameters have been defined to alter the inner workings of the pool :
The number of items that the thread-pool will attempt to dequeue has an impact on performances. To fine tune this impact, we have defined a few constants that you can use as hints for the thread-pool to use when dequeuing callables. Note that when dequeued by workers, callables will be stored on the worker thread's stack, make sure that whatever hint you use, it can fit your thread stack size.
-
WORK_PARTITIONING_LIGHT- Hint indicating the use of lightweight processing of tasks within a worker (less elements will be dequeued). -
WORK_PARTITIONING_SPARSE- Hint indicating the use of sparse processing of tasks within a worker. -
WORK_PARTITIONING_BALANCED- Hint indicating the use of a balanced processing of tasks within a worker. -
WORK_PARTITIONING_HEAVY- Hint indicating the use of heavy processing of tasks within a worker. -
WORK_PARTITIONING_HEAVIER- Hint indicating the use of heavier processing of tasks within a worker.
You can pass one of these constants as a first template parameter to the parameterized_pool_t when creating it (more elements will be dequeued).
// Configuring the pool for lightweight processing per worker thread.
thread::pool::parameterized_pool_t<thread::pool::WORK_PARTITIONING_LIGHT> pool(
std::thread::hardware_concurrency() + 1
);If you wish to have absolute control on this number, rather than using hints, you can safely pass a custom integer instead of one of the provided constants.
Before the internal worker threads actually executes the enqueued callables, they will block awaiting for an element in the internal queue to be available. To allow clients of the thread-pool to stop it, a maximum amount of time that the worker thread spends waiting is used, and each time we'll reach that timeout, the worker thread will unblock and check whether it needs to stop its processing.
This parameter can have both an impact on performances and on the reactivity of the pool. If the timeout is too low, a stop operation will be quicker, but at the expense of worker threads consuming more CPU cycles unblocking and checking whether they should stop their process. If it is too high, CPU cycles waiste will be lower, but the worker threads will react slower to a stop operation.
The parameterized_pool_t can take a second optional template parameter to set this timeout manually.
thread::pool::parameterized_pool_t<thread::pool::WORK_PARTITIONING_LIGHT, 2 * 1000> pool(
std::thread::hardware_concurrency() + 1
);In the above example, worker threads will wait 2 seconds for elements in the queue before unblocking.
When you want to interrupt your workers and stop all the threads currently running in your thread-pool explicitely, you can use the stop method. This method will indicate to the running threads that they should stop their current work. The await method will synchronously wait for them to finish.
// Scheduling the interruption of threads execution,
// and awaiting for them to have completed.
pool.stop().await();As explained in the Maximum time to block section, the stop method will not have immediate effect on the running threads, as the time they spend blocking on the queue affects the reactiveness of the stop request.
This thread-pool implementation follow the RAII pattern, meaning that upon destruction, the thread pool object will stop the running thread. Note that because the thread in which the thread-pool is initialized owns the worker threads, the destructor will synchronously wait for the termination of the threads upon destruction.
In this context, you should in fact rarely be invoking the stop or await methods explicitely since this will be taken care of by the destructor of a thread pool instance.
{
// The thread pool creates 5 worker threads upon construction.
thread::pool::pool_t pool(5);
// Scheduling workers.
pool.schedule_bulk(workers, length);
}
// The thread pool and the worker threads are now terminated.In the above example, the execution of all the given workers is not guaranteed. Upon destruction, the thread-pool instance will prompt the worker threads to finish their execution, which may happen before every elements in the internal queue have been processed.
Different sample usages of the thread-pool can be found in the examples directory.
Different unit tests and benchmarks are available under the tests directory. In order to build the tests and the examples, you can simply run make in the project directory. To execute tests, run make tests.