# 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>
{
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'
{
try
{
promise.set_value(result);
}
catch (...)
{
promise.set_exception(std::current_exception());
}
});

return future;
}


The remaining code.

#ifndef THREADPOOL_H

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

#include <boost/lockfree/queue.hpp>

//#define USE_YIELD

{
public:
//  constructors
//
//
//  calls threadpool(size_t concurrency, size_t queue_size) with:
//
//  concurrency - concurrency
//  queue_size  - 128, arbitary value, should be sufficient for most
//                use cases.
//  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
//
//  Argument
//    task - function or functor to be called on the
//
//  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 T, typename... 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'
{
try
{
promise.set_value(result);
}
catch (...)
{
promise.set_exception(std::current_exception());
}
});

return future;
}
#endif

private:
{
public:
std::promise<void> completion_promise;
};

//  Have to use 'task_package *' since a trivial destructor is
//  do not satisfy.
std::atomic<bool> shutdown_flag;

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

};

#endif


#include "threadpool.hpp"

#include <algorithm>
#include <exception>
#include <utility>

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
{
try
{
}
catch (...)
{
// try and tell the owner that something bad has happened
}
}
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
{
try
{
}
catch (...) { }
}
};

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

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

package->completion_promise = std::move(promise);

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

{

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


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

• 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.
– dyp
Jun 23 '14 at 18:11
• Thanks, @Yuushi also provided a link below that has helped clear that up for me, it should be fixed by the next round! Jun 24 '14 at 3:10

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) :
shutdown_flag(false),
#ifndef USE_YIELD
,
wakeup_signal(),
wakeup_mutex()
#endif


I would do it this way:

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


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

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


I would change this as follows.

struct task_package
{
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>
{
: promise(std::forward<std::promise<R>>(promise))
{}
virtual void run() {
}
virtual void exception(std::exception_ptr eptr) {
promise.set_exception(eptr);
}
std::promise<R>     promise;
};


This simplifies the work done by the thread too:

    threads.emplace_back([this]()  {
// STUFF
}
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>
{
std::promise<R>     promise;
std::future<R>      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.release();

#ifndef USE_YIELD
wakeup_signal.notify_one();
#endif

return future;
}

• 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? Jun 24 '14 at 3:05

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>


(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...)>;


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

auto promise = std::promise<T>{};
auto future = promise.get_future();
// 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.

• 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. Jun 23 '14 at 6:21
• 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. Jun 23 '14 at 6:33
• 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. Jun 23 '14 at 7:18
• 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/… Jun 23 '14 at 8:04
• 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. Jun 24 '14 at 8:52