thread_pool.hpp

// MIT License
//
// Copyright (c) 2016-2017 Simon Ninon <simon.ninon@gmail.com>
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.

#pragma once

#include <atomic>
#include <condition_variable>
#include <functional>
#include <list>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>

namespace tacopie {

namespace utils {

class thread_pool {
public:
  explicit thread_pool(std::size_t nb_threads);

  ~thread_pool(void);

  thread_pool(const thread_pool&) = delete;
  thread_pool& operator=(const thread_pool&) = delete;

public:
  typedef std::function<void()> task_t;

  void add_task(const task_t& task);

  thread_pool& operator<<(const task_t& task);

  void stop(void);

public:
  bool is_running(void) const;

public:
  void set_nb_threads(std::size_t nb_threads);

private:
  void run(void);

  std::pair<bool, task_t> fetch_task_or_stop(void);

  bool should_stop(void) const;

private:
  std::list<std::thread> m_workers;

  std::atomic<std::size_t> m_max_nb_threads = ATOMIC_VAR_INIT(0);

  std::atomic<std::size_t> m_nb_running_threads = ATOMIC_VAR_INIT(0);

  std::atomic<bool> m_should_stop = ATOMIC_VAR_INIT(false);

  std::queue<task_t> m_tasks;

  std::mutex m_tasks_mtx;

  std::condition_variable m_tasks_condvar;
};

} // namespace utils

} // namespace tacopie