Merge 'tokio-1.47.5' into 'tokio-1.51.x' (#8123)

tokio/CHANGELOG.md

@@ -330,6 +330,20 @@ The MSRV is increased to 1.71.
 [#7672]: https://github.com/tokio-rs/tokio/pull/7672
 [#7675]: https://github.com/tokio-rs/tokio/pull/7675
 
+# 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

tokio/src/sync/mpsc/block.rs

@@ -220,11 +220,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

@@ -1853,14 +1853,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 }
     }
 
@@ -1919,21 +1917,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

@@ -436,7 +436,9 @@ impl<T, S: Semaphore> Rx<T, S> {
                         }
                         TryPopResult::Closed => return Err(TryRecvError::Disconnected),
                         // If close() was called, an empty queue should report Disconnected.
-                        TryPopResult::Empty if rx_fields.rx_closed => {
+                        TryPopResult::Empty
+                            if rx_fields.rx_closed && self.inner.semaphore.is_idle() =>
+                        {
                             return Err(TryRecvError::Disconnected)
                         }
                         TryPopResult::Empty => return Err(TryRecvError::Empty),

tokio/src/sync/mpsc/list.rs

@@ -239,15 +239,6 @@ impl<T> Tx<T> {
             let _ = unsafe { 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> {
@@ -271,11 +262,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

@@ -137,10 +137,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

@@ -265,12 +265,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"
@@ -366,11 +367,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 {
@@ -773,6 +779,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.
@@ -911,6 +918,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.
@@ -975,6 +983,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(())),
@@ -1033,6 +1042,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/sync_mpsc.rs

@@ -788,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);
@@ -999,6 +1084,19 @@ fn try_recv_after_receiver_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() {