Platform independant thread pool v3

Question

This is a continuation of this question, v4 can be found here

Taking the advise given previously, the deadlock was fixed in the non-yielding code. The next problem comes from the suggestion to make the interface, now renamed enqueue_task able to handle more generic tasks. The goal was to have the ability to return a result as well as take arguments via variadic tempting, but clang will not compile it.

I have tried multiple combinations of the below, but none compile if the function is called.

template<typename T, typename... Args>
std::future<T> enqueue_task(std::function<T(Args...)> && task, Args... args)
{
    auto promise = std::promise<T>{};
    auto future = promise.get_future();

    //  Cannot get this to work, clang gives:
    //  error: expression contains unexpanded parameter packs 'args' and 'Args'
    enqueue_task([promise = std::move(promise), task = std::forward< std::function<T(Args...)> >(task), args...]()
        {
            try
            {
                auto result = task(std::forward<Args>(args...));
                promise.set_value(result);
            }
            catch (...)
            {
                promise.set_exception(std::current_exception());
            }
        });

    return future;
}

The remaining code.

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
//#define ALLOW_GENERIC_TASKS

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;

    //  enqueue_task
    //
    //  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> enqueue_task(std::function<void()> && task);
#ifdef ALLOW_GENERIC_TASKS
    template<typename T, typename... Args>
    std::future<T> enqueue_task(std::function<T(Args...)> && task, Args... args)
    {
        auto promise = std::promise<T>{};
        auto future = promise.get_future();

        //  Cannot get this to work, clang gives:
        //  error: expression contains unexpanded parameter packs 'args' and 'Args'
        enqueue_task([promise = std::move(promise), task = std::forward< std::function<T(Args...)> >(task), args...]()
            {
                try
                {
                    auto result = task(std::forward<Args>(args...));
                    promise.set_value(result);
                }
                catch (...)
                {
                    promise.set_exception(std::current_exception());
                }
            });

        return future;
    }
#endif

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
    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 <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_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))
                    {
                        try
                        {
                            current_task_package->task();
                            current_task_package->completion_promise.set_value();
                        }
                        catch (...)
                        {
                            // try and tell the owner that something bad has happened
                            current_task_package->completion_promise.set_exception(std::current_exception());
                        }
                    }
                    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))
    {
        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::enqueue_task(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
    wakeup_signal.notify_one();
#endif

    return future;
};

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

Any more issues, or a solution to the generic task problem you can see?

Answer 1

When doing the initializer list. I always put the punctuation at the beginning of the line. This avoids this situation.

threadpool::threadpool(size_t concurrency, size_t queue_size) :
    tasks(queue_size),
    shutdown_flag(false),
    threads()
#ifndef USE_YIELD
    ,
    wakeup_signal(),
    wakeup_mutex()
#endif

I would do it this way:

threadpool::threadpool(size_t concurrency, size_t queue_size)
    : tasks(queue_size)
    , shutdown_flag(false)
    , threads()
#ifndef USE_YIELD
    , wakeup_signal()
    , wakeup_mutex()
#endif

When you comment a member out you take all the stuff you need with it.

I would also move the work for your jobs into your task object.

struct task_package
{
public:     // Extra note. public not needed here.
            // default permission is public.
    std::promise<void> completion_promise;
    std::function<void()> task;
};

I would change this as follows.

struct task_package
{
    virtual ~task_package(){}
    void runTask() noexcept {
        try         {  this->run();}
        catch (...) {  this->exception(std::current_exception());}
    }
    virtual void run() = 0;                              // Run and assign value to promise.
    virtual void exception(std::exception_ptr eptr) = 0; // Assign exception to promise.
};

// We defer the work to a templated class.
// As the promise is dependent on the return type.
//
template<typename R, typename F>
struct task_package_return: public task_package
{
    task_package_return(F&& task, std::promise<R>&& promise)
        : promise(std::forward<std::promise<R>>(promise))
        , task(std::forward<F>(task))
    {}
    virtual void run() {
        promise.set_value(task());
    }
    virtual void exception(std::exception_ptr eptr) {
        promise.set_exception(eptr);
    }
    std::promise<R>     promise;
    F                   task;
};

This simplifies the work done by the thread too:

    threads.emplace_back([this]()  {
          // STUFF
                if (pop_task(current_task_package)) {
                    current_task_package->runTask();
                }
                else {
                    // STUFF
                }
           // STUFF
        });

Now when you create a task object you can create an object that takes any parameters and returns any types.

template<typename Func, typename ...Args>
auto enqueue_task(Func&& task, Args&&... args) -> std::future<decltype(task(std::forward<Args>(args)...))> 
{
    using R = decltype(task(std::forward<Args>(args)...));
    std::promise<R>     promise;
    std::future<R>      future = promise.get_future();
    auto boundTask = std::bind(std::forward<std::function<R(Args...)>>(task), std::forward<Args>(args)...);

    // ensures no memory leak if push throws (it shouldn't but to be safe)
    std::unique_ptr<task_package> package(new task_package_return<R, decltype(boundTask)>(std::move(boundTask), std::move(promise)));

    tasks.push(package.get());

    // no longer in danger, can revoke ownership so
    // tasks is not left with dangling reference
    package.release();

#ifndef USE_YIELD
    wakeup_signal.notify_one();
#endif

    return future;
}

Answer 2

Technically, this is off-topic as this isn't working code.

That being said, it's quite a tricky problem, so I want to answer it anyway.

First off, your args should be passed by Args&&:

template<typename T, typename... Args>
std::future<T> enqueue_task(std::function<T(Args...)> && task, Args&&... args) 

(I suspect this is a minor oversight, however).

Parameter pack expansions in lambdas seem to be difficult to get right, so I'm going to take a slightly different route. Instead of writing this:

enqueue_task([promise = std::move(promise), task = std::forward< std::function<T(Args...)> >(task), args...]()

We'll instead split it into this:

using task_t = std::function<T(Args...)>;
auto bound = std::bind(std::forward<task_t>(task), std::forward<Args>(args)...);

We should then be able to write:

enqueue_task([promise = std::move(promise), bound = std::move(bound)]() { ... }

However, this still doesn't quite work, as promise is treated as const:

'this' argument has type 'const std::__1::promise', but method is not marked const

void set_value(const _Rp& __r);

Unfortunately, marking the lambda as mutable also doesn't work, as this (for some reason) triggers an attempted call to the copy constructor of promise. As a final workaround, we can instead use:

enqueue_task([&, bound = std::move(bound)]() {
        auto prom(std::move(promise));
        ...
};

The whole method ends up being:

template<typename T, typename... Args>
std::future<T> enqueue_task(std::function<T(Args...)>&& task, Args&&... args)
{
    using task_t = std::function<T(Args...)>;

    auto promise = std::promise<T>{};
    auto future = promise.get_future();
    auto bound = std::bind(std::forward<task_t>(task), std::forward<Args>(args)...);
    enqueue_task([&, bound = std::move(bound)]() {
        // Note: this leaves a dangling reference, see below!
        auto prom(std::move(promise));
        try {
            auto result = bound();
            prom.set_value(result);
        } catch (...)
        {
            prom.set_exception(std::current_exception());
        }
    });

    return future;
}

Edit: As per comments below, using auto prom(std::move(promise)) above will leave a dangling reference. A solution for this is to wrap the promise in a std::shared_ptr.