diff --git a/tokio/src/io/util/mem.rs b/tokio/src/io/util/mem.rs index 74c8cf149..beb9d6ca3 100644 --- a/tokio/src/io/util/mem.rs +++ b/tokio/src/io/util/mem.rs @@ -250,6 +250,10 @@ impl SimplexStream { cx: &mut task::Context<'_>, buf: &mut ReadBuf<'_>, ) -> Poll> { + 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()); diff --git a/tokio/tests/io_mem_stream.rs b/tokio/tests/io_mem_stream.rs index e79ec4e9d..9c4304203 100644 --- a/tokio/tests/io_mem_stream.rs +++ b/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);