watch: add check() method on sender

surban · surban · commit f160490db97f · 2024-03-13T21:35:54.000+01:00
This allows simple checking whether an item-specific send
error has occurred after the channel has been closed.
diff --git a/remoc/src/rch/watch/mod.rs b/remoc/src/rch/watch/mod.rs
@@ -63,9 +63,6 @@ mod sender;
 pub use receiver::{ChangedError, Receiver, ReceiverStream, RecvError};
 pub use sender::{SendError, Sender};
 
-/// Length of queuing for storing errors that occurred during remote send.
-const ERROR_QUEUE: usize = 16;
-
 /// Returns a reference to the inner value.
 pub struct Ref<'a, T>(tokio::sync::watch::Ref<'a, Result<T, RecvError>>);
 
@@ -94,7 +91,7 @@ where
     T: RemoteSend,
 {
     let (tx, rx) = tokio::sync::watch::channel(Ok(init));
-    let (remote_send_err_tx, remote_send_err_rx) = tokio::sync::mpsc::channel(ERROR_QUEUE);
+    let (remote_send_err_tx, remote_send_err_rx) = tokio::sync::mpsc::unbounded_channel();
 
     let sender = Sender::new(tx, remote_send_err_tx.clone(), remote_send_err_rx, DEFAULT_MAX_ITEM_SIZE);
     let receiver = Receiver::new(rx, remote_send_err_tx, None);
@@ -104,7 +101,7 @@ where
 /// Send implementation for deserializer of Sender and serializer of Receiver.
 async fn send_impl<T, Codec>(
     mut rx: tokio::sync::watch::Receiver<Result<T, RecvError>>, raw_tx: chmux::Sender,
-    mut raw_rx: chmux::Receiver, remote_send_err_tx: tokio::sync::mpsc::Sender<RemoteSendError>,
+    mut raw_rx: chmux::Receiver, remote_send_err_tx: tokio::sync::mpsc::UnboundedSender<RemoteSendError>,
     max_item_size: usize,
 ) where
     T: Serialize + Send + Clone + 'static,
@@ -124,7 +121,7 @@ async fn send_impl<T, Codec>(
                 match backchannel_msg {
                     Ok(Some(mut msg)) if msg.remaining() >= 1 => {
                         if msg.get_u8() == BACKCHANNEL_MSG_ERROR {
-                            let _ = remote_send_err_tx.try_send(RemoteSendError::Forward);
+                            let _ = remote_send_err_tx.send(RemoteSendError::Forward);
                         }
                     }
                     _ => break,
@@ -137,7 +134,7 @@ async fn send_impl<T, Codec>(
                     Ok(()) => {
                         let value = rx.borrow_and_update().clone();
                         if let Err(err) = remote_tx.send(value).await {
-                            let _ = remote_send_err_tx.try_send(RemoteSendError::Send(err.kind.clone()));
+                            let _ = remote_send_err_tx.send(RemoteSendError::Send(err.kind.clone()));
                             if err.is_item_specific() {
                                 break
                             }
@@ -153,7 +150,7 @@ async fn send_impl<T, Codec>(
 /// Receive implementation for serializer of Sender and deserializer of Receiver.
 async fn recv_impl<T, Codec>(
     tx: tokio::sync::watch::Sender<Result<T, RecvError>>, mut raw_tx: chmux::Sender, raw_rx: chmux::Receiver,
-    mut remote_send_err_rx: tokio::sync::mpsc::Receiver<RemoteSendError>,
+    mut remote_send_err_rx: tokio::sync::mpsc::UnboundedReceiver<RemoteSendError>,
     mut current_err: Option<RemoteSendError>, max_item_size: usize,
 ) where
     T: DeserializeOwned + Send + 'static,
diff --git a/remoc/src/rch/watch/receiver.rs b/remoc/src/rch/watch/receiver.rs
@@ -15,7 +15,7 @@ use super::{
         base::{self, PortDeserializer, PortSerializer},
         RemoteSendError, DEFAULT_MAX_ITEM_SIZE,
     },
-    Ref, ERROR_QUEUE,
+    Ref,
 };
 use crate::{chmux, codec, RemoteSend};
 
@@ -87,7 +87,7 @@ impl Error for ChangedError {}
 #[derive(Clone)]
 pub struct Receiver<T, Codec = codec::Default, const MAX_ITEM_SIZE: usize = DEFAULT_MAX_ITEM_SIZE> {
     rx: tokio::sync::watch::Receiver<Result<T, RecvError>>,
-    remote_send_err_tx: tokio::sync::mpsc::Sender<RemoteSendError>,
+    remote_send_err_tx: tokio::sync::mpsc::UnboundedSender<RemoteSendError>,
     remote_max_item_size: Option<usize>,
     _codec: PhantomData<Codec>,
 }
@@ -115,7 +115,8 @@ pub(crate) struct TransportedReceiver<T, Codec> {
 impl<T, Codec, const MAX_ITEM_SIZE: usize> Receiver<T, Codec, MAX_ITEM_SIZE> {
     pub(crate) fn new(
         rx: tokio::sync::watch::Receiver<Result<T, RecvError>>,
-        remote_send_err_tx: tokio::sync::mpsc::Sender<RemoteSendError>, remote_max_item_size: Option<usize>,
+        remote_send_err_tx: tokio::sync::mpsc::UnboundedSender<RemoteSendError>,
+        remote_max_item_size: Option<usize>,
     ) -> Self {
         Self { rx, remote_send_err_tx, remote_max_item_size, _codec: PhantomData }
     }
@@ -201,7 +202,7 @@ where
                 let (raw_tx, raw_rx) = match connect.await {
                     Ok(tx_rx) => tx_rx,
                     Err(err) => {
-                        let _ = remote_send_err_tx.try_send(RemoteSendError::Connect(err));
+                        let _ = remote_send_err_tx.send(RemoteSendError::Connect(err));
                         return;
                     }
                 };
@@ -248,7 +249,7 @@ where
 
         // Create channels.
         let (tx, rx) = tokio::sync::watch::channel(data);
-        let (remote_send_err_tx, remote_send_err_rx) = tokio::sync::mpsc::channel(ERROR_QUEUE);
+        let (remote_send_err_tx, remote_send_err_rx) = tokio::sync::mpsc::unbounded_channel();
 
         PortDeserializer::accept(port, |local_port, request| {
             async move {
diff --git a/remoc/src/rch/watch/sender.rs b/remoc/src/rch/watch/sender.rs
@@ -1,20 +1,14 @@
-use futures::{task::noop_waker, FutureExt};
+use futures::FutureExt;
 use serde::{Deserialize, Serialize};
-use std::{
-    error::Error,
-    fmt,
-    marker::PhantomData,
-    sync::Mutex,
-    task::{Context, Poll},
-};
+use std::{error::Error, fmt, marker::PhantomData, sync::Mutex};
 
 use super::{
     super::{
         base::{self, PortDeserializer, PortSerializer},
         RemoteSendError, SendErrorExt,
     },
     receiver::RecvError,
-    Receiver, Ref, ERROR_QUEUE,
+    Receiver, Ref,
 };
 use crate::{chmux, codec, RemoteSend};
 
@@ -118,8 +112,8 @@ impl<T, Codec> fmt::Debug for Sender<T, Codec> {
 
 pub(crate) struct SenderInner<T, Codec> {
     tx: tokio::sync::watch::Sender<Result<T, RecvError>>,
-    remote_send_err_tx: tokio::sync::mpsc::Sender<RemoteSendError>,
-    remote_send_err_rx: Mutex<tokio::sync::mpsc::Receiver<RemoteSendError>>,
+    remote_send_err_tx: tokio::sync::mpsc::UnboundedSender<RemoteSendError>,
+    remote_send_err_rx: Mutex<tokio::sync::mpsc::UnboundedReceiver<RemoteSendError>>,
     current_err: Mutex<Option<RemoteSendError>>,
     max_item_size: usize,
     _codec: PhantomData<Codec>,
@@ -150,8 +144,8 @@ where
     /// Creates a new sender.
     pub(crate) fn new(
         tx: tokio::sync::watch::Sender<Result<T, RecvError>>,
-        remote_send_err_tx: tokio::sync::mpsc::Sender<RemoteSendError>,
-        remote_send_err_rx: tokio::sync::mpsc::Receiver<RemoteSendError>, max_item_size: usize,
+        remote_send_err_tx: tokio::sync::mpsc::UnboundedSender<RemoteSendError>,
+        remote_send_err_rx: tokio::sync::mpsc::UnboundedReceiver<RemoteSendError>, max_item_size: usize,
     ) -> Self {
         let inner = SenderInner {
             tx,
@@ -267,6 +261,28 @@ where
         *current_err = None;
     }
 
+    /// Checks that no item-specific send error has occurred.
+    ///
+    /// This method clears non-item-specific errors present on the channel.
+    ///
+    /// # Error reporting
+    /// Sending and error reporting are done asynchronously.
+    /// Thus, the reporting of an error may be delayed.
+    ///
+    /// To verify that no item-specific send error has occurred during the lifetime of
+    /// the channel, call this method after the channel is closed, i.e.
+    /// [`closed`](Self::closed) has returned or [`is_closed`](Self::is_closed) is
+    /// `true`.
+    pub fn check(&mut self) -> Result<(), SendError> {
+        while let Some(err) = self.error() {
+            if err.is_item_specific() {
+                return Err(err);
+            }
+            self.clear_error();
+        }
+        Ok(())
+    }
+
     /// Maximum allowed item size in bytes.
     pub fn max_item_size(&self) -> usize {
         self.inner.as_ref().unwrap().max_item_size
@@ -361,7 +377,7 @@ where
 
         // Create internal communication channels.
         let (tx, rx) = tokio::sync::watch::channel(data);
-        let (remote_send_err_tx, remote_send_err_rx) = tokio::sync::mpsc::channel(ERROR_QUEUE);
+        let (remote_send_err_tx, remote_send_err_rx) = tokio::sync::mpsc::unbounded_channel();
         let remote_send_err_tx2 = remote_send_err_tx.clone();
 
         // Accept chmux port request.
@@ -371,7 +387,7 @@ where
                 let (raw_tx, raw_rx) = match request.accept_from(local_port).await {
                     Ok(tx_rx) => tx_rx,
                     Err(err) => {
-                        let _ = remote_send_err_tx.try_send(RemoteSendError::Listen(err));
+                        let _ = remote_send_err_tx.send(RemoteSendError::Listen(err));
                         return;
                     }
                 };
diff --git a/remoc/tests/rch/watch.rs b/remoc/tests/rch/watch.rs
@@ -15,7 +15,7 @@ async fn simple() {
     let end_value = 124;
 
     println!("Sending remote mpsc channel receiver");
-    let (tx, rx) = watch::channel(start_value);
+    let (mut tx, rx) = watch::channel(start_value);
     a_tx.send(rx).await.unwrap();
     println!("Receiving remote mpsc channel receiver");
     let mut rx = b_rx.recv().await.unwrap().unwrap();
@@ -47,6 +47,8 @@ async fn simple() {
             sleep(Duration::from_millis(20)).await;
         }
     }
+
+    tx.check().unwrap();
     drop(tx);
 
     println!("Waiting for receive task");
@@ -62,7 +64,7 @@ async fn simple_stream() {
     let end_value = 124;
 
     println!("Sending remote mpsc channel receiver");
-    let (tx, rx) = watch::channel(start_value);
+    let (mut tx, rx) = watch::channel(start_value);
     a_tx.send(rx).await.unwrap();
     println!("Receiving remote mpsc channel receiver");
     let rx = b_rx.recv().await.unwrap().unwrap();
@@ -94,6 +96,8 @@ async fn simple_stream() {
             prev_value -= 1;
         }
     }
+
+    tx.check().unwrap();
     drop(tx);
 
     println!("Waiting for receive task");
@@ -109,7 +113,7 @@ async fn close() {
     let (tx, rx) = watch::channel(123);
     a_tx.send(tx).await.unwrap();
     println!("Receiving remote mpsc channel sender");
-    let tx = b_rx.recv().await.unwrap().unwrap();
+    let mut tx = b_rx.recv().await.unwrap().unwrap();
 
     println!("Cloning receiver");
     let rx2 = rx.clone();
@@ -126,6 +130,7 @@ async fn close() {
     println!("Waiting for close notification");
     tx.closed().await;
     assert!(tx.is_closed());
+    tx.check().unwrap();
 
     println!("Attempting to send");
     match tx.send(15) {
@@ -144,7 +149,7 @@ async fn conn_failure() {
     let (tx, rx) = watch::channel(123);
     a_tx.send(tx).await.unwrap();
     println!("Receiving remote mpsc channel sender");
-    let tx = b_rx.recv().await.unwrap().unwrap();
+    let mut tx = b_rx.recv().await.unwrap().unwrap();
 
     println!("Cloning receiver");
     let _rx2 = rx.clone();
@@ -157,6 +162,7 @@ async fn conn_failure() {
     println!("Waiting for close notification");
     tx.closed().await;
     assert!(tx.is_closed());
+    tx.check().unwrap();
 
     println!("Attempting to send");
     match tx.send(15) {
@@ -229,4 +235,65 @@ async fn max_item_size_exceeded() {
     assert!(matches!(tx.error(), Some(SendError::RemoteSend(SendErrorKind::MaxItemSizeExceeded))));
     tx.clear_error();
     assert!(matches!(tx.error(), None));
+    tx.check().unwrap();
+}
+
+#[tokio::test]
+async fn max_item_size_exceeded_check() {
+    crate::init();
+    let ((mut a_tx, _), (_, mut b_rx)) = loop_channel::<watch::Receiver<Vec<u8>>>().await;
+
+    println!("Sending remote mpsc channel receiver");
+    let (mut tx, rx) = watch::channel(Vec::new());
+    a_tx.send(rx).await.unwrap();
+    println!("Receiving remote mpsc channel receiver");
+    let mut rx = b_rx.recv().await.unwrap().unwrap();
+
+    assert_eq!(tx.max_item_size(), rx.max_item_size());
+    let max_item_size = tx.max_item_size();
+    println!("Maximum send and recv item size is {max_item_size}");
+
+    {
+        let value = rx.borrow().unwrap();
+        println!("Initial value: {value:?}");
+    }
+
+    let recv_task = tokio::spawn(async move {
+        loop {
+            let res = rx.changed().await;
+            println!("RX changed result: {res:?}");
+            if res.is_err() {
+                break res;
+            }
+
+            let value = rx.borrow_and_update().unwrap().clone();
+            println!("Received value change: {} elements", value.len());
+        }
+    });
+
+    // Happy case: sent data size is under limit.
+    // JSON encoding will result in much larger transfer size.
+    let elems = max_item_size / 10;
+    println!("Sending {elems} elements");
+    let value = vec![100; elems];
+    tx.send(value.clone()).unwrap();
+    assert_eq!(*tx.borrow(), value);
+
+    sleep(Duration::from_millis(100)).await;
+
+    // Failure case: sent data size exceeds limits.
+    let elems = max_item_size * 10;
+    println!("Sending {elems} elements");
+    let value = vec![100; elems];
+    tx.send(value.clone()).unwrap();
+    assert_eq!(*tx.borrow(), value);
+
+    println!("Wait for sender close");
+    tx.closed().await;
+    let res = tx.check();
+    println!("Sender check result: {res:?}");
+    assert!(matches!(res, Err(SendError::RemoteSend(SendErrorKind::MaxItemSizeExceeded))));
+
+    println!("Waiting for receive task");
+    assert!(matches!(recv_task.await.unwrap(), Err(ChangedError::Closed)));
 }
