ThreadPool.h

Go to the documentation of this file.

#ifndef AMPGEN_THREADPOOL_H
#define AMPGEN_THREADPOOL_H
 
/* @class ThreadPool ThreadPool.h AmpGen/ThreadPool.h
 * Thread pool implementation taken from https://github.com/progschj/ThreadPool
 * Modified to allow explicit clearing of queues, and to make
 * the constructor explicit
 *
 * Copyright (c) 2012 Jakob Progsch, Václav Zeman
 *
 * This software is provided 'as-is', without any express or implied
 * warranty. In no event will the authors be held liable for any damages
 * arising from the use of this software.
 *
 * Permission is granted to anyone to use this software for any purpose,
 * including commercial applications, and to alter it and redistribute it
 * freely, subject to the following restrictions:
 *
 * 1. The origin of this software must not be misrepresented; you must not
 * claim that you wrote the original software. If you use this software
 * in a product, an acknowledgment in the product documentation would be
 * appreciated but is not required.
 *
 * 2. Altered source versions must be plainly marked as such, and must not be
 * misrepresented as being the original software.
 *
 * 3. This notice may not be removed or altered from any source distribution.
 */
 
#include <memory.h>
#include <stddef.h>
#include <atomic>
#include <condition_variable>
#include <functional>
#include <future>
#include <iostream>
#include <memory>
#include <mutex>
#include <queue>
#include <stdexcept>
#include <thread>
#include <type_traits>
#include <utility>
#include <vector>
#include "AmpGen/Property.h"
 
namespace AmpGen {
  class ThreadPool {
  public:
    explicit ThreadPool(const size_t &nt);
    ~ThreadPool();
    template <typename F, typename... Args> auto enqueue(F &&f, Args &&...args) -> std::future<typename std::invoke_result_t<F, Args...>>;
    void waitForStoppedThreads();
 
  private:
    std::vector<std::thread> m_workers;
    std::queue<std::function<void()>> m_tasks;
    std::mutex m_queue_mutex;
    std::condition_variable m_condition;
    bool m_stop = {false};
    Property<bool> m_enableThreadPool{this, "ThreadPool::Enable", true};
  };
 
  template <typename F, typename... Args> auto ThreadPool::enqueue(F &&f, Args &&...args) -> std::future<typename std::invoke_result_t<F, Args...>> {
    using return_type = typename std::invoke_result_t<F, Args...>;
    auto task = std::make_shared<std::packaged_task<return_type()>>(f, args...);
    std::future<return_type> res = task->get_future();
    if(!m_enableThreadPool) {
      f(args...);
      return res;
    } else {
      {
        std::unique_lock<std::mutex> lock(m_queue_mutex);
        if(m_stop) throw std::runtime_error("enqueue on stopped ThreadPool");
        m_tasks.emplace([task]() { (*task)(); });
      }
      m_condition.notify_one();
    }
    return res;
  }
} // namespace AmpGen;
#endif