10
\$\begingroup\$

v2 of this question is here and v3 is here

To get a better understanding of C++11/C++14, I thought I would develop a thread pool, even if it has been done to death!

The only dependency outside of the standard library is boost::lockfree::queue.

threadpool.hpp

#ifndef THREADPOOL_H
#define THREADPOOL_H

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

#include <boost/lockfree/queue.hpp>

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;

    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()
{
    // 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
                        std::this_thread::yield();
                    }
                }
            });

    }
};

threadpool::~threadpool()
{
    // signal that threads should not perform any new work
    shutdown_flag.store(true);

    // 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 = task;

    tasks.push(package.get());

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

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

Are there any issues or improvements that you can see?

\$\endgroup\$
4
\$\begingroup\$

This does not really stop them spinning:

// rather than spinning, give up thread time to other things
std::this_thread::yield();

It may give up the core for another thread (temporarily). But it will come back just as quickly to check for more work. So I am not convinced this actually buys you anything.

I would use a condition variable and block threads on it while there is no work. Then you know they are not taking up any cycles when there is no work.

When creating the work package.

package->task = task;

I would forward the task (remember that named variables can not activate move semantics).

package->task = std::forward<std::function<void()>>(task);
\$\endgroup\$
4
  • \$\begingroup\$ Interesting, I will have to do a couple tests on using and not using yield. With respect to the std::forward, I get a compile error when using clang-503.0.40, threadpool.cpp:108:21: error: no matching function for call to 'forward' considering this would it be better to have run task task by value then std::move it into the package? \$\endgroup\$ – Tom Myles Jun 21 '14 at 23:19
  • \$\begingroup\$ Update the std::forward(). Should have been: std::forward<std::function<void()>>() \$\endgroup\$ – Martin York Jun 22 '14 at 1:14
  • \$\begingroup\$ Ok, I've had a crack at implementing your suggestions, you can see the results here (hint: it is broken!) \$\endgroup\$ – Tom Myles Jun 22 '14 at 11:33
  • \$\begingroup\$ @TomMyles: Added answer that should fix your problem. \$\endgroup\$ – Martin York Jun 22 '14 at 17:09

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.