Continuable Futures
Continuations are the most common extension to future objects. A continuable shared state provides the future instance with write access to attach continuations to a task. When the task sets the value of the shared state, all continuations are run.
+
Continuations)] F -.-> |attach continuation|S T[Task] --> |set and continue|S end
Motivation
Non-continuable tasks
Consider what happens when we launch a task with C++11 std::async:
std::future<int> f = std::async([]() {
// Parallel work
return 65;
});
// Main work
assert(f.get() == 65);
When we only need a single parallel task with future types, the main thread is allowed to wait or do some other work while the task is running.
In this example, the main thread spent some time waiting but this is often OK, as long as it had nothing better to do but to wait for the asynchronous task. This is common in user interfaces that need to be refreshed while a longer background task is running.
Now let's say we want to execute a sequence of asynchronous tasks as simple as:
As we shall see, std::async does not provide the mechanisms to make this happen properly. The first alternative that usually comes to mind is waiting for one task after launching the next.
std::future<int> A = std::async([]() { return 65; });
std::future<char> B = std::
async([](int v) { return static_cast<char>(v); }, A.get());
std::future<void>
C = std::async([](char c) { assert(c == 'A'); }, B.get());
C.wait();
The code might look reasonable but, in that case, we would have:
We have a number of problems here. The more tasks we have, and the shorter the tasks, the less time the main thread has to do any useful work before waiting and the more time it spends waiting for tasks. Even worse, we already know how these tasks should continue: we are just waiting to attach this continuation. At a certain point, it might not even be worth using asynchronous code at all.
Polling
The second alternative to solve this problem is polling. In this case, we would make task B wait for A before doing its work. The same for task B and C.
std::future<int> A = std::async([]() { return 65; });
std::future<char> B = std::async([&A]() {
// B waits for A in its turn
int v = A.get();
// Use the value
return static_cast<char>(v);
});
std::future<void> C = std::async([&B]() {
assert(B.get() == 'A');
});
C.wait();
And now we have:
This might look more reasonable from the perspective of the main thread. We are not waiting for so long inline, and we have more time to do work in parallel. However, this outsources the cost of waiting to other threads even though we know the initial task is not ready. Note for how long the tasks A, B, and C are active in this example.
Thus, the biggest problem with this strategy is it cannot scale properly. For every task in our application, we would need one idle thread waiting for the previous task. In an application with 2000 tasks, we would need 1999 threads for polling antecedent tasks and only one thread would to execute real work.
Continuations in C++ libraries
The act of waiting for a std::future result is synchronous, which is not appropriate in communication-intensive code.
In the original std::future model, if a continuation task B depends on the result of the first task A, we only
have two options:
- waiting for the first task synchronously
- polling the first task asynchronously.
If we always wait for the first task to start its continuation, the asynchronicity has no purpose. If we always poll for the first task, we waste resources and an extra thread to repeatedly check the status of the first task.
For this reason, the most common extension proposed for std::future is continuations, such as implemented in Microsoft's PPL, async++, continuable.
std::future A = std::async([]() { return 2; });
int A_result = A.get();
std::cout << A_result << std::endl;
std::future A = std::async([]() { return 2; });
std::future B = std::async([&]() {
int A_result = A.get();
std::cout << A_result << std::endl;
});
B.wait();
auto A = std::experimental::async([]() { return 2; });
auto B = A.then([](int A_result) {
std::cout << A_result << std::endl;
});
B.wait();
Continuations are the foundation for composing task graphs, with operations such as std::experimental::when_all and std::experimental::when_any. These conjunction and disjunction operations depend on continuations so that previous tasks can inform the operation result when they are ready without polling.
Continuable futures
Continuable futures allow us to launch a second task as a continuation to the first task, instead of an independent task.
cfuture<int> A = async([]() { return 65; });
cfuture<char> B = A.then([](int v) {
return static_cast<char>(v);
});
cfuture<void> C = then(B, [](char c) { assert(c == 'A'); });
C.wait();
Note we can use both the member function basic_future::then or the free function then. basic_future::then allows chaining while the free function then allows interoperability between future types.
cfuture<void> C = async([]() {
return 65;
}).then([](int v) {
return static_cast<char>(v);
}).then([](char c) {
assert(c == 'A');
});
C.wait();
In these examples,
- it's up to the continuation to wait for the previous task, and
- it's up to the previous task to launch its own continuations
In other words, task B does not have to pool task A because task A is launching task B. Task B knows A is ready and can just take it from there.
In this solution, the main thread has more time to do useful work, such as scheduling other tasks, and the processing time we spend on waiting is minimized.
Continuation Unwrapping
You might notice that continuation functions also unwrap the previous future value. A continuation function for
future<A> might have a parameter future<A> or A. More patterns of continuation unwrapping are described in
Section adaptors/continuations.
For this reason, continuations are one of the most common proposed extensions for std::future, including the original model presented by the Microsoft PPL Library which inspired the C++ Extensions for Concurrency.
Eager Future Continuations in C++
Eager futures with eager continuation chaining, such as in std::experimental::future, allow us to asynchronously register a second operation and pass data to it. The first task might already be running eagerly.
- The continuation is attached after the first task is scheduled
- The continuation is scheduled as soon as, but not before, the first task is ready
The process does not consume any polling threads. The continuation can also have its continuations and so on. In this scenario, attaching a continuation has its own synchronization cost.
Deferred continuations
The process of attaching continuations to a future whose main task is potentially executing has a synchronization cost. When attaching a continuation, we need to check if the future is not currently attempting to run the continuations and vice-versa. This synchronization cost was identified in N3747.
The library implements this procedure with atomic operations to avoid this cost. However, in some contexts, the cost of continuations can be further minimized by launching deferred futures.
auto A = schedule([]() { return 65; });
auto B = then(A, [](int v) { return static_cast<char>(v); });
auto C = then(B, [](char c) { assert(c == 'A'); });
C.wait(); // launch A now!
The continuation to a deferred shared state created with schedule is also deferred by default. When the task related to any shared state is deferred, we have a different sequence of events:
In this case, the synchronization cost can be completely removed because the task will only be sent to the executor once its continuations have already been attached to it. Notice how tasks B and C are never waiting at the same time.
Deferred futures can also avoid the list of continuations all together. When requesting a deferred continuation to start, it can simply wait for the previous task inline before sending its own task to the executor.
While an eager future stores its continuations:
A deferred continuation can store the previous task:
Thus, deferred futures without explicit continuation lists can still have lazy continuations, as the continuation task will store its previous task, forming a chain of tasks in the shared state of these objects. This allows deferred futures to have the member function basic_future::then defined even we no continuation list is available.
In fact, this is safer than continuation lists for deferred futures. Let A and B be deferred tasks. Because they are not eager, task A will not be launched before task B and then post task B to the executor. When we wait for task B, it will be launched, which would take room in the executor. Task B, already running, will need to wait for task A. At the point, however, the executor might not have the enough capacity for launching task A because B is already running and polling A:
The library solves this problem by checking for previous continuations of A before starting B. But deferred continuations can solve the problem natively:
For this reason, by default, then attaches the previous future to its deferred continuation instead of attaching the continuation to the antecedent future.
Lazy future continuations in C++
Lazy futures with lazy continuation chaining store the continuation in the shared state of the first task before the task is scheduled.
- The continuation is attached before the first task starts to execute
- The continuation is scheduled as soon as, but not before, the first task is ready
As usual, all futures are programmed to run its internal continuations when they finish their main task. This also avoids blocking waits and wasting threads pooling for the results of the antecedent task. In this scenario, attaching a continuation has no synchronization cost.
auto A = std::experimental::async([]() { return 2; });
auto B = A.then([](int A_result) {
// This task is not scheduled until A completes
std::cout << A_result << std::endl;
});
B.wait();
Exceptions and continuations in C++
If the antecedent future throws an exception, attempting to retrieve the result usually rethrows the error. Some models besides C++ Extensions for Concurrency, such as Continuable, allow the continuation to also catch this error:
async([]{ /* operation that might throw an error */ })
.then([] {
throw std::exception("Some error");
})
.fail([] (std::exception_ptr ptr) {
try {
std::rethrow_exception(ptr);
} catch(std::exception const& e) {
// Handle the exception or error code here
}
});
Continuations and Executors in C++
By default, the first task usually includes an executor handle and the continuation inherits it unless some other executor is requested for the continuation. Futures with continuations can also be used as components of resumable functions.