# 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.

#ifndef THREADPOOL_H

#include <atomic>
#include <condition_variable>
#include <functional>
#include <future>
#include <mutex>
#include <vector>

#include <boost/lockfree/queue.hpp>

#define USE_YIELD

{
public:
//  constructors
//
//
//  calls threadpool(size_t concurrency, size_t queue_size) with:
//
//  concurrency - concurrency
//  queue_size  - 128, arbitary value, should be sufficient for most
//                use cases.
//  a maximum number of queued tasks.
//
//  Argument
//    concurrency - the guaranteed number of threads used in the
//                  on concurrently.
//    queue_size  - the maximum number of tasks that can be queued
//                  for completion, currently running tasks do not
//                  count towards this total.

//  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

//  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
//
//  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.

private:
{
public:
std::promise<void> completion_promise;
};

//  Have to use 'task_package *' since a trivial destructor is
//  do not satisfy.
std::atomic<bool> shutdown_flag;

#ifndef USE_YIELD
volatile bool wakeup_flag;
std::condition_variable wakeup_signal;
std::mutex wakeup_mutex;
#endif

};

#endif


#include "threadpool.hpp"

#include <algorithm>
#include <exception>
#include <utility>

template<typename T>
constexpr T zero(T)
{
return 0;
}

{ };

{ };

shutdown_flag(false),
#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

for (auto a = zero(concurrency); a < concurrency; ++a)
{
// emplace_back so thread is constructed in place
{
// checks whether parent threadpool is being destroyed,
// if it is, stop running.
{

// use pop_task so we only ever have one reference to the
{
try
{
}
catch (...)
{
// try and tell the owner that something bad has happened...
try
{
// ...but this can also throw, so stay protected
}
catch (...) { }
}
}
else
{
// rather than spinning, give up thread time to other things
#ifdef USE_YIELD
#else
auto lock = std::unique_lock<std::mutex>(wakeup_mutex);

wakeup_flag = false;

wakeup_signal.wait(lock, [this](){ return wakeup_flag; });
#endif
}
}
});

}
};

{
// 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
{
}

// signal to each uncomplete task that it will not complete due to
{
try
{
}
catch (...) { }
}
};

{
auto promise = std::promise<void>{};
auto future = promise.get_future();

// ensures no memory leak if push throws (it shouldn't but to be safe)

package->completion_promise = std::move(promise);

// 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;
};

{

{
return true;
}
return false;
}


### main.cpp

#include <iostream>
#include <chrono>
#include <queue>
#include <numeric>

//  pool3 is used as a parasite, increasing the number threads it uses
//  pronounces the negative effect of spinning

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)) ;
cout_flag.clear(std::memory_order_release);
};
auto lambda2 = []()
{
while (cout_flag.test_and_set(std::memory_order_acquire)) ;
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;

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();
}

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?

• "an implementation of the threadpool using a std::continuation_variable" You mean condition_variable? – dyp Jun 22 '14 at 14:40
• std::forward<std::function<void()> >(task) A && is only a universal reference if a) it is an auto&& or b) if it's a function parameter of a function template T&&, where T is a template parameter. – dyp Jun 22 '14 at 14:50
• wakeup_flag = false; you should only set this to false if there's only a single element in the queue. – dyp Jun 22 '14 at 14:58
• out = std::unique_ptr<task_package>(temp_ptr); or out.reset(temp_ptr) – dyp Jun 22 '14 at 15:03
• Btw you can hide the definition of class task_package in your cpp. In the header file, only a (forward) declaration is required, since you only use task_package*. – dyp Jun 23 '14 at 6:52

OK some fixes to stop you getting trapped:

You were getting stuck because you could notify_all() and wake the threads in the condition variable. Then after the notification another thread can set wakeup_flag to false thus trapping the threads in the condition variable.

### Example

• Thread 1: is at the line:

auto lock = std::unique_lock(wakeup_mutex);

