time: cumulative minor improvements (#7358)

tokio/src/runtime/local_runtime/runtime.rs

@@ -376,7 +376,6 @@ impl LocalRuntime {
     }
 }
 
-#[allow(clippy::single_match)] // there are comments in the error branch, so we don't want if-let
 impl Drop for LocalRuntime {
     fn drop(&mut self) {
         if let LocalRuntimeScheduler::CurrentThread(current_thread) = &mut self.scheduler {

tokio/src/runtime/runtime.rs

@@ -471,7 +471,6 @@ impl Runtime {
     }
 }
 
-#[allow(clippy::single_match)] // there are comments in the error branch, so we don't want if-let
 impl Drop for Runtime {
     fn drop(&mut self) {
         match &mut self.scheduler {

tokio/src/runtime/time/entry.rs

@@ -63,6 +63,7 @@ use crate::sync::AtomicWaker;
 use crate::time::Instant;
 use crate::util::linked_list;
 
+use pin_project_lite::pin_project;
 use std::task::{Context, Poll, Waker};
 use std::{marker::PhantomPinned, pin::Pin, ptr::NonNull};
 
@@ -276,29 +277,36 @@ impl StateCell {
     }
 }
 
-/// A timer entry.
+pin_project! {
-///
+    // A timer entry.
-/// This is the handle to a timer that is controlled by the requester of the
+    //
-/// timer. As this participates in intrusive data structures, it must be pinned
+    // This is the handle to a timer that is controlled by the requester of the
-/// before polling.
+    // timer. As this participates in intrusive data structures, it must be pinned
-#[derive(Debug)]
+    // before polling.
-pub(crate) struct TimerEntry {
+    #[derive(Debug)]
-    /// Arc reference to the runtime handle. We can only free the driver after
+    pub(crate) struct TimerEntry {
-    /// deregistering everything from their respective timer wheels.
+        // Arc reference to the runtime handle. We can only free the driver after
-    driver: scheduler::Handle,
+        // deregistering everything from their respective timer wheels.
-    /// Shared inner structure; this is part of an intrusive linked list, and
+        driver: scheduler::Handle,
-    /// therefore other references can exist to it while mutable references to
+        // Shared inner structure; this is part of an intrusive linked list, and
-    /// Entry exist.
+        // therefore other references can exist to it while mutable references to
-    ///
+        // Entry exist.
-    /// This is manipulated only under the inner mutex.
+        //
-    inner: Option<TimerShared>,
+        // This is manipulated only under the inner mutex.
-    /// Deadline for the timer. This is used to register on the first
+        #[pin]
-    /// poll, as we can't register prior to being pinned.
+        inner: Option<TimerShared>,
-    deadline: Instant,
+        // Deadline for the timer. This is used to register on the first
-    /// Whether the deadline has been registered.
+        // poll, as we can't register prior to being pinned.
-    registered: bool,
+        deadline: Instant,
-    /// Ensure the type is !Unpin
+        // Whether the deadline has been registered.
-    _m: std::marker::PhantomPinned,
+        registered: bool,
+    }
+
+    impl PinnedDrop for TimerEntry {
+        fn drop(this: Pin<&mut Self>) {
+            this.cancel();
+        }
+    }
 }
 
 unsafe impl Send for TimerEntry {}
@@ -480,7 +488,6 @@ impl TimerEntry {
             inner: None,
             deadline,
             registered: false,
-            _m: std::marker::PhantomPinned,
         }
     }
 
@@ -488,10 +495,10 @@ impl TimerEntry {
         self.inner.as_ref()
     }
 
-    fn init_inner(&mut self) {
+    fn init_inner(self: Pin<&mut Self>) {
         match self.inner {
             Some(_) => {}
-            None => self.inner = Some(TimerShared::new()),
+            None => self.project().inner.set(Some(TimerShared::new())),
         }
     }
 
@@ -504,7 +511,29 @@ impl TimerEntry {
             return false;
         };
 
-        !inner.state.might_be_registered() && self.registered
+        // Is this timer still in the timer wheel?
+        let deregistered = !inner.might_be_registered();
+
+        // Once the timer has expired,
+        // it will be taken out of the wheel and be fired.
+        //
+        // So if we have already registered the timer into the wheel,
+        // but now it is not in the wheel, it means that it has been
+        // fired.
+        //
+        // +--------------+-----------------+----------+
+        // | deregistered | self.registered |  output  |
+        // +--------------+-----------------+----------+
+        // |     true     |      false      |  false   | <- never been registered
+        // +--------------+-----------------+----------+
+        // |     false    |      false      |  false   | <- never been registered
+        // +--------------+-----------------+----------+
+        // |     true     |      true       |  true    | <- registered into the wheel,
+        // |              |                 |          |    and then taken out of the wheel.
+        // +--------------+-----------------+----------+
+        // |     false    |      true       |  false   | <- still registered in the wheel
+        // +--------------+-----------------+----------+
+        deregistered && self.registered
     }
 
     /// Cancels and deregisters the timer. This operation is irreversible.
@@ -540,16 +569,16 @@ impl TimerEntry {
     }
 
     pub(crate) fn reset(mut self: Pin<&mut Self>, new_time: Instant, reregister: bool) {
-        let this = unsafe { self.as_mut().get_unchecked_mut() };
+        let this = self.as_mut().project();
-        this.deadline = new_time;
+        *this.deadline = new_time;
-        this.registered = reregister;
+        *this.registered = reregister;
 
-        let tick = this.driver().time_source().deadline_to_tick(new_time);
+        let tick = self.driver().time_source().deadline_to_tick(new_time);
-        let inner = match this.inner() {
+        let inner = match self.inner() {
             Some(inner) => inner,
             None => {
-                this.init_inner();
+                self.as_mut().init_inner();
-                this.inner()
+                self.inner()
                     .expect("inner should already be initialized by `this.init_inner()`")
             }
         };
@@ -560,8 +589,8 @@ impl TimerEntry {
 
         if reregister {
             unsafe {
-                this.driver()
+                self.driver()
-                    .reregister(&this.driver.driver().io, tick, inner.into());
+                    .reregister(&self.driver.driver().io, tick, inner.into());
             }
         }
     }
@@ -656,9 +685,3 @@ impl TimerHandle {
         self.inner.as_ref().state.fire(completed_state)
     }
 }
-
-impl Drop for TimerEntry {
-    fn drop(&mut self) {
-        unsafe { Pin::new_unchecked(self) }.as_mut().cancel();
-    }
-}

tokio/src/runtime/time/mod.rs

@@ -246,7 +246,6 @@ impl Driver {
 }
 
 impl Handle {
-    /// Runs timer related logic, and returns the next wakeup time
     pub(self) fn process(&self, clock: &Clock) {
         let now = self.time_source().now(clock);
 

tokio/src/runtime/time/tests/mod.rs

@@ -62,9 +62,7 @@ fn single_timer() {
         let time = handle.inner.driver().time();
         let clock = handle.inner.driver().clock();
 
-        // This may or may not return Some (depending on how it races with the
+        // advance 2s
-        // thread). If it does return None, however, the timer should complete
-        // synchronously.
         time.process_at_time(time.time_source().now(clock) + 2_000_000_000);
 
         jh.join().unwrap();
@@ -170,7 +168,6 @@ fn reset_future() {
 
         let handle = handle.inner.driver().time();
 
-        // This may or may not return a wakeup time.
         handle.process_at_time(
             handle
                 .time_source()