Thread Pools Implementation

• The thread_pool class will contain

  • A vector of std::thread objects
  • A concurrent queue to store incoming tasks as callable objects

• Each thread will execute as an infinite loop

  • Call pop() on the queue
  • Invoke the task returned by pop()
  • Call pop() on the queue again

Interface

• The thread_pool class will have a submit() member function

  • Takes a callable object as argument

• Users will call submit() with a task as argument

• The task is pushed on the queue

  • It will be performed by the next thread that is ready

Makefile

CXX = g++
CFLAGS = -Wall

all: thread_pool_main

thread_pool_main.o: thread_pool_main.cpp
	$(CXX) $(CFLAGS) -c thread_pool_main.cpp

thr_pool.o : thr_pool.cpp
	$(CXX) $(CFLAGS) -c thr_pool.cpp

thread_pool_main: thread_pool_main.o thr_pool.o
	$(CXX) -o thread_pool_main thread_pool_main.o thr_pool.o

thr_pool.h

// reuse the concurrent_queue from previous lecture
#include "concurrent_queue.h"
#include <condition_variable>
#include <functional>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>

// Type alias to simplify the code
// All the task functions will have this type
using Func = std::function<void()>;

class thread_pool {
  concurrent_queue<Func> work_queue; // Queue of tasks
  std::vector<std::thread> threads;  // Vector of thread objects
  void worker();                     // Their entry point function
  unsigned int thread_count;
  bool completed;

public:
  thread_pool();
  ~thread_pool();

  // Add a task to the queue
  void submit(Func func);
};

thr_pool.cpp

#include "thr_pool.h"
#include <iostream>
#include <thread>

thread_pool::thread_pool() {
  thread_count = std::thread::hardware_concurrency() - 1;
  std::cout << "Creating a thread pool with " << thread_count << " threads\n";
  completed = false;

  // Start the threads
  for (unsigned int i = 0; i < thread_count; ++i) {
    threads.push_back(std::thread{&thread_pool::worker, this});
  }
}

thread_pool::~thread_pool() {
  // Wait for the threads to finish
  for (auto &thr : threads)
    thr.join();
}

// Entry point function for the threads
void thread_pool::worker() {
  while (!completed) {
    Func task;

    // Take a task function off the queue
    work_queue.pop(task);
    task();

    if (work_queue.empty())
      completed = true;
  }
}

// Add a task to the queue
void thread_pool::submit(Func func) { work_queue.push(func); }

thread_pool_main.cpp

#include "thr_pool.h"
#include <iostream>
#include <thread>

using namespace std::chrono_literals;

// A task function
void task() {
  std::cout << "Thread id: " << std::this_thread::get_id() << " starting a task"
            << std::endl;
  std::this_thread::sleep_for(100ms);
  std::cout << "Thread id: " << std::this_thread::get_id()
            << " finishing a task" << std::endl;
}

int main() {
  // Create the thread pool
  thread_pool pool;

  // Send some tasks to the thread pool
  for (int i = 0; i < 20; ++i)
    pool.submit(task);

  pool.submit([&pool]() {
    std::this_thread::sleep_for(1s);
    std::cout << "All tasks completed \n";
  });
}