thread_pool.h

 
#ifndef CTRL_UTILS_THREAD_POOL_H_
#define CTRL_UTILS_THREAD_POOL_H_
 
#include <ctrl_utils/atomic_queue.h>
 
#include <functional>  // std::function
#include <future>      // std::future, std::promise
#include <thread>      // std::thread
#include <utility>     // std::make_pair, std::move, std::pair
#include <vector>      // std::vector
 
namespace ctrl_utils {
 
template <typename T>
class ThreadPool {
 public:
  ThreadPool(size_t num_threads) {
    if (num_threads == 0) {
      num_threads = std::thread::hardware_concurrency();
    }
    threads_.reserve(num_threads);
    for (size_t i = 0; i < num_threads; i++) {
      threads_.emplace_back([this]() { InfiniteLoop(); });
      // threads_.back().detach();
    }
  }
 
  virtual ~ThreadPool() {
    Terminate();
    for (std::thread& t : threads_) {
      if (t.joinable()) t.join();
    }
  }
 
  std::future<T> Submit(std::function<T()>&& job) {
    auto promise = std::make_shared<std::promise<T>>();
    jobs_.Push(std::make_pair(std::move(job), promise));
    return promise->get_future();
  }
 
  std::future<T> Submit(std::function<T()>& job) {
    auto promise = std::make_shared<std::promise<T>>();
    jobs_.Push(std::make_pair(job, promise));
    return promise->get_future();
  }
 
  void Terminate() {
    terminate_ = true;
    jobs_.Terminate();
  }
 
 private:
  using Promise = std::shared_ptr<std::promise<T>>;
 
  void InfiniteLoop() {
    while (!terminate_) {
      std::pair<std::function<T()>, Promise> job_promise = jobs_.Pop();
      std::function<T()>& job = job_promise.first;
      Promise& promise = job_promise.second;
 
      // If terminated, raise an exception instead of processing.
      if (terminate_) {
        if (!promise) return;
        promise->set_exception(std::make_exception_ptr(
            std::runtime_error("ThreadPool::Terminate() called.")));
        return;
      }
 
      // Run the job and return the promised value.
      ExecuteJob(promise, job);
    }
  }
 
  void ExecuteJob(Promise& promise, std::function<T()>& job) {
    promise->set_value(job());
  }
 
  std::vector<std::thread> threads_;
 
  AtomicQueue<std::pair<std::function<T()>, Promise>> jobs_;
 
  std::sig_atomic_t terminate_ = false;
};
 
template <>
void ThreadPool<void>::ExecuteJob(Promise& promise,
                                  std::function<void()>& job) {
  job();
  promise->set_value();
}
 
}  // namespace ctrl_utils
 
#endif  // CTRL_UTILS_THREAD_POOL_H_