Platform independant thread pool v2

Question

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?

Answer 1

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

Answer 2

A deadlock scenario

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.

Consider making this a free function with internal linkage instead.

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
current_task_package->completion_promise.set_exception(std::current_exception());
}
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.

Thread-safe shutdown flag

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.