Waiting for multiple futures?

Question

I'd like to run tasks (worker threads) of the same type, but not more than a certain number of tasks at a time. When a task finishes, its result is an input for a new task which, then, can be started.

Is there any good way to implement this with async/future paradigm in C++11?

At first glance, it looks straight forward, you just spawn multiple tasks with:

std::future<T> result = std::async(...);

and, then, run result.get() to get an async result of a task.

However, the problem here is that the future objects has to be stored in some sort of queue and be waited one by one. It is, though, possible to iterate over the future objects over and over again checking if any of them are ready, but it's not desired due to unnecessary CPU load.

Is it possible somehow to wait for any future from a given set to be ready and get its result?

The only option I can think of so far is an old-school approach without any async/future. Specifically, spawning multiple worker threads and at the end of each thread push its result into a mutex-protected queue notifying the waiting thread via a condition variable that the queue has been updated with more results.

Is there any other better solution with async/future possible?

Strutting

In C++14, this would be when_any(futures...).then(foo), but for now, you're kinda out of luck. – Heavyset

With Boost you get wait_for_any. With Vanilla C++11 you are out of luck for now. – Straightaway

@Xeo, @ComicSansMS, thanks for the hints! More info on C++14 when_any here. Boost's wait_for_any is exactly what I'm looking for. AFAIU, it's not compatible with std::future which does not support registration of external waiters (cvs), right? – Strutting

spawning multiple worker threads and at the end of each thread push its result into a mutex-protected queue sounds fine to me. You don't even need the notification if the worker thread simply checks for a queue size ≥ 1 once in a while. – Lunn

@Heavyset sorry to disappoint, but I don't think when_any/.then(...) made it into C++14. I think they're headed for the Concurrency TS that was kicked off in Chicago. – Magdaleno

@Magdaleno And now not seemingly coming for 2017 either... – Rangefinder

@je4d, do you know why it did not make it for 2017? will it ever be added ? – Sneakers

I'm not even optimistic for C++20, the new concurrency TS was pushed back again – Lunatic

Answer 1

Thread support in C++11 was just a first pass, and while std::future rocks, it does not support multiple waiting as yet.

You can fake it relatively inefficiently, however. You end up creating a helper thread for each std::future (ouch, very expensive), then gathering their "this future is ready" into a synchronized many-producer single-consumer message queue, then setting up a consumer task that dispatches the fact that a given std::future is ready.

The std::future in this system doesn't add much functionality, and having tasks that directly state that they are ready and sticks their result into the above queue would be more efficient. If you go this route, you could write wrapper that match the pattern of std::async or std::thread, and return a std::future like object that represents a queue message. This basically involves reimplementing a chunk of the the concurrency library.

If you want to stay with std::future, you could create shared_futures, and have each dependent task depend on the set of shared_futures: ie, do it without a central scheduler. This doesn't permit things like abort/shutdown messages, which I consider essential for a robust multi threaded task system.

Finally, you can wait for C++2x, or whenever the concurrency TS is folded into the standard, to solve the problem for you.

Saccharose

Is it resolved in C++14, if that is the case can you please update your answer? – Breadthways

@Breadthways Sorry: it wasn't in C++14, I was ridiculously optimistic. Link to paper which I think would solve the problem added. Not only does it have when_any and when_all, you can also attach .thens to a bunch of futures, and in them signal a condition variable, which you wait on: works a lot like the "old school" approach in the OP, but the producer of the content doesn't need to know about the queue mechanism, they just need to supply a future with .then. Emulating .then can only be done with (up to) an extra thread of overhead per future, which is far from ideal. – Saccharose

C++1z won't solve the problem. While it has been agreed upon it is apparently too late to make it in to 2017:-( – Rangefinder

@Rangefinder nod. I mean, it was being worked on/talked about in 2013, you can understand that I figured it might make it in C++17. ;) Which standards-track TS is it in, if you have it on hand? I could improve the answer with that information. – Saccharose

The extensions for futures got moved into the Concurrency TS in early-2015 when they were pretty much complete. – Rangefinder

Seems to be not in C++20 either. – Lunatic

Answer 2

You could create all the futures of "generation 1", and give all those futures to your generation 2 tasks, who will then wait for their input themselves.

Patriarchate

+1 for the nice idea! unfortunately it won't work in my program where the logic behind scheduling the tasks is a bit more complicated than I described in the question. – Strutting

Answer 3

Given that the "Wating for multiple futures" title attracts folks with questions like "is there a wait all for a list of futures?". You can do that adequately by keeping track of the pending threads:

unsigned pending = 0;
for (size_t i = 0; i < N; ++i) {
    ++pending;
    auto callPause =
        [&pending, i, &each, &done]()->unsigned {
            unsigned ret = each();
            results[i] = ret;
            if (!--pending)
                // called in whatever thread happens to finish last
                done(results);
            return ret;
        };
    futures[i] = std::async(std::launch::async, each);
}

full example

It might be possible to use std::experimental::when_all with a spread operator

Meltage

This answer needs a better explanation, as it is very tedious to see what it solves, and how, and how to use it. Your code snippet is also incomplete and only makes sense as part of the full example you linked. – Drabbet

Answer 4

facebook's folly has collectAny/collectN/collectAll on futures, I haven't try it yet, but looks promising.

Ishtar

promising - I see what you did there – Petepetechia