io: complete zero-length memory stream operations (#8323)

tokio/src/io/util/mem.rs

@@ -250,6 +250,10 @@ impl SimplexStream {
         cx: &mut task::Context<'_>,
         buf: &mut ReadBuf<'_>,
     ) -> Poll<std::io::Result<()>> {
+        if buf.remaining() == 0 {
+            return Poll::Ready(Ok(()));
+        }
+
         if self.buffer.has_remaining() {
             let max = self.buffer.remaining().min(buf.remaining());
             buf.put_slice(&self.buffer[..max]);
@@ -278,6 +282,10 @@ impl SimplexStream {
         if self.is_closed {
             return Poll::Ready(Err(std::io::ErrorKind::BrokenPipe.into()));
         }
+        if buf.is_empty() {
+            return Poll::Ready(Ok(0));
+        }
+
         let avail = self.max_buf_size - self.buffer.len();
         if avail == 0 {
             self.write_waker = Some(cx.waker().clone());
@@ -300,6 +308,10 @@ impl SimplexStream {
         if self.is_closed {
             return Poll::Ready(Err(std::io::ErrorKind::BrokenPipe.into()));
         }
+        if bufs.iter().all(|buf| buf.is_empty()) {
+            return Poll::Ready(Ok(0));
+        }
+
         let avail = self.max_buf_size - self.buffer.len();
         if avail == 0 {
             self.write_waker = Some(cx.waker().clone());

tokio/tests/io_mem_stream.rs

@@ -1,6 +1,8 @@
 #![warn(rust_2018_idioms)]
 #![cfg(feature = "full")]
 
+use futures::FutureExt;
+use std::io::IoSlice;
 use tokio::io::{duplex, AsyncReadExt, AsyncWriteExt};
 
 #[tokio::test]
@@ -101,6 +103,38 @@ async fn max_write_size() {
     drop(b);
 }
 
+#[tokio::test]
+async fn zero_length_operations() {
+    let (mut reader, _peer) = duplex(1);
+    assert!(matches!(reader.read(&mut []).now_or_never(), Some(Ok(0))));
+
+    let (mut writer, _peer) = duplex(1);
+    writer.write_all(b"x").await.unwrap();
+    assert!(matches!(writer.write(&[]).now_or_never(), Some(Ok(0))));
+
+    let (mut writer, _peer) = duplex(1);
+    writer.write_all(b"x").await.unwrap();
+    let bufs = [IoSlice::new(&[]), IoSlice::new(&[])];
+    assert!(matches!(
+        writer.write_vectored(&bufs).now_or_never(),
+        Some(Ok(0))
+    ));
+}
+
+#[tokio::test]
+async fn zero_length_writes_to_closed_stream() {
+    let (mut writer, peer) = duplex(1);
+    drop(peer);
+    let err = writer.write(&[]).await.unwrap_err();
+    assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe);
+
+    let (mut writer, peer) = duplex(1);
+    drop(peer);
+    let bufs = [IoSlice::new(&[]), IoSlice::new(&[])];
+    let err = writer.write_vectored(&bufs).await.unwrap_err();
+    assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe);
+}
+
 #[tokio::test]
 async fn duplex_is_cooperative() {
     let (mut tx, mut rx) = tokio::io::duplex(1024 * 8);