thread_pool.cpp

#include "thread_pool.h"

#include <algorithm>

ThreadPool::ThreadPool(unsigned num_threads) {
  if (num_threads == 0) num_threads = 1;
  // Pre-size stats so each worker writes to its own slot (no reallocation, no
  // sharing between workers => no data races on the stats vector).
  stats_.resize(num_threads);
  workers_.reserve(num_threads);
  for (unsigned i = 0; i < num_threads; ++i) {
    workers_.emplace_back([this, i] { worker_loop(i); });
  }
}

ThreadPool::~ThreadPool() { close_and_join(); }

void ThreadPool::submit(const Chunk& chunk) { queue_.push(chunk); }

void ThreadPool::worker_loop(unsigned id) {
  Chunk chunk;
  while (queue_.pop(chunk)) {
    process_chunk(chunk, stats_[id]);
  }
}

void ThreadPool::close_and_join() {
  if (joined_) return;
  joined_ = true;
  queue_.close();
  for (std::thread& t : workers_) {
    if (t.joinable()) t.join();
  }
}