WIP: Add task::scope #2153

Matthias247 · 2020-01-22T08:23:08Z

This change adds task::scope as a mechanism for supporting
structured concurrency as described in #1879.

The change adds a task::scope function which will forcefully cancel all
child tasks when the scope is exited, as well as a
task::scope_with_options function which allows to override the default
cancellation and drop behavior.

The scope implementations makes use of 2 primitives:

CancellationToken: This allows to signal an arbitrary amount of tasks
to cancel
WaitGroup: This allows to wait for outstanding tasks to complete

Both primitives are implemented using mechansims and code from
futures-intrusive.

The current PR is work-in-progress and mainly up for discussions.
One thing that definitely needs to be changed is the dependency on futures::executor::block_on for dropping a scope. Besides that tests and docs are missing.

This change implements scope in a way where a forced cancellation of child tasks is the default. However graceful cancellation is still possible if applications prefer the behavior. In order to achieve this the automatic cancellation can be disabled, and users can utilize their own cancellation tokens in order to perform a graceful cancellation.

This change adds `task::scope` as a mechanism for supporting structured concurrency as described in tokio-rs#1879. The change adds a `task::scope` function which will forcefully cancel all child tasks when the scope is exited, as well as a `task::scope_with_options` function which allows to override the default cancellation and drop behavior. The `scope` implementations makes use of 2 primitives: - CancellationToken: This allows to signal an arbitrary amount of tasks to cancel - WaitGroup: This allows to wait for outstanding tasks to complete Both primitives are implemented using mechansims and code from futures-intrusive.

hawkw

This is very cool, and I'd like to take a closer look. I commented on a few questions & thoughts.

hawkw · 2020-01-23T00:10:44Z

tokio/src/task/mod.rs

+    mod scope;
+    pub use scope::{scope, scope_with_options, ScopeOptions, ScopedJoinHandle, ScopeHandle, ScopeCancelBehavior, ScopeDropBehavior};


WDYT about, instead of calling these types ScopeOptions, ScopeHandle, ScopeCancelBehavior, etc, we make the scope module public, and just refer to them as scope::Options, scope::Handle, etc...

I'm flexible regarding those. However one concern I have with names like Option and Handle is that they are already used in some places. And if users need both of them (e.g. runtime Handle and scope Handle - which is not too unreasonable - they would either need to rename them or only import the module. But maybe that problem doesn't exist, since we expect them to import only scope and use the individual structs in a qualified fashion as you mentioned?

We encourage people to import only the module in tracing-subscriber and I was shocked as to how well that worked. For instance, we now have a fmt::Layer and a layer::Layer and I don’t find them to be confusing in the slightest.

hawkw · 2020-01-23T00:11:04Z

tokio/src/task/scope/cancellation_token.rs

@@ -0,0 +1,255 @@
+//! An asynchronously awaitable event for signalization between tasks


I don't think "signalization" is a word :)

For non native speakers it works fine 😂
I first wanted to say "please provide a recommendation for an update" - but it's wrong anyway - the description was copied from futures-intrusive ManualResetEvent, but this is a CancellationToken

hawkw · 2020-01-23T00:12:26Z

tokio/src/task/scope/cancellation_token.rs

+// The Event is can be sent to other threads as long as it's not borrowed
+unsafe impl Send for CancellationToken {}
+// The Event is thread-safe as long as the utilized Mutex is thread-safe
+unsafe impl Sync for CancellationToken {}


I'm assuming this is inherited from futures-intrusive — AFAICT, we should automatically be Send + Sync without these impls, since we are not generic over a mutex?

Afaik the issue was the raw pointer for the linked waiter list which is stored inside the struct. As long this is there the struct can't be Send, and the Mutex can't make it sync.
I can check if those are still necessary, but I guess so.

hawkw · 2020-01-23T00:14:32Z

tokio/src/task/scope/cancellation_token.rs

