From ebf61b45b5184018f00bc666887ebccf3d4fe51b Mon Sep 17 00:00:00 2001 From: Alice Ryhl Date: Thu, 7 May 2026 09:29:33 +0200 Subject: [PATCH 1/5] sync: fix underflow in mpsc channel `len()` (#8062) --- tokio/src/sync/mpsc/block.rs | 5 --- tokio/src/sync/mpsc/list.rs | 62 +++++++++++++++++++++++++------ tokio/src/sync/mpsc/mod.rs | 6 +-- tokio/src/sync/tests/loom_mpsc.rs | 55 ++++++++++++++++++++++++++- 4 files changed, 107 insertions(+), 21 deletions(-) diff --git a/tokio/src/sync/mpsc/block.rs b/tokio/src/sync/mpsc/block.rs index 927c45664..7d7e59afd 100644 --- a/tokio/src/sync/mpsc/block.rs +++ b/tokio/src/sync/mpsc/block.rs @@ -211,11 +211,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/list.rs b/tokio/src/sync/mpsc/list.rs index 118bac856..1f3d23839 100644 --- a/tokio/src/sync/mpsc/list.rs +++ b/tokio/src/sync/mpsc/list.rs @@ -224,15 +224,6 @@ impl Tx { 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 fmt::Debug for Tx { @@ -256,11 +247,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 3df612ca4..e724f62a7 100644 --- a/tokio/src/sync/mpsc/mod.rs +++ b/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; diff --git a/tokio/src/sync/tests/loom_mpsc.rs b/tokio/src/sync/tests/loom_mpsc.rs index 039b87a77..3e6c84798 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); +} From 9fccf5339d41c1f2f863f97b9133bc8a5a10bc28 Mon Sep 17 00:00:00 2001 From: Alice Ryhl Date: Thu, 7 May 2026 09:30:50 +0200 Subject: [PATCH 2/5] sync: return `Empty` from `try_recv()` when mpsc is closed with outstanding permits (#8074) --- tokio/src/sync/mpsc/chan.rs | 4 +++- tokio/tests/sync_mpsc.rs | 13 +++++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) 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/tests/sync_mpsc.rs b/tokio/tests/sync_mpsc.rs index 048b94eb5..27157288e 100644 --- a/tokio/tests/sync_mpsc.rs +++ b/tokio/tests/sync_mpsc.rs @@ -999,6 +999,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); From 30d25ccb8bc91ca811773ee243e71e31772275d2 Mon Sep 17 00:00:00 2001 From: Alice Ryhl Date: Thu, 7 May 2026 09:31:12 +0200 Subject: [PATCH 3/5] sync: require that an `RwLock` has `max_readers != 0` (#8076) --- tokio/src/sync/rwlock.rs | 12 +++++++++++- tokio/tests/sync_rwlock.rs | 12 ++++++++++++ 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/tokio/src/sync/rwlock.rs b/tokio/src/sync/rwlock.rs index d94b65143..b6bf7faca 100644 --- a/tokio/src/sync/rwlock.rs +++ b/tokio/src/sync/rwlock.rs @@ -266,12 +266,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" @@ -367,11 +368,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 { @@ -771,6 +777,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. @@ -906,6 +913,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. @@ -970,6 +978,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(())), @@ -1028,6 +1037,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/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() { From f085b6211b8ebb6aba21f1f1f91e7b8b243aa815 Mon Sep 17 00:00:00 2001 From: Alice Ryhl Date: Thu, 7 May 2026 09:32:14 +0200 Subject: [PATCH 4/5] sync: notify receivers in mpsc `OwnedPermit::release()` method (#8075) --- tokio/src/sync/mpsc/bounded.rs | 19 ++------ tokio/tests/sync_mpsc.rs | 85 ++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 16 deletions(-) diff --git a/tokio/src/sync/mpsc/bounded.rs b/tokio/src/sync/mpsc/bounded.rs index 06eeffc3f..f7b0081c6 100644 --- a/tokio/src/sync/mpsc/bounded.rs +++ b/tokio/src/sync/mpsc/bounded.rs @@ -1844,14 +1844,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 } } @@ -1910,21 +1908,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/tests/sync_mpsc.rs b/tokio/tests/sync_mpsc.rs index 27157288e..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); From 11bfc1345bbd5e901187e2b3702de10b0efbffdc Mon Sep 17 00:00:00 2001 From: Alice Ryhl Date: Thu, 7 May 2026 13:55:35 +0200 Subject: [PATCH 5/5] chore: prepare Tokio v1.47.5 (#8122) --- Cargo.lock | 6 +++--- README.md | 2 +- tokio/CHANGELOG.md | 14 ++++++++++++++ tokio/Cargo.toml | 2 +- tokio/README.md | 2 +- 5 files changed, 20 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d1a338d43..6d093ee83 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1050,9 +1050,9 @@ checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" [[package]] name = "rand" -version = "0.9.2" +version = "0.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1" +checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" dependencies = [ "rand_chacha", "rand_core", @@ -1405,7 +1405,7 @@ dependencies = [ [[package]] name = "tokio" -version = "1.47.4" +version = "1.47.5" dependencies = [ "async-stream", "backtrace", diff --git a/README.md b/README.md index a0e2e47ae..76c7fad12 100644 --- a/README.md +++ b/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.4", features = ["full"] } +tokio = { version = "1.47.5", features = ["full"] } ``` Then, on your main.rs: diff --git a/tokio/CHANGELOG.md b/tokio/CHANGELOG.md index 128fbd0d4..70acb0b87 100644 --- a/tokio/CHANGELOG.md +++ b/tokio/CHANGELOG.md @@ -1,3 +1,17 @@ +# 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/Cargo.toml b/tokio/Cargo.toml index e0e83a19d..843a32384 100644 --- a/tokio/Cargo.toml +++ b/tokio/Cargo.toml @@ -6,7 +6,7 @@ name = "tokio" # - README.md # - Update CHANGELOG.md. # - Create "v1.x.y" git tag. -version = "1.47.4" +version = "1.47.5" edition = "2021" rust-version = "1.70" authors = ["Tokio Contributors "] diff --git a/tokio/README.md b/tokio/README.md index a0e2e47ae..76c7fad12 100644 --- a/tokio/README.md +++ b/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.4", features = ["full"] } +tokio = { version = "1.47.5", features = ["full"] } ``` Then, on your main.rs: