Simplify ForceSendError, and discard oldest message rather than most recent one first

Note that this is a very naive implementation of force_send, that works when used within a single thread, but can easily fail when multiple threads/cores are involved.

Author: bergkvist

crossbeam-channel/src/channel.rs

@@ -418,29 +418,38 @@ impl<T> Sender<T> {
     /// the channel is full. The returned error contains the original message.
     ///
     /// If called on a zero-capacity channel, this method will send the message only if there
-    /// happens to be a receive operation on the other side of the channel at the same time.
+    /// happens to be a pending receive operation on the other side of the channel at the same time,
+    /// otherwise it will return the value to the sender.
     ///
     /// ```
     /// use crossbeam_channel::{bounded, ForceSendError};
     ///
     /// let (s, r) = bounded(3);
+    ///
     /// assert_eq!(s.force_send(0), Ok(None));
     /// assert_eq!(s.force_send(1), Ok(None));
     /// assert_eq!(s.force_send(2), Ok(None));
-    /// assert_eq!(s.force_send(3), Ok(Some(2)));
-    /// assert_eq!(r.recv(), Ok(0));
-    /// assert_eq!(s.force_send(4), Ok(None));
+    /// assert_eq!(s.force_send(3), Ok(Some(0)));
+    ///
     /// assert_eq!(r.recv(), Ok(1));
+    ///
+    /// assert_eq!(s.force_send(4), Ok(None));
+    ///
+    /// assert_eq!(r.recv(), Ok(2));
     /// assert_eq!(r.recv(), Ok(3));
+    ///
     /// assert_eq!(s.force_send(5), Ok(None));
     /// assert_eq!(s.force_send(6), Ok(None));
-    /// assert_eq!(s.force_send(7), Ok(Some(6)));
-    /// assert_eq!(s.force_send(8), Ok(Some(7)));
-    /// assert_eq!(r.recv(), Ok(4));
-    /// assert_eq!(r.recv(), Ok(5));
+    /// assert_eq!(s.force_send(7), Ok(Some(4)));
+    /// assert_eq!(s.force_send(8), Ok(Some(5)));
+    ///
+    /// assert_eq!(r.recv(), Ok(6));
+    /// assert_eq!(r.recv(), Ok(7));
     /// assert_eq!(r.recv(), Ok(8));
+    ///
     /// drop(r);
-    /// assert_eq!(s.force_send(9), Err(ForceSendError::Disconnected(9)));
+    ///
+    /// assert_eq!(s.force_send(9), Err(ForceSendError(9)));
     /// ``````
     pub fn force_send(&self, msg: T) -> Result<Option<T>, ForceSendError<T>> {
         match &self.flavor {

crossbeam-channel/src/err.rs

@@ -13,14 +13,13 @@ pub struct SendError<T>(pub T);
 
 /// An error returned from the [`force_send`] method.
 ///
-/// The error contains the message being sent so it can be recovered.
+/// The message could not be sent because the channel is disconnected.
+///
+/// The error contains the message so it can be recovered.
 ///
 /// [`force_send`]: super::Sender::force_send
 #[derive(PartialEq, Eq, Clone, Copy)]
-pub enum ForceSendError<T> {
-    /// The message could not be sent because the channel is disconnected.
-    Disconnected(T),
-}
+pub struct ForceSendError<T>(pub T);
 
 /// An error returned from the [`try_send`] method.
 ///
@@ -223,15 +222,15 @@ impl<T> TrySendError<T> {
 impl<T> fmt::Debug for ForceSendError<T> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match *self {
-            Self::Disconnected(..) => "Disconnected(..)".fmt(f),
+            Self(..) => "ForceSendError(..)".fmt(f),
         }
     }
 }
 
 impl<T> fmt::Display for ForceSendError<T> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match *self {
-            Self::Disconnected(..) => "sending on a disconnected channel".fmt(f),
+            Self(..) => "sending on a disconnected channel".fmt(f),
         }
     }
 }
@@ -241,7 +240,7 @@ impl<T: Send> error::Error for ForceSendError<T> {}
 impl<T> From<SendError<T>> for ForceSendError<T> {
     fn from(err: SendError<T>) -> Self {
         match err {
-            SendError(t) => Self::Disconnected(t),
+            SendError(t) => Self(t),
         }
     }
 }
@@ -263,14 +262,9 @@ impl<T> ForceSendError<T> {
     /// ```
     pub fn into_inner(self) -> T {
         match self {
-            Self::Disconnected(v) => v,
+            Self(v) => v,
         }
     }
-
-    /// Returns `true` if the send operation failed because the channel is disconnected.
-    pub fn is_disconnected(&self) -> bool {
-        matches!(self, Self::Disconnected(_))
-    }
 }
 
 impl<T> fmt::Debug for SendTimeoutError<T> {

crossbeam-channel/src/flavors/array.rs

@@ -326,24 +326,26 @@ impl<T> Channel<T> {
     }
 
     /// Force send a message into the channel. Only fails if the channel is disconnected
-    pub(crate) fn force_send(&self, mut msg: T) -> Result<Option<T>, ForceSendError<T>> {
-        let mut token = Token::default();
-        if self.start_send(&mut token) {
-            match unsafe { self.write(&mut token, msg) } {
-                Ok(()) => Ok(None),
-                Err(msg) => Err(ForceSendError::Disconnected(msg)),
-            }
-        } else {
-            let tail = self.tail.load(Ordering::Acquire);
-            let prev_index = match tail & (self.mark_bit - 1) {
-                0 => self.cap() - 1,
-                x => x - 1,
-            };
-            let queued_msg =
-                unsafe { (*self.buffer.get_unchecked(prev_index).msg.get()).assume_init_mut() };
-            std::mem::swap(&mut msg, queued_msg);
-            Ok(Some(msg))
+    ///
+    /// Note that this is currently a naive implementation to make the sequential test pass
+    pub(crate) fn force_send(&self, msg: T) -> Result<Option<T>, ForceSendError<T>> {
+        if self.is_disconnected() {
+            return Err(ForceSendError(msg));
         }
+
+        let old_msg = if self.is_full() {
+            let Ok(old_msg) = self.try_recv() else {
+                return Err(ForceSendError(msg));
+            };
+            Some(old_msg)
+        } else {
+            None
+        };
+
+        self.try_send(msg)
+            .map_err(|e| ForceSendError(e.into_inner()))?;
+
+        Ok(old_msg)
     }
 
     /// Sends a message into the channel.

crossbeam-channel/src/flavors/list.rs

@@ -421,7 +421,7 @@ impl<T> Channel<T> {
     pub(crate) fn force_send(&self, msg: T) -> Result<Option<T>, ForceSendError<T>> {
         match self.send(msg, None) {
             Ok(()) => Ok(None),
-            Err(SendTimeoutError::Disconnected(err)) => Err(ForceSendError::Disconnected(err)),
+            Err(SendTimeoutError::Disconnected(err)) => Err(ForceSendError(err)),
             Err(SendTimeoutError::Timeout(_)) => unreachable!(),
         }
     }

crossbeam-channel/src/flavors/zero.rs

@@ -230,7 +230,7 @@ impl<T> Channel<T> {
             }
             Ok(None)
         } else if inner.is_disconnected {
-            Err(ForceSendError::Disconnected(msg))
+            Err(ForceSendError(msg))
         } else {
             Ok(Some(msg))
         }