Prevent blocking all virtual serial ports on full PTS

README.md

@@ -78,10 +78,18 @@ In terminal 2
 picocom 2
 ```
 
-Characters entered in terminal 0 will be sent to terminal 2 only.
-Characters entered in terminal 1 will be sent to terminal 2 only.
+Characters entered in terminal 0 will be sent to terminal 2 only.\
+Characters entered in terminal 1 will be sent to terminal 2 only.\
 Characters entered in terminal 2 will be sent to both terminals 0 and 1.
 
+## Connection to a Virtual Serial Port
+
+Virtual serial ports created by vsp-router behave a bit differently from physical serial ports:
+* You can connect to them using any baud rate. You are not forced to use the same baud rate as the physical serial port you are multiplexing.
+* When you don't read from them they accumulate data in a buffer. When this buffer becomes full new data will be discarded and a warning message will be shown in the logs. Any buffered data will get returned when you next read from them.
+
+To avoid reading stale data accumulated in the buffer when you want to read from the virtual serial port it is recommended to flush its input buffer before you first read from it. This can be done using `tcflush(fd, TCIFLUSH)` (or equivalent on your platform) on the file descriptor of the virtual serial port.
+
 ## Comparison to TTYBUS
 
 vsp-router is similar to [TTYBUS](https://github.com/danielinux/ttybus).

src/lib.rs

@@ -1,16 +1,20 @@
+use bytes::{Buf, Bytes};
 use camino::{Utf8Path, Utf8PathBuf};
 use thiserror::Error;
-use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
+use tokio::io::{AsyncRead, AsyncWrite};
 #[cfg(unix)]
 use tokio_serial::SerialPort;
 use tokio_serial::SerialPortBuilderExt;
 use tokio_serial::SerialStream;
 use tokio_stream::{StreamExt, StreamMap};
 use tokio_util::io::ReaderStream;
-use tracing::{error, info};
+use tracing::{error, info, warn};
 
 use std::collections::HashMap;
 use std::fs;
+use std::io::ErrorKind;
+use std::pin::Pin;
+use std::task::Poll::Ready;
 
 #[cfg(unix)]
 use std::os::unix;
@@ -108,15 +112,41 @@ where
             let bytes = result.map_err(Error::Read)?;
             info!(?src_id, ?dst_ids, ?bytes, "read");
             for dst_id in dst_ids {
-                // This unwrap is OK as long as we validate all route IDs exist first
-                // Route IDs are validated in Args::check_route_ids()
-                let dst = sinks.get_mut(dst_id).unwrap();
-                let mut buf = bytes.clone();
-                dst.write_all_buf(&mut buf).await.map_err(Error::Write)?;
-                info!(?dst_id, ?bytes, "wrote");
+                if let Some(dst) = sinks.get_mut(dst_id) {
+                    let mut buf = bytes.clone();
+                    if let Err(e) = write_non_blocking(dst, &mut buf).await {
+                        if let Error::Write(io_err) = &e {
+                            if io_err.kind() == ErrorKind::WouldBlock {
+                                warn!(?dst_id, ?bytes, "discarded");
+                            } else {
+                                error!(?dst_id, ?e, "write error");
+                            }
+                        }
+                    } else {
+                        info!(?dst_id, ?bytes, "wrote");
+                    }
+                }
             }
         }
     }
 
     Ok(())
 }
+
+async fn write_non_blocking<W: AsyncWrite + Unpin>(dst: &mut W, buf: &mut Bytes) -> Result<()> {
+    let waker = futures::task::noop_waker();
+    let mut cx = futures::task::Context::from_waker(&waker);
+
+    let pinned_dst = Pin::new(dst);
+    match pinned_dst.poll_write(&mut cx, buf) {
+        Ready(Ok(bytes_written)) => {
+            buf.advance(bytes_written);
+            Ok(())
+        }
+        Ready(Err(e)) => Err(Error::Write(e)),
+        _ => Err(Error::Write(std::io::Error::new(
+            ErrorKind::WouldBlock,
+            "Would block",
+        ))),
+    }
+}