rt: remove unparker from time driver (#5013)

Currently, the various resource drivers use a layered approach where
each driver contains the next one. When the scheduler calls park on the
top-most driver, it does work then calls park on the inner driver it
holds. The handles do the same with unparking.

This patch is a step towards refactoring the runtime to move away from
the nested approach towards keeping all drivers and handles together in
a single runtime driver/handle. The unparker is removed from the time
handle and placed in the runtime handle and is passed into the time
driver as needed.

tokio/src/runtime/driver.rs

@@ -2,7 +2,7 @@
 
 // Eventually, this file will see significant refactoring / cleanup. For now, we
 // don't need to worry much about dead code with certain feature permutations.
-#![cfg_attr(not(feature = "rt"), allow(dead_code))]
+#![cfg_attr(not(feature = "full"), allow(dead_code))]
 
 use crate::park::thread::{ParkThread, UnparkThread};
 
@@ -13,7 +13,7 @@ use std::time::Duration;
 
 cfg_io_driver! {
     pub(crate) type IoDriver = crate::runtime::io::Driver;
-    pub(crate) type IoHandle = Option<crate::runtime::io::Handle>;
+    pub(crate) type IoHandle = IoUnpark;
 
     #[derive(Debug)]
     pub(crate) enum IoStack {
@@ -21,6 +21,7 @@ cfg_io_driver! {
         Disabled(ParkThread),
     }
 
+    #[derive(Debug, Clone)]
     pub(crate) enum IoUnpark {
         Enabled(crate::runtime::io::Handle),
         Disabled(UnparkThread),
@@ -37,9 +38,11 @@ cfg_io_driver! {
             let (signal_driver, signal_handle) = create_signal_driver(io_driver)?;
             let process_driver = create_process_driver(signal_driver);
 
-            (IoStack::Enabled(process_driver), Some(io_handle), signal_handle)
+            (IoStack::Enabled(process_driver), IoUnpark::Enabled(io_handle), signal_handle)
         } else {
-            (IoStack::Disabled(ParkThread::new()), Default::default(), Default::default())
+            let park_thread = ParkThread::new();
+            let unpark_thread = park_thread.unpark();
+            (IoStack::Disabled(park_thread), IoUnpark::Disabled(unpark_thread), Default::default())
         };
 
         Ok(ret)
@@ -83,16 +86,35 @@ cfg_io_driver! {
                 IoUnpark::Disabled(v) => v.unpark(),
             }
         }
+
+        #[track_caller]
+        pub(crate) fn expect(self, msg: &'static str) -> crate::runtime::io::Handle {
+            match self {
+                IoUnpark::Enabled(v) => v,
+                IoUnpark::Disabled(..) => panic!("{}", msg),
+            }
+        }
+
+        cfg_unstable! {
+            pub(crate) fn as_ref(&self) -> Option<&crate::runtime::io::Handle> {
+                match self {
+                    IoUnpark::Enabled(v) => Some(v),
+                    IoUnpark::Disabled(..) => None,
+                }
+            }
+        }
     }
 }
 
 cfg_not_io_driver! {
-    pub(crate) type IoHandle = ();
+    pub(crate) type IoHandle = IoUnpark;
     pub(crate) type IoStack = ParkThread;
     pub(crate) type IoUnpark = UnparkThread;
 
     fn create_io_stack(_enabled: bool) -> io::Result<(IoStack, IoHandle, SignalHandle)> {
-        Ok((ParkThread::new(), Default::default(), Default::default()))
+        let park_thread = ParkThread::new();
+        let unpark_thread = park_thread.unpark();
+        Ok((park_thread, unpark_thread, Default::default()))
     }
 }
 

tokio/src/runtime/time/entry.rs

@@ -549,7 +549,8 @@ impl TimerEntry {
         }
 
         unsafe {
-            self.driver().reregister(tick, self.inner().into());
+            self.driver()
+                .reregister(&self.driver.inner.io_handle, tick, self.inner().into());
         }
     }
 

tokio/src/runtime/time/mod.rs

@@ -107,9 +107,6 @@ struct Inner {
 
     /// True if the driver is being shutdown.
     pub(super) is_shutdown: AtomicBool,
-
-    /// Unparker that can be used to wake the time driver.
-    unpark: IoUnpark,
 }
 
 /// Time state shared which must be protected by a `Mutex`
@@ -137,7 +134,7 @@ impl Driver {
     pub(crate) fn new(park: IoStack, clock: Clock) -> (Driver, Handle) {
         let time_source = TimeSource::new(clock);
 
-        let inner = Inner::new(time_source.clone(), park.unpark());
+        let inner = Inner::new(time_source.clone());
         let handle = Handle::new(Arc::new(inner));
 
         let driver = Driver {
@@ -340,7 +337,12 @@ impl Handle {
     /// driver. No other threads are allowed to concurrently manipulate the
     /// timer at all (the current thread should hold an exclusive reference to
     /// the `TimerEntry`)
-    pub(self) unsafe fn reregister(&self, new_tick: u64, entry: NonNull<TimerShared>) {
+    pub(self) unsafe fn reregister(
+        &self,
+        unpark: &IoUnpark,
+        new_tick: u64,
+        entry: NonNull<TimerShared>,
+    ) {
         let waker = unsafe {
             let mut lock = self.get().lock();
 
@@ -368,7 +370,7 @@ impl Handle {
                             .map(|next_wake| when < next_wake.get())
                             .unwrap_or(true)
                         {
-                            self.inner.unpark.unpark();
+                            unpark.unpark();
                         }
 
                         None
@@ -419,7 +421,7 @@ impl TimerUnpark {
 // ===== impl Inner =====
 
 impl Inner {
-    pub(self) fn new(time_source: TimeSource, unpark: IoUnpark) -> Self {
+    pub(self) fn new(time_source: TimeSource) -> Self {
         Inner {
             state: Mutex::new(InnerState {
                 time_source,
@@ -427,7 +429,6 @@ impl Inner {
                 next_wake: None,
                 wheel: wheel::Wheel::new(),
             }),
-            unpark,
             is_shutdown: AtomicBool::new(false),
         }
     }