tokio - cargo Package Compare versions

Comparing version 1.53.0 to 1.53.1

.cargo_vcs_info.json

 {
   "git": {
-    "sha1": "be689a35f5ade5a39e507f79d3ec85cdab27806f"
+    "sha1": "75fef53d0a8590c2d1dbb63672aa7b7d1ef51155"
   },
   "path_in_vcs": "tokio"
 }

Cargo.lock

@@ -1088,5 +1088,5 @@ # This file is automatically @generated by Cargo.
 name = "tokio"
-version = "1.52.4"
+version = "1.53.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "317fafbbe3f02fc663dad00ea6186197de963cd4190e86a26d8d0fae095539af"
+checksum = "d988bcd52dbe076d3d46903332f58c912b87a2c49b1428419a5845154762ffee"
 dependencies = [
@@ -1098,3 +1098,3 @@ "pin-project-lite",
 name = "tokio"
-version = "1.53.0"
+version = "1.53.1"
 dependencies = [
@@ -1151,3 +1151,3 @@ "async-stream",
  "pin-project-lite",
- "tokio 1.52.4",
+ "tokio 1.53.0",
 ]
@@ -1162,3 +1162,3 @@
  "futures-core",
- "tokio 1.52.4",
+ "tokio 1.53.0",
  "tokio-stream",
@@ -1178,3 +1178,3 @@ ]
  "pin-project-lite",
- "tokio 1.52.4",
+ "tokio 1.53.0",
 ]
@@ -1181,0 +1181,0 @@

Cargo.toml

@@ -16,3 +16,3 @@ # THIS FILE IS AUTOMATICALLY GENERATED BY CARGO
 name = "tokio"
-version = "1.53.0"
+version = "1.53.1"
 authors = ["Tokio Contributors <[e-mail]>"]
@@ -19,0 +19,0 @@ build = false

README.md

