6
\$\begingroup\$

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;
}
\$\endgroup\$
  • \$\begingroup\$ Do you normally put all includes up top and only broke the rule for the test code? \$\endgroup\$ – yuri Mar 12 '18 at 9:42
  • \$\begingroup\$ @yuri - Yes.... \$\endgroup\$ – Jive Dadson Mar 12 '18 at 9:45
2
\$\begingroup\$

Observations

Without actually timing things these are just speculations ...

It looks like you're basically using deferred as the strategy when you think that the pool is oversubscribed, which means that the task will be executed when someone calls get() on the future. I am not sure if this is really an improvement over letting the OS handle the oversubscription.

For example if there is one thread setting off a whole bunch of smaller tasks, these will be scheduled async until the limit is hit, then they'll get deferred. As per cppreference (http://en.cppreference.com/w/cpp/thread/async) those will get executed on the first wait() or get() call on the future, so once the main thread wants those results it'll call the futures, which means the deferred ones will be called single threaded from the main thread. Or the futures have to be passed to other threads for execution, which doesn't seem very convenient.

num_threads is a global, if dj:async is only used from a small number of threads that might be ok, if it's used from a larger number of threads the cost of having to share this value between cores might have an impact on performance.

If you are executing a loop around compare and exchange you might as well think about using weak rather than strong.

\$\endgroup\$
  • \$\begingroup\$ Thank you for your suggestions. I changed the compare_exchange from strong to weak. The use-cases I have in mind do not have "one thread setting off a whole bunch of smaller tasks." For example, divide-and-conquer algorithms have new threads spawning threads. No shuffling of futures required. The threads run completely parallel with no concurrence except at the futr.get(). So it's all good. I whupped up a parallel version of quicksort. I haven't as yet bench-marked it. \$\endgroup\$ – Jive Dadson Mar 14 '18 at 3:11
  • \$\begingroup\$ ... Divide-and-conquer does not use a large number of threads. The goal is log(n) futures. The global atomic gets touched very gently. What else? I've done some testing of the VC++ threading system, and I heard Sean Parent say some scary things about it. I am not quite convinced I want to "let the OS handle the oversubscription." \$\endgroup\$ – Jive Dadson Mar 14 '18 at 3:25
  • \$\begingroup\$ BTW, in the original post, I had missed a bug that pretty much made the thing ineffectual. I wiped it off the face of this page. It lives on only in the revision history. Latest and greatest updated in Revision 2 (probably below). \$\endgroup\$ – Jive Dadson Mar 14 '18 at 3:29
  • \$\begingroup\$ @JiveDadson I can’t find any mention of what the bug was ... you should note that somewhere \$\endgroup\$ – Harald Scheirich Mar 14 '18 at 18:15
  • \$\begingroup\$ OK, I confessed. \$\endgroup\$ – Jive Dadson Mar 16 '18 at 7:06
0
\$\begingroup\$

Revision 3

This is probably the Director's Cut. The only thing I might do is to allow the user to choose whether to do the in-thread call (when necessary) either immediately or deferred. That would allow use in some algorithms that would deadlock one way or the other, but not both.

I took Harald's advice and used exchange-weak. I solved the problem of scheduling functions that return void by, uh, avoiding storing a temporary result in the templates.

I tested it using a hand-rolled, quick and dirty quicksort. To my surprise, on vectors large and small, it was only 14% slower than the VC++ parallel version in the experimental STL.

Next I noticed a subtlety that could affect performance. The num_threads count was not being decremented until the user code called get(). I changed things around so that it decrements as soon as the function finishes. The quicksort test improved to only 12% slower than the VC++ std::experimental benchmark. Notice how the new template function par decrements the thread count after the return statement, using an RAII "finally" destructor.

#include <future>
#include <atomic>
#include <thread>

namespace dj {
    namespace global { 
        inline std::atomic<unsigned> num_threads{ 1 };
    };
    namespace {
        template<typename F, typename... Args>
        auto par(F& f, Args&&... args) {
            struct fini { ~fini(){ global::num_threads--; }} finally;
            return f(std::forward<Args>(args)...);
        };
    }
    template<typename F, typename... Args>
    auto async(F& f, Args&&... args)  {
        auto current = global::num_threads.load();
        while(current < std::thread::hardware_concurrency()) {
            if (global::num_threads.compare_exchange_weak(current, current+1)) {
                return std::async(std::launch::async, 
                    [&] { return par(f, std::forward<Args>(args)...); });
            }
        }
        return std::async(std::launch::deferred, f, std::forward<Args>(args)...); 
    }
}

That's all, folks!

\$\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.