+        // WaitForCancellationFuture only needs to get removed if it has been added to
+        // the wait queue of the Event. This has happened in the PollState::Waiting case.
+        if let PollState::Waiting = wait_node.state {
+            if !unsafe { self.waiters.remove(wait_node) } {


In this method, we have exclusive mutable access to self.waiters. Naively, I would expect to be able to remove from a list with a &mut ref safely...i've not gotten to read the linked list implementation yet, though.

It's indeed safe as long as the list is consistent. But the outside API of the list at this point just forwards the unsafe annotations that are required internally. And since it mutates a raw pointer of non-owned elements, it is unsafe.

hawkw · 2020-01-23T00:17:48Z

tokio/src/task/scope/intrusive_double_linked_list.rs

@@ -0,0 +1,628 @@
+//! An intrusive double linked list of data


Kind of amusing to note that if we merge this PR as-is, tokio would now contain at least 4 separate linked-list implementations. :)

Am I correct that this linked list is inherently not thread safe? This should probably be stated in this comment...

Re 4 list implementations: Sounds indeed not ideal :-)
I confess I didn't perform an in-depth review on what is available up to now, since the main focus was to get scope working. This one here was mostly copy/paste to enable the necessary synchronization primitives.

Re thread-safety: It indeed is not. I thought not being Sync is indicator enough :-)

hawkw · 2020-01-23T00:26:38Z

tokio/src/task/scope/scope.rs

+    pub struct ScopedJoinHandle<'scope, T> {
+        #[pin]
+        handle: JoinHandle<Result<T, CancellableFutureError>>,
+        phantom: core::marker::PhantomData<&'scope ()>,


I think that, since the phantomdata is unused, this should have a leading _?

hawkw · 2020-01-23T00:27:56Z

tokio/src/task/scope/scope.rs

+
+impl ScopeHandle {
+    /// spawns a task on the scope
+    pub fn spawn<'inner, T, R>(&'inner self, task: T) -> ScopedJoinHandle<'inner, R>


For ergonomics reasons, I think we probably ought to have a free-fn spawn (or spawn_scoped if it's exported at the top level?) that's available inside the scope, similar to spawn_local...

The non-free function has an advantage: It also constrains the lifetime of the join handles to the ones of the ScopeHandle. Since the ScopeHandle doesn't allow to spawn outside of the scope, the ScopedJoinHandle can thereby also never resolve to a cancelled variant. If we would make this a free function there could either be no current scope available, or the scope might already have been cancelled.

Another thing is that ScopeHandles can currently be cloned and stored in structs for spawning later - at least as long as that struct lives within the lifetime of a scope.

If we would make this a free function there could either be no current scope available, or the scope might already have been cancelled.

I think it would be fine for a free function to panic in these cases; that's what the rest of tokio does...

It also constrains the lifetime of the join handles to the ones of the ScopeHandle.

This, on the other hand, is a compelling reason to use the handle only. AFAICT, there isn't really any other way to constrain the scoped join handle lifetime. I think the ergonomics of a free function would be better, but I think the ability to return a JoinHandle that only lives as long as the scope is super valuable, so this may be the best approach!

hawkw · 2020-01-23T00:28:47Z

tokio/src/task/scope/scope.rs

+            futures::executor::block_on(async {
+                let _ = child_task.await;
+            });
+            panic!("Spawn on cancelled Scope");


Since this panic will signal incorrect API use to a user, it would be nice if the panic message was a little more descriptive.

hawkw · 2020-01-23T00:30:38Z

tokio/src/task/scope/scope.rs

+    /// Whether tasks should be cancelled once the scope is exited
+    pub cancel_behavior: ScopeCancelBehavior,
+    /// How the scope should behave if it gets dropped instead of being `await`ed
+    pub drop_behavior: ScopeDropBehavior,


To avoid breaking changes, I think we should either add an empty private field here, so the struct has to be constructed like

ScopeOptions { cancel_behavior: // whatever drop_behavior: // whatever ..ScopeOptions::default(), }

or replace the public fields with a builder. That way, adding new options isn't a breaking change.

I think a builder is the way to go! One open question I had regarding this was whether the builder at the end should directly build the scope or just returns the Option.

The first option would then look along:

ScopeBuilder::new() .set_cancel_behavior(ScopeCancelBehavior::Panic) .build(|scope| async move { scope.spawn(...); }).await;

I think that looks a bit heavy - maybe rather just build the Options

Hmm, do you think there's a use-case for building multiple scopes with the same Options?

tokio/src/task/scope/cancellation_token.rs

mikeando · 2020-01-30T03:03:02Z

tokio/src/task/scope/cancellation_token.rs

+    }
+}
+
+/// Internal state of the `CancellationToken` pair above


