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?

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.

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.

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