Thread Pool In C++

A thread pool consists of multiple threads, each assigned with specific tasks. Consequently, different threads handle unique types of work. Every thread is dedicated to executing a particular group of similar operations, while another thread carries out a different group of similar operations, and this sequence extends to more threads. It is essential for each thread to concentrate on similar functions due to the varying parameters across different threads.

A thread pool is essential in C++ programming and requires a library to handle the creation and supervision of thread pools. This necessity arises due to the various approaches available for establishing a thread pool. Consequently, a C++ developer must tailor a thread pool to meet specific project needs.

What is a thread?

A thread is an entity generated by defining the thread class. Typically, when instantiating, the initial parameter passed to the thread constructor is commonly the name of a high-level method. Subsequent inputs to the thread constructor consist of function arguments. The execution of the function commences immediately upon the thread's instantiation. The primary function in C++, main, operates at a higher level. Tasks performed within the global scope are considered top-level operations. Unlike other threads, the main function functions as a thread that does not necessitate explicit declaration.

What properties should a thread pool have?

A thread pool is essentially a group of threads that are available for tasks. In C++, it can be represented as an array of std::thread or as a vector. This setup is suitable for scalability and accommodating future growth.

At some time, each thread within the thread pool may be assigned a job. When the thread of work is created, the particular job is unknown. In C++, this indicates that a thread in a thread pool:

• ✅ It should be able to perform arbitrary functions that allows any set of arguments and any return value type.
• ✅ It should be allowed to communicate task execution results back to the task's publisher.
• ✅ It should be able to be woken up to do a job when needed while not using excessive CPU resources when not required.
• ✅ When necessary, the controller thread must be able to regulate it to halt tasks, stop accepting tasks, reject unfinished jobs, etc.

The current technique in C++ for the initial step involves utilizing the framework provided by functions within the header file (such as std::bind, std::function, etc.) in combination with template parameterization. In contrast, the traditional method for the second step entails developing the callback function at the same time as the task execution; the more modern C++ strategy would entail utilizing std::packagedtask along with std::future. When dealing with less frequent tasks, it is advisable to consider using std::conditionvariable for the third step, while for more frequent tasks, employing std::this_thread::yield could be more suitable. Lastly, for the fourth step, it can be achieved by designating an internal variable as a token representation and ensuring that each worker thread consistently verifies this token.

The Thread Pool Requirements and Architecture in C++

The design for the thread pool explained in the previous paragraph is simple. It implies that the thread pool does not have any unusual needs. Thus, we may assume:

• ✅ We can pass any function that accepts any form of input parameter . For ease of use, we may argue for our tasks std:: variants in this scenario.
• ✅ The queue must be thread-safe because the thread pool requires a queue to contain jobs and their parameters.
• ✅ The thread pool will only stop running after all the jobs in the queue have been completed.

The key ideas required for our execution are summarized in the list below.

• ✅ Tasks - A structure that represents a task that our thread pool may execute. In its most basic form, this object contains a lambda (or function pointer) of the operation that must be performed and its arguments.
• ✅ A task queue- It is a container for jobs that will be picked out from the thread pool.
• ✅ The thread pool - It is a struct or class that contains the thread pool functionality. It contains the task queue, a std::vector of std::jthreads , and the logic for pushing and popping from the task queues.

Implementing a Thread Pool in C++

Examining the execution and essential aspects of the code subsequent to that.

class ThreadPool
{
public:
 ThreadPool(std::size_t n_threads)
 {
 for (std::size_t i = 0; i < n_threads; ++i)
 {
 _threads.push_back(make_thread_handler(_queue));
 }
 }

 ~ThreadPool()
 {
 // Task = {Execute/Stop, function, args}
 Task const stop_task{TaskType::Stop, {}, {}};
 for (std::size_t i = 0; i < _threads.size(); ++i)
 {
 push(stop_task);
 }
 }

 bool push(Task const& task)
 {
 _queue.push(task);
 return true;
 }
private:
 TsQueue<Task> _queue;
 std::vector<std::jthread> _threads;
}

• ✅ The thread pool is a management layer over the task queue.
• ✅ Pushing to the thread pool is the same as adding a job to the queue.
• ✅ There is no need for a popping mechanism because our basic architecture requires the thread pool to terminate only after completing all tasks.
• ✅ The destructor has to ensure that all threads within the pool are stopped. As a result, it should send a stop task to every thread in the pool.

How do the Threads execute Tasks?

We can observe nthreads std::jthread instances within the threads container in the provided thread pool implementation. Each thread is responsible for executing the logic to dequeue tasks from the queue and terminate when required.

auto make_thread_handler(TsQueue<Task>& queue)
{
 return std::jthread{
 [&queue]{
 while (true)
 {
 auto const elem = queue.pop();
 switch (elem.type) {
 case TaskType::Execute:
 elem.task(elem.arguments);
 break;
 case TaskType::Stop:
 return;
 }
 }
 }
 };
}

Using the Thread Pool

A basic program that showcases an increasing integer with the assistance of a thread pool. Additionally, it reveals the unique thread ID to verify the execution of each increment on separate threads.

int main()
{
 ThreadPool my_pool{4};
 std::mutex iostream_m;

 int job = 0;
 for (int i = 0; i < 100; ++i)
 {
 my_pool.push({
 TaskType::Execute, // declaring the TaskType
 [&iostream_m, job](std::vector<Param> const&) // Lambda
 {
 {
 std::lock_guard cout_guard{iostream_m};
 std::cout << "Hi from the thread " << std::this_thread::get_id()
 << " requesting the job " << job << '\n';
 }
 std::this_thread::sleep_for(std::chrono::milliseconds{50});
 },
 {} // Parameters
 });
 job++;
}