time: positively check the deadline before reading the state of TimerEntry in Sleep

tokio/src/runtime/time/entry.rs

@@ -577,7 +577,6 @@ impl TimerEntry {
  self.driver.driver().time()
  }
 
- #[cfg(all(tokio_unstable, feature = "tracing"))]
  pub(crate) fn clock(&self) -> &super::Clock {
  self.driver.driver().clock()
  }

tokio/src/time/sleep.rs

@@ -412,6 +412,19 @@ impl Sleep {
  #[cfg(any(not(tokio_unstable), not(feature = "tracing")))]
  let coop = ready!(crate::runtime::coop::poll_proceed(cx));
 
+ let time_source = &me.entry.driver().time_source();
+ let now_tick = time_source.now(me.entry.clock());
+ let deadline_tick = time_source.deadline_to_tick(me.entry.deadline());
+
+ // If the deadline has been reached, then we return here without checking the state of TimerEntry.
+ // This ensures that even if the state of TimerEntry is not maintained by the Timer Driver,
+ // it can expire as expected.
+ // See <https://github.com/tokio-rs/tokio/issues/6545>
+ if deadline_tick <= now_tick {
+ me.entry.cancel();
+ return Poll::Ready(Ok(()));
+ }
+
  let result = me.entry.poll_elapsed(cx).map(move |r| {
  coop.made_progress();
  r

tokio/tests/rt_metrics.rs

@@ -137,15 +137,15 @@ fn worker_park_count() {
  let rt = current_thread();
  let metrics = rt.metrics();
  rt.block_on(async {
- time::sleep(Duration::from_millis(1)).await;
+ time::sleep(Duration::from_millis(100)).await;
  });
  drop(rt);
  assert!(1 <= metrics.worker_park_count(0));
 
  let rt = threaded();
  let metrics = rt.metrics();
  rt.block_on(async {
- time::sleep(Duration::from_millis(1)).await;
+ time::sleep(Duration::from_millis(100)).await;
  });
  drop(rt);
  assert!(1 <= metrics.worker_park_count(0));
@@ -160,15 +160,15 @@ fn worker_noop_count() {
  let rt = current_thread();
  let metrics = rt.metrics();
  rt.block_on(async {
- time::sleep(Duration::from_millis(1)).await;
+ time::sleep(Duration::from_millis(100)).await;
  });
  drop(rt);
  assert!(0 < metrics.worker_noop_count(0));
 
  let rt = threaded();
  let metrics = rt.metrics();
  rt.block_on(async {
- time::sleep(Duration::from_millis(1)).await;
+ time::sleep(Duration::from_millis(100)).await;
  });
  drop(rt);
  assert!(0 < metrics.worker_noop_count(0));

tokio/tests/time_rt.rs

@@ -88,3 +88,35 @@ async fn timeout_value() {
  assert!(res.is_err());
  assert!(Instant::now() >= now + dur);
 }
+
+#[cfg(not(target_os = "wasi"))] // Wasi doesn't support threads
+#[test]
+fn ready_in_busy_loop() {
+ async fn never_pending() {}
+
+ let handle = std::thread::spawn(|| {
+ tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .unwrap()
+ .block_on(async {
+ tokio::spawn(async move {
+ let mut timer = tokio::time::interval(tokio::time::Duration::from_millis(2));
+ loop {
+ tokio::select! {
+ _= timer.tick() => {
+ break;
+ }
+ _ = never_pending() => {}
+ }
+ }
+ })
+ .await
+ .unwrap();
+ });
+ });
+
+ std::thread::sleep(std::time::Duration::from_secs(2));
+
+ assert!(handle.is_finished());
+}