fix: port fix discard all message on receiver droped

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 \

crossbeam-channel/src/channel.rs

@@ -675,7 +675,7 @@ impl<T> Drop for Sender<T> {
     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<T> Drop for Receiver<T> {
     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(_) => {}

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},
 };
@@ -31,7 +31,8 @@ struct Slot<T> {
     /// 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<MaybeUninit<T>>,
 }
 
@@ -480,21 +481,99 @@ impl<T> Channel<T> {
         Some(self.cap())
     }
 
-    /// Disconnects the channel 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(&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 {
             false
         }
     }
 
+    /// 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.snooze();
+            }
+        }
+    }
+
     /// 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 +604,6 @@ impl<T> Channel<T> {
     }
 }
 
-impl<T> Drop for Channel<T> {
-    fn drop(&mut self) {
-        if mem::needs_drop::<T>() {
-            // 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<T>);
 

crossbeam-channel/tests/array.rs

@@ -2,6 +2,7 @@
 
 use std::{
     any::Any,
+    rc::Rc,
     sync::atomic::{AtomicUsize, Ordering},
     thread,
     time::Duration,
@@ -690,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);
@@ -707,3 +712,15 @@ fn panic_on_drop() {
     // Elements after the panicked element will leak.
     assert!(!b);
 }
+
+#[test]
+fn drop_unreceived() {
+    let (tx, rx) = bounded::<Rc<()>>(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.
+    assert!(weak.upgrade().is_none());
+    drop(tx);
+}