RFC: structured concurrency via task::scope

[carllerche]

This proposal is mostly the result of @Matthias247's work (#1879, #2579, #2576).
I took this work, summarized the API and added a few tweaks as proposals.

---

Summary

A specific proposal for structured concurrency. The background info and
motivation have already been discussed in #1879, so this will be skipped here.

There are also two PRs with specific proposals: #2579, #2576.

This RFC build upon this work.

Also related: #2576

The RFC proposes breaking changes for 0.3. "Initial steps" proposes a way to add
the functionality without breaking changes in 0.2.

Requirements

At a high level, structured concurrency requires that:

• No tasks are leaked from a scope.
• No error (Result and panic) goes unhandled.

Again, see #2579 for more details.

Proposed Changes

• All tasks must be spawned within a scope.
• Each task has an implicit scope.
• Explicit scopes may be created with task::scope.
• task::signalled().await waits until the task is signalled, indicating it
  should gracefully terminate.
• JoinHandle forcefully cancels the task on drop.
• JoinHandle gains the following methods:
  • background(self): run the task in the background until the scope drops.
  • try_background(self): run the task in the background until the scope
    drops. If the task completes with Err, forcefully cancel the owning scope.
  • signal(&self) signals to the task to gracefully terminate.

Terminology

forceful cancellation: The runtime drops the spawned asap without providing
an ability to gracefully complete. All cleanup is done synchronously in drop
implementations.

graceful cancellation: Signal to a task it should stop processing and clean
up any resources itself.

Details

Scopes

All spawned tasks must be spawned within a scope. The task is bound to the
scope.

task::scope(async {

  let join_handle = tokio::spawn(async {
    // do work
  });

}).await;

This creates a new scope and executes the provided block within the context of
this scope. All calls to tokio::spawn link the task to the current scope. When
the block provided to scope completes, any spawned tasks that have not yet
completed are forcefully cancelled. Once all tasks are cancelled, the call to
task::scope(...).await completes.

Scopes do not attempt to handle graceful cancellation. Graceful cancellation
is provided by a separate set of utilities described below.

All tasks come with an implicit scope. In other words, spawning a task is
equivalent to:

tokio::spawn(async {
  task::scope(async {
    // task body
  }).await
})

There could also be a global scope that is used as a catch-all for tasks that
need to run in the background without being tied to a specific scope. THis
global scope would be the "runtime global" scope.

Error propagation

As @Matthias247 pointed out in #1879, when using JoinHandle values, error
propagation is naturally handled using Rust's ? operator:

task::scope(async {

  let t1 = tokio::spawn(async { ... });
  let t2 = tokio::spawn(async { ... });

  t1.await?;
  t2.await?;

}).await?;

In this case, if t1 completes with Err, t1.await? will return from the
async { ... } block passed to task::scope. Once this happens, all
outstanding tasks are forcibly cancelled, resulting in no tasks leaking.

Dropping JoinHandle forcefully cancel the task

Sub task errors must be handled. This requires avoiding to drop them without
processing them. To do this, the return value of the task must be processed
somehow. This is done by using the JoinHandle returned from
tokio::spawn. In order to ensure the return value is handled, dropping
JoinHandle results in the associated task being forcefully canceled.
JoinHandle is also annotated with #[must_use].

However, there are cases in which the caller does not need the return value. For
example, take the TcpListener accept loop pattern:

while let Some(stream) = listener.accept().await? {
    tokio::spawn(async move {
        process(stream).await
    });
}

In this case, the caller does not need to track each JoinHandle. To handle
these cases, JoinHandle gains two new functions (naming TBD):

impl JoinHandle<()> {
    fn background(self) { ... }
}

impl<E> JoinHandle<Result<(), E>> {
    fn try_background(self) { ... }
}

In the first case, the type signature indicates the task can only fail due to a
panic. In the second, the task may fail due to Err being returned by the task.
When a task that is moved to the scope "background" fails due to a panic or
Err return, the scope is forcibly canceled, resulting in any outstanding task
associated with the scope to be forcibly canceled.

Now, the TcpListener accept loop becomes:

while let Some(stream) = listener.accept().await? {
    tokio::spawn(async move {
        process(stream).await
    }).try_background();
}

