4
\$\begingroup\$

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?

\$\endgroup\$
2
  • 1
    \$\begingroup\$ enqueue_task(std::function<T(Args...)> && task, Args... args) -- task is not a universal reference. Universal references are only formed when using template_parameter_name&&. Not even const. This interface is also problematic since it requires either a) not supplying a std::function as the first argument or b) matching exactly the types of the std::function and the arguments and it will fail if passing a not-a-std::function with no arguments. \$\endgroup\$
    – dyp
    Jun 23 '14 at 18:11
  • \$\begingroup\$ Thanks, @Yuushi also provided a link below that has helped clear that up for me, it should be fixed by the next round! \$\endgroup\$
    – Tom Myles
    Jun 24 '14 at 3:10
3
\$\begingroup\$

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;
}
\$\endgroup\$
1
  • \$\begingroup\$ Interesting, I have been working on removing heap allocations by having a set size buffer of task_package objects and two queues to manage which are alive and which are spent as was suggested in the comments of this answer. I can not think of a way to combine these systems without knowing the largest derived task_package_return beforehand, any ideas? \$\endgroup\$
    – Tom Myles
    Jun 24 '14 at 3:05
3
\$\begingroup\$

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.

\$\endgroup\$
5
  • \$\begingroup\$ I've been doing some work on this, and came to the point where std::promise was trying to be copied, I believe this happens because std::function must be CopyConstructible, hence the lambda must be the same, as well as any value not captured by reference, so we cannot capture it directly even if we use std::move. I didn't think to use std::bind, will this allow perfect forwarding of arguments? My solution does not. \$\endgroup\$
    – Tom Myles
    Jun 23 '14 at 6:21
  • \$\begingroup\$ std::bind has the signature template< class F, class... Args > /*unspecified*/ bind( F&& f, Args&&... args ) hence yes, it should support perfect forwarding (as above). Further, the result of std::bind is movable iff all objects are movable - hence the above solution would only work for the case where all args are movable. \$\endgroup\$
    – Yuushi
    Jun 23 '14 at 6:33
  • 1
    \$\begingroup\$ Great, although what you did to get the promise inside the lambda will not work, unless the task is started before enqueue_task is complete as it will go out of scope and be destroyed before it is accessed, i.e. a dangling reference. I ended up using a std::shared_ptr to hold the std::promise and std::move'ing that into the lambda. Also when I use && in the definition of enqueue_task clang says error: no matching member function for call to 'enqueue_task' when I use a std::function<int(int,int)> with arguments. \$\endgroup\$
    – Tom Myles
    Jun 23 '14 at 7:18
  • 1
    \$\begingroup\$ Yes, my apologies, you're quite right that you'll get a dangling reference there, not sure what I was thinking. Using a shared_ptr seems to be the way to go here, I agree. How are you calling enqueue_task? In this instance, it will need an actual rvalue reference (so either an inline lambda, or a call to std::move(f) where f is some type implicitly convertible to std::function). I'd recommend reading isocpp.org/blog/2012/11/… \$\endgroup\$
    – Yuushi
    Jun 23 '14 at 8:04
  • \$\begingroup\$ Thanks, that was really helpful. I'll go with the other answer since they managed to make it such that a single function handles almost everything. \$\endgroup\$
    – Tom Myles
    Jun 24 '14 at 8:52

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.