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.

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());
-}