Nesting scopes

Usually, structured concurrency descripes a "tree" of tasks. Scopes have N
associated tasks. Each one of those tasks may have M sub tasks, etc. When an
error happens at any level, all decendent tasks are canceled and all ancestors
are canceled until the error is handled.

Instead of explicitly linking scopes, Rust's ownership system is used. A
scope is nested by virtue of the return value of task::scope (Scope) being held in a
task owned by a parent scope. If a Scope is dropped it forcefully cancels all
associated sub tasks. This, in turn, results in the sub tasks being dropped and
any Scope values held by the sub task to be dropped.

However, Scope::drop must be synchronous yet cancelling a task and waiting
for all sub tasks to drop is asynchronous. This is handled has follows:

Every Scope has a "wait set" which is the set of tasks that the scope needs to
wait on when it is .awaited

When Scope is dropped:

• All sub tasks are forcefully cancelled (they are not dropped yet).
• The current scope context is inspected to find the containing scope.
• The dropped scope's wait set is added to the containing scope's wait set.

By doing this, containing_scope.await does not complete until all descendent
tasks are completely terminated.

If a scope block terminates early due to an unhandled sub task error,
task::scope(async { ... }).await completes with an error. In this case, the
task either handles the error or completes the task with an error. In the latter
case, the task's containing scope will receive the error and the process
repeats up the task hierarchy until the error is handled.

Graceful task cancellation

Graceful task cancellation requires "signalling" a task, then allowing the task
to terminate on its own time. To do this, JoinHandle gains a new function,
signal. This sends a signal to the task. While this signal could be used for
any purpose, it is implied to indicate graceful cancellation.

From the spawned task, the signal is received using: task::signalled().await.
Putting it all together:

let join_handle = tokio::spawn(async {
    tokio::select! {
      _ => do_work() => {}
      _ => task::signaled() => {
        // Do cleanup work here
      }
    }
});

join_handle.signal();
join_handle.await;

Initial steps

The majority of this work can be done without any breaking changes. As an
initial step, all behavior in Tokio 0.2 would remain as is. Tasks would not
get an implicit scope. Instead, task::scope(...) must always be called.
Secondly, there would be tokio::spawn_scoped(...) which would spawn tasks
within the current scope. In 0.3, spawn_scoped would become tokio::spawn.

[jonhoo]

try_background is a bit of a misnomer to me, though I don't currently have a better name for it. try_background implies to me that it will try to background the task, but the backgrounding may fail. But that's not really what this method is doing. It's more like background_try I think?

For "nested scopes", a couple of things stood out to me:

• The first paragraph says

  When an error happens at any level, all decendent tasks are canceled and all ancestors are canceled until the error is handled.

  The rest of the section didn't really address that as far as I could tell?

• If all of the sub-tasks are forcefully terminated, why can't they be dropped synchronously? Isn't that what forceful termination means, or am I missing something?

• All of the dropped scope's sub tasks are added to the containing scope. By doing this, containing_scope.await does not complete until all descendent tasks are completely terminated.

  This (perhaps naively) to me sounds like there is a race here, where the descendant tasks are allowed to continue running (terminating?) after their parent scope has been dropped. That is, in the span of time between when the parent scope is dropped and containing_scope finishes. Is that not a problem?

[hawkw]

All tasks come with an implicit scope. In other words, spawning a task is
equivalent to:

Just to clarify that I'm understanding this correctly: does this then mean that any call to tokio::spawn inside a spawned task will spawn a new child of that task's scope, and there would need to be a separate spawn_global (or something) to spawn a task in a new, separate scope?

[carllerche]

@jonhoo

To address:

all ancestors are canceled until the error is handled.

I have added the following paragraph:

If a scope block terminates early due to an unhandled sub task error,
task::scope(async { ... }).await completes with an error. In this case, the
task either handles the error or completes the task with an error. In the latter
case, the task's containing scope will receive the error and the process
repeats up the task hierarchy until the error is handled.

Does that answer your question? I will address the other points shortly.

[carllerche]

@jonhoo

If all of the sub-tasks are forcefully terminated, why can't they be dropped synchronously? Isn't that what forceful termination means, or am I missing something?

