async LineWriter fix #2131

futures-util/src/io/buf_writer.rs

@@ -6,6 +6,7 @@ use pin_project_lite::pin_project;
 use std::fmt;
 use std::io::{self, Write};
 use std::pin::Pin;
+use std::ptr;
 
 pin_project! {
     /// Wraps a writer and buffers its output.
@@ -49,7 +50,7 @@ impl<W: AsyncWrite> BufWriter<W> {
         Self { inner, buf: Vec::with_capacity(cap), written: 0 }
     }
 
-    fn flush_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+    pub(super) fn flush_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
         let mut this = self.project();
 
         let len = this.buf.len();
@@ -83,6 +84,42 @@ impl<W: AsyncWrite> BufWriter<W> {
     pub fn buffer(&self) -> &[u8] {
         &self.buf
     }
+
+    /// TODO WIP
+    pub(super) fn capacity(&self) -> usize {
+        self.buf.capacity()
+    }
+
+    #[inline]
+    /// TODO WIP
+    pub(super) fn spare_capacity(&self) -> usize {
+        self.buf.capacity() - self.buf.len()
+    }
+
+    /// TODO WIP
+    pub(super) fn write_to_buf(mut self: Pin<&mut Self>, buf: &[u8]) -> usize {
+        let available = self.spare_capacity();
+        let amt_to_buffer = available.min(buf.len());
+
+        // SAFETY: `amt_to_buffer` is <= buffer's spare capacity by construction.
+        unsafe {
+            self.write_to_buffer_unchecked(&buf[..amt_to_buffer]);
+        }
+
+        amt_to_buffer
+    }
+
+    #[inline]
+    /// TODO WIP
+    unsafe fn write_to_buffer_unchecked(mut self: Pin<&mut Self>, buf: &[u8]) {
+        debug_assert!(buf.len() <= self.spare_capacity());
+        let old_len = self.buf.len();
+        let buf_len = buf.len();
+        let src = buf.as_ptr();
+        let dst = self.buf.as_mut_ptr().add(old_len);
+        ptr::copy_nonoverlapping(src, dst, buf_len);
+        self.buf.set_len(old_len + buf_len);
+    }
 }
 
 impl<W: AsyncWrite> AsyncWrite for BufWriter<W> {

futures-util/src/io/line_writer.rs

@@ -0,0 +1,127 @@
+use super::buf_writer::BufWriter;
+use futures_core::ready;
+use futures_core::task::{Context, Poll};
+use futures_io::AsyncWrite;
+use pin_project_lite::pin_project;
+use pin_utils::pin_mut;
+use std::io;
+use std::pin::Pin;
+
+pin_project! {
+#[derive(Debug)]
+struct LineWriterShim<W: AsyncWrite> {
+    #[pin]
+    buffer: BufWriter<W>, // TODO HELP what's this field's type suppossed to be?
+}
+}
+
+impl<W: AsyncWrite> LineWriterShim<W> {
+    /// TODO WIP
+    fn buffered(&self) -> &[u8] {
+        self.buffer.buffer()
+    }
+    /// TODO WIP
+    fn flush_if_completed_line(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        let this = self.project();
+        //let this = &mut *self;
+        match self.buffered().last().copied() {
+            Some(b'\n') => this.buffer.flush_buf(cx),
+            _ => Poll::Ready(Ok(())),
+        }
+    }
+}
+
+impl<W: AsyncWrite> AsyncWrite for LineWriterShim<W> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        let this = self.project();
+        let newline_index = match memchr::memrchr(b'\n', buf) {
+            None => {
+                ready!(self.flush_if_completed_line(cx)?);
+                return this.buffer.poll_write(cx, buf);
+            }
+            Some(newline_index) => newline_index + 1,
+        };
+
+        this.buffer.poll_flush(cx)?;
+
+        let lines = &buf[..newline_index];
+
+        let flushed = ready!(this.buffer.as_mut().poll_write(cx, lines))?;
+
+        if flushed == 0 {
+            return Poll::Ready(Ok(0));
+        }
+
+        let tail = if flushed >= newline_index {
+            &buf[flushed..]
+        } else if newline_index - flushed <= self.buffer.capacity() {
+            &buf[flushed..newline_index]
+        } else {
+            let scan_area = &buf[flushed..];
+            let scan_area = &scan_area[..self.buffer.capacity()];
+            match memchr::memrchr(b'\n', scan_area) {
+                Some(newline_index) => &scan_area[..newline_index + 1],
+                None => scan_area,
+            }
+        };
+
+        let buffered = self.buffer.write_to_buf(tail);
+        Poll::Ready(Ok(flushed + buffered))
+    }
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.project().buffer.poll_flush(cx)
+    }
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.project().buffer.poll_close(cx)
+    }
+}
+pin_project! {
+/// TODO: WIP
+#[derive(Debug)]
+pub struct LineWriter<W: AsyncWrite> {
+    #[pin]
+    inner: BufWriter<W>,
+}
+}
+
+impl<W: AsyncWrite> LineWriter<W> {
+    /// TODO: WIP
+    pub fn new(inner: W) -> LineWriter<W> {
+        // 1024 is taken from std::io::buffered::LineWriter
+        LineWriter::with_capacity(1024, inner)
+    }
+    /// TODO: WIP
+    pub fn with_capacity(capacity: usize, inner: W) -> LineWriter<W> {
+        LineWriter { inner: BufWriter::with_capacity(capacity, inner) }
+    }
+}
+
+impl<W: AsyncWrite> AsyncWrite for LineWriter<W> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
+        let lws = LineWriterShim { buffer: Box::new(&self.inner).as_mut() };
+        pin_mut!(lws);
+        lws.poll_write(cx, buf)
+
+        //lws.poll_write(cx, buf)
+
+        //Pin::new(&mut lws).poll_write(cx, buf)
+
+        //Box::pin(&mut lws).poll_write(cx, buf)
+
+        //Pin::new(&mut *lws).poll_write(cx, buf)
+    }
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.project().inner.poll_flush(cx)
+    }
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+        self.project().inner.poll_close(cx)
+    }
+}

futures-util/src/io/mod.rs

@@ -61,6 +61,9 @@ pub use self::buf_reader::BufReader;
 mod buf_writer;
 pub use self::buf_writer::BufWriter;
 
+mod line_writer;
+pub use self::line_writer::LineWriter;
+
 mod chain;
 pub use self::chain::Chain;