src/ThreadPool.cpp

   1 // Copyright (C) 2019-2020 Joel Rosdahl and other contributors
   2 //
   3 // See doc/AUTHORS.adoc for a complete list of contributors.
   4 //
   5 // This program is free software; you can redistribute it and/or modify it
   6 // under the terms of the GNU General Public License as published by the Free
   7 // Software Foundation; either version 3 of the License, or (at your option)
   8 // any later version.
   9 //
  10 // This program is distributed in the hope that it will be useful, but WITHOUT
  11 // ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12 // FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  13 // more details.
  14 //
  15 // You should have received a copy of the GNU General Public License along with
  16 // this program; if not, write to the Free Software Foundation, Inc., 51
  17 // Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  18
  19 #include "ThreadPool.hpp"
  20
  21 ThreadPool::ThreadPool(size_t number_of_threads, size_t task_queue_max_size)
  22   : m_task_queue_max_size(task_queue_max_size)
  23 {
  24   m_worker_threads.reserve(number_of_threads);
  25   for (size_t i = 0; i < number_of_threads; ++i) {
  26     m_worker_threads.emplace_back(&ThreadPool::worker_thread_main, this);
  27   }
  28 }
  29
  30 ThreadPool::~ThreadPool()
  31 {
  32   shut_down();
  33 }
  34
  35 void
  36 ThreadPool::enqueue(std::function<void()> function)
  37 {
  38   {
  39     std::unique_lock<std::mutex> lock(m_mutex);
  40     if (m_task_queue.size() >= m_task_queue_max_size) {
  41       m_task_popped_condition.wait(
  42         lock, [this] { return m_task_queue.size() < m_task_queue_max_size; });
  43     }
  44     m_task_queue.emplace(function);
  45   }
  46   m_task_enqueued_or_shutting_down_condition.notify_one();
  47 }
  48
  49 void
  50 ThreadPool::shut_down()
  51 {
  52   {
  53     std::unique_lock<std::mutex> lock(m_mutex);
  54     m_shutting_down = true;
  55   }
  56   m_task_enqueued_or_shutting_down_condition.notify_all();
  57   for (auto& thread : m_worker_threads) {
  58     if (thread.joinable()) {
  59       thread.join();
  60     }
  61   }
  62 }
  63
  64 void
  65 ThreadPool::worker_thread_main()
  66 {
  67   while (true) {
  68     std::function<void()> task;
  69
  70     {
  71       std::unique_lock<std::mutex> lock(m_mutex);
  72       m_task_enqueued_or_shutting_down_condition.wait(
  73         lock, [this] { return m_shutting_down || !m_task_queue.empty(); });
  74       if (m_shutting_down && m_task_queue.empty()) {
  75         return;
  76       }
  77       task = std::move(m_task_queue.front());
  78       m_task_queue.pop();
  79     }
  80
  81     m_task_popped_condition.notify_all();
  82     task();
  83   }
  84 }