A generic thread pool for modern C++.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

200 lines
5.8 KiB

#pragma once
//
// main.cpp
// thread_pool v0.2a
//
// Created by Kristof Toth on 26/06/15.
//
// Copyright © 2015 Kristof Toth <mrtoth@strongds.hu>
// This program is free software. It comes without any warranty, to
// the extent permitted by applicable law. You can redistribute it
// and/or modify it under the terms of the Do What The Fuck You Want
// To Public License, Version 2, as published by Sam Hocevar. See
// http://www.wtfpl.net/ for more details.
//
// Generalized thread pool class.
// ( READ BEFORE USING IN PRODUCTION CODE )
// User's notes:
// - the recommended way to add a task to the queue is to use the
// add_task(F&& f, Args&&... args) format, where f is any kind of
// functor and (args...) are the parameters to the functor (any number/kind).
// - copying & moving not allowed (not sure if they'd make sense)
// - if a task is added to the pool via add_task(std::packaged_task<R()>& ),
// the client must grant, that the packaged_task object is NOT
// destructed before the task is finished. (otherwise it's undefined behaviour)
// this way of adding tasks is only recommended for micro-operations,
// where the overhead of add_task(F&& f, Args&&... args)'s use of operator new
// is too much (this is rarely the case).
//
/* necessary includes */
#include <thread>
#include <future>
#include <mutex>
#include <atomic>
#include <vector>
#include <deque>
#include <condition_variable>
#include <functional>
#include <memory>
class thread_pool
{
/* under the hood */
std::mutex mu;
std::condition_variable cond;
std::vector<std::thread> workers;
std::deque<std::function<void()> > queue;
std::atomic<bool> fin;
/* the main loop of the threads */
void loop();
template <bool front, typename F, typename... Args>
auto _add_task_internal(F&& f, Args&&... args)
-> std::future<decltype(f(std::forward<Args>(args)...))>;
public:
/* construction & destruction */
explicit thread_pool(size_t);
~thread_pool();
/* disallowed operations */
thread_pool(const thread_pool&) = delete;
thread_pool& operator=(const thread_pool&) = delete;
thread_pool(thread_pool&&) = delete;
thread_pool& operator=(thread_pool&&) = delete;
template <typename R> void add_task(std::packaged_task<R()>&& ) = delete;
template <typename R> void priority_task(std::packaged_task<R()>&& ) = delete;
/* adding tasks to the queue */
template <typename F, typename... Args>
auto add_task(F&& f, Args&&... args) -> std::future<decltype(f(std::forward<Args>(args)...))>;
template <typename F, typename... Args>
auto priority_task(F&& f, Args&&... args) -> std::future<decltype(f(std::forward<Args>(args)...))>;
template <typename R> void add_task(std::packaged_task<R()>& );
template <typename R> void priority_task(std::packaged_task<R()>& );
// template <typename R> void add_task(std::function<R()> ); << think about this
void add_task(std::function<void()> );
/* other operations */
void add_thread(size_t);
/* means of getting information */
inline size_t get_thread_num() const { return workers.size(); }
inline size_t get_queue_size() const { return queue.size(); }
};
template <bool front, typename F, typename... Args>
auto thread_pool::_add_task_internal(F&& f, Args&&... args)
-> std::future<decltype(f(std::forward<Args>(args)...))>
{
auto pckgd_tsk = std::make_shared<std::packaged_task<decltype(f(std::forward<Args>(args)...))()> >
(std::bind(std::forward<F>(f), std::forward<Args>(args)...));
{
std::lock_guard<std::mutex> lock(mu);
if (front)
queue.emplace_front([pckgd_tsk](){ (*pckgd_tsk)(); });
else
queue.emplace_back([pckgd_tsk](){ (*pckgd_tsk)(); });
}
cond.notify_one();
return pckgd_tsk->get_future();
}
thread_pool::thread_pool(size_t thcount):
fin(false)
{
for (size_t i = 0; i < thcount ; ++i)
workers.emplace_back(&thread_pool::loop, this);
}
thread_pool::~thread_pool()
{
fin = true;
cond.notify_all();
for (auto& i : workers)
i.join();
}
template <typename F, typename... Args>
auto thread_pool::add_task(F&& f, Args&&... args)
-> std::future<decltype(f(std::forward<Args>(args)...))>
{
return _add_task_internal<false>(std::forward<F>(f), std::forward<Args>(args)...);
}
template <typename F, typename... Args>
auto thread_pool::priority_task(F&& f, Args&&... args)
-> std::future<decltype(f(std::forward<Args>(args)...))>
{
return _add_task_internal<true>(std::forward<F>(f), std::forward<Args>(args)...);
}
template <typename R>
void thread_pool::add_task(std::packaged_task<R()>& arg)
{
{
std::lock_guard<std::mutex> lock(mu);
queue.emplace_back([&arg](){ arg(); });
}
cond.notify_one();
}
template <typename R>
void thread_pool::priority_task(std::packaged_task<R()>& arg)
{
{
std::lock_guard<std::mutex> lock(mu);
queue.emplace_front([&arg](){ arg(); });
}
cond.notify_one();
}
void thread_pool::add_task(std::function<void()> func)
{
{
std::lock_guard<std::mutex> lock(mu);
queue.push_back(std::move(func));
}
cond.notify_one();
}
void thread_pool::add_thread(size_t num = 1)
{
for (size_t i = 0; i < num; ++i)
{
workers.emplace_back(&thread_pool::loop, this);
}
}
void thread_pool::loop()
{
std::function<void()> fun;
while (true)
{
{
std::unique_lock<std::mutex> lock(mu);
while (!fin && queue.empty())
cond.wait(lock);
if (fin)
return;
fun = std::move(queue.front());
queue.pop_front();
}
fun();
fun = nullptr;
}
}