This is a continuation of this question.
Following the previous advise, the thread pool now can handle almost all types of input, except for one key form, where the function/functor requires one of its arguments to be std::move'd as shown below.
main.cpp
#include <iostream>
#include "threadpool.hpp"
threadpool pool;
int main()
{
auto ptr1 = std::make_unique<unsigned>();
*ptr1 = 10;
auto lambda1 = [](std::unique_ptr<unsigned> ptr) { return *ptr; };
auto ptr2 = std::make_unique<unsigned>();
*ptr2 = 10;
auto lambda2 = [ptr = std::move(ptr2)]() { return *ptr; };
// fails with below compile error
auto future1 = pool.enqueue_task(lambda1, std::move(ptr1));
std::cout << future1.get() << std::endl;
// compiles fine
auto future2 = pool.enqueue_task(std::move(lambda2));
std::cout << future2.get() << std::endl;
}
You get this error message:
./task_package.hpp:36:27: error: no matching function for call to object of type
'std::__1::__bind<<lambda at main.cpp:12:20> &,
std::__1::unique_ptr<unsigned int, std::__1::default_delete<unsigned int>
> >'
promise.set_value(func());
^~~~
As usual, here is the bulk of the code, any more improvements/comments, especially with respect to optimisations, are more than welcome!
threadpool.hpp
#ifndef THREADPOOL_HPP
#define THREADPOOL_HPP
#include <atomic>
#include <condition_variable>
#include <functional>
#include <future>
#include <mutex>
#include <thread>
#include <vector>
#include <boost/lockfree/queue.hpp>
#include "task_package.hpp"
//#define USE_YIELD
class threadpool
{
public:
// constructor
//
// calls threadpool(size_t concurrency) with:
//
// concurrency - std::thread::hardware_concurrency()
threadpool();
// constructor
//
// 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);
// constructor
//
// 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;
// enqueue_task
//
// Runs the given function on one of the thread pool
// threads in First In First Out (FIFO) order
//
// Arguments
// task - Function or functor to be called on the
// thread pool, takes an arbitary number of
// arguments and arbitary return type.
// args - Arguments for task, cannot be std::move'ed
// if such parameters must be used, use a
// lambda and capture via move then move
// the lambda.
//
// 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.
template<typename Func, typename ... Args>
auto enqueue_task(Func&& task, Args&&... args) -> std::future<decltype(task(std::forward<Args>(args)...))>
{
// Return type of the functor, can be void via
// specilisation of task_package_impl
using R = decltype(task(std::forward<Args>(args)...));
auto promise = std::promise<R>{ };
auto future = promise.get_future();
auto bound_task = std::bind(std::forward<Func>(task), std::forward<Args>(args)...);
// ensures no memory leak if push throws (it shouldn't but to be safe)
auto package_ptr = std::make_unique<task_package_impl<R, decltype(bound_task)>>(std::move(bound_task), std::move(promise));
tasks.push(static_cast<task_package *>(package_ptr.get()));
// no longer in danger, can revoke ownership so
// tasks is not left with dangling reference
package_ptr.release();
#ifndef USE_YIELD
wakeup_signal.notify_one();
#endif
return future;
};
private:
std::vector<std::thread> threads;
std::atomic<bool> shutdown_flag;
boost::lockfree::queue<task_package *> tasks;
#ifndef USE_YIELD
std::condition_variable wakeup_signal;
std::mutex wakeup_mutex;
#endif
bool pop_task(std::unique_ptr<task_package> & out);
};
#endif
threadpool.cpp
#include "threadpool.hpp"
#include <exception>
#include <utility>
#include <iostream>
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()
#ifndef USE_YIELD
, wakeup_signal()
, wakeup_mutex()
#endif
{
// 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(std::memory_order_relaxed))
{
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))
{
current_task_package->run_task();
}
else
{
// rather than spinning, give up thread time to other things
#ifdef USE_YIELD
std::this_thread::yield();
#else
auto lock = std::unique_lock<std::mutex>(wakeup_mutex);
wakeup_signal.wait(lock, [this](){ return !tasks.empty() || shutdown_flag; });
#endif
}
}
});
}
};
threadpool::~threadpool()
{
// signal that threads should not perform any new work
shutdown_flag.store(true);
#ifndef USE_YIELD
wakeup_signal.notify_all();
#endif
// 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))
{
auto except = std::runtime_error("Could not perform task before threadpool destruction");
current_task_package->set_exception(std::make_exception_ptr(except));
}
};
bool threadpool::pop_task(std::unique_ptr<task_package> & out)
{
task_package * temp_ptr = nullptr;
if (tasks.pop(temp_ptr))
{
out.reset(temp_ptr);
return true;
}
return false;
}
task_package.hpp
#ifndef TASK_PACKAGE_HPP
#define TASK_PACKAGE_HPP
#include <future>
struct task_package
{
virtual ~task_package() { };
void run_task() noexcept
{
try
{
run();
}
catch (...)
{
set_exception(std::current_exception());
}
}
virtual void run() = 0;
virtual void set_exception(std::exception_ptr except_ptr) = 0;
};
template<typename R, typename Func>
struct task_package_impl : public task_package
{
task_package_impl(Func&& func, std::promise<R>&& promise)
: promise(std::forward<std::promise<R>>(promise))
, func(std::forward<Func>(func))
{ };
virtual void run()
{
promise.set_value(func());
}
virtual void set_exception(std::exception_ptr except_ptr)
{
promise.set_exception(except_ptr);
}
std::promise<R> promise;
Func func;
};
template<typename Func>
struct task_package_impl<void, Func> : public task_package
{
task_package_impl(Func&& func, std::promise<void>&& promise)
: promise(std::forward<std::promise<void>>(promise))
, func(std::forward<Func>(func))
{ };
virtual void run()
{
func();
promise.set_value();
}
virtual void set_exception(std::exception_ptr except_ptr)
{
promise.set_exception(except_ptr);
}
std::promise<void> promise;
Func func;
};
#endif
static_casthere?static_cast<task_package *>(package_ptr.get())\$\endgroup\$Funcinsidetask_package_impl). \$\endgroup\$