From 72ff1700eec7fc0b586601ff0ba546b1ae3a5504 Mon Sep 17 00:00:00 2001 From: driftluo Date: Wed, 26 Jun 2024 11:06:35 +0800 Subject: [PATCH 1/4] fix: port fix discard all message on receiver droped --- crossbeam-channel/src/channel.rs | 4 +- crossbeam-channel/src/flavors/array.rs | 128 ++++++++++++++++--------- crossbeam-channel/tests/array.rs | 12 +++ 3 files changed, 99 insertions(+), 45 deletions(-) diff --git a/crossbeam-channel/src/channel.rs b/crossbeam-channel/src/channel.rs index b185b6e69..2a7cfe490 100644 --- a/crossbeam-channel/src/channel.rs +++ b/crossbeam-channel/src/channel.rs @@ -675,7 +675,7 @@ impl Drop for Sender { fn drop(&mut self) { unsafe { match &self.flavor { - SenderFlavor::Array(chan) => chan.release(|c| c.disconnect()), + SenderFlavor::Array(chan) => chan.release(|c| c.disconnect_senders()), SenderFlavor::List(chan) => chan.release(|c| c.disconnect_senders()), SenderFlavor::Zero(chan) => chan.release(|c| c.disconnect()), } @@ -1185,7 +1185,7 @@ impl Drop for Receiver { fn drop(&mut self) { unsafe { match &self.flavor { - ReceiverFlavor::Array(chan) => chan.release(|c| c.disconnect()), + ReceiverFlavor::Array(chan) => chan.release(|c| c.disconnect_receivers()), ReceiverFlavor::List(chan) => chan.release(|c| c.disconnect_receivers()), ReceiverFlavor::Zero(chan) => chan.release(|c| c.disconnect()), ReceiverFlavor::At(_) => {} diff --git a/crossbeam-channel/src/flavors/array.rs b/crossbeam-channel/src/flavors/array.rs index e8c5f81da..663e46b83 100644 --- a/crossbeam-channel/src/flavors/array.rs +++ b/crossbeam-channel/src/flavors/array.rs @@ -11,7 +11,7 @@ use alloc::boxed::Box; use core::{ cell::UnsafeCell, - mem::{self, MaybeUninit}, + mem::MaybeUninit, ptr, sync::atomic::{self, AtomicUsize, Ordering}, }; @@ -480,14 +480,13 @@ impl Channel { Some(self.cap()) } - /// Disconnects the channel and wakes up all blocked senders and receivers. + /// Disconnects senders and wakes up all blocked senders and receivers. /// /// Returns `true` if this call disconnected the channel. - pub(crate) fn disconnect(&self) -> bool { + pub(crate) fn disconnect_senders(&self) -> bool { let tail = self.tail.fetch_or(self.mark_bit, Ordering::SeqCst); if tail & self.mark_bit == 0 { - self.senders.disconnect(); self.receivers.disconnect(); true } else { @@ -495,6 +494,88 @@ impl Channel { } } + /// Disconnects receivers and wakes up all blocked senders. + /// + /// Returns `true` if this call disconnected the channel. + /// + /// # Safety + /// May only be called once upon dropping the last receiver. The + /// destruction of all other receivers must have been observed with acquire + /// ordering or stronger. + pub(crate) unsafe fn disconnect_receivers(&self) -> bool { + let tail = self.tail.fetch_or(self.mark_bit, Ordering::SeqCst); + let disconnected = if tail & self.mark_bit == 0 { + self.senders.disconnect(); + true + } else { + false + }; + + unsafe { + self.discard_all_messages(tail); + } + + disconnected + } + + /// Discards all messages. + /// + /// `tail` should be the current (and therefore last) value of `tail`. + /// + /// # Panicking + /// If a destructor panics, the remaining messages are leaked, matching the + /// behaviour of the unbounded channel. + /// + /// # Safety + /// This method must only be called when dropping the last receiver. The + /// destruction of all other receivers must have been observed with acquire + /// ordering or stronger. + unsafe fn discard_all_messages(&self, tail: usize) { + debug_assert!(self.is_disconnected()); + + // Only receivers modify `head`, so since we are the last one, + // this value will not change and will not be observed (since + // no new messages can be sent after disconnection). + let mut head = self.head.load(Ordering::Relaxed); + let tail = tail & !self.mark_bit; + + let backoff = Backoff::new(); + loop { + // Deconstruct the head. + let index = head & (self.mark_bit - 1); + let lap = head & !(self.one_lap - 1); + + // Inspect the corresponding slot. + debug_assert!(index < self.buffer.len()); + let slot = unsafe { self.buffer.get_unchecked(index) }; + let stamp = slot.stamp.load(Ordering::Acquire); + + // If the stamp is ahead of the head by 1, we may drop the message. + if head + 1 == stamp { + head = if index + 1 < self.cap() { + // Same lap, incremented index. + // Set to `{ lap: lap, mark: 0, index: index + 1 }`. + head + 1 + } else { + // One lap forward, index wraps around to zero. + // Set to `{ lap: lap.wrapping_add(1), mark: 0, index: 0 }`. + lap.wrapping_add(self.one_lap) + }; + + unsafe { + (*slot.msg.get()).assume_init_drop(); + } + // If the tail equals the head, that means the channel is empty. + } else if tail == head { + return; + // Otherwise, a sender is about to write into the slot, so we need + // to wait for it to update the stamp. + } else { + backoff.spin(); + } + } + } + /// Returns `true` if the channel is disconnected. pub(crate) fn is_disconnected(&self) -> bool { self.tail.load(Ordering::SeqCst) & self.mark_bit != 0 @@ -525,45 +606,6 @@ impl Channel { } } -impl Drop for Channel { - fn drop(&mut self) { - if mem::needs_drop::() { - // Get the index of the head. - let head = *self.head.get_mut(); - let tail = *self.tail.get_mut(); - - let hix = head & (self.mark_bit - 1); - let tix = tail & (self.mark_bit - 1); - - let len = if hix < tix { - tix - hix - } else if hix > tix { - self.cap() - hix + tix - } else if (tail & !self.mark_bit) == head { - 0 - } else { - self.cap() - }; - - // Loop over all slots that hold a message and drop them. - for i in 0..len { - // Compute the index of the next slot holding a message. - let index = if hix + i < self.cap() { - hix + i - } else { - hix + i - self.cap() - }; - - unsafe { - debug_assert!(index < self.buffer.len()); - let slot = self.buffer.get_unchecked_mut(index); - (*slot.msg.get()).assume_init_drop(); - } - } - } - } -} - /// Receiver handle to a channel. pub(crate) struct Receiver<'a, T>(&'a Channel); diff --git a/crossbeam-channel/tests/array.rs b/crossbeam-channel/tests/array.rs index 080a391de..c40a120af 100644 --- a/crossbeam-channel/tests/array.rs +++ b/crossbeam-channel/tests/array.rs @@ -707,3 +707,15 @@ fn panic_on_drop() { // Elements after the panicked element will leak. assert!(!b); } + +#[test] +fn drop_unreceived() { + let (tx, rx) = bounded::>(1); + let msg = std::rc::Rc::new(()); + let weak = std::rc::Rc::downgrade(&msg); + assert!(tx.send(msg).is_ok()); + drop(rx); + // Messages should be dropped immediately when the last receiver is destroyed. + assert!(weak.upgrade().is_none()); + drop(tx); +} From 5e1c7b5e158a9bc1f0a9b5a506d8e20420da1199 Mon Sep 17 00:00:00 2001 From: Taiki Endo Date: Mon, 8 Jun 2026 02:47:59 +0900 Subject: [PATCH 2/4] handle leaks in drop_unreceived test --- ci/miri.sh | 2 +- crossbeam-channel/tests/array.rs | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/ci/miri.sh b/ci/miri.sh index 6c912c07b..46d08ef2d 100755 --- a/ci/miri.sh +++ b/ci/miri.sh @@ -21,7 +21,7 @@ case "${group}" in # -Zmiri-ignore-leaks is needed because we use detached threads in tests in tests/golang.rs: https://github.com/rust-lang/miri/issues/1371 MIRIFLAGS="${MIRIFLAGS} -Zmiri-ignore-leaks" \ cargo miri test --all-features \ - -p crossbeam-channel --test golang 2>&1 | ts -i '%.s ' + -p crossbeam-channel --test golang --test array 2>&1 | ts -i '%.s ' ;; others) cargo miri test --all-features \ diff --git a/crossbeam-channel/tests/array.rs b/crossbeam-channel/tests/array.rs index c40a120af..f07d8b096 100644 --- a/crossbeam-channel/tests/array.rs +++ b/crossbeam-channel/tests/array.rs @@ -710,6 +710,9 @@ fn panic_on_drop() { #[test] fn drop_unreceived() { + if option_env!("MIRI_LEAK_CHECK").is_some() || option_env!("ASAN_OPTIONS").is_some() { + return; + } let (tx, rx) = bounded::>(1); let msg = std::rc::Rc::new(()); let weak = std::rc::Rc::downgrade(&msg); From f462b470e5927d71d02b78c2c06939802211b7df Mon Sep 17 00:00:00 2001 From: Taiki Endo Date: Mon, 8 Jun 2026 02:55:33 +0900 Subject: [PATCH 3/4] fully port rust-lang/rust's change Co-authored-by: joboet --- crossbeam-channel/src/flavors/array.rs | 12 +++++------- crossbeam-channel/tests/array.rs | 7 ++++--- 2 files changed, 9 insertions(+), 10 deletions(-) diff --git a/crossbeam-channel/src/flavors/array.rs b/crossbeam-channel/src/flavors/array.rs index 663e46b83..74a9d04a6 100644 --- a/crossbeam-channel/src/flavors/array.rs +++ b/crossbeam-channel/src/flavors/array.rs @@ -31,7 +31,8 @@ struct Slot { /// The current stamp. stamp: AtomicUsize, - /// The message in this slot. + /// The message in this slot. Either read out in `read` or dropped through + /// `discard_all_messages`. msg: UnsafeCell>, } @@ -480,7 +481,7 @@ impl Channel { Some(self.cap()) } - /// Disconnects senders and wakes up all blocked senders and receivers. + /// Disconnects senders and wakes up all blocked receivers. /// /// Returns `true` if this call disconnected the channel. pub(crate) fn disconnect_senders(&self) -> bool { @@ -511,10 +512,7 @@ impl Channel { false }; - unsafe { - self.discard_all_messages(tail); - } - + unsafe { self.discard_all_messages(tail) } disconnected } @@ -571,7 +569,7 @@ impl Channel { // Otherwise, a sender is about to write into the slot, so we need // to wait for it to update the stamp. } else { - backoff.spin(); + backoff.snooze(); } } } diff --git a/crossbeam-channel/tests/array.rs b/crossbeam-channel/tests/array.rs index f07d8b096..4ae379991 100644 --- a/crossbeam-channel/tests/array.rs +++ b/crossbeam-channel/tests/array.rs @@ -2,6 +2,7 @@ use std::{ any::Any, + rc::Rc, sync::atomic::{AtomicUsize, Ordering}, thread, time::Duration, @@ -713,9 +714,9 @@ fn drop_unreceived() { if option_env!("MIRI_LEAK_CHECK").is_some() || option_env!("ASAN_OPTIONS").is_some() { return; } - let (tx, rx) = bounded::>(1); - let msg = std::rc::Rc::new(()); - let weak = std::rc::Rc::downgrade(&msg); + let (tx, rx) = bounded::>(1); + let msg = Rc::new(()); + let weak = Rc::downgrade(&msg); assert!(tx.send(msg).is_ok()); drop(rx); // Messages should be dropped immediately when the last receiver is destroyed. From 6cf4d2d0c4f53da46c4e104074bd1f806bb85f66 Mon Sep 17 00:00:00 2001 From: Taiki Endo Date: Mon, 8 Jun 2026 03:16:03 +0900 Subject: [PATCH 4/4] fix "handle leaks in drop_unreceived test" --- crossbeam-channel/tests/array.rs | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/crossbeam-channel/tests/array.rs b/crossbeam-channel/tests/array.rs index 4ae379991..d653d0143 100644 --- a/crossbeam-channel/tests/array.rs +++ b/crossbeam-channel/tests/array.rs @@ -691,6 +691,10 @@ fn panic_on_drop() { assert!(a); assert!(b); + if option_env!("MIRI_LEAK_CHECK").is_some() || option_env!("ASAN_OPTIONS").is_some() { + return; + } + // panic on drop let (s, r) = bounded(2); let (mut a, mut b) = (false, false); @@ -711,9 +715,6 @@ fn panic_on_drop() { #[test] fn drop_unreceived() { - if option_env!("MIRI_LEAK_CHECK").is_some() || option_env!("ASAN_OPTIONS").is_some() { - return; - } let (tx, rx) = bounded::>(1); let msg = Rc::new(()); let weak = Rc::downgrade(&msg);