diff --git a/tokio/CHANGELOG.md b/tokio/CHANGELOG.md index 613172870..93b9e7f8f 100644 --- a/tokio/CHANGELOG.md +++ b/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 diff --git a/tokio/src/sync/mpsc/block.rs b/tokio/src/sync/mpsc/block.rs index cf86922bd..e231b8e57 100644 --- a/tokio/src/sync/mpsc/block.rs +++ b/tokio/src/sync/mpsc/block.rs @@ -220,11 +220,6 @@ impl Block { 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. /// diff --git a/tokio/src/sync/mpsc/bounded.rs b/tokio/src/sync/mpsc/bounded.rs index c5ebb4749..db50535ff 100644 --- a/tokio/src/sync/mpsc/bounded.rs +++ b/tokio/src/sync/mpsc/bounded.rs @@ -1853,14 +1853,12 @@ impl OwnedPermit { /// /// [`Sender`]: Sender pub fn release(mut self) -> Sender { - 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 OwnedPermit { impl Drop for OwnedPermit { 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. diff --git a/tokio/src/sync/mpsc/chan.rs b/tokio/src/sync/mpsc/chan.rs index 7f7f8b16d..f9a72de60 100644 --- a/tokio/src/sync/mpsc/chan.rs +++ b/tokio/src/sync/mpsc/chan.rs @@ -436,7 +436,9 @@ impl Rx { } 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), diff --git a/tokio/src/sync/mpsc/list.rs b/tokio/src/sync/mpsc/list.rs index 289be0bcf..c5f21b8b2 100644 --- a/tokio/src/sync/mpsc/list.rs +++ b/tokio/src/sync/mpsc/list.rs @@ -239,15 +239,6 @@ impl Tx { 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 fmt::Debug for Tx { @@ -271,11 +262,58 @@ impl Rx { 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, 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) -> 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. diff --git a/tokio/src/sync/mpsc/mod.rs b/tokio/src/sync/mpsc/mod.rs index 46363f933..1947013e6 100644 --- a/tokio/src/sync/mpsc/mod.rs +++ b/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; diff --git a/tokio/src/sync/rwlock.rs b/tokio/src/sync/rwlock.rs index 9bc0a1146..b1aae6db2 100644 --- a/tokio/src/sync/rwlock.rs +++ b/tokio/src/sync/rwlock.rs @@ -265,12 +265,13 @@ impl RwLock { /// /// # 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 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 RwLock { /// /// static LOCK: RwLock = 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 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 RwLock { /// ``` 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 RwLock { 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 RwLock { /// # } /// ``` pub fn try_write(&self) -> Result, 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 RwLock { /// # } /// ``` pub fn try_write_owned(self: Arc) -> Result, 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(())), diff --git a/tokio/src/sync/tests/loom_mpsc.rs b/tokio/src/sync/tests/loom_mpsc.rs index 620a9638a..468177143 100644 --- a/tokio/src/sync/tests/loom_mpsc.rs +++ b/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); +} diff --git a/tokio/tests/sync_mpsc.rs b/tokio/tests/sync_mpsc.rs index 048b94eb5..93804581b 100644 --- a/tokio/tests/sync_mpsc.rs +++ b/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); diff --git a/tokio/tests/sync_rwlock.rs b/tokio/tests/sync_rwlock.rs index 5a58b4971..2dc7b0a62 100644 --- a/tokio/tests/sync_rwlock.rs +++ b/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() {