chore: prepare Tokio v1.47.5 (#8122)

Co-authored-by: Mattia Pitossi <[email protected]>

.cirrus.yml

@@ -1,58 +0,0 @@
-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
-env:
-  RUST_STABLE: stable
-  RUST_NIGHTLY: nightly-2025-01-25
-  RUSTFLAGS: -D warnings
-
-# Test FreeBSD in a full VM on cirrus-ci.com.  Test the i686 target too, in the
-# same VM.  The binary will be built in 32-bit mode, but will execute on a
-# 64-bit kernel and in a 64-bit environment.  Our tests don't execute any of
-# the system's binaries, so the environment shouldn't matter.
-task:
-  name: FreeBSD 64-bit
-  setup_script:
-    - pkg install -y bash
-    - curl https://sh.rustup.rs -sSf --output rustup.sh
-    - sh rustup.sh -y --profile minimal --default-toolchain $RUST_STABLE
-    - . $HOME/.cargo/env
-    - |
-      echo "~~~~ rustc --version ~~~~"
-      rustc --version
-  test_script:
-    - . $HOME/.cargo/env
-    - cargo test --all --all-features
-
-task:
-  name: FreeBSD docs
-  env:
-    RUSTFLAGS: --cfg docsrs --cfg tokio_unstable
-    RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable -Dwarnings
-  setup_script:
-    - pkg install -y bash
-    - curl https://sh.rustup.rs -sSf --output rustup.sh
-    - sh rustup.sh -y --profile minimal --default-toolchain $RUST_NIGHTLY
-    - . $HOME/.cargo/env
-    - |
-      echo "~~~~ rustc --version ~~~~"
-      rustc --version
-  test_script:
-    - . $HOME/.cargo/env
-    - cargo doc --lib --no-deps --all-features --document-private-items
-
-task:
-  name: FreeBSD 32-bit
-  setup_script:
-    - pkg install -y bash
-    - curl https://sh.rustup.rs -sSf --output rustup.sh
-    - sh rustup.sh -y --profile minimal --default-toolchain $RUST_STABLE
-    - . $HOME/.cargo/env
-    - rustup target add i686-unknown-freebsd
-    - |
-      echo "~~~~ rustc --version ~~~~"
-      rustc --version
-  test_script:
-    - . $HOME/.cargo/env
-    - cargo test --all --all-features --target i686-unknown-freebsd

.github/workflows/ci.yml

@@ -1038,10 +1038,10 @@ jobs:
     runs-on: ubuntu-latest
     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
       - name: Install wasm-pack
         uses: taiki-e/install-action@wasm-pack
 
@@ -1061,10 +1061,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
@@ -1078,21 +1078,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
@@ -1109,7 +1109,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:
@@ -1231,3 +1231,74 @@ jobs:
               echo 'Please remove trailing whitespace from these lines.'
               exit 1
           fi
+
+  freebsd-x86_64:
+    name: FreeBSD x86_64
+    needs: basics
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+      - name: Test in FreeBSD
+        uses: vmactions/freebsd-vm@v1
+        with:
+          release: '14.4'
+          envs: "RUSTFLAGS"
+          prepare: |
+            pkg install -y curl
+            curl https://sh.rustup.rs -sSf --output rustup.sh
+            sh rustup.sh -y --profile minimal --default-toolchain ${{ env.rust_stable }}
+          run: |
+            . $HOME/.cargo/env
+            cargo test --workspace --features full,test-util
+            # Enable all unstable features except `io_uring` and `taskdump`,
+            # which are Linux-only features.
+            RUSTFLAGS="$RUSTFLAGS --cfg tokio_unstable" \
+            cargo test \
+                --features full,test-util,tracing
+
+  freebsd-docs:
+    name: FreeBSD docs
+    needs: basics
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+      - name: Test in FreeBSD
+        uses: vmactions/freebsd-vm@v1
+        env:
+          RUSTFLAGS: --cfg docsrs --cfg tokio_unstable
+          RUSTDOCFLAGS: --cfg docsrs --cfg tokio_unstable -Dwarnings
+        with:
+          release: '14.4'
+          envs: "RUST_NIGHTLY RUSTDOCFLAGS RUSTFLAGS"
+          prepare: |
+            pkg install -y curl
+            curl https://sh.rustup.rs -sSf --output rustup.sh
+            sh rustup.sh -y --profile minimal --default-toolchain ${{ env.rust_nightly }}
+          run: |
+            . $HOME/.cargo/env
+            # We use `--features full,test-util,io-uring,tracing` instead of
+            # `--all-features` to exclude `taskdump` and `io_uring`, which are Linux-only
+            # features.
+            cargo doc --lib --no-deps --features full,test-util,tracing \
+              --document-private-items
+
+  freebsd-i686:
+    name: FreeBSD i686
+    needs: basics
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+      - name: Test in FreeBSD
+        uses: vmactions/freebsd-vm@v1
+        with:
+          release: '14.4'
+          envs: "RUSTFLAGS"
+          prepare: |
+            pkg install -y curl
+            curl https://sh.rustup.rs -sSf --output rustup.sh
+            sh rustup.sh -y --profile minimal --default-toolchain ${{ env.rust_stable }}
+          run: |
+            . $HOME/.cargo/env
+            rustup target add i686-unknown-freebsd
+            cargo test --workspace --features full,test-util \
+              --target i686-unknown-freebsd

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.47.1", features = ["full"] }
+tokio = { version = "1.47.5", features = ["full"] }
 ```
 Then, on your main.rs:
 

netlify.toml

@@ -1,6 +1,15 @@
 [build]
+  # TODO: unfreeze toolchain
+  # error[E0557]: feature has been removed
+  #   --> /opt/buildhome/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/lock_api-0.4.13/src/lib.rs:89:29
+  #    |
+  # 89 | #![cfg_attr(docsrs, feature(doc_auto_cfg))]
+  #    |                             ^^^^^^^^^^^^ feature has been removed
+  #    |
+  #    = note: removed in 1.58.0; see <https://github.com/rust-lang/rust/pull/138907; for more information
+  #    = note: merged into `doc_cfg`
   command = """
-    rustup install nightly --profile minimal && cargo doc --no-deps --all-features
+    rustup install nightly-2025-01-25 --profile minimal && cargo doc --no-deps --all-features
     """
   publish = "target/doc"
 

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,45 @@
+# 1.47.5 (May 7th, 2026)
+
+### Fixed
+
+* sync: fix underflow in mpsc channel `len()` ([#8062])
+* sync: notify receivers in mpsc `OwnedPermit::release()` method ([#8075])
+* sync: require that an `RwLock` has `max_readers != 0` ([#8076])
+* sync: return `Empty` from `try_recv()` when mpsc is closed with outstanding permits ([#8074])
+
+[#8062]: https://github.com/tokio-rs/tokio/pull/8062
+[#8074]: https://github.com/tokio-rs/tokio/pull/8074
+[#8075]: https://github.com/tokio-rs/tokio/pull/8075
+[#8076]: https://github.com/tokio-rs/tokio/pull/8076
+
+# 1.47.4 (April 2nd, 2026)
+
+### Fixed
+
+* sync: fix panic in `Chan::recv_many` when called with non-empty vector on closed channel ([#7991])
+
+[#7991]: https://github.com/tokio-rs/tokio/pull/7991
+
+# 1.47.3 (January 3rd, 2026)
+
+### Fixed
+
+* sync: return `TryRecvError::Disconnected` from `Receiver::try_recv` after `Receiver::close` ([#7686])
+
+# 1.47.2 (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])
+- macros: fix hygiene issue in `join!` and `try_join!` ([#7638])
+- 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
+[#7638]: https://github.com/tokio-rs/tokio/pull/7638
+[#7672]: https://github.com/tokio-rs/tokio/pull/7672
+
 # 1.47.1 (August 1st, 2025)
 
 ### Fixed
@@ -249,6 +291,26 @@ comment on [#7172].
 [#7186]: https://github.com/tokio-rs/tokio/pull/7186
 [#7192]: https://github.com/tokio-rs/tokio/pull/7192
 
+# 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

tokio/Cargo.toml

@@ -6,7 +6,7 @@ name = "tokio"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v1.x.y" git tag.
-version = "1.47.1"
+version = "1.47.5"
 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.47.1", features = ["full"] }
+tokio = { version = "1.47.5", features = ["full"] }
 ```
 Then, on your main.rs:
 

tokio/src/macros/join.rs

@@ -113,7 +113,7 @@ doc! {macro_rules! join {
     (@ {
         // Type of rotator that controls which inner future to start with
         // when polling our output future.
-        rotator=$rotator:ty;
+        rotator_select=$rotator_select:ty;
 
         // One `_` for each branch in the `join!` macro. This is not used once
         // normalization is complete.
@@ -126,7 +126,7 @@ doc! {macro_rules! join {
         $( ( $($skip:tt)* ) $e:expr, )*
 
     }) => {{
-        use $crate::macros::support::{maybe_done, poll_fn, Future, Pin};
+        use $crate::macros::support::{maybe_done, poll_fn, Future, Pin, RotatorSelect};
         use $crate::macros::support::Poll::{Ready, Pending};
 
         // Safety: nothing must be moved out of `futures`. This is to satisfy
@@ -143,14 +143,14 @@ doc! {macro_rules! join {
         // <https://internals.rust-lang.org/t/surprising-soundness-trouble-around-pollfn/17484>
         let mut futures = &mut futures;
 
-        const COUNT: u32 = $($total)*;
-
         // Each time the future created by poll_fn is polled, if not using biased mode,
         // a different future is polled first to ensure every future passed to join!
         // can make progress even if one of the futures consumes the whole budget.
-        let mut rotator = <$rotator>::default();
+        let mut rotator = <$rotator_select as RotatorSelect>::Rotator::<{$($total)*}>::default();
 
         poll_fn(move |cx| {
+            const COUNT: u32 = $($total)*;
+
             let mut is_pending = false;
             let mut to_run = COUNT;
 
@@ -205,17 +205,17 @@ doc! {macro_rules! join {
 
     // ===== Normalize =====
 
-    (@ { rotator=$rotator:ty; ( $($s:tt)* ) ( $($n:tt)* ) $($t:tt)* } $e:expr, $($r:tt)* ) => {
-        $crate::join!(@{ rotator=$rotator; ($($s)* _) ($($n)* + 1) $($t)* ($($s)*) $e, } $($r)*)
+    (@ { rotator_select=$rotator_select:ty; ( $($s:tt)* ) ( $($n:tt)* ) $($t:tt)* } $e:expr, $($r:tt)* ) => {
+        $crate::join!(@{ rotator_select=$rotator_select; ($($s)* _) ($($n)* + 1) $($t)* ($($s)*) $e, } $($r)*)
     };
 
     // ===== Entry point =====
     ( biased; $($e:expr),+ $(,)?) => {
-        $crate::join!(@{ rotator=$crate::macros::support::BiasedRotator; () (0) } $($e,)*)
+        $crate::join!(@{ rotator_select=$crate::macros::support::SelectBiased; () (0) } $($e,)*)
     };
 
     ( $($e:expr),+ $(,)?) => {
-        $crate::join!(@{ rotator=$crate::macros::support::Rotator<COUNT>; () (0) } $($e,)*)
+        $crate::join!(@{ rotator_select=$crate::macros::support::SelectNormal; () (0) } $($e,)*)
     };
 
     (biased;) => { async {}.await };
@@ -223,6 +223,30 @@ doc! {macro_rules! join {
     () => { async {}.await }
 }}
 
+/// Helper trait to select which type of `Rotator` to use.
+// We need this to allow specifying a const generic without
+// colliding with caller const names due to macro hygiene.
+pub trait RotatorSelect {
+    type Rotator<const COUNT: u32>: Default;
+}
+
+/// Marker type indicating that the starting branch should
+/// rotate each poll.
+#[derive(Debug)]
+pub struct SelectNormal;
+/// Marker type indicating that the starting branch should
+/// be the first declared branch each poll.
+#[derive(Debug)]
+pub struct SelectBiased;
+
+impl RotatorSelect for SelectNormal {
+    type Rotator<const COUNT: u32> = Rotator<COUNT>;
+}
+
+impl RotatorSelect for SelectBiased {
+    type Rotator<const COUNT: u32> = BiasedRotator;
+}
+
 /// Rotates by one each [`Self::num_skip`] call up to COUNT - 1.
 #[derive(Default, Debug)]
 pub struct Rotator<const COUNT: u32> {

tokio/src/macros/support.rs

@@ -3,7 +3,7 @@ cfg_macros! {
 
     pub use std::future::poll_fn;
 
-    pub use crate::macros::join::{BiasedRotator, Rotator};
+    pub use crate::macros::join::{BiasedRotator, Rotator, RotatorSelect, SelectNormal, SelectBiased};
 
     #[doc(hidden)]
     pub fn thread_rng_n(n: u32) -> u32 {

tokio/src/macros/try_join.rs

@@ -166,7 +166,7 @@ doc! {macro_rules! try_join {
     (@ {
         // Type of rotator that controls which inner future to start with
         // when polling our output future.
-        rotator=$rotator:ty;
+        rotator_select=$rotator_select:ty;
 
         // One `_` for each branch in the `try_join!` macro. This is not used once
         // normalization is complete.
@@ -179,7 +179,7 @@ doc! {macro_rules! try_join {
         $( ( $($skip:tt)* ) $e:expr, )*
 
     }) => {{
-        use $crate::macros::support::{maybe_done, poll_fn, Future, Pin};
+        use $crate::macros::support::{maybe_done, poll_fn, Future, Pin, RotatorSelect};
         use $crate::macros::support::Poll::{Ready, Pending};
 
         // Safety: nothing must be moved out of `futures`. This is to satisfy
@@ -196,14 +196,14 @@ doc! {macro_rules! try_join {
         // <https://internals.rust-lang.org/t/surprising-soundness-trouble-around-pollfn/17484>
         let mut futures = &mut futures;
 
-        const COUNT: u32 = $($total)*;
-
         // Each time the future created by poll_fn is polled, if not using biased mode,
         // a different future is polled first to ensure every future passed to try_join!
         // can make progress even if one of the futures consumes the whole budget.
-        let mut rotator = <$rotator>::default();
+        let mut rotator = <$rotator_select as RotatorSelect>::Rotator::<{$($total)*}>::default();
 
         poll_fn(move |cx| {
+            const COUNT: u32 = $($total)*;
+
             let mut is_pending = false;
             let mut to_run = COUNT;
 
@@ -264,17 +264,17 @@ doc! {macro_rules! try_join {
 
     // ===== Normalize =====
 
-    (@ { rotator=$rotator:ty;  ( $($s:tt)* ) ( $($n:tt)* ) $($t:tt)* } $e:expr, $($r:tt)* ) => {
-      $crate::try_join!(@{ rotator=$rotator; ($($s)* _) ($($n)* + 1) $($t)* ($($s)*) $e, } $($r)*)
+    (@ { rotator_select=$rotator_select:ty;  ( $($s:tt)* ) ( $($n:tt)* ) $($t:tt)* } $e:expr, $($r:tt)* ) => {
+      $crate::try_join!(@{ rotator_select=$rotator_select; ($($s)* _) ($($n)* + 1) $($t)* ($($s)*) $e, } $($r)*)
     };
 
     // ===== Entry point =====
     ( biased; $($e:expr),+ $(,)?) => {
-        $crate::try_join!(@{ rotator=$crate::macros::support::BiasedRotator;  () (0) } $($e,)*)
+        $crate::try_join!(@{ rotator_select=$crate::macros::support::SelectBiased;  () (0) } $($e,)*)
     };
 
     ( $($e:expr),+ $(,)?) => {
-        $crate::try_join!(@{ rotator=$crate::macros::support::Rotator<COUNT>; () (0) } $($e,)*)
+        $crate::try_join!(@{ rotator_select=$crate::macros::support::SelectNormal; () (0) } $($e,)*)
     };
 
     (biased;) => { async { Ok(()) }.await };

tokio/src/process/unix/pidfd_reaper.rs

@@ -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
     }

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

@@ -566,7 +566,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/block.rs

@@ -211,11 +211,6 @@ impl<T> Block<T> {
         self.header.ready_slots.fetch_or(TX_CLOSED, Release);
     }
 
-    pub(crate) unsafe fn is_closed(&self) -> bool {
-        let ready_bits = self.header.ready_slots.load(Acquire);
-        is_tx_closed(ready_bits)
-    }
-
     /// Resets the block to a blank state. This enables reusing blocks in the
     /// channel.
     ///

tokio/src/sync/mpsc/bounded.rs

@@ -1844,14 +1844,12 @@ impl<T> OwnedPermit<T> {
     ///
     /// [`Sender`]: Sender
     pub fn release(mut self) -> Sender<T> {
-        use chan::Semaphore;
-
         let chan = self.chan.take().unwrap_or_else(|| {
             unreachable!("OwnedPermit channel is only taken when the permit is moved")
         });
 
         // Add the permit back to the semaphore
-        chan.semaphore().add_permit();
+        drop(Permit { chan: &chan });
         Sender { chan }
     }
 
@@ -1910,21 +1908,10 @@ impl<T> OwnedPermit<T> {
 
 impl<T> Drop for OwnedPermit<T> {
     fn drop(&mut self) {
-        use chan::Semaphore;
-
         // Are we still holding onto the sender?
         if let Some(chan) = self.chan.take() {
-            let semaphore = chan.semaphore();
-
-            // Add the permit back to the semaphore
-            semaphore.add_permit();
-
-            // If this `OwnedPermit` is holding the last sender for this
-            // channel, wake the receiver so that it can be notified that the
-            // channel is closed.
-            if semaphore.is_closed() && semaphore.is_idle() {
-                chan.wake_rx();
-            }
+            // Reuse Drop impl of non-owned Permit.
+            drop(Permit { chan: &chan });
         }
 
         // Otherwise, do nothing.

tokio/src/sync/mpsc/chan.rs

@@ -306,13 +306,11 @@ impl<T, S: Semaphore> Rx<T, S> {
                             return Ready(Some(value));
                         }
                         Some(Read::Closed) => {
-                            // TODO: This check may not be required as it most
-                            // likely can only return `true` at this point. A
-                            // channel is closed when all tx handles are
+                            // A channel is closed when all tx handles are
                             // dropped. Dropping a tx handle releases memory,
                             // which ensures that if dropping the tx handle is
                             // visible, then all messages sent are also visible.
-                            assert!(self.inner.semaphore.is_idle());
+                            debug_assert!(self.inner.semaphore.is_idle());
                             coop.made_progress();
                             return Ready(None);
                         }
@@ -380,13 +378,11 @@ impl<T, S: Semaphore> Rx<T, S> {
                                 if number_added > 0 {
                                     self.inner.semaphore.add_permits(number_added);
                                 }
-                                // TODO: This check may not be required as it most
-                                // likely can only return `true` at this point. A
-                                // channel is closed when all tx handles are
+                                // A channel is closed when all tx handles are
                                 // dropped. Dropping a tx handle releases memory,
                                 // which ensures that if dropping the tx handle is
                                 // visible, then all messages sent are also visible.
-                                assert!(self.inner.semaphore.is_idle());
+                                debug_assert!(self.inner.semaphore.is_idle());
                                 coop.made_progress();
                                 return Ready(number_added);
                             }
@@ -415,7 +411,7 @@ impl<T, S: Semaphore> Rx<T, S> {
             try_recv!();
 
             if rx_fields.rx_closed && self.inner.semaphore.is_idle() {
-                assert!(buffer.is_empty());
+                debug_assert_eq!(buffer.len(), initial_length);
                 coop.made_progress();
                 Ready(0usize)
             } else {
@@ -439,6 +435,12 @@ 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 && self.inner.semaphore.is_idle() =>
+                        {
+                            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,
@@ -218,15 +224,6 @@ impl<T> Tx<T> {
             let _ = Box::from_raw(block.as_ptr());
         }
     }
-
-    pub(crate) fn is_closed(&self) -> bool {
-        let tail = self.block_tail.load(Acquire);
-
-        unsafe {
-            let tail_block = &*tail;
-            tail_block.is_closed()
-        }
-    }
 }
 
 impl<T> fmt::Debug for Tx<T> {
@@ -250,11 +247,58 @@ impl<T> Rx<T> {
         self.len(tx) == 0
     }
 
+    // Guaranteed to return true if `slot_index` is the fake message sent on channel close.
+    // Guaranteed to return false if `slot_index` is a fully sent message.
+    //
+    // For messages that are partially sent, may return either true or false.
+    fn is_maybe_closed(&self, tx: &Tx<T>, slot_index: usize) -> bool {
+        let start_index = block::start_index(slot_index);
+
+        let tail = tx.block_tail.load(Acquire);
+        // SAFETY: Only the receiver frees blocks, so since we are the receiver, this will not be
+        // freed right now.
+        let tail_ref = unsafe { &*tail };
+        if tail_ref.is_at_index(start_index) {
+            return !tail_ref.has_value(slot_index);
+        }
+
+        // This method is optimized for checking whether the last value is present, so most of the
+        // time it is in `block_tail`. However, this isn't always the case since it's possible
+        // that the list was grown with an empty block, in which case `block_tail` points one block
+        // too far. To handle this case, we walk the list from the head.
+        let mut block_ptr = Some(self.head);
+
+        while let Some(block) = block_ptr {
+            // SAFETY: Only the receiver frees blocks, so since we are the receiver, this will not
+            // be freed right now.
+            let block_ref = unsafe { block.as_ref() };
+            if block_ref.is_at_index(start_index) {
+                return !block_ref.has_value(slot_index);
+            }
+            block_ptr = block_ref.load_next(Acquire);
+        }
+        true
+    }
+
     pub(crate) fn len(&self, tx: &Tx<T>) -> usize {
-        // When all the senders are dropped, there will be a last block in the tail position,
-        // but it will be closed
         let tail_position = tx.tail_position.load(Acquire);
-        tail_position - self.index - (tx.is_closed() as usize)
+        let mut len = tail_position.wrapping_sub(self.index);
+        debug_assert!(0 <= len as isize);
+        if len == 0 {
+            return 0;
+        }
+        // There are messages present in the queue. However, it's possible that the last message is
+        // a fake "closed" message that we do not wish to count. To avoid counting it, we do not
+        // count the last message if the ready bit is unset.
+        //
+        // Note that it is also possible for the ready bit to be unset on a normal message, but
+        // this happens only if that message is currently being sent *right now* in parallel on
+        // another thread. That is okay because it is optional to count messages that are currently
+        // being sent.
+        if self.is_maybe_closed(tx, tail_position.wrapping_sub(1)) {
+            len -= 1;
+        }
+        len
     }
 
     /// Pops the next value off the queue.

tokio/src/sync/mpsc/mod.rs

@@ -135,10 +135,10 @@ pub mod error;
 /// This value must be a power of 2. It also must be smaller than the number of
 /// bits in `usize`.
 #[cfg(all(target_pointer_width = "64", not(loom)))]
-const BLOCK_CAP: usize = 32;
+pub(crate) const BLOCK_CAP: usize = 32;
 
 #[cfg(all(not(target_pointer_width = "64"), not(loom)))]
-const BLOCK_CAP: usize = 16;
+pub(crate) const BLOCK_CAP: usize = 16;
 
 #[cfg(loom)]
-const BLOCK_CAP: usize = 2;
+pub(crate) const BLOCK_CAP: usize = 2;

tokio/src/sync/rwlock.rs

@@ -266,12 +266,13 @@ impl<T: ?Sized> RwLock<T> {
     ///
     /// # Panics
     ///
-    /// Panics if `max_reads` is more than `u32::MAX >> 3`.
+    /// Panics if `max_reads` is `0` or is bigger than `u32::MAX >> 3`.
     #[track_caller]
     pub fn with_max_readers(value: T, max_reads: u32) -> RwLock<T>
     where
         T: Sized,
     {
+        assert_ne!(max_reads, 0, "a RwLock may not be created with 0 readers");
         assert!(
             max_reads <= MAX_READS,
             "a RwLock may not be created with more than {MAX_READS} readers"
@@ -367,11 +368,16 @@ impl<T: ?Sized> RwLock<T> {
     ///
     /// static LOCK: RwLock<i32> = RwLock::const_with_max_readers(5, 1024);
     /// ```
+    ///
+    /// # Panics
+    ///
+    /// Panics if `max_reads` is `0` or is bigger than `u32::MAX >> 3`.
     #[cfg(not(all(loom, test)))]
     pub const fn const_with_max_readers(value: T, max_reads: u32) -> RwLock<T>
     where
         T: Sized,
     {
+        assert!(max_reads != 0, "a RwLock may not be created with 0 readers");
         assert!(max_reads <= MAX_READS);
 
         RwLock {
@@ -771,6 +777,7 @@ impl<T: ?Sized> RwLock<T> {
     /// ```
     pub async fn write(&self) -> RwLockWriteGuard<'_, T> {
         let acquire_fut = async {
+            debug_assert_ne!(self.mr, 0);
             self.s.acquire(self.mr as usize).await.unwrap_or_else(|_| {
                 // The semaphore was closed. but, we never explicitly close it, and we have a
                 // handle to it through the Arc, which means that this can never happen.
@@ -906,6 +913,7 @@ impl<T: ?Sized> RwLock<T> {
         let resource_span = self.resource_span.clone();
 
         let acquire_fut = async {
+            debug_assert_ne!(self.mr, 0);
             self.s.acquire(self.mr as usize).await.unwrap_or_else(|_| {
                 // The semaphore was closed. but, we never explicitly close it, and we have a
                 // handle to it through the Arc, which means that this can never happen.
@@ -970,6 +978,7 @@ impl<T: ?Sized> RwLock<T> {
     /// }
     /// ```
     pub fn try_write(&self) -> Result<RwLockWriteGuard<'_, T>, TryLockError> {
+        debug_assert_ne!(self.mr, 0);
         match self.s.try_acquire(self.mr as usize) {
             Ok(permit) => permit,
             Err(TryAcquireError::NoPermits) => return Err(TryLockError(())),
@@ -1028,6 +1037,7 @@ impl<T: ?Sized> RwLock<T> {
     /// }
     /// ```
     pub fn try_write_owned(self: Arc<Self>) -> Result<OwnedRwLockWriteGuard<T>, TryLockError> {
+        debug_assert_ne!(self.mr, 0);
         match self.s.try_acquire(self.mr as usize) {
             Ok(permit) => permit,
             Err(TryAcquireError::NoPermits) => return Err(TryLockError(())),

tokio/src/sync/tests/loom_mpsc.rs

@@ -1,4 +1,4 @@
-use crate::sync::mpsc;
+use crate::sync::mpsc::{self, BLOCK_CAP};
 
 use loom::future::block_on;
 use loom::sync::Arc;
@@ -222,3 +222,56 @@ fn nonempty_after_send() {
         join.join().unwrap();
     });
 }
+
+#[test]
+fn is_empty_during_close() {
+    loom::model(|| {
+        let (tx, rx) = mpsc::channel::<()>(1);
+
+        let th1 = thread::spawn(move || {
+            assert!(rx.is_empty());
+        });
+
+        drop(tx);
+
+        th1.join().unwrap();
+    });
+}
+
+fn len_during_close_helper(n: usize) {
+    loom::model(move || {
+        let (tx, rx) = mpsc::channel::<()>(n + 1);
+
+        for _ in 0..n {
+            tx.try_send(()).unwrap();
+        }
+
+        let th1 = thread::spawn(move || {
+            assert_eq!(rx.len(), n);
+        });
+
+        drop(tx);
+
+        th1.join().unwrap();
+    });
+}
+
+#[test]
+fn len_during_close_0() {
+    len_during_close_helper(0);
+}
+
+#[test]
+fn len_during_close_1() {
+    len_during_close_helper(1);
+}
+
+#[test]
+fn len_during_close_block_cap() {
+    len_during_close_helper(BLOCK_CAP);
+}
+
+#[test]
+fn len_during_close_block_cap_plus_1() {
+    len_during_close_helper(BLOCK_CAP + 1);
+}

tokio/tests/macros_join.rs

@@ -234,3 +234,23 @@ async fn join_into_future() {
 
     tokio::join!(NotAFuture);
 }
+
+// Regression test for: https://github.com/tokio-rs/tokio/issues/7637
+// We want to make sure that the `const COUNT: u32` declaration
+// inside the macro body doesn't leak to the caller to cause compiler failures
+// or variable shadowing.
+#[tokio::test]
+async fn caller_names_const_count() {
+    let (tx, rx) = oneshot::channel::<u32>();
+
+    const COUNT: u32 = 2;
+
+    let mut join = task::spawn(async { tokio::join!(async { tx.send(COUNT).unwrap() }) });
+    assert_ready!(join.poll());
+
+    let res = rx.await.unwrap();
+
+    // This passing demonstrates that the const in the macro is
+    // not shadowing the caller-specified COUNT value
+    assert_eq!(2, res);
+}

tokio/tests/macros_try_join.rs

@@ -247,3 +247,23 @@ async fn empty_try_join() {
     assert_eq!(tokio::try_join!() as Result<_, ()>, Ok(()));
     assert_eq!(tokio::try_join!(biased;) as Result<_, ()>, Ok(()));
 }
+
+// Regression test for: https://github.com/tokio-rs/tokio/issues/7637
+// We want to make sure that the `const COUNT: u32` declaration
+// inside the macro body doesn't leak to the caller to cause compiler failures
+// or variable shadowing.
+#[tokio::test]
+async fn caller_names_const_count() {
+    let (tx, rx) = oneshot::channel::<u32>();
+
+    const COUNT: u32 = 2;
+
+    let mut try_join = task::spawn(async { tokio::try_join!(async { tx.send(COUNT) }) });
+    assert_ready!(try_join.poll()).unwrap();
+
+    let res = rx.await.unwrap();
+
+    // This passing demonstrates that the const in the macro is
+    // not shadowing the caller-specified COUNT value
+    assert_eq!(2, res);
+}

tokio/tests/sync_broadcast.rs

@@ -706,3 +706,17 @@ fn broadcast_sender_closed_with_extra_subscribe() {
     assert!(task3.is_woken());
     assert_ready!(task3.poll());
 }
+
+#[tokio::test]
+async fn broadcast_sender_new_must_be_closed() {
+    let capacity = 1;
+    let tx: broadcast::Sender<()> = broadcast::Sender::new(capacity);
+
+    let mut task = task::spawn(tx.closed());
+    assert_ready!(task.poll());
+
+    let _rx = tx.subscribe();
+
+    let mut task2 = task::spawn(tx.closed());
+    assert_pending!(task2.poll());
+}

tokio/tests/sync_mpsc.rs

@@ -358,6 +358,30 @@ async fn send_recv_many_unbounded_capacity() {
     assert_eq!(expected, buffer);
 }
 
+#[maybe_tokio_test]
+async fn recv_many_with_non_empty_buffer_bounded_rx_closed_and_idle() {
+    let (_tx, mut rx) = mpsc::channel::<i32>(1);
+
+    let mut buffer: Vec<i32> = vec![1];
+
+    rx.close();
+
+    assert_eq!(0, rx.recv_many(&mut buffer, 1).await);
+    assert_eq!(vec![1], buffer);
+}
+
+#[maybe_tokio_test]
+async fn recv_many_with_non_empty_buffer_unbounded_rx_closed_and_idle() {
+    let (_tx, mut rx) = mpsc::unbounded_channel::<i32>();
+
+    let mut buffer: Vec<i32> = vec![1];
+
+    rx.close();
+
+    assert_eq!(0, rx.recv_many(&mut buffer, 1).await);
+    assert_eq!(vec![1], buffer);
+}
+
 #[tokio::test]
 #[cfg(feature = "full")]
 async fn async_send_recv_unbounded() {
@@ -764,6 +788,91 @@ async fn drop_permit_iterator_releases_permits() {
     }
 }
 
+#[test]
+fn dropping_last_permit_wakes_closed_receiver() {
+    let (tx, mut rx) = mpsc::channel::<()>(100);
+
+    let permit = tx.try_reserve().unwrap();
+    rx.close();
+
+    let mut recv = tokio_test::task::spawn(rx.recv());
+    assert_pending!(recv.poll());
+    drop(permit);
+    assert!(recv.is_woken());
+    assert_ready!(recv.poll());
+}
+
+#[test]
+fn dropping_last_owned_permit_wakes_closed_receiver() {
+    let (tx, mut rx) = mpsc::channel::<()>(100);
+
+    let permit = tx.try_reserve_owned().unwrap();
+    rx.close();
+
+    let mut recv = tokio_test::task::spawn(rx.recv());
+    assert_pending!(recv.poll());
+    drop(permit);
+    assert!(recv.is_woken());
+    assert_ready!(recv.poll());
+}
+
+#[test]
+fn dropping_last_permit_iterator_wakes_closed_receiver() {
+    let (tx, mut rx) = mpsc::channel::<()>(100);
+
+    let permits = tx.try_reserve_many(1).unwrap();
+    rx.close();
+
+    let mut recv = tokio_test::task::spawn(rx.recv());
+    assert_pending!(recv.poll());
+    drop(permits);
+    assert!(recv.is_woken());
+    assert_ready!(recv.poll());
+}
+
+#[test]
+fn sending_last_permit_wakes_closed_receiver() {
+    let (tx, mut rx) = mpsc::channel::<()>(100);
+
+    let permit = tx.try_reserve().unwrap();
+    rx.close();
+
+    let mut recv = tokio_test::task::spawn(rx.recv());
+    assert_pending!(recv.poll());
+    permit.send(());
+    assert!(recv.is_woken());
+    assert_ready!(recv.poll());
+}
+
+#[test]
+fn sending_last_owned_permit_wakes_closed_receiver() {
+    let (tx, mut rx) = mpsc::channel::<()>(100);
+
+    let permit = tx.try_reserve_owned().unwrap();
+    rx.close();
+
+    let mut recv = tokio_test::task::spawn(rx.recv());
+    assert_pending!(recv.poll());
+    permit.send(());
+    assert!(recv.is_woken());
+    assert_ready!(recv.poll());
+}
+
+#[test]
+fn releasing_last_owned_permit_wakes_closed_receiver() {
+    let (tx, mut rx) = mpsc::channel::<()>(100);
+
+    let permit = tx.try_reserve_owned().unwrap();
+    rx.close();
+
+    let mut recv = tokio_test::task::spawn(rx.recv());
+    assert_pending!(recv.poll());
+    let inert_sender = permit.release();
+    assert!(recv.is_woken());
+    assert_ready!(recv.poll());
+    drop(inert_sender);
+}
+
 #[maybe_tokio_test]
 async fn dropping_rx_closes_channel() {
     let (tx, rx) = mpsc::channel(100);
@@ -966,6 +1075,28 @@ 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_after_receiver_close_with_permit() {
+    let (tx, mut rx) = mpsc::channel::<()>(5);
+
+    let permit = tx.try_reserve().unwrap();
+
+    assert_eq!(Err(TryRecvError::Empty), rx.try_recv());
+    rx.close();
+    assert_eq!(Err(TryRecvError::Empty), rx.try_recv());
+    drop(permit);
+    assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv());
+}
+
 #[test]
 fn try_recv_close_while_empty_bounded() {
     let (tx, mut rx) = mpsc::channel::<()>(5);

tokio/tests/sync_rwlock.rs

@@ -78,6 +78,18 @@ fn exhaust_reading() {
     let _g1 = assert_ready!(t1.poll());
 }
 
+#[test]
+#[should_panic(expected = "a RwLock may not be created with 0 readers")]
+fn zero_max_readers() {
+    RwLock::with_max_readers(100, 0);
+}
+
+#[test]
+#[should_panic(expected = "a RwLock may not be created with 0 readers")]
+fn zero_max_readers_const() {
+    RwLock::const_with_max_readers(100, 0);
+}
+
 // When there is an active exclusive owner, subsequent exclusive access should not be possible
 #[test]
 fn write_exclusive_pending() {