The sub task may be concurrently executing. In that case, cancellation must wait for execution to complete then drop.

This (perhaps naively) to me sounds like there is a race here, where the descendant tasks are allowed to continue running (terminating?) after their parent scope has been dropped. That is, in the span of time between when the parent scope is dropped and containing_scope finishes. Is that not a problem?

I should probably call this out. Given that there is no AsyncDrop, you are correct. If a Scope is dropped without .await, it is possible for it to drop before all sub tasks are cancelled. I consider this acceptable as 1) Scope is annotated as #[must_use] and 2) the user explicitly did not express interest in the scope completion by not .awaiting it.

However, the steps I outlined ensure that the parent scope's .await does not complete until the child scope is dropped and all descendant tasks are dropped.

The goal is that, at any level where scope(...).await is called, this will not return until all descendant tasks are dropped. This is because, there is recursive process that happens where, when each sub-task drops, all containing scopes move its tasks up a level.

@jonhoo did this answer your question?

[Matthias247]

This document is missing sections on what benefits it adds over the existing implementations at #2153, #2576, #2579. What are the shortcomings of the other proposals? What are the main benefits this adds over those?

Here are my comments for this proposal.

First of all, this seems like a variation of the initial implementation at #2153, which defaulted to forceful cancellation. The main difference here seems to be that there is no explicit ScopeHandle type, and the mechanism is deeper integrated into the task system.

However already in #2153 it turned out that that forceful cancellation does not compose. You can't do:

task::scope(async {
  task::scope(async {
    tokio::spawn(...)
  }.await;
}).await;

without issues, since the cancelling the outer scope would always lead to a force drop of the inner scope, for which no great solution is available. In this proposal the solution for this is attached the tasks now to the parent scope.

Now my short comment for this is: This breaks structured concurrency. If tasks can migrate up the stack there simply exists no structured concurrency. Application programmers can not rely on the fact that some task does not outlive its containing scope anymore. This is a super powerful guarantee - you can be 100% sure that no unexpected things are running in the background anymore. E.g. some code might assume that a certain file that is worked on in the background task will always be closed after the scope exits. Then code after the scope can always successfully reopen it - or clean it - whatever it wants. Without structured concurrency we don't have this guarantee - the code might still be running.

Besides this:

Forceful cancellation is the wrong default

As pointed out there are no great ways to deal with forceful cancellation. Any component that can not be forcefully cancelled (e.g. because it needs to perform some cleanup work, finish a transaction, etc) can never run in a forcefully cancelled environment. Clean cancellation using our favorite new fancy io_uring topic is one of those examples, gracefully cancellating database transactions, asynchronous calls, threadpool work, etc. are other examples.

Now if the top level system only supports forceful cancellation, there is no sane way to integrate those concepts. They can only rely on hacks (like a global spawn on drop) - which are exactly the kind of hacks that were intended to be avoided with structured concurrency.

Real world cancellation is asynchronous. Even though rust Futures are always cancellable, they can't magically make all operations on top of it magically synchronously cancellable.

What do I do in a pure forceful cancellation world if my requirement is to run a pending transaction always to the end, and just stop starting new transactions?

In a world with graceful cancellation I can do:

scope::enter(async move {
    while let transaction = transaction_provider.accept_for_next_transaction_with_token(scope::current_cancellation_token()) await {
        scope::spawn(async move {
            // This would always run to completion inside the outer scope
            transaction.execute().await;
        });
    }
}).await;

In a world with forceful cancellation as the primary mechanism this wouldn't work.
With #2579 you can even give the transaction a timeout and then make sure the whole parent scope exists after that, while still allowing for a soft shutdown.

scope::enter(async move {
    while let transaction = transaction_provider.accept_for_next_transaction_with_token(scope::current_cancellation_token()) await {
        scope::spawn(async move {
            // This would always run to completion inside the outer scope
            transaction.execute().timeout(time::duration::from_secs(60).await;
        });
    }
}).await;

You can use the same mechanism to e.g. stop a TCP listener from accepting incoming connections, while still giving the active requests some time to finish (draining phase). And then only return from the listener scope when you are 100% sure that everything shut down.

I don't see this covered with this proposal

This makes the task system even more complicated instead of less

tokios task implementation together with shutdown and co are already very high on the complexity scale. This implementation adds yet more complexity to them, in terms of a cancellation states, more joinhandle functionality, etc. The fact that everything needs to deal with forceful cancellation certainly contributes a lot to the complexity. Compared to this #2579 would have allowed to simplify the task system instead up to the point that just run to completion tasks without any cancellation abilities were required, since graceful and forceful cancellation was running on top of the system. This also allowed to drop the annoying error type on the JoinHandle that most users will never require any just unwrap. Here I feel both JoinHandles and tasks get more complicated.

[carllerche]

What are the main benefits this adds over those?

I would say the majority of this proposal is the same as yours. This attempts to outline the API / behavior standalone without having to worry about implementation detail. The main difference is avoiding the need for builders/configuration when it comes to core functionality. For example, instead of controlling how cancellation signals get propagated to descendants via builder options, it is done via Rust ownership.

The main difference here seems to be that there is no explicit ScopeHandle

Yes, it is done via context.

the mechanism is deeper integrated into the task system.

I would disagree, the proposal could be implemented in a separate crate. Any integration with the core task system would be for performance reasons.

This breaks structured concurrency. If tasks can migrate up the stack there simply exists no structured concurrency.

I do not believe this is correct. All code outside of drop fns can rely on sub tasks termination. Only drop fns cannot as tasks can be dropped at anytime. However, this is true for all proposals. Shutting down the Tokio runtime will result in all tasks being forcefully terminated regardless of the state they are in. So, in terms of guarantees, this proposal is in line with the ones you linked.

Forceful cancellation is the wrong default

All tasks need to be able to handle forceful cancellation in a safe way. The Tokio runtime may forcefully cancel all tasks at any point. Even with graceful cancellation, parent scopes will not allow sub tasks to gracefully cancel for an indefinite amount of time. Parent scopes will signal cancellation, wait for a fixed amount of time, then forcefully cancel. Sub-tasks must be able to tolerate the forceful cancellation.

Given that forceful cancellation is a given, this proposal takes a "forceful first" approach. It is based on forceful cancellation and provides a strategy to layer on best-effort graceful cancellation.

This makes the task system even more complicated instead of less

This is difficult to respond to as it is fairly subjective. I'm not entirely sure what specifics you think are more complicated. You mention "forceful cancellation" as a root issue. As mentioned in the above section, no matter the proposal, handling forceful cancellation is a requirement as it is a core part of Rust's future model. Are there any other aspects you think are more complex?

[jonhoo]

@carllerche
Yes, your comments address most of my concerns. I think your answers from #2592 (comment) would be good to include in the proposal.

I still believe try_background isn't a great name.

[carllerche]

@jonhoo I'm not set on any name. If you have suggestions, please share :)

[davidbarsky]

Thanks for writing this up! I have somewhat related two comments.

In the TCP listener accept loop pattern, am I correct in thinking that the expected usage of the new methods on the join handle be:

while let Some(stream) = listener.accept().await? {
    tokio::spawn(async move {
        process(stream).await
    }).try_background().await?;
}

If so, can you add the above example to the initial RFC comment?

My second comment echo's @jonhoo's dislike of the try_background() method name. Here's a possibly bad idea: instead of introducing try_background()/background() on JoinHandle, could the notion of background/unstructured concurrency be moved to the spawn function itself? e.g., Tokio 0.3 will a scoped tokio::spawn; could there be a complementary tokio::spawn_unscoped function as well?

[Matthias247]

What are the main benefits this adds over those?

I would say the majority of this proposal is the same as yours.

No, it is not. To start with, this is not structured concurrency. If you leak tasks in error cases then you simply don't have structured concurrency. This is the same as if the borrow checker sometimes is imprecise - we also don't accept this.

The goal of #1879 was to implement support for it. Now I have no idea why we throw that goal now away in favor of a "tokio-shutdown shutdown system" which has other properties.

Besides this throws a few hundred hours of work away, since it's built on different primitives. The existing propoals built on a hierarchical cancellation token and a wait group, which are industry standard tools to achieve this behavior. As far as I understand this now requires none of those, and instead a Wait Group which bubbles tasks up on cancellation.

The main difference is avoiding the need for builders/configuration when it comes to core functionality. For example, instead of controlling how cancellation signals get propagated to descendants via builder options, it is done via Rust ownership.

The latest implementation in #2579 (which still received no feedback) does not require any mandatory builder options. It's APIs surface is not more than what is described here. It was merely mentioned that it's functionality could be expanded in the future by providing additional options to the scope.

This breaks structured concurrency. If tasks can migrate up the stack there simply exists no structured concurrency.

I do not believe this is correct. All code outside of drop fns can rely on sub tasks termination.

In this proposal things outside of the scope can rely on cancellation having been initiated. They have no guarantees about it having finished. The background task could still run. The goal of structured concurrency is to have those guarantees.

Shutting down the Tokio runtime will result in all tasks being forcefully terminated regardless of the state they are in. So, in terms of guarantees, this proposal is in line with the ones you linked.

No. This might be the current state of things. However that doesn't mean that things have to stay that way. With a follow-up addition to #2579 any forceful runtime shutdown could be have been removed, and users would have been relieved from worrying about having to support 2 different cancellation mechanisms as they would do today. There exists even an RFC for being able to indicate the run-to-completion vs force-cancellable behavior using Rusts type system.

I still think this proposal is even worse than the current state of things. Currently authors of library code can rely on their code running to completion as long as:

• the code lives inside a spawn block
• the author knows the runtime won't be cancelled

Both are fairly typical in long running application.

If you implement this behavior there is however no safe place for code left that does not want to deal with forced cancellation. The only thing people could do is spawn their own runtime that they have more control about.

Forceful cancellation is the wrong default

All tasks need to be able to handle forceful cancellation in a safe way. The Tokio runtime may forcefully cancel all tasks at any point

See above. This is a choice. And I don't think it is the right choice. People should be able to think about tasks as they think about threads. All the differences between those just make tasks harder to work with. I do think lots of code out there would be buggy when encountering forced cancellation in the same way that most code doesn't deal very well with panic!ing. Those are not the execution path people look at, and they might not even see them in code reviews.

Being able to specify that those path should never be taken for a controlled large subset of an application and to reduce the scope of force-cancellability to tasks which have been built to work with it is a big plus.

[carllerche]

@davidbarsky

The accept loop becomes:

while let Some(stream) = listener.accept().await? {
    tokio::spawn(async move {
        process(stream).await
    }).try_background();
}

There is no await on try_background(). I updated the initial post with the snippet.

dislike of the try_background()

I'm not attached to the name.

could the notion of background/unstructured concurrency be moved to the spawn function itself

They could, but we would still need to name them. unscoped isn't accurate as the spawns are still scoped, they just do not have JoinHandles anymore. So, regardless of whether the fns are on JoinHandle or they are a free fns, they would need names.

[carllerche]

No, it is not. To start with, this is not structured concurrency. If you leak tasks in error cases then you simply don't have structured concurrency. This is the same as if the borrow checker sometimes is imprecise - we also don't accept this.

I think it would be more productive to discuss specific examples and how they are handled in both cases. I am not following your logic of how one option is less precise than the other. Perhaps I misread your PR somehow. It is a large chunk of code. Backing out behavior details can be tricky.

Are you saying that there should be no force cancellation of tasks at all? Also, what happens if a scope is dropped from a select! branch?

[carllerche]

I've been thinking more about what I think are the important bits. I think the "nesting by ownership" is less important.

Nesting

I think nested tasks need to have a very clear and deterministic shutdown order.

To forcefully cancel a task, the following recursive steps are taken:

1. Task stops executing (obtain a run lock)
2. Cancel all sub tasks (for each sub task, repeat steps from 1).
3. Drop task.
4. Signal that the task has been cancelled.

To gracefully cancel a task, I believe it is important to only signal the task being gracefully cancelled itself. It should be that task's responsibility to gracefully cancel its sub tasks.

Consider a task that has a hyper client and that hyper client has a sub task to manage the socket. If the parent task gracefully cancels by first sending an HTTP request, it is important that the sub task continues to operate normally until the parent task is ready for it to terminate.

The implicit nesting via ownership enables being able to treat scopes as values that can be sent around (see the hyper example in the RFC's root comment), but I don't think that is critical.

Forceful vs. graceful cancellation

Given that a scope may be dropped at any time, I don't see a way to enable an API where the user does not have to consider the forceful cancellation scenario. I may be missing a detail that makes this assumption wrong, if so that would need to be explained.

[carllerche]

@tokio-rs/maintainers I would love for others to weigh in on the various points here that still lack consensus.

[carllerche]

@Matthias247 Ah, I remember the reason why I ended up going the path of linking scopes on drop.

task::scope(async { ... }) returns a value that is awaited. That value can be sent to other tasks. If the scope is linked with the task that created the scope, sent to a different task, then awaited... what does that even mean?

[Matthias247]

No, it is not. To start with, this is not structured concurrency. If you leak tasks in error cases then you simply don't have structured concurrency. This is the same as if the borrow checker sometimes is imprecise - we also don't accept this.

I think it would be more productive to discuss specific examples and how they are handled in both cases. I am not following your logic of how one option is less precise than the other. Perhaps I misread your PR somehow. It is a large chunk of code. Backing out behavior details can be tricky.

Are you saying that there should be no force cancellation of tasks at all? Also, what happens if a scope is dropped from a select! branch?

This is described in the API docs of #2579: Scopes can only be created inside other scopes which support graceful cancellation. If you don't adhere to this principle they will panic when dropped. Thinking about that now that should maybe changed in panicing when they are created to indicate the API violation earlier.

While this isn't great there are no really better options for this. And there were now really 6 month of research on this. Force cancelling and not waiting (as proposed here) breaks structured concurrency. Forceful cancellation and doing a blocking wait on join might deadlock - and won't work on a singlethreaded eventloop to start with.

#2597 supports forceful cancellation in all blocks which are explicitely marked as supporting to it. To recap #2597, the API surface isn't really big:

• scope::enter creates a new scope and will run all code and subtasks of it as part of the scope. If the scope (or the task containing the scope) gets cancelled, the cancellation request will propagate to all subtasks.
• scope::spawn creates a subtask which is part of the scope. The task will run to completion. If the scope it cancelled, it will merely get a cancellation request which can be obtained via scope::current_cancellation_token().cancelled() or also scope::cancelled() if that shortcut is more convenient to users. The task currently returns a JoinHandle which evaluates to Result<T, JoinError> but it could be simplified to evaluating just to T, since it is not possible that the task gets force interrupted through the runtime anymore.
• scope::spawn_cancellable() spawns a subtask which will get force cancelled when the parent gets cancelled. The requirement here is to only run code inside this task which supports force cancellation. Therefore this is always at the leaf of a task graph.
• scope::current_cancellation_token() returns the CancellationToken which is associcated with the current scope. It can be used to gracefully shut down the current task by listening on the token whenever convient. It can also be used to initiate cancellation.

To gracefully cancel a task, I believe it is important to only signal the task being gracefully cancelled itself. It should be that task's responsibility to gracefully cancel its sub tasks.

This doesn't really work if the task is blocked on the subtask. E.g. consider this example:

scope::enter(async move {
   scope::spawn(async move {
       wait_for_message_from_peer().await;
   }).await; 
}).await;

Now if you would just gracefully cancel the parent without the cancellation signal propagating this would potentially never terminate. Therefore all cancellation systems (and it doesn't matter now whether we look at Kotlin, Go, C# or anything else) went on to perform cancellation in depth.

Of course you could make the example above working by instead writing:

scope::enter(async move {   
   let handle = scope::spawn(async move {
       wait_for_message_from_peer().await;
   });

   select! {
       r = &handle => {
           return r;
       }
       _ = scope::current_cancellation_token()=> {
            handle.cancel();
            return handle.await;
       }
   } 
}).await;

But that really leads to a lot of boilerplate code for the common case.

[carllerche]

@Matthias247

Ok, just to make sure that I understand, scopes as you propose cannot be used from within select! or try_join! statements, nor can they be used inside any combinator that does not have a "drop" protocol?

[carllerche]

@Matthias247 If it is such a critical requirement, shouldn't we just say that a scope is the task and you can't enter sub scopes within the task? That would resolve most of the issues we are talking about, no?

[carllerche]

Closed in favor of #2596