chore: prepare Tokio v1.43.4 (#7821)

(cherry picked from commit 1b17a7e241)

.cirrus.yml

@@ -1,7 +1,7 @@
 only_if: $CIRRUS_TAG == '' && ($CIRRUS_PR != '' || $CIRRUS_BRANCH == 'master' || $CIRRUS_BRANCH =~ 'tokio-.*')
 auto_cancellation: $CIRRUS_BRANCH != 'master' && $CIRRUS_BRANCH !=~ 'tokio-.*'
 freebsd_instance:
-  image_family: freebsd-14-2
+  image_family: freebsd-14-3
 env:
   RUST_STABLE: stable
   RUST_NIGHTLY: nightly-2024-05-05

.github/workflows/ci.yml

@@ -991,10 +991,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-      - name: Install Rust 1.81
+      - name: Install Rust 1.88.0
         uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: 1.81
+          toolchain: 1.88.0
       - name: Install wasm-pack
         uses: taiki-e/install-action@wasm-pack
 
@@ -1014,10 +1014,10 @@ jobs:
           - wasm32-wasip1-threads
     steps:
       - uses: actions/checkout@v4
-      - name: Install Rust ${{ env.rust_stable }}
+      - name: Install Rust 1.88.0
         uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: ${{ env.rust_stable }}
+          toolchain: 1.88.0
           targets: ${{ matrix.target }}
 
       # Install dependencies
@@ -1031,21 +1031,21 @@ jobs:
         run: cargo test -p tokio --target ${{ matrix.target }} --features full
         env:
           CARGO_TARGET_WASM32_WASIP1_RUNNER: "wasmtime run --"
-          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
+          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -W shared-memory=y -S threads=y --"
           RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
 
       - name: WASI test tokio-util full
         run: cargo test -p tokio-util --target ${{ matrix.target }} --features full
         env:
           CARGO_TARGET_WASM32_WASIP1_RUNNER: "wasmtime run --"
-          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
+          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -W shared-memory=y -S threads=y --"
           RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
 
       - name: WASI test tokio-stream
         run: cargo test -p tokio-stream --target ${{ matrix.target }} --features time,net,io-util,sync
         env:
           CARGO_TARGET_WASM32_WASIP1_RUNNER: "wasmtime run --"
-          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
+          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -W shared-memory=y -S threads=y --"
           RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
 
       - name: test tests-integration --features wasi-rt
@@ -1062,7 +1062,7 @@ jobs:
         if: matrix.target == 'wasm32-wasip1-threads'
         working-directory: tests-integration
         env:
-          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -S threads=y --"
+          CARGO_TARGET_WASM32_WASIP1_THREADS_RUNNER: "wasmtime run -W bulk-memory=y -W threads=y -W shared-memory=y -S threads=y --"
           RUSTFLAGS: --cfg tokio_unstable -Dwarnings -C target-feature=+atomics,+bulk-memory -C link-args=--max-memory=67108864
 
   check-external-types:

README.md

@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
 
 ```toml
 [dependencies]
-tokio = { version = "1.43.1", features = ["full"] }
+tokio = { version = "1.43.4", features = ["full"] }
 ```
 Then, on your main.rs:
 

tests-build/tests/fail/macros_type_mismatch.stderr

@@ -25,7 +25,7 @@ error[E0308]: mismatched types
                    found enum `Result<(), _>`
 help: a return type might be missing here
    |
-9  | async fn missing_return_type() -> _ {
+ 9 | async fn missing_return_type() -> _ {
    |                                ++++
 help: consider using `Result::expect` to unwrap the `Result<(), _>` value, panicking if the value is a `Result::Err`
    |

tokio/CHANGELOG.md

@@ -1,3 +1,31 @@
+# 1.43.4 (January 3rd, 2026)
+
+### Fixed
+
+* sync: return `TryRecvError::Disconnected` from `Receiver::try_recv` after `Receiver::close` ([#7686])
+
+[#7686]: https://github.com/tokio-rs/tokio/pull/7686
+
+# 1.43.3 (October 14th, 2025)
+
+### Fixed
+
+- runtime: use release ordering in `wake_by_ref()` even if already woken ([#7622])
+- sync: close the `broadcast::Sender` in `broadcast::Sender::new()` ([#7629])
+- process: fix error when runtime is shut down on nightly-2025-10-12 ([#7672])
+
+[#7622]: https://github.com/tokio-rs/tokio/pull/7622
+[#7629]: https://github.com/tokio-rs/tokio/pull/7629
+[#7672]: https://github.com/tokio-rs/tokio/pull/7672
+
+# 1.43.2 (August 1st, 2025)
+
+### Fixed
+
+- process: fix panic from spurious pidfd wakeup ([#7494])
+
+[#7494]: https://github.com/tokio-rs/tokio/pull/7494
+
 # 1.43.1 (April 5th, 2025)
 
 This release fixes a soundness issue in the broadcast channel. The channel

tokio/Cargo.toml

@@ -6,7 +6,7 @@ name = "tokio"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v1.x.y" git tag.
-version = "1.43.1"
+version = "1.43.4"
 edition = "2021"
 rust-version = "1.70"
 authors = ["Tokio Contributors <[email protected]>"]

tokio/README.md

@@ -56,7 +56,7 @@ Make sure you activated the full features of the tokio crate on Cargo.toml:
 
 ```toml
 [dependencies]
-tokio = { version = "1.43.1", features = ["full"] }
+tokio = { version = "1.43.4", features = ["full"] }
 ```
 Then, on your main.rs:
 

tokio/src/io/poll_evented.rs

@@ -125,7 +125,7 @@ impl<E: Source> PollEvented<E> {
     }
 
     /// Returns a reference to the registration.
-    #[cfg(feature = "net")]
+    #[cfg(any(feature = "net", all(feature = "process", target_os = "linux")))]
     pub(crate) fn registration(&self) -> &Registration {
         &self.registration
     }
@@ -138,14 +138,6 @@ impl<E: Source> PollEvented<E> {
         Ok(inner)
     }
 
-    #[cfg(all(feature = "process", target_os = "linux"))]
-    pub(crate) fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
-        self.registration
-            .poll_read_ready(cx)
-            .map_err(io::Error::from)
-            .map_ok(|_| ())
-    }
-
     /// Re-register under new runtime with `interest`.
     #[cfg(all(feature = "process", target_os = "linux"))]
     pub(crate) fn reregister(&mut self, interest: Interest) -> io::Result<()> {

tokio/src/process/unix/pidfd_reaper.rs

@@ -19,7 +19,7 @@ use std::{
     pin::Pin,
     process::ExitStatus,
     sync::atomic::{AtomicBool, Ordering::Relaxed},
-    task::{ready, Context, Poll},
+    task::{Context, Poll},
 };
 
 #[derive(Debug)]
@@ -95,14 +95,41 @@ where
     pidfd: PollEvented<Pidfd>,
 }
 
+fn display_eq(d: impl std::fmt::Display, s: &str) -> bool {
+    use std::fmt::Write;
+
+    struct FormatEq<'r> {
+        remainder: &'r str,
+        unequal: bool,
+    }
+
+    impl<'r> Write for FormatEq<'r> {
+        fn write_str(&mut self, s: &str) -> std::fmt::Result {
+            if !self.unequal {
+                if let Some(new_remainder) = self.remainder.strip_prefix(s) {
+                    self.remainder = new_remainder;
+                } else {
+                    self.unequal = true;
+                }
+            }
+            Ok(())
+        }
+    }
+
+    let mut fmt_eq = FormatEq {
+        remainder: s,
+        unequal: false,
+    };
+    let _ = write!(fmt_eq, "{d}");
+    fmt_eq.remainder.is_empty() && !fmt_eq.unequal
+}
+
 #[allow(deprecated)]
 fn is_rt_shutdown_err(err: &io::Error) -> bool {
     if let Some(inner) = err.get_ref() {
-        // Using `Error::description()` is more efficient than `format!("{inner}")`,
-        // so we use it here even if it is deprecated.
         err.kind() == io::ErrorKind::Other
             && inner.source().is_none()
-            && inner.description() == RUNTIME_SHUTTING_DOWN_ERROR
+            && display_eq(inner, RUNTIME_SHUTTING_DOWN_ERROR)
     } else {
         false
     }
@@ -117,17 +144,21 @@ where
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = Pin::into_inner(self);
 
-        match ready!(this.pidfd.poll_read_ready(cx)) {
-            Err(err) if is_rt_shutdown_err(&err) => {
-                this.pidfd.reregister(Interest::READABLE)?;
-                ready!(this.pidfd.poll_read_ready(cx))?
+        match this.pidfd.registration().poll_read_ready(cx) {
+            Poll::Ready(Ok(evt)) => {
+                if let Some(exit_code) = this.inner.try_wait()? {
+                    return Poll::Ready(Ok(exit_code));
+                }
+                this.pidfd.registration().clear_readiness(evt);
             }
-            res => res?,
-        }
-        Poll::Ready(Ok(this
-            .inner
-            .try_wait()?
-            .expect("pidfd is ready to read, the process should have exited")))
+            Poll::Ready(Err(err)) if is_rt_shutdown_err(&err) => {}
+            Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+            Poll::Pending => return Poll::Pending,
+        };
+
+        this.pidfd.reregister(Interest::READABLE)?;
+        cx.waker().wake_by_ref();
+        Poll::Pending
     }
 }
 

tokio/src/runtime/task/state.rs

@@ -252,9 +252,15 @@ impl State {
     /// Transitions the state to `NOTIFIED`.
     pub(super) fn transition_to_notified_by_ref(&self) -> TransitionToNotifiedByRef {
         self.fetch_update_action(|mut snapshot| {
-            if snapshot.is_complete() || snapshot.is_notified() {
-                // There is nothing to do in this case.
+            if snapshot.is_complete() {
+                // The complete state is final
                 (TransitionToNotifiedByRef::DoNothing, None)
+            } else if snapshot.is_notified() {
+                // Even hough we have nothing to do in this branch,
+                // wake_by_ref() should synchronize-with the task starting execution,
+                // therefore we must use an Release store (with the same value),
+                // to pair with the Acquire in transition_to_running.
+                (TransitionToNotifiedByRef::DoNothing, Some(snapshot))
             } else if snapshot.is_running() {
                 // If the task is running, we mark it as notified, but we should
                 // not submit as the thread currently running the future is

tokio/src/sync/broadcast.rs

@@ -522,7 +522,7 @@ impl<T> Sender<T> {
             tail: Mutex::new(Tail {
                 pos: 0,
                 rx_cnt: receiver_count,
-                closed: false,
+                closed: receiver_count == 0,
                 waiters: LinkedList::new(),
             }),
             num_tx: AtomicUsize::new(1),

tokio/src/sync/mpsc/chan.rs

@@ -439,6 +439,10 @@ impl<T, S: Semaphore> Rx<T, S> {
                             return Ok(value);
                         }
                         TryPopResult::Closed => return Err(TryRecvError::Disconnected),
+                        // If close() was called, an empty queue should report Disconnected.
+                        TryPopResult::Empty if rx_fields.rx_closed => {
+                            return Err(TryRecvError::Disconnected)
+                        }
                         TryPopResult::Empty => return Err(TryRecvError::Empty),
                         TryPopResult::Busy => {} // fall through
                     }

tokio/src/sync/mpsc/list.rs

@@ -35,8 +35,14 @@ pub(crate) enum TryPopResult<T> {
     /// Successfully popped a value.
     Ok(T),
     /// The channel is empty.
+    ///
+    /// Note that `list.rs` only tracks the close state set by senders. If the
+    /// channel is closed by `Rx::close()`, then `TryPopResult::Empty` is still
+    /// returned, and the close state needs to be handled by `chan.rs`.
     Empty,
     /// The channel is empty and closed.
+    ///
+    /// Returned when the send half is closed (all senders dropped).
     Closed,
     /// The channel is not empty, but the first value is being written.
     Busy,

tokio/tests/process_issue_7144.rs

@@ -0,0 +1,28 @@
+#![cfg(feature = "process")]
+#![warn(rust_2018_idioms)]
+#![cfg(target_os = "linux")]
+#![cfg(not(miri))]
+
+use tokio::process::Command;
+use tokio::time::{sleep, Duration};
+
+#[tokio::test]
+async fn issue_7144() {
+    let mut threads = vec![];
+    for _ in 0..20 {
+        threads.push(tokio::spawn(test_one()));
+    }
+    for thread in threads {
+        thread.await.unwrap();
+    }
+}
+
+async fn test_one() {
+    let mut t = Command::new("strace")
+        .args("-o /dev/null -D sleep 5".split(' '))
+        .spawn()
+        .unwrap();
+    sleep(Duration::from_millis(100)).await;
+    unsafe { libc::kill(t.id().unwrap() as _, libc::SIGINT) };
+    t.wait().await.unwrap();
+}

tokio/tests/sync_broadcast.rs

@@ -7,9 +7,7 @@ use wasm_bindgen_test::wasm_bindgen_test as test;
 
 use tokio::sync::broadcast;
 use tokio_test::task;
-use tokio_test::{
-    assert_err, assert_ok, assert_pending, assert_ready, assert_ready_err, assert_ready_ok,
-};
+use tokio_test::{assert_err, assert_ok, assert_pending, assert_ready_err, assert_ready_ok};
 
 use std::sync::Arc;
 
@@ -656,3 +654,15 @@ async fn receiver_recv_is_cooperative() {
         _ = tokio::task::yield_now() => {},
     }
 }
+
+#[tokio::test]
+async fn broadcast_sender_new_must_be_closed() {
+    let capacity = 1;
+    let tx: broadcast::Sender<()> = broadcast::Sender::new(capacity);
+
+    tx.send(()).unwrap_err();
+
+    let _rx = tx.subscribe();
+
+    tx.send(()).unwrap();
+}

tokio/tests/sync_mpsc.rs

@@ -966,6 +966,15 @@ fn try_recv_unbounded() {
     }
 }
 
+#[test]
+fn try_recv_after_receiver_close() {
+    let (_tx, mut rx) = mpsc::channel::<()>(5);
+
+    assert_eq!(Err(TryRecvError::Empty), rx.try_recv());
+    rx.close();
+    assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv());
+}
+
 #[test]
 fn try_recv_close_while_empty_bounded() {
     let (tx, mut rx) = mpsc::channel::<()>(5);