v2 of this question is here and v3 is here
To get a better understanding of C++11/C++14, I thought I would develop a thread pool, even if it has been done to death!
The only dependency outside of the standard library is boost::lockfree::queue.
threadpool.hpp
#ifndef THREADPOOL_H
#define THREADPOOL_H
#include <atomic>
#include <functional>
#include <future>
#include <thread>
#include <vector>
#include <boost/lockfree/queue.hpp>
class threadpool
{
public:
// constructors
//
// calls threadpool(size_t concurrency) with:
//
// concurrency - std::thread::hardware_concurrency()
threadpool();
// calls threadpool(size_t concurrency, size_t queue_size) with:
//
// concurrency - concurrency
// queue_size - 128, arbitary value, should be sufficient for most
// use cases.
threadpool(size_t concurrency);
// creates a threadpool with a specific number of threads and
// a maximum number of queued tasks.
//
// Argument
// concurrency - the guaranteed number of threads used in the
// threadpool, ie. maximum number of tasks worked
// on concurrently.
// queue_size - the maximum number of tasks that can be queued
// for completion, currently running tasks do not
// count towards this total.
threadpool(size_t concurrency, size_t queue_size);
// destructor
//
// Will complete any currently running task as normal, then
// signal to any other tasks that they were not able to run
// through a std::runtime_error exception
~threadpool();
threadpool(const threadpool &) = delete;
threadpool(threadpool &&) = delete;
threadpool & operator=(const threadpool &) = delete;
threadpool & operator=(threadpool &&) = delete;
// run
//
// Runs the given function on one of the thread pool
// threads in First In First Out (FIFO) order
//
// Argument
// task - function or functor to be called on the
// thread pool.
//
// Result
// signals when the task has completed with either
// success or an exception. Also results in an
// exception if the thread pool is destroyed before
// execution has begun.
std::future<void> run(std::function<void()> && task);
private:
struct task_package
{
public:
std::promise<void> completion_promise;
std::function<void()> task;
};
// Have to use 'task_package *' since a trivial destructor is
// required, 'task_package' and 'std::unique_ptr<task_package>'
// do not satisfy.
boost::lockfree::queue<task_package *> tasks;
std::vector<std::thread> threads;
std::atomic<bool> shutdown_flag;
inline bool pop_task(std::unique_ptr<task_package> & out);
};
#endif
threadpool.cpp
#include "threadpool.hpp"
#include <algorithm>
#include <exception>
#include <utility>
template<typename T>
constexpr T zero(T)
{
return 0;
}
threadpool::threadpool() :
threadpool(std::thread::hardware_concurrency())
{ };
threadpool::threadpool(size_t concurrency) :
threadpool(concurrency, 128)
{ };
threadpool::threadpool(size_t concurrency, size_t queue_size) :
tasks(queue_size),
shutdown_flag(false),
threads()
{
// This is more efficient than creating the 'threads' vector with
// size constructor and populating with std::generate since
// std::thread objects will be constructed only to be replaced
threads.reserve(concurrency);
for (auto a = zero(concurrency); a < concurrency; ++a)
{
// emplace_back so thread is constructed in place
threads.emplace_back([this]()
{
// checks whether parent threadpool is being destroyed,
// if it is, stop running.
while (!shutdown_flag.load())
{
auto current_task_package = std::unique_ptr<task_package>{nullptr};
// use pop_task so we only ever have one reference to the
// task_package
if (pop_task(current_task_package))
{
try
{
current_task_package->task();
current_task_package->completion_promise.set_value();
}
catch (...)
{
// try and tell the owner that something bad has happened...
try
{
// ...but this can also throw, so stay protected
current_task_package->completion_promise.set_exception(std::current_exception());
}
catch (...) { }
}
}
else
{
// rather than spinning, give up thread time to other things
std::this_thread::yield();
}
}
});
}
};
threadpool::~threadpool()
{
// signal that threads should not perform any new work
shutdown_flag.store(true);
// wait for work to complete then destroy thread
for (auto && thread : threads)
{
thread.join();
}
auto current_task_package = std::unique_ptr<task_package>{nullptr};
// signal to each uncomplete task that it will not complete due to
// threadpool destruction
while (pop_task(current_task_package))
{
try
{
auto except = std::runtime_error("Could not perform task before threadpool destruction");
current_task_package->completion_promise.set_exception(std::make_exception_ptr(except));
}
catch (...) { }
}
};
std::future<void> threadpool::run(std::function<void()> && task)
{
auto promise = std::promise<void>{};
auto future = promise.get_future();
// ensures no memory leak if push throws (it shouldn't but to be safe)
auto package = std::make_unique<task_package>();
package->completion_promise = std::move(promise);
package->task = task;
tasks.push(package.get());
// no longer in danger, can revoke ownership so
// tasks is not left with dangling reference
package.release();
return future;
};
inline bool threadpool::pop_task(std::unique_ptr<task_package> & out)
{
task_package * temp_ptr = nullptr;
if (tasks.pop(temp_ptr))
{
out = std::unique_ptr<task_package>(temp_ptr);
return true;
}
return false;
}
Are there any issues or improvements that you can see?
