Dynamic execution policy did not make the cut for C++17. Submitted for your criticism, a minimalist dynamic task scheduler to use until the std::async sorts it out...
I am interested not only in code quality and correctness, but also whether the strategy itself can be improved while maintaining simplicity.
Tasks will be done via std::async. The strategy is to keep up with how many new threads have been launched using the scheduler, but not yet completed. If that number+1 equals or exceeds the number of cores, a new task will be scheduled as launch::deferred, otherwise launch::async. Obviously, that cannot be done perfectly, but if a task infrequently gets scheduled "the wrong way", it's not a disaster, so long as things keep chugging along.
I defined a wrapper for std::future called dj::t_future. It keeps a memo of which policy was used to start the thread. (Is that necessary? Is some better manner of RAII possible?) There are two constructors for t_future, all cut-and-pasted, because I didn't figure out how to use only one that covered both functions that return values and void functions. The battle field in dj::t_future<T>::get() shows evidence of a skirmish involving propagation of exceptions. (Did the good guys win?)
If you study the code, please tell me what you think, even in the unlikely event that you see nothing to criticize.
SEE UPDATED CODE in an Answer I posted.
Here is original, except I fixed one stupid bug that could rear its ugly head in "header only" mode. I had decorated the num_threads with static inline, which makes no sense. If static, inline is redundant.
#include <future>
#include <atomic>
#include <thread>
namespace dj {
namespace global { // Wasa bug. This was "static". I blame my fingers.
inline std::atomic<unsigned> num_threads{ 1 };
};
namespace {
struct finish {
std::launch policy;
finish(std::launch p) : policy(p) {}
~finish() {
if (policy != std::launch::deferred) {
global::num_threads -= 1;
}
}
};
}
// Wrapper for std::future that remembers whether thread was spawned
// and decrements thread count on "get" completion when appropriate.
template<class T>
class t_future {
std::future<T> fut;
std::launch policy;
public:
t_future(std::future<T> &&fut, std::launch policy)
: policy(policy)
, fut(std::move(fut))
{}
T get() {
struct finish finally(policy); // In case get() throws exception
T ret;
try { ret = fut.get(); } catch (...) { throw; } // See notes below
return ret;
}
};
// Copy-paste-edit of the above for functions returning void.
// Is there a way to avoid this?
template<>
class t_future<void> {
std::future<void> fut;
std::launch policy;
public:
t_future(std::future<void> &&fut, std::launch policy)
: policy(policy)
, fut(std::move(fut))
{}
void get() {
struct finish finally(policy); // In case get() throws exception
// Why do I need this try/catch/throw?
try { fut.get(); } catch (...) { throw; }
// without it, the program does not abort when
// exception is not caught in user code.
}
};
// dj::async sets std::launch policy automatically
// based on how many threads have
// already been launched.
template<typename F, typename... Args>
auto async(F& f, Args&&... args) {
using ret_t = decltype(f(std::forward<Args>(args)...));
auto policy{ std::launch::deferred };
unsigned count{ 0 };
unsigned current { global::num_threads };
while(current < std::thread::hardware_concurrency()
&& ++count < 1'000) {
if (global::num_threads.compare_exchange_strong(current, current + 1)) {
policy = std::launch::async;
break;
}
}
auto fut = std::async(policy, f, std::forward<Args>(args)...);
return t_future<ret_t> {std::move(fut), policy};
}
}
// Code to review ends here
// Minimal test code begins here...
#include <chrono>
static void busy_sleep(long double time) noexcept
{
using duration_t = std::chrono::duration<long long, std::nano>;
const auto end = std::chrono::high_resolution_clock::now()
+ duration_t(static_cast<long long> (time * 1e9));
do {
;
} while (std::chrono::high_resolution_clock::now() < end);
}
#include <random>
std::default_random_engine re;
std::uniform_real_distribution<double> ud(.02, .5);
#include <iostream>
int main() {
auto f = []() {
std::cout << "(";
auto spin = ud(re);
busy_sleep(spin);
std::cout << ")";
};
std::vector<dj::t_future<void>> threads;
for (unsigned i = 0; i < 2*std::thread::hardware_concurrency() ; ++i) {
threads.emplace_back(dj::async(f));
}
for (auto& t: threads) {
t.get();
}
return 0;
}
