# 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>

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
std::cout << future1.get() << std::endl;

//  compiles fine
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!

#ifndef THREADPOOL_HPP

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

#include <boost/lockfree/queue.hpp>

//#define USE_YIELD

{
public:
//  constructor
//
//
//  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.
//  constructor
//
//  a maximum number of queued tasks.
//
//  Argument
//    concurrency - the guaranteed number of threads used in the
//                  on concurrently.
//    queue_size  - the maximum number of tasks that can be queued
//                  for completion, currently running tasks do not
//                  count towards this total.

//  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

//
//  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>
{
//  Return type of the functor, can be void via

auto promise = std::promise<R>{ };
auto future = promise.get_future();

// ensures no memory leak if push throws (it shouldn't but to be safe)

// 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::atomic<bool> shutdown_flag;

#ifndef USE_YIELD
std::condition_variable wakeup_signal;
std::mutex wakeup_mutex;
#endif

};

#endif


#include "threadpool.hpp"

#include <exception>
#include <utility>
#include <iostream>

template<typename T>
constexpr T zero(T)
{
return 0;
}

{ };

{ };

, shutdown_flag(false)
#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

for (auto a = zero(concurrency); a < concurrency; ++a)
{
// emplace_back so thread is constructed in place
{
// checks whether parent threadpool is being destroyed,
// if it is, stop running.
{

// use pop_task so we only ever have one reference to the
{
}
else
{
// rather than spinning, give up thread time to other things
#ifdef USE_YIELD
#else
auto lock = std::unique_lock<std::mutex>(wakeup_mutex);

wakeup_signal.wait(lock, [this](){ return !tasks.empty() || shutdown_flag; });
#endif
}
}
});
}
};

{
// 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
{
}

// signal to each uncomplete task that it will not complete due to
{
}
};

{

{
out.reset(temp_ptr);
return true;
}
return false;
}


#ifndef TASK_PACKAGE_HPP

#include <future>

{

{
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>
{
: 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>
{
: 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

• Why do you use a static_cast here? static_cast<task_package *>(package_ptr.get()) – dyp Jun 27 '14 at 9:14
• 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). – dyp Jun 27 '14 at 9:25

You give 3 threadpool constructors:

threadpool::threadpool()
{ };

{ };



You could use one threadpool constructor with default arguments:

threadpool::threadpool(
, size_t queue_size=128
)


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.

• 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. – Tom Myles Jun 24 '14 at 9:57
• @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 :) – Morwenn Jun 24 '14 at 11:45

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.