Current thread executor #639

carllerche · 2017-11-10T21:08:33Z

Provides a current thread executor allowing users to spawn tasks that are guaranteed to run on the current thread.

This is based on the Tokio reform RFC but made some API tweaks.

Remaining

This commit extracts the scheduling logic from FuturesUnordered and moves it to a private `scheduler` module. This module will be used by the CurrentThread executor from the Tokio reform RFC.

The `CurrentThread` executor is an event loop that allows the user to spawn tasks that are guaranteed to run on the current thread. This is based on the Tokio reform RFC, but also includes some API tweaks.

cramertj · 2017-11-10T22:53:30Z

src/executor/current_thread.rs

+    _p: ::std::marker::PhantomData<Rc<()>>,
+}
+
+/// Executes dameonized tasks on the current thread.


Here and elsewhere it would be good to clarify what it means for a task to be a "daemon"-- i.e. it executes indefinitely, and it does not prevent functions like block_on_all from returning.

(nit: also, spelling)

cramertj · 2017-11-10T22:54:28Z

src/executor/current_thread.rs

+    }
+
+    /// Execute the given future *synchronously* on the current thread, blocking
+    /// until it (and all spawned tasks) completes and returning its result.


*all non-daemon tasks.

cramertj · 2017-11-10T22:54:50Z

src/executor/current_thread.rs

+    /// Execute the given closure, then block until all spawned tasks complete.
+    ///
+    /// In more detail, this function will block until:
+    /// - All spawned tasks are complete, or


*non-daemon

cramertj · 2017-11-10T22:55:57Z

src/executor/current_thread.rs

+        })
+    }
+
+    /// Spawns a daemon, which does *not* block the pending `block_on_all` call.


...which is a task that does not block....

cramertj · 2017-11-10T23:04:11Z

src/scheduler.rs

+
+pub struct Notify<'a, T: 'a, W: 'a>(&'a Arc<Node<T, W>>);
+
+/// Wakeup a sleeper


Probably worth clarifying what a "sleeper" is. My understanding is its something that is in charge of polling the scheduler (either a Task in the FuturesUnordered case or a thread in the current-thread executor).

cramertj · 2017-11-10T23:06:38Z

src/scheduler.rs

+use std::sync::{Arc, Weak};
+use std::usize;
+
+/// A generic task-aware scheduler.


Could you add a more in-depth comment? Something to the effect of: Scheduler is used to run many child tasks inside of another thing that can be woken up via a "Wakeup".

I would definitely appreciate a patch that improves on the docs 👍

cramertj · 2017-11-10T23:09:06Z

src/scheduler.rs

+}
+
+unsafe fn hide_lt<T, W: Wakeup>(p: *mut ArcNode<T, W>) -> *mut UnsafeNotify {
+    mem::transmute(p as *mut UnsafeNotify)


I think this transmute call isn't actually doing anything?

It is hiding the lifetime :)

That said, I think that W needs to be 'static.

aturon

PR largely looks good, although there's at least one significant (but easy to fix) bug.

I was impressed with the way you got a modicum of safety checking around re-entrancy with the scheduler data structure; would definitely like a bit more docs laying out how this works, as it took me some time to glean it from the code.

aturon · 2017-11-10T22:45:42Z

src/scheduler.rs

+    queued: AtomicBool,
+}
+
+pub enum Tick<T> {


Would be nice to have a comment saying this is an enum for the return of tick below.

aturon · 2017-11-10T22:45:54Z

src/scheduler.rs

+    Inconsistent,
+}
+
+enum Dequeue<T, W> {


Similar: short comment would be great.

aturon · 2017-11-13T17:50:35Z

src/executor/current_thread.rs

@@ -0,0 +1,361 @@
+//! Execute tasks on the current thread


The docs here and elsewhere need significant expansion. We can do that in a follow-up PR if you prefer.

aturon · 2017-11-13T21:45:19Z

src/executor/current_thread.rs

+            }
+
+            let res = future.as_mut()
+                .map(|f| f.poll_future_notify(thread_notify, 0));


This future appears to be polled outside of the scheduler being active, meaning that attempts to spawn etc. will panic.

Would maybe be best to provide a helper to use her and below in poll_all.

aturon · 2017-11-13T21:47:45Z

src/executor/current_thread.rs