CancellationToken is now below, not above, and is not a pair. Suggest you just make this

/// Internal state of the `CancellationToken`

carllerche · 2020-01-30T18:24:57Z

tokio/src/task/scope/scope.rs

+                //   the current executor thread to make progress, due to dependening on
+                //   its IO handles. We need to do something along task::block_in_place
+                //   to solve this.
+                futures::executor::block_on(wait_fut);


Internals are able to use our implementation of block_on: https://github.com/tokio-rs/tokio/blob/master/tokio/src/runtime/enter.rs#L83

carllerche · 2020-01-30T18:35:35Z

I haven't done a detailed review yet, but we should aim to get this merged sooner than later flagged w/ a #[cfg(tokio_unstable)], especially since it is pretty standalone.

What if we move this to a top-level module: tokio::scope. I think we should aim for a flatter module structure in general (other tokio modules should be flattened as well).

hawkw · 2020-01-30T18:56:21Z

@carllerche

What if we move this to a top-level module: tokio::scope. I think we should aim for a flatter module structure in general (other tokio modules should be flattened as well).

+1 for a flatter module structure (and, IMO, shortening the names by referring them to scope::JoinHandle etc).

However, there is one minor issue exposing a top-level scope module: the free function scope for creating a scope would be tokio::scope::scope(...), I find a little unpleasant...I'd prefer it to be exposed as tokio::task::scope(...) or something. Or just tokio::scope(...), but I feel like it might be a little weird to export a scope function and a scope module (although the compiler doesn't mind this since functions and modules occupy different namespaces and after looking at it for a bit, I think it's actually kind of nice...).

carllerche · 2020-01-30T18:56:39Z

tokio/src/task/scope/scope.rs

+
+/// A handle to the scope, which allows to spawn child tasks
+#[derive(Clone)]
+pub struct ScopeHandle {


This probably could just be named Scope?

carllerche · 2020-01-30T19:02:49Z

tokio/src/task/scope/scope.rs

+pub async fn scope<F, Fut, R>(scope_func: F) -> R
+where
+    F: FnOnce(ScopeHandle) -> Fut,
+    Fut: Future<Output = R> + Send,


Could you clarify why Send is required here?

udoprog

Reviewed first batch of code. Only minor nits so far!

udoprog · 2020-01-30T18:47:04Z

tokio/src/task/scope/intrusive_double_linked_list.rs

+    /// The function is only safe as long as valid pointers are stored inside
+    /// the linked list.
+    pub(crate) unsafe fn add_front(&mut self, item: *mut ListNode<T>) {
+        assert!(!item.is_null(), "Can not add null pointers");


Either change to NonNull, or this runtime assertion could be changed into a debug_assert! and added as a Safety invariant in documentation.

udoprog · 2020-01-30T18:50:38Z

tokio/src/task/scope/intrusive_double_linked_list.rs

+    }
+
+    /// Consumes the list and creates an iterator over the linked list.
+    /// This function is only safe as long as all pointers which are stored inside


Nice with these comments, but it would be nice if it (and others like it) followed rustdoc convention and lived under a # Safety section.

udoprog · 2020-01-30T18:51:49Z

tokio/src/task/scope/cancellation_token.rs

+            // further side effects.
+
+            let waiters = self.waiters.take();
+


Please document what assumptions are made here to make the unsafe use sound. Some places you already have, but same for the ones you haven't. Even if they are trivial it makes them easier to review and maintain in case the assumptions change!

udoprog · 2020-01-30T18:57:03Z

tokio/src/task/scope/intrusive_double_linked_list.rs

+            return false;
+        }
+
+        assert!(self.tail.is_null());


