From 3f379abda4f980086ab44f4912eff877bdb75120 Mon Sep 17 00:00:00 2001 From: Carl Lerche Date: Wed, 14 Sep 2022 12:45:21 -0700 Subject: [PATCH] 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 | 34 +++++++++++++++++++++++++++------ tokio/src/runtime/time/entry.rs | 3 ++- tokio/src/runtime/time/mod.rs | 17 +++++++++-------- 3 files changed, 39 insertions(+), 15 deletions(-) diff --git a/tokio/src/runtime/driver.rs b/tokio/src/runtime/driver.rs index 5c92fe97a..1d90c7945 100644 --- a/tokio/src/runtime/driver.rs +++ b/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; + 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())) } } diff --git a/tokio/src/runtime/time/entry.rs b/tokio/src/runtime/time/entry.rs index cf90bb117..12ff1202f 100644 --- a/tokio/src/runtime/time/entry.rs +++ b/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()); } } diff --git a/tokio/src/runtime/time/mod.rs b/tokio/src/runtime/time/mod.rs index 07704e61f..798858841 100644 --- a/tokio/src/runtime/time/mod.rs +++ b/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) { + pub(self) unsafe fn reregister( + &self, + unpark: &IoUnpark, + new_tick: u64, + entry: NonNull, + ) { 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), } }