+        let mut result = None;
+        let mut future = Some(executor::spawn(future));
+
+        while future.is_some() || self.non_daemons > 0 {


Could rewrite this as a loop and break at the end (removing the replicated condition).

aturon · 2017-11-13T21:56:12Z

src/executor/current_thread.rs

+#[derive(Debug)]
+struct TaskRunner {
+    /// Number of non-daemon tasks being executed by this `TaskRunner`.
+    non_daemons: usize,


As far as I can tell, this field is never actually updated. It also can't be accessed as part of CurrentRunner, which is what you really need.

aturon · 2017-11-13T22:03:00Z

src/executor/current_thread.rs

+    }
+
+    /// Enter a new `TaskRunner` context
+    fn enter<F, A>(f: F) -> Result<A::Item, A::Error>


Would appreciate a bit more descriptive comments on this and the next couple top-level functions.

aturon · 2017-11-13T22:03:51Z

src/executor/current_thread.rs

+                // TODO: This probably can be improved
+                current.cancel.set(false);
+
+                debug_assert!(current.scheduler.get().is_null());


This assert initially caught me by surprise, because I hadn't caught that the scheduler was only temporarily established in enter. Can we make this flow more clear?

aturon · 2017-11-13T22:14:04Z

src/scheduler.rs

+    /// This function should be called whenever the caller is notified via a
+    /// wakeup.
+    pub fn tick<F, R>(&mut self, mut f: F) -> Tick<R>
+    where F: FnMut(&mut Self, &mut T) -> Async<R>


Very clever to thread the &mut issue by passing it back to the callback.

aturon · 2017-11-13T22:18:22Z

src/executor/current_thread.rs

+        loop {
+            let res = self.scheduler.tick(|scheduler, spawned, notify| {
+                current.set_scheduler(scheduler, || {
+                    match spawned.inner.0.poll_future_notify(notify, 0) {


I'd love some commentary here about the fact that tick gives us &mut access back to the scheduler, which provides some amount of safety checking against re-entrancy issues.

carllerche · 2017-11-14T02:50:36Z

@aturon thanks for the review. I’m currently out of town but will try to address shortly. Docs should definitely be improved before merging. I just wanted to get the code up ASAP for initial review. I should have included a doc checkbox in the original Pr.

The current thread executor should be set to the thread-local variable before polling any futures.

Only pull in the files when the std feature is set.

carllerche · 2017-11-15T21:40:39Z

I'm going to remove CurrentThread::block_on_all for now. It is pretty similar to block_with_init, so I think we should be conservative for now with the API.

The original reason for block_on_all was to drive a single future to completion, but that has been handled with future::blocking(fut).wait().

Either way, it can be added back later.

carllerche · 2017-11-15T22:35:00Z

I don't love the name cancel_all_spawned as I think we have mixed up our terminology here.

carllerche · 2017-11-15T22:35:44Z

In general, we have "executor" and "execute", but CurrentThread::spawn. We probably should converge to execute and move away from spawn

cramertj · 2017-11-15T23:02:21Z

In general, we have "executor" and "execute", but CurrentThread::spawn. We probably should converge to execute and move away from spawn

If I saw execute i'd personally expect it to be a call to the method on the Executor trait.

carllerche · 2017-11-15T23:06:25Z

@cramertj That is basically what it is.

carllerche · 2017-11-15T23:13:45Z

I have made a number of API changes in this PR compared to the RFC:

CurrentThread instead of free functions

The PR just proposes a bucket of free functions. Instead, I put all of these functions on executor::CurrentThread. This type also is the implementor of the Executor trait.

I assume that at some point, there will also be executor::Pool as a facade over a customizable thread pool implementation.

Naming polish

I renamed spawn -> execute to bring it inline with the Executor trait. There are probably more tweaks besides this.

**Merged block_for_all and block_with_init.

Now there is only one blocking function. The original split makes more sense without the future::blocking helper.

**Pass &mut current_thread::Context to the block_with_init closure.

The RFC calls for passing &mut executor::Enter directly, however, this isn't very future proof. Instead, I define a new type that will eventually contain &mut executor::Enter. This way, if the current thread executor wants to eventually provide more context, it is able to do so without breaking changes.

I personally just picked Context without putting much thought into it, so if there are other ideas for naming, please post them.

aturon

Changes (especially docs) largely look good!

I'm not sure about the "execute" vs "spawn" terminology. Spawn is more evocative of what's happening -- you're creating a new, independent task that will be concurrently executed, and that should be true of all executors. I'm not sure we should give much weight to the naming within the futures crate (and if anything that should change, IMO).

aturon · 2017-11-15T23:50:03Z

benches/current_thread_executor.rs

+
+#[bench]
+#[ignore]
+fn spawn_daisy(b: &mut Bencher) {


Should finish or remove before landing.

aturon · 2017-11-15T23:53:24Z

src/executor/current_thread.rs

@@ -1,4 +1,69 @@
 //! Execute tasks on the current thread
+//!
+//! [`CurrentThread`] provides an executor that keeps spawned futures keeps


Repeated phrase "keeps spawned futures".

Maybe rephrase as: "is an executor that runs all spawned futures on a single thread".

aturon · 2017-11-16T00:09:20Z

src/executor/current_thread.rs

+//! [`block_with_init`]: struct.CurrentThread.html#method.block_with_init
+//! [`block_on_all`]: struct.CurrentThread.html#method.block_on_all
+//! [`CurrentThread::spawn`]: struct.CurrentThread.html#method.spawn
+//! [`CurrentThread::spawn_daemon`]: struct.CurrentThread.html#method.spawn_daemon


Docs look great 👍

aturon · 2017-11-16T00:10:57Z

src/executor/current_thread.rs

@@ -14,13 +79,25 @@ use std::rc::Rc;
 use std::sync::Arc;

 /// Executes tasks on the current thread.
+///
+/// All tasks spawned using this executor will be executed on the current thread


This is confusing given the inherent methods (which can spawn daemons). Maybe you want to say "when used as an Executor?

aturon · 2017-11-16T00:17:51Z

Other than the spawn/execute issue, the other departures from the RFC are largely cosmetic and are fine by me.

carllerche · 2017-11-16T05:40:43Z

@aturon I personally prefer the spawn terminology as well, however I think that having disjoint terminology is worse. Precise and consistent terminology is valuable. I made note of the naming issue in a tracking PR (#645).

carllerche · 2017-11-16T06:21:35Z

@aturon OK, this PR should be good to go (assuming CI passes). Please provide final PR review.

The `CurrentThread` executor is an event loop that allows the user to spawn tasks that are guaranteed to run on the current thread. This is done by extracting the scheduling logic from FuturesUnordered and moving it to a `scheduler` module that can be used by both `FuturesUnordered` and `CurrentThread` executor. This is based on the Tokio reform RFC, but also includes some API tweaks.

carllerche requested a review from aturon November 10, 2017 21:08

carllerche changed the base branch from master to tokio-reform November 10, 2017 21:08

carllerche added 2 commits November 10, 2017 13:16

Extract scheduling logic from FuturesUnordered

8c42591

This commit extracts the scheduling logic from FuturesUnordered and moves it to a private `scheduler` module. This module will be used by the CurrentThread executor from the Tokio reform RFC.

Implement CurrentThread executor.

f7eeb61

The `CurrentThread` executor is an event loop that allows the user to spawn tasks that are guaranteed to run on the current thread. This is based on the Tokio reform RFC, but also includes some API tweaks.

carllerche force-pushed the current-thread-executor branch from e34ee3f to f7eeb61 Compare November 10, 2017 21:18

cramertj reviewed Nov 10, 2017

View reviewed changes

aturon suggested changes Nov 13, 2017

View reviewed changes

carllerche added 9 commits November 14, 2017 13:47

Add benchmarks for CurrentThread executor

6ddb5dd

Fix CurrentThread bug.

7fe1f4d

The current thread executor should be set to the thread-local variable before polling any futures.

CurrentThread requires std

77c0ab3

Only pull in the files when the std feature is set.

Add test for blocking non-daemon tasks

59fb3ce

Fix blocking for non-daemon tasks

1e9133b

Add a spawn_many test

0011132

Explicitly impl Debug for ThreadNotify

ebb4022

Start documenting current_thread

2f2893a

Remove CurrentThread::block_on_all for now.

78e27af

carllerche force-pushed the current-thread-executor branch from 444e29f to 78e27af Compare November 15, 2017 21:31

Simplify current thread executor

d80dca7

Add some more inline documentation

144f9c1

Try to converge terminology

77f59fe

aturon reviewed Nov 16, 2017

View reviewed changes

carllerche mentioned this pull request Nov 16, 2017

Tokio reform changes #645

Closed

3 tasks

carllerche added 3 commits November 15, 2017 21:42

Fix benches

d167c7b

Add daisy execute bench

9f274ae

Fix bug with exiting executor with spinning task

45a9f58

aturon approved these changes Nov 16, 2017

View reviewed changes

aturon merged commit 1a2f97f into tokio-reform Nov 16, 2017

cramertj mentioned this pull request Nov 16, 2017

Rename execute to spawn #646

Closed

carllerche deleted the current-thread-executor branch November 29, 2017 21:28


		pub struct Notify<'a, T: 'a, W: 'a>(&'a Arc<Node<T, W>>);

		/// Wakeup a sleeper

Current thread executor #639

Current thread executor #639

Uh oh!

Conversation

carllerche commented Nov 10, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Remaining

Uh oh!

cramertj Nov 10, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aturon left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

carllerche commented Nov 14, 2017

Uh oh!

carllerche commented Nov 15, 2017

Uh oh!

carllerche commented Nov 15, 2017

Uh oh!

carllerche commented Nov 15, 2017

Uh oh!

cramertj commented Nov 15, 2017

Uh oh!

carllerche commented Nov 15, 2017

Uh oh!

carllerche commented Nov 15, 2017

Uh oh!

aturon left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aturon commented Nov 16, 2017

Uh oh!

carllerche commented Nov 10, 2017 •

edited

Loading

cramertj Nov 10, 2017 •

edited

Loading