feat(io): add duplex forwarding for BufReader/BufWriter (#695)

... and optimize repeat::read_vectored

* feat(io): add duplex forwarding for BufReader and BufWriter

* perf(io): add read_vectored fast path for repeat

Author: Aalivexy

compio-io/src/read/buf.rs

@@ -1,6 +1,6 @@
-use compio_buf::{BufResult, IntoInner, IoBuf, IoBufMut, IoVectoredBufMut, buf_try};
+use compio_buf::{BufResult, IntoInner, IoBuf, IoBufMut, IoVectoredBuf, IoVectoredBufMut, buf_try};
 
-use crate::{AsyncRead, IoResult, buffer::Buffer, util::DEFAULT_BUF_SIZE};
+use crate::{AsyncRead, AsyncWrite, IoResult, buffer::Buffer, util::DEFAULT_BUF_SIZE};
 /// # AsyncBufRead
 ///
 /// Async read with buffered content.
@@ -34,6 +34,10 @@ impl<A: AsyncBufRead + ?Sized> AsyncBufRead for &mut A {
 /// times. It also provides no advantage when reading from a source that is
 /// already in memory, like a `Vec<u8>`.
 ///
+/// If the underlying reader also implements [`AsyncWrite`], `BufReader<R>`
+/// forwards write operations directly to the inner writer without touching the
+/// read buffer.
+///
 /// When the `BufReader<R>` is dropped, the contents of its buffer will be
 /// discarded. Reading from the underlying reader after unwrapping the
 /// `BufReader<R>` with [`BufReader::into_inner`] can cause data loss.
@@ -112,6 +116,24 @@ impl<R: AsyncRead> AsyncBufRead for BufReader<R> {
     }
 }
 
+impl<R: AsyncRead + AsyncWrite> AsyncWrite for BufReader<R> {
+    async fn write<B: IoBuf>(&mut self, buf: B) -> BufResult<usize, B> {
+        self.reader.write(buf).await
+    }
+
+    async fn write_vectored<B: IoVectoredBuf>(&mut self, buf: B) -> BufResult<usize, B> {
+        self.reader.write_vectored(buf).await
+    }
+
+    async fn flush(&mut self) -> IoResult<()> {
+        self.reader.flush().await
+    }
+
+    async fn shutdown(&mut self) -> IoResult<()> {
+        self.reader.shutdown().await
+    }
+}
+
 impl<R> IntoInner for BufReader<R> {
     type Inner = R;
 

compio-io/src/util/repeat.rs

@@ -1,6 +1,6 @@
 use std::mem::MaybeUninit;
 
-use compio_buf::BufResult;
+use compio_buf::{BufResult, IoVectoredBufMut};
 
 use crate::{AsyncBufRead, AsyncRead, IoResult};
 
@@ -35,10 +35,26 @@ impl AsyncRead for Repeat {
 
         let len = slice.len();
         slice.fill(MaybeUninit::new(self.0));
+        // SAFETY: we just initialized exactly `len` bytes in `buf` from index 0.
         unsafe { buf.advance_to(len) };
 
         BufResult(Ok(len), buf)
     }
+
+    async fn read_vectored<V: IoVectoredBufMut>(&mut self, mut buf: V) -> BufResult<usize, V> {
+        let mut len: usize = 0;
+        for slice in buf.iter_uninit_slice() {
+            len = len
+                .checked_add(slice.len())
+                .expect("total vectored buffer length overflow");
+            slice.fill(MaybeUninit::new(self.0));
+        }
+        debug_assert_eq!(len, buf.total_capacity());
+        // SAFETY: every byte counted in `len` is initialized in the loop above.
+        unsafe { buf.advance_vec_to(len) };
+
+        BufResult(Ok(len), buf)
+    }
 }
 
 impl AsyncBufRead for Repeat {

compio-io/src/write/buf.rs

@@ -1,7 +1,7 @@
-use compio_buf::{BufResult, IntoInner, IoBuf, IoBufMut, IoVectoredBuf, buf_try};
+use compio_buf::{BufResult, IntoInner, IoBuf, IoBufMut, IoVectoredBuf, IoVectoredBufMut, buf_try};
 
 use crate::{
-    AsyncWrite, IoResult,
+    AsyncBufRead, AsyncRead, AsyncWrite, IoResult,
     buffer::Buffer,
     util::{DEFAULT_BUF_SIZE, slice_to_buf},
 };
@@ -18,6 +18,10 @@ use crate::{
 /// times. It also provides no advantage when writing to a destination that is
 /// in memory, like a `Vec<u8>`.
 ///
+/// If the underlying writer also implements [`AsyncRead`] or [`AsyncBufRead`],
+/// `BufWriter<W>` forwards read operations directly to the inner reader without
+/// flushing the write buffer.
+///
 /// Dropping `BufWriter<W>` also discards any bytes left in the buffer, so it is
 /// critical to call [`flush`] before `BufWriter<W>` is dropped. Calling
 /// [`flush`] ensures that the buffer is empty and thus no data is lost.
@@ -116,6 +120,26 @@ impl<W: AsyncWrite> AsyncWrite for BufWriter<W> {
     }
 }
 
+impl<W: AsyncRead + AsyncWrite> AsyncRead for BufWriter<W> {
+    async fn read<B: IoBufMut>(&mut self, buf: B) -> BufResult<usize, B> {
+        self.writer.read(buf).await
+    }
+
+    async fn read_vectored<V: IoVectoredBufMut>(&mut self, buf: V) -> BufResult<usize, V> {
+        self.writer.read_vectored(buf).await
+    }
+}
+
+impl<W: AsyncBufRead + AsyncWrite> AsyncBufRead for BufWriter<W> {
+    async fn fill_buf(&mut self) -> IoResult<&'_ [u8]> {
+        self.writer.fill_buf().await
+    }
+
+    fn consume(&mut self, amount: usize) {
+        self.writer.consume(amount)
+    }
+}
+
 impl<W> IntoInner for BufWriter<W> {
     type Inner = W;