time: delay the cancellation of timers

.github/workflows/loom.yml

@@ -52,7 +52,7 @@ jobs:
             toolchain: ${{ env.rust_stable }}
       - uses: Swatinem/rust-cache@v2
       - name: run tests
-        run: cargo test --lib --release --features full -- --nocapture runtime::time::tests
+        run: cargo test --lib --release --features full -- --nocapture runtime::time
         working-directory: tokio
 
   loom-current-thread:

spellcheck.dic

@@ -1,4 +1,4 @@
-307
+309
 &
 +
 <
@@ -163,6 +163,7 @@ Lauck
 libc
 lifecycle
 lifo
+LLVM
 lookups
 macOS
 MacOS
@@ -306,3 +307,4 @@ Wakers
 wakeup
 wakeups
 workstealing
+ZST

tokio-util/tests/time_delay_queue.rs

@@ -3,7 +3,7 @@
 #![cfg(feature = "full")]
 
 use futures::StreamExt;
-use tokio::time::{self, sleep, sleep_until, Duration, Instant};
+use tokio::time::{self, sleep, Duration, Instant};
 use tokio_test::{assert_pending, assert_ready, task};
 use tokio_util::time::DelayQueue;
 
@@ -82,8 +82,6 @@ async fn single_short_delay() {
 
     sleep(ms(5)).await;
 
-    assert!(queue.is_woken());
-
     let entry = assert_ready_some!(poll!(queue));
     assert_eq!(*entry.get_ref(), "foo");
 
@@ -221,7 +219,7 @@ async fn reset_much_later() {
 
     sleep(ms(20)).await;
 
-    assert!(queue.is_woken());
+    assert_ready_some!(poll!(queue));
 }
 
 // Reproduces tokio-rs/tokio#849.
@@ -248,7 +246,7 @@ async fn reset_twice() {
 
     sleep(ms(20)).await;
 
-    assert!(queue.is_woken());
+    assert_ready_some!(poll!(queue));
 }
 
 /// Regression test: Given an entry inserted with a deadline in the past, so
@@ -412,8 +410,6 @@ async fn expire_first_key_when_reset_to_expire_earlier() {
 
     sleep(ms(100)).await;
 
-    assert!(queue.is_woken());
-
     let entry = assert_ready_some!(poll!(queue)).into_inner();
     assert_eq!(entry, "one");
 }
@@ -435,8 +431,6 @@ async fn expire_second_key_when_reset_to_expire_earlier() {
 
     sleep(ms(100)).await;
 
-    assert!(queue.is_woken());
-
     let entry = assert_ready_some!(poll!(queue)).into_inner();
     assert_eq!(entry, "two");
 }
@@ -457,8 +451,6 @@ async fn reset_first_expiring_item_to_expire_later() {
     queue.reset_at(&one, now + ms(300));
     sleep(ms(250)).await;
 
-    assert!(queue.is_woken());
-
     let entry = assert_ready_some!(poll!(queue)).into_inner();
     assert_eq!(entry, "two");
 }
@@ -522,43 +514,6 @@ async fn insert_after_ready_poll() {
     assert_eq!("3", res[2]);
 }
 
-#[tokio::test]
-async fn reset_later_after_slot_starts() {
-    time::pause();
-
-    let mut queue = task::spawn(DelayQueue::new());
-
-    let now = Instant::now();
-
-    let foo = queue.insert_at("foo", now + ms(100));
-
-    assert_pending!(poll!(queue));
-
-    sleep_until(now + Duration::from_millis(80)).await;
-
-    assert!(!queue.is_woken());
-
-    // At this point the queue hasn't been polled, so `elapsed` on the wheel
-    // for the queue is still at 0 and hence the 1ms resolution slots cover
-    // [0-64).  Resetting the time on the entry to 120 causes it to get put in
-    // the [64-128) slot.  As the queue knows that the first entry is within
-    // that slot, but doesn't know when, it must wake immediately to advance
-    // the wheel.
-    queue.reset_at(&foo, now + ms(120));
-    assert!(queue.is_woken());
-
-    assert_pending!(poll!(queue));
-
-    sleep_until(now + Duration::from_millis(119)).await;
-    assert!(!queue.is_woken());
-
-    sleep(ms(1)).await;
-    assert!(queue.is_woken());
-
-    let entry = assert_ready_some!(poll!(queue)).into_inner();
-    assert_eq!(entry, "foo");
-}
-
 #[tokio::test]
 async fn reset_inserted_expired() {
     time::pause();
@@ -584,43 +539,6 @@ async fn reset_inserted_expired() {
     assert_eq!(queue.len(), 0);
 }
 
-#[tokio::test]
-async fn reset_earlier_after_slot_starts() {
-    time::pause();
-
-    let mut queue = task::spawn(DelayQueue::new());
-
-    let now = Instant::now();
-
-    let foo = queue.insert_at("foo", now + ms(200));
-
-    assert_pending!(poll!(queue));
-
-    sleep_until(now + Duration::from_millis(80)).await;
-
-    assert!(!queue.is_woken());
-
-    // At this point the queue hasn't been polled, so `elapsed` on the wheel
-    // for the queue is still at 0 and hence the 1ms resolution slots cover
-    // [0-64).  Resetting the time on the entry to 120 causes it to get put in
-    // the [64-128) slot.  As the queue knows that the first entry is within
-    // that slot, but doesn't know when, it must wake immediately to advance
-    // the wheel.
-    queue.reset_at(&foo, now + ms(120));
-    assert!(queue.is_woken());
-
-    assert_pending!(poll!(queue));
-
-    sleep_until(now + Duration::from_millis(119)).await;
-    assert!(!queue.is_woken());
-
-    sleep(ms(1)).await;
-    assert!(queue.is_woken());
-
-    let entry = assert_ready_some!(poll!(queue)).into_inner();
-    assert_eq!(entry, "foo");
-}
-
 #[tokio::test]
 async fn insert_in_past_after_poll_fires_immediately() {
     time::pause();

tokio/src/macros/cfg.rs

@@ -711,3 +711,27 @@ macro_rules! cfg_io_uring {
         )*
     };
 }
+
+macro_rules! cfg_rt_and_time{
+    ($($item:item)*) => {
+        $(
+            #[cfg(all(
+                feature = "rt",
+                feature = "time",
+            ))]
+            $item
+        )*
+    };
+}
+
+macro_rules! cfg_rt_or_time{
+    ($($item:item)*) => {
+        $(
+            #[cfg(any(
+                feature = "rt",
+                feature = "time",
+            ))]
+            $item
+        )*
+    };
+}

tokio/src/runtime/driver.rs

@@ -113,6 +113,13 @@ impl Handle {
                 .expect("A Tokio 1.x context was found, but timers are disabled. Call `enable_time` on the runtime builder to enable timers.")
         }
 
+        pub(crate) fn with_time<F, R>(&self, f: F) -> R
+        where
+            F: FnOnce(Option<&crate::runtime::time::Handle>) -> R,
+        {
+            f(self.time.as_ref())
+        }
+
         pub(crate) fn clock(&self) -> &Clock {
             &self.clock
         }