This is a continuation of this question, v3 can be found here
Taking into account the advise given by Loki, an implementation of the threadpool using a std::condition_variable
to control when threads wakeup is presented below. Using this, the test program eventually deadlocks, the time before deadlock occurs is directly related to the number of tasks in pre-allocation. This must mean that there is a problem in waking the threads when work has arrived, but I cannot identify why it is happening.
This behaviour has currently been disabled by the USE_YIELD
pre-compiler define.
threadpool.hpp
#ifndef THREADPOOL_H
#define THREADPOOL_H
#include <atomic>
#include <condition_variable>
#include <functional>
#include <future>
#include <mutex>
#include <thread>
#include <vector>
#include <boost/lockfree/queue.hpp>
#define USE_YIELD
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;
#ifndef USE_YIELD
volatile bool wakeup_flag;
std::condition_variable wakeup_signal;
std::mutex wakeup_mutex;
#endif
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()
#ifndef USE_YIELD
,wakeup_flag(false),
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())
{
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
#ifdef USE_YIELD
std::this_thread::yield();
#else
auto lock = std::unique_lock<std::mutex>(wakeup_mutex);
wakeup_flag = false;
wakeup_signal.wait(lock, [this](){ return wakeup_flag; });
#endif
}
}
});
}
};
threadpool::~threadpool()
{
// signal that threads should not perform any new work
shutdown_flag.store(true);
#ifndef USE_YIELD
{
std::lock_guard<std::mutex> lock(wakeup_mutex);
wakeup_flag = true;
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))
{
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 = std::forward<std::function<void()> >(task);
tasks.push(package.get());
// no longer in danger, can revoke ownership so
// tasks is not left with dangling reference
package.release();
#ifndef USE_YIELD
{
std::lock_guard<std::mutex> lock(wakeup_mutex);
wakeup_flag = true;
wakeup_signal.notify_all();
}
#endif
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;
}
main.cpp
#include <iostream>
#include <chrono>
#include <queue>
#include <numeric>
#include "threadpool.hpp"
threadpool pool1;
threadpool pool2;
// pool3 is used as a parasite, increasing the number threads it uses
// pronounces the negative effect of spinning
threadpool pool3(4 * std::thread::hardware_concurrency());
std::atomic_flag cout_flag = ATOMIC_FLAG_INIT;
int main()
{
auto results1 = std::queue< std::future<void> >();
auto results2 = std::queue< std::future<void> >();
auto lambda1 = []()
{
while (cout_flag.test_and_set(std::memory_order_acquire)) ;
std::cout << "running on pool1 threadid: " << std::this_thread::get_id() << std::endl;
cout_flag.clear(std::memory_order_release);
};
auto lambda2 = []()
{
while (cout_flag.test_and_set(std::memory_order_acquire)) ;
std::cout << "running on pool2 threadid: " << std::this_thread::get_id() << std::endl;
cout_flag.clear(std::memory_order_release);
};
auto times = std::vector<std::chrono::nanoseconds>{};
times.reserve(10000);
for (int j = 0; j < 10000; ++j)
{
cout_flag.test_and_set(std::memory_order_acquire);
std::cout << "round " << j << std::endl;
for (unsigned u = 0; u < 3; ++u)
{
results1.push(pool1.run(lambda1));
results2.push(pool2.run(lambda2));
}
int i = 0;
auto start = std::chrono::steady_clock::now();
cout_flag.clear(std::memory_order_release);
// main loop
while (i++ < 100)
{
auto & future1 = results1.front();
future1.get();
results1.pop();
results1.push(pool1.run(lambda1));
auto & future2 = results2.front();
future2.get();
results2.pop();
results2.push(pool2.run(lambda2));
}
while (!results1.empty())
{
results1.front().get();
results1.pop();
}
while (!results2.empty())
{
results2.front().get();
results2.pop();
}
auto finish = std::chrono::steady_clock::now();
times.push_back(finish - start);
}
auto average = std::accumulate(std::begin(times), std::end(times), std::chrono::nanoseconds{}) / (206 * times.size());
std::cout << "Average time per task: " << static_cast<double>(average.count()) / 1000 << " us" << std::endl;
}
Are there any issues, improvements or fixes to the non-yield code that you can see?
std::continuation_variable
" You meancondition_variable
? \$\endgroup\$std::forward<std::function<void()> >(task)
A&&
is only a universal reference if a) it is anauto&&
or b) if it's a function parameter of a function templateT&&
, whereT
is a template parameter. \$\endgroup\$wakeup_flag = false;
you should only set this tofalse
if there's only a single element in the queue. \$\endgroup\$out = std::unique_ptr<task_package>(temp_ptr);
orout.reset(temp_ptr)
\$\endgroup\$task_package
in your cpp. In the header file, only a (forward) declaration is required, since you only usetask_package*
. \$\endgroup\$