9
\$\begingroup\$

This is a continuation of this question.

Following the previous advise, the thread pool now can handle almost all types of input, except for one key form, where the function/functor requires one of its arguments to be std::move'd as shown below.

main.cpp

#include <iostream>

#include "threadpool.hpp"

threadpool pool;

int main()
{
    auto ptr1 = std::make_unique<unsigned>();
    *ptr1 = 10;

    auto lambda1 = [](std::unique_ptr<unsigned> ptr) { return *ptr; };

    auto ptr2 = std::make_unique<unsigned>();
    *ptr2 = 10;

    auto lambda2 = [ptr = std::move(ptr2)]() { return *ptr; };

    //  fails with below compile error
    auto future1 = pool.enqueue_task(lambda1, std::move(ptr1));
    std::cout << future1.get() << std::endl;

    //  compiles fine
    auto future2 = pool.enqueue_task(std::move(lambda2));
    std::cout << future2.get() << std::endl;
}

You get this error message:

./task_package.hpp:36:27: error: no matching function for call to object of type
      'std::__1::__bind<<lambda at main.cpp:12:20> &,
      std::__1::unique_ptr<unsigned int, std::__1::default_delete<unsigned int>
      > >'
        promise.set_value(func());
                          ^~~~

As usual, here is the bulk of the code, any more improvements/comments, especially with respect to optimisations, are more than welcome!

threadpool.hpp

#ifndef THREADPOOL_HPP
#define THREADPOOL_HPP

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

#include <boost/lockfree/queue.hpp>

#include "task_package.hpp"

//#define USE_YIELD

class threadpool
{
public:
    //  constructor
    //
    //  calls threadpool(size_t concurrency) with:
    //
    //  concurrency - std::thread::hardware_concurrency()
    threadpool();
    //  constructor
    //
    //  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);
    //  constructor
    //
    //  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
    //
    //  Arguments
    //    task - Function or functor to be called on the
    //           thread pool, takes an arbitary number of
    //           arguments and arbitary return type.
    //    args - Arguments for task, cannot be std::move'ed
    //           if such parameters must be used, use a
    //           lambda and capture via move then move
    //           the lambda.
    //
    //  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.
    template<typename Func, typename ... Args>
    auto enqueue_task(Func&& task, Args&&... args) -> std::future<decltype(task(std::forward<Args>(args)...))>
    {
        //  Return type of the functor, can be void via
        //  specilisation of task_package_impl
        using R = decltype(task(std::forward<Args>(args)...));

        auto promise = std::promise<R>{ };
        auto future = promise.get_future();
        auto bound_task = std::bind(std::forward<Func>(task), std::forward<Args>(args)...);

        // ensures no memory leak if push throws (it shouldn't but to be safe)
        auto package_ptr = std::make_unique<task_package_impl<R, decltype(bound_task)>>(std::move(bound_task), std::move(promise));

        tasks.push(static_cast<task_package *>(package_ptr.get()));

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

#ifndef USE_YIELD
        wakeup_signal.notify_one();
#endif

        return future;
    };

private:
    std::vector<std::thread> threads;

    std::atomic<bool> shutdown_flag;

    boost::lockfree::queue<task_package *> tasks;

#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 <exception>
#include <utility>
#include <iostream>

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))
                    {
                        current_task_package->run_task();
                    }
                    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))
    {
        auto except = std::runtime_error("Could not perform task before threadpool destruction");
        current_task_package->set_exception(std::make_exception_ptr(except));
    }
};

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

task_package.hpp

#ifndef TASK_PACKAGE_HPP
#define TASK_PACKAGE_HPP

#include <future>

struct task_package
{
    virtual ~task_package() { };

    void run_task() noexcept
    {
        try
        {
            run();
        }
        catch (...)
        {
            set_exception(std::current_exception());
        }
    }

    virtual void run() = 0;
    virtual void set_exception(std::exception_ptr except_ptr) = 0;
};

template<typename R, typename Func>
struct task_package_impl : public task_package
{
    task_package_impl(Func&& func, std::promise<R>&& promise)
        : promise(std::forward<std::promise<R>>(promise))
        , func(std::forward<Func>(func))
    { };

    virtual void run()
    {
        promise.set_value(func());
    }

    virtual void set_exception(std::exception_ptr except_ptr)
    {
        promise.set_exception(except_ptr);
    }

    std::promise<R> promise;
    Func func;
};

template<typename Func>
struct task_package_impl<void, Func> : public task_package
{
    task_package_impl(Func&& func, std::promise<void>&& promise)
        : promise(std::forward<std::promise<void>>(promise))
        , func(std::forward<Func>(func))
    { };

    virtual void run()
    {
        func();
        promise.set_value();
    }

    virtual void set_exception(std::exception_ptr except_ptr)
    {
        promise.set_exception(except_ptr);
    }

    std::promise<void> promise;
    Func func;
};

#endif
\$\endgroup\$
  • \$\begingroup\$ Why do you use a static_cast here? static_cast<task_package *>(package_ptr.get()) \$\endgroup\$ – dyp Jun 27 '14 at 9:14
  • \$\begingroup\$ If performance is important, you might be able to move arguments from the task by using a tuple and tuple expansion via the indices trick (store them alongside the original Func inside task_package_impl). \$\endgroup\$ – dyp Jun 27 '14 at 9:25
6
\$\begingroup\$

You give 3 threadpool constructors:

threadpool::threadpool()
    : threadpool(std::thread::hardware_concurrency())
{ };

threadpool::threadpool(size_t concurrency)
    : threadpool(concurrency, 128)
{ };

threadpool::threadpool(size_t concurrency, size_t queue_size)

You could use one threadpool constructor with default arguments:

threadpool::threadpool(
        size_t concurrency=std::thread::hardware_concurrency()
      , size_t queue_size=128
)
\$\endgroup\$
3
\$\begingroup\$

I would make the overriding behaviour explicit in task_package_impl by adding the override specifier to run and set_exception:

virtual void run() override
{
    promise.set_value(func());
}

virtual void set_exception(std::exception_ptr except_ptr) override
{
    promise.set_exception(except_ptr);
}

That way, if you ever intend to change the signature of run or set_exception in the base class (breaking changes in an interface are hardly a good idea, but they sometimes happen) and forget to change one of the derived classes, you will get a compilation error stating that run and/or set_exception are not overriding anything but merely hiding the base class function name.

\$\endgroup\$
  • \$\begingroup\$ Thanks, I completely forgot about the override and final keywords from C++11. I think using final makes more sense as deriving from task_package_impl wouldn't have much use and still has the benefits of using override. \$\endgroup\$ – Tom Myles Jun 24 '14 at 9:57
  • \$\begingroup\$ @TomMyles You can use final and it could actually help the compiler to devirtualize the calls. However, marking a class final is not really a good idea since many libraries derive from user-defined classes to use the empty base class optimization :) \$\endgroup\$ – Morwenn Jun 24 '14 at 11:45
3
\$\begingroup\$

I think the reason for your compilation error is neither a bug in your code nor a compiler error - it's just a consequence of the current standard. It occurs, because the compiler would like to pass a copy the result of std::move as an input parameter to the given func, instead of taking it as a reference. As far as I remember this is the exact reason why the 'Lambda capture expressions' have been improved with the C++14 standard - so you could pass a unique_ptr, which couldn't be passed by value otherwise.

\$\endgroup\$

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.