rt: cleanup and simplify scheduler (scheduler v2.5) (#2273)

A refactor of the scheduler internals focusing on simplifying and
reducing unsafety. There are no fundamental logic changes.

* The state transitions of the core task component are refined and
reduced.
* `basic_scheduler` has most unsafety removed.
* `local_set` has most unsafety removed.
* `threaded_scheduler` limits most unsafety to its queue implementation.

2 files changed, 16 insertions, 18 deletions

diff --git a/tokio/src/runtime/blocking/pool.rs b/tokio/src/runtime/blocking/pool.rs
index 0b9d2209..a3b208d1 100644
--- a/tokio/src/runtime/blocking/pool.rs
+++ b/tokio/src/runtime/blocking/pool.rs
@@ -5,8 +5,8 @@ use crate::loom::thread;
 use crate::runtime::blocking::schedule::NoopSchedule;
 use crate::runtime::blocking::shutdown;
 use crate::runtime::blocking::task::BlockingTask;
+use crate::runtime::task::{self, JoinHandle};
 use crate::runtime::{Builder, Callback, Handle};
-use crate::task::{self, JoinHandle};
 
 use std::collections::VecDeque;
 use std::fmt;
@@ -53,7 +53,7 @@ struct Shared {
     shutdown_tx: Option<shutdown::Sender>,
 }
 
-type Task = task::Task<NoopSchedule>;
+type Task = task::Notified<NoopSchedule>;
 
 const KEEP_ALIVE: Duration = Duration::from_secs(10);
 
@@ -227,7 +227,7 @@ impl Inner {
             // BUSY
             while let Some(task) = shared.queue.pop_front() {
                 drop(shared);
-                run_task(task);
+                task.run();
 
                 shared = self.shared.lock().unwrap();
             }
@@ -305,9 +305,3 @@ impl fmt::Debug for Spawner {
         fmt.debug_struct("blocking::Spawner").finish()
     }
 }
-
-fn run_task(f: Task) {
-    let scheduler: &'static NoopSchedule = &NoopSchedule;
-    let res = f.run(|| Some(scheduler.into()));
-    assert!(res.is_none());
-}
diff --git a/tokio/src/runtime/blocking/schedule.rs b/tokio/src/runtime/blocking/schedule.rs
index 5d2cd5f5..e10778d5 100644
--- a/tokio/src/runtime/blocking/schedule.rs
+++ b/tokio/src/runtime/blocking/schedule.rs
@@ -1,20 +1,24 @@
-use crate::task::{Schedule, ScheduleSendOnly, Task};
+use crate::runtime::task::{self, Task};
 
 /// `task::Schedule` implementation that does nothing. This is unique to the
 /// blocking scheduler as tasks scheduled are not really futures but blocking
 /// operations.
+///
+/// We avoid storing the task by forgetting it in `bind` and re-materializing it
+/// in `release.
 pub(super) struct NoopSchedule;
 
-impl Schedule for NoopSchedule {
-    fn bind(&self, _task: &Task<Self>) {}
-
-    fn release(&self, _task: Task<Self>) {}
+impl task::Schedule for NoopSchedule {
+    fn bind(_task: Task<Self>) -> NoopSchedule {
+        // Do nothing w/ the task
+        NoopSchedule
+    }
 
-    fn release_local(&self, _task: &Task<Self>) {}
+    fn release(&self, _task: &Task<Self>) -> Option<Task<Self>> {
+        None
+    }
 
-    fn schedule(&self, _task: Task<Self>) {
+    fn schedule(&self, _task: task::Notified<Self>) {
         unreachable!();
     }
 }
-
-impl ScheduleSendOnly for NoopSchedule {}