But is unscheduled before it starts executing this line (for another processes).

• Thread 2: (the main thread) now enters the destructor (set shutdown_flag) and the then locks mutex wakeup_mutex.

• Thread 1: Even if thread 1 wakes up now. It can not proceed because of the lock.

• Thread 2: Proceeds. Sets wakeup_flag to true and notify_all() waiking all threads on the condition variable; thus releasing them. It releases the lock. and continues in the destructor.

• Thread 1: Can now proceed. It set wakeup_flag to false. And then wait() on the condition variable. It will never be woken up as it missed the notify_all().

• Thread 3/4/5/6/7/8: Wake up (as they were waiting on condition variable and did receive the notify_all()). So they start to execute. But before they are released they all check wakeup_flag which is now false so go back to waiting on the condition variable.

The same scenario can apply to run(). In the above scenario replace destructor and run() and you get jobs left on the queue with nobody working on them.

Don't use another variable wakeup_flag to keep the state where you should run in. You already have two of these. tasks.empty() and shutdown_flag. These should be what you test against.

    auto lock = std::unique_lock<std::mutex>(wakeup_mutex);

wakeup_signal.wait(lock, [this]()
{ return !tasks.empty()         // Wake if there are jobs
||  shutdown_flag.load(); // Or we are shutting down
});


Once you do this you don't need to wake all the threads when you add a new task. Just wake one of them.

std::future<void> threadpool::run(std::function<void()> && task) {

// STUFF

// Don't need this anymore (as you are not manipulating state)
// std::lock_guard<std::mutex> lock(wakeup_mutex);
wakeup_signal.notify_one();   // Changed from notify_all


Assume two threads are in the pool (one thread = pathological case).

1. Both threads are running. The queue is empty.
2. Both threads fail to pop a task and enter the else-branch.
3. A new job is added, wakeup_flag is set, the threads (none) waiting for the condition variable are notified. (nothing happens)
4. One thread grabs the mutex, unsets the wakeup flag, and waits for the condition variable. Then, the other thread does the same.
5. The job remains in the job queue. If it is the last job added before waiting for the results, you'll wait forever.

## The interface of threadpool

std::future<void> run(std::function<void()> && task);


I would allow pushing arbitrary function objects with arbitrary result types. Even if your queue has to have a fixed packaged_task-like type, you can internally wrap the passed function object e.g. in a lambda.

Also, I don't quite like the name of this function. It doesn't run a job, it merely enqueues a job.

## volatile

volatile bool wakeup_flag;


I don't quite understand what this volatile is supposed to do. There's no hardware entity changing its value (as far as I can see).

## inline

inline bool threadpool::pop_task(std::unique_ptr<task_package> & out)


I see you're only using this function in a single translation unit. So it should be OK. However, the name of this function is visible in other translation units. I would remove this inline unless you can measure that it makes a difference.

## Task packages on the heap

I think I understand the problem you want to solve with that. But your solution requires a memory allocation for each new job. Consider using a pre-allocated (lock-free) queue as a free list. (measure performance)

## Setting the exception

// try and tell the owner that something bad has happened...
try
{
// ...but this can also throw, so stay protected
}
catch (...) { }


set_exception only throws if [n3797 future.promise]/19:

Throws: future_error if its shared state already has a stored value or exception.

So if it throws, that's a bug in your program. Do not just ignore this case.

while (!shutdown_flag.load())

The load as far as I can see, is superfluous. Since you don't need sequential consistency between all loads/stores, you could use relaxed atomics, as far as I can see.
• @TomMyles "The only reason I am using the heap is that the queue requires a trivially destructible type" I know. What I suggest is an additional lockfree::queue<buffer*> as a free list (where buffer is large enough to hold a task_package). Instead of allocating, you could pop from the free list and construct in-place; later instead of deleting, you could manually call the dtor and push to the free list. – dyp Jun 23 '14 at 6:50