time: wake DelayQueue when removing last item (#6752)

tokio-util/src/time/delay_queue.rs

@@ -766,6 +766,12 @@ impl<T> DelayQueue<T> {
             }
         }
 
+        if self.slab.is_empty() {
+            if let Some(waker) = self.waker.take() {
+                waker.wake();
+            }
+        }
+
         Expired {
             key: Key::new(key.index),
             data: data.inner,

tokio-util/tests/time_delay_queue.rs

@@ -880,6 +880,19 @@ async fn peek() {
     assert!(queue.peek().is_none());
 }
 
+#[tokio::test(start_paused = true)]
+async fn wake_after_remove_last() {
+    let mut queue = task::spawn(DelayQueue::new());
+    let key = queue.insert("foo", ms(1000));
+
+    assert_pending!(poll!(queue));
+
+    queue.remove(&key);
+
+    assert!(queue.is_woken());
+    assert!(assert_ready!(poll!(queue)).is_none());
+}
+
 fn ms(n: u64) -> Duration {
     Duration::from_millis(n)
 }