Could be a debug assertion? Unless this has safety implications which could arise from safe use at runtime. If that's the case, the panic should be documented - especially how it could arise from misuse.

udoprog · 2020-01-30T19:05:16Z

tokio/src/task/scope/intrusive_double_linked_list.rs

+
+    /// Removes the last item from the linked list and returns it
+    #[allow(dead_code)]
+    pub(crate) unsafe fn remove_last(&mut self) -> *mut ListNode<T> {


This function doesn't look like it has to be unsafe. It has exclusive access, and all invariants are internally checked as far as I can see / understand. Otherwise, please document safety :D.

udoprog · 2020-01-30T19:30:06Z

tokio/src/task/scope/intrusive_double_linked_list.rs

+}
+
+#[cfg(test)]
+#[cfg(feature = "std")] // Tests make use of Vec at the moment


Feature std doesn't exist (yet), so can't run tests. I'm guessing probably just remove this?

udoprog · 2020-01-30T19:30:39Z

tokio/src/task/scope/intrusive_double_linked_list.rs

+    }
+
+    #[test]
+    fn add_sorted() {


Testing function which doesn't exist add_sorted, guessing copy-paste mistake so should probably be removed?

Matthias247 · 2020-04-29T16:39:47Z

This is currently paused, and will be rebased on top of #2263 once ready.

Mygod · 2020-05-22T15:44:12Z

Any updates on this?

Matthias247 · 2020-05-31T02:21:48Z

Superseeded by #2576

Matthias247 mentioned this pull request Jan 22, 2020

Structured Concurrency Support #1879

Closed

Matthias247 force-pushed the scope branch from 74e2439 to 465b5b4 Compare January 22, 2020 15:53

hawkw reviewed Jan 23, 2020

View reviewed changes

mikeando reviewed Jan 30, 2020

View reviewed changes

carllerche reviewed Jan 30, 2020

View reviewed changes

udoprog reviewed Jan 30, 2020

View reviewed changes

Matthias247 mentioned this pull request Feb 3, 2020

Refactor the intrusive linked list Matthias247/futures-intrusive#30

Merged

carllerche mentioned this pull request Feb 3, 2020

sync: adds Notify for basic task notification #2210

Merged

Matthias247 mentioned this pull request Feb 20, 2020

Add CancellationToken #2263

Merged

Darksonn added A-tokio Area: The main tokio crate C-enhancement Category: A PR with an enhancement or bugfix. M-task Module: tokio/task S-waiting-on-author Status: awaiting some action (such as code changes) from the PR or issue author. labels Apr 20, 2020

Darksonn added S-blocked Status: marked as blocked ❌ on something else such as a PR or other implementation work. and removed S-waiting-on-author Status: awaiting some action (such as code changes) from the PR or issue author. labels Apr 29, 2020

Mygod mentioned this pull request May 23, 2020

Connect UDP sockets shadowsocks/shadowsocks-rust#265

Merged

Matthias247 mentioned this pull request May 31, 2020

WIP: task::scope #2576

Closed

Matthias247 closed this May 31, 2020

Matthias247 mentioned this pull request Jun 6, 2020

RFC: structured concurrency via task::scope #2592

Closed

		mod scope;
		pub use scope::{scope, scope_with_options, ScopeOptions, ScopedJoinHandle, ScopeHandle, ScopeCancelBehavior, ScopeDropBehavior};

		@@ -0,0 +1,255 @@
		//! An asynchronously awaitable event for signalization between tasks

		@@ -0,0 +1,628 @@
		//! An intrusive double linked list of data

WIP: Add task::scope #2153

WIP: Add task::scope #2153

Conversation

Matthias247 commented Jan 22, 2020

hawkw left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

carllerche commented Jan 30, 2020

hawkw commented Jan 30, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

udoprog left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

udoprog Jan 30, 2020 • edited Loading

Choose a reason for hiding this comment

Matthias247 commented Apr 29, 2020

Mygod commented May 22, 2020

Matthias247 commented May 31, 2020

udoprog Jan 30, 2020 •

edited

Loading