@@ -63,3 +63,3 @@ *[TokioConf 2026 program and tickets are now available!](https://tokioconf.com)*
 [dependencies]
-tokio = { version = "1.53.0", features = ["full"] }
+tokio = { version = "1.53.1", features = ["full"] }
 ```
@@ -66,0 +66,0 @@ Then, on your main.rs:

src/runtime/runtime.rs

@@ -337,4 +337,2 @@ use super::BOX_FUTURE_THRESHOLD;
     /// ```
-    ///
-    /// [handle]: fn@Handle::block_on
     #[track_caller]
@@ -341,0 +339,0 @@ pub fn block_on<F: Future>(&self, future: F) -> F::Output {

src/runtime/time_alt/entry.rs

@@ -213,3 +213,4 @@ use super::cancellation_queue::Sender;
 
-    pub(crate) fn register_cancel_tx(&self, cancel_tx: Sender) {
+    /// Returns `false` if the `self` has already been cancelled or woken up.
+    pub(crate) fn register_cancel_tx(&self, cancel_tx: Sender) -> bool {
         let mut lock = self.entry.state.lock();
@@ -220,2 +221,5 @@ if !lock.cancelled && !lock.woken_up {
             assert!(old_tx.is_none(), "cancel_tx is already registered");
+            true
+        } else {
+            false
         }
@@ -222,0 +226,0 @@ }

src/runtime/time_alt/tests.rs

@@ -15,4 +15,8 @@ use super::*;
 fn new_handle() -> (EntryHandle, AwokenCount) {
+    new_handle_with_deadline(0)
+}
+
+fn new_handle_with_deadline(deadline: u64) -> (EntryHandle, AwokenCount) {
     let (waker, count) = new_count_waker();
-    let entry = EntryHandle::new(0);
+    let entry = EntryHandle::new(deadline);
     _ = entry.poll(&mut Context::from_waker(&waker));
@@ -70,3 +74,3 @@ (entry, count)
             let (hdl, count) = new_handle();
-            hdl.register_cancel_tx(cancel_tx.clone());
+            assert!(hdl.register_cancel_tx(cancel_tx.clone()));
             counts.push(count);
@@ -97,2 +101,95 @@ unsafe {
 #[test]
+fn insert_of_already_cancelled_entry_does_not_enter_wheel() {
+    model(|| {
+        let (cancel_tx, mut cancel_rx) = cancellation_queue::new();
+        let mut wheel = Wheel::new();
+
+        // do not expire during the test
+        let far_future = 10_000_000;
+        let (hdl, awoken_count) = new_handle_with_deadline(far_future);
+
+        // cancel the timer before inserting it into the wheel
+        hdl.cancel();
+        assert!(hdl.is_cancelled());
+
+        // try to insert the cancelled entry into the wheel
+        unsafe {
+            wheel.insert(hdl, cancel_tx);
+        }
+
+        // a cancelled entry should not be inserted into the wheel
+        assert!(
+            wheel.next_expiration_time().is_none(),
+            "an already-cancelled entry leaked into the wheel on insert"
+        );
+
+        // It also must not have been queued for cancellation removal, since
+        // it was never actually placed in the wheel for `remove` to find.
+        assert_eq!(cancel_rx.recv_all().count(), 0);
+        assert_eq!(awoken_count.get(), 0);
+
+        // drain the wheel unconditionally, otherwise loom will complain
+        // about the leaked entry, which confuses developers in case
+        // this test fails for some other reason.
+        let mut wake_queue = WakeQueue::new();
+        wheel.take_expired(u64::MAX, &mut wake_queue);
+        wake_queue.wake_all();
+    });
+}
+
+#[test]
+fn cancel_races_with_insert() {
+    model(|| {
+        let (cancel_tx, mut cancel_rx) = cancellation_queue::new();
+        let mut wheel = Wheel::new();
+
+        // do not expire during the test
+        let far_future = 10_000_000;
+        let (hdl, count) = new_handle_with_deadline(far_future);
+
+        let hdl2 = hdl.clone();
+        let jh = thread::spawn(move || {
+            // cancel the timer concurrently with insertion into the wheel
+            hdl2.cancel();
+        });
+
+        // try to insert the entry into the wheel concurrently with cancellation
+        unsafe {
+            wheel.insert(hdl, cancel_tx);
+        }
+
+        // ensure the cancellation thread has exited
+        jh.join().unwrap();
+
+        // Whichever way the race went, the entry must end up in exactly one
+        // of two consistent end states -- never "in the wheel with no way
+        // to ever remove it again":
+        //
+        // (a) `cancel()` won the race and ran first: `insert` sees it's
+        //     already cancelled and refuses to add it to the wheel.
+        // (b) `insert` won the race and registered `cancel_tx` first:
+        //     `cancel()` then finds `cancel_tx` and pushes the entry into
+        //     the cancellation queue for later removal from the wheel.
+        //
+        // Either way, the entry is not left stuck in the wheel with no
+        // corresponding entry in the cancellation queue.
+        let cancelled_via_queue = cancel_rx.recv_all().count();
+        let leaked_in_wheel = wheel.next_expiration_time().is_some();
+
+        assert!(
+            !leaked_in_wheel || cancelled_via_queue == 1,
+            "entry is stuck in the wheel with no way to ever be removed"
+        );
+        assert_eq!(count.get(), 0, "a cancelled entry must never be woken");
+
+        // drain the wheel unconditionally, otherwise loom will complain
+        // about the leaked entry, which confuses developers in case
+        // this test fails for some other reason.
+        let mut wake_queue = WakeQueue::new();
+        wheel.take_expired(u64::MAX, &mut wake_queue);
+        wake_queue.wake_all();
+    });
+}
+
+#[test]
 fn wake_up_in_the_different_thread() {
@@ -143,3 +240,3 @@ model(|| {
             let (hdl, count) = new_handle();
-            hdl.register_cancel_tx(cancel_tx.clone());
+            assert!(hdl.register_cancel_tx(cancel_tx.clone()));
             counts.push(count);
@@ -146,0 +243,0 @@ hdls.push(hdl.clone());

src/runtime/time_alt/wheel/mod.rs

@@ -57,3 +57,3 @@ mod level;
 
-    /// Inserts an entry into the timing wheel.
+    /// Inserts an entry into the timing wheel
     ///
@@ -74,3 +74,6 @@ /// # Arguments
 
-        hdl.register_cancel_tx(cancel_tx);
+        if !hdl.register_cancel_tx(cancel_tx) {
+            // `hdl` has been cancelled or woken up concurrently
+            return;
+        }
 
@@ -77,0 +80,0 @@ // Get the level at which the entry should be stored

src/signal/windows/sys.rs

@@ -60,11 +60,16 @@ use std::io;
 fn new(signal: SignalKind) -> io::Result<RxFuture> {
-    let registry = REGISTRY
+    // Initialize the registry BEFORE registering the OS handler: the
+    // handler thread can then always observe an initialized `REGISTRY`
+    // (`SetConsoleCtrlHandler` happens-after the initialization below),
+    // so it needs no blocking wait.
+    let registry = REGISTRY.get_or_init(Registry::default);
+
+    HANDLER_RESULT
         .get_or_init(
             || match unsafe { console::SetConsoleCtrlHandler(Some(handler), 1) } {
                 0 => Err(Error::last_os_error().raw_os_error().expect("unreachable")),
-                _ => Ok(Registry::default()),
+                _ => Ok(()),
             },
         )
-        .as_ref()
-        .map_err(|&code| Error::from_raw_os_error(code))?;
+        .map_err(Error::from_raw_os_error)?;
 
@@ -98,4 +103,8 @@ let rx = registry[signal].subscribe();
 
-static REGISTRY: OnceLock<Result<Registry, i32>> = OnceLock::new();
+static REGISTRY: OnceLock<Registry> = OnceLock::new();
 
+/// Whether `SetConsoleCtrlHandler` succeeded, initialized (once) only
+/// after `REGISTRY` — see `new` for the ordering argument.
+static HANDLER_RESULT: OnceLock<Result<(), i32>> = OnceLock::new();
+
 unsafe extern "system" fn handler(ty: u32) -> BOOL {
@@ -112,7 +121,9 @@ let signal = match ty {
 
-    // Note that `OnceLock::get` does not handle the small window between calling
-    // `SetConsoleCtrlHandler` and `REGISTRY` being initialized.
-    let Ok(registry) = REGISTRY.wait().as_ref() else {
-        // Technically unreachable since `handler` is only called if
-        // `SetConsoleCtrlHandler` succeeded.
+    // `new` initializes `REGISTRY` before it registers this handler with
+    // the OS, so an invoked handler always finds it initialized —
+    // `get()` suffices and no blocking wait is needed (using `get`
+    // also keeps the crate's MSRV: `OnceLock::wait` needs Rust 1.86).
+    let Some(registry) = REGISTRY.get() else {
+        // Unreachable by the ordering above; kept as a defensive
+        // fallback that lets the OS run the next handler.
         return 0;
@@ -119,0 +130,0 @@ };

No alert changes

No metric changes