Thread Pools Multiple Queues

• The queue can become a bottleneck

  • When a thread takes a task off the queue, it locks the queue
  • Other threads are blocked until this operation is complete
  • If there are many small tasks, this can affect performance

• An alternative is to use a separate queue for each thread

  • A thread never has to wait to get to the next task
  • Uses more memory

Work Sharing

• Can perform better than a single-queue pool
  • When there are many small tasks
• If a thread's queue is empty, the thread is idle

Work Sharing Implementation

• Replace the queue by a fixed size vector of queues

  • One element for each thread

• "Round robin" scheduling

  • Put a new task on the next thread's queue
  • After the last element of the vector, go to the front element

Updated thr_pool.h and thr_pool.cpp

// reuse the concurrent_queue from previous lecture
#include "concurrent_queue.h"
#include <cstddef>
#include <functional>
#include <memory>
#include <thread>
#include <vector>

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

// Alias for concurrent_queue type
using Queue = concurrent_queue<Func>;

class thread_pool {
  // Each thread has its own queue of task functions
  std::unique_ptr<Queue[]> work_queues;
  std::vector<std::thread> threads; // Vector of thread objects

  void worker(size_t index); // Their entry point function

  unsigned int thread_count;
  size_t pos{0};

public:
  thread_pool();
  ~thread_pool();

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

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

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

  work_queues = std::make_unique<Queue[]>(thread_count);

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

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(size_t index) {
  while (true) {
    Func task;

    // Take a task function off the queue
    work_queues[index].pop(task);
    task();
  }
}

// Add a task to the queue
void thread_pool::submit(Func func) {
  work_queues[pos].push(func);

  pos = (pos + 1) % thread_count;
}