Merge #148

148: bufread::generic::Decoder: don't reinitialize on reader EOF r=Nemo157 a=eeeeeta

If `multiple_members` is enabled, the `bufread::generic::Decoder` will
attempt to reinitialise the decoder inside `State::Flushing`, even if
the reason it entered that state was due to the reader returning an EOF.
This will result in an attempt to read past EOF, which is highly
undesirable to say the least.

To fix this, force `multiple_members` to `false` when we get an EOF
condition from the reader.

Co-authored-by: eta <[email protected]>
Co-authored-by: Wim Looman <[email protected]>

Author: 3 people

src/futures/bufread/generic/decoder.rs

@@ -70,6 +70,9 @@ impl<R: AsyncBufRead, D: Decode> Decoder<R, D> {
                 State::Decoding => {
                     let input = ready!(this.reader.as_mut().poll_fill_buf(cx))?;
                     if input.is_empty() {
+                        // Avoid attempting to reinitialise the decoder if the reader
+                        // has returned EOF.
+                        *this.multiple_members = false;
                         State::Flushing
                     } else {
                         let mut input = PartialBuffer::new(input);

src/tokio/bufread/generic/decoder.rs

@@ -70,6 +70,9 @@ impl<R: AsyncBufRead, D: Decode> Decoder<R, D> {
                 State::Decoding => {
                     let input = ready!(this.reader.as_mut().poll_fill_buf(cx))?;
                     if input.is_empty() {
+                        // Avoid attempting to reinitialise the decoder if the reader
+                        // has returned EOF.
+                        *this.multiple_members = false;
                         State::Flushing
                     } else {
                         let mut input = PartialBuffer::new(input);

src/tokio_02/bufread/generic/decoder.rs

@@ -70,6 +70,9 @@ impl<R: AsyncBufRead, D: Decode> Decoder<R, D> {
                 State::Decoding => {
                     let input = ready!(this.reader.as_mut().poll_fill_buf(cx))?;
                     if input.is_empty() {
+                        // Avoid attempting to reinitialise the decoder if the reader
+                        // has returned EOF.
+                        *this.multiple_members = false;
                         State::Flushing
                     } else {
                         let mut input = PartialBuffer::new(input);

src/tokio_03/bufread/generic/decoder.rs

@@ -70,6 +70,9 @@ impl<R: AsyncBufRead, D: Decode> Decoder<R, D> {
                 State::Decoding => {
                     let input = ready!(this.reader.as_mut().poll_fill_buf(cx))?;
                     if input.is_empty() {
+                        // Avoid attempting to reinitialise the decoder if the reader
+                        // has returned EOF.
+                        *this.multiple_members = false;
                         State::Flushing
                     } else {
                         let mut input = PartialBuffer::new(input);

tests/utils/impls.rs

@@ -13,13 +13,13 @@ pub mod futures {
     pub mod bufread {
         pub use futures::io::AsyncBufRead;
 
-        use crate::utils::InputStream;
+        use crate::utils::{InputStream, TrackEof};
         use futures::stream::{StreamExt as _, TryStreamExt as _};
 
         pub fn from(input: &InputStream) -> impl AsyncBufRead {
             // By using the stream here we ensure that each chunk will require a separate
             // read/poll_fill_buf call to process to help test reading multiple chunks.
-            input.stream().map(Ok).into_async_read()
+            TrackEof::new(input.stream().map(Ok).into_async_read())
         }
     }
 
@@ -100,13 +100,13 @@ pub mod tokio_02 {
     pub mod bufread {
         pub use tokio_02::io::AsyncBufRead;
 
-        use crate::utils::InputStream;
+        use crate::utils::{InputStream, TrackEof};
         use tokio_02::io::stream_reader;
 
         pub fn from(input: &InputStream) -> impl AsyncBufRead {
             // By using the stream here we ensure that each chunk will require a separate
             // read/poll_fill_buf call to process to help test reading multiple chunks.
-            stream_reader(input.bytes_05_stream())
+            TrackEof::new(stream_reader(input.bytes_05_stream()))
         }
     }
 
@@ -169,14 +169,14 @@ pub mod tokio_02 {
 #[cfg(feature = "tokio-03")]
 pub mod tokio_03 {
     pub mod bufread {
-        use crate::utils::InputStream;
+        use crate::utils::{InputStream, TrackEof};
         pub use tokio_03::io::AsyncBufRead;
         use tokio_util_04::io::StreamReader;
 
         pub fn from(input: &InputStream) -> impl AsyncBufRead {
             // By using the stream here we ensure that each chunk will require a separate
             // read/poll_fill_buf call to process to help test reading multiple chunks.
-            StreamReader::new(input.bytes_05_stream())
+            TrackEof::new(StreamReader::new(input.bytes_05_stream()))
         }
     }
 
@@ -239,7 +239,7 @@ pub mod tokio_03 {
 #[cfg(feature = "tokio")]
 pub mod tokio {
     pub mod bufread {
-        use crate::utils::InputStream;
+        use crate::utils::{InputStream, TrackEof};
         use bytes::Bytes;
         use futures::stream::StreamExt;
         pub use tokio::io::AsyncBufRead;
@@ -248,7 +248,9 @@ pub mod tokio {
         pub fn from(input: &InputStream) -> impl AsyncBufRead {
             // By using the stream here we ensure that each chunk will require a separate
             // read/poll_fill_buf call to process to help test reading multiple chunks.
-            StreamReader::new(input.stream().map(Bytes::from).map(std::io::Result::Ok))
+            TrackEof::new(StreamReader::new(
+                input.stream().map(Bytes::from).map(std::io::Result::Ok),
+            ))
         }
     }
 

tests/utils/mod.rs

@@ -8,13 +8,14 @@ mod tokio_03_ext;
 #[cfg(feature = "tokio")]
 mod tokio_ext;
 mod track_closed;
+mod track_eof;
 #[macro_use]
 mod test_cases;
 
 pub mod algos;
 pub mod impls;
 
-pub use self::{input_stream::InputStream, track_closed::TrackClosed};
+pub use self::{input_stream::InputStream, track_closed::TrackClosed, track_eof::TrackEof};
 pub use async_compression::Level;
 pub use futures::{executor::block_on, pin_mut, stream::Stream};
 pub use std::{future::Future, io::Result, iter::FromIterator, pin::Pin};

tests/utils/track_eof.rs

@@ -0,0 +1,184 @@
+#[cfg_attr(not(feature = "all-implementations"), allow(unused))]
+use std::{
+    io::Result,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+pub struct TrackEof<R> {
+    inner: R,
+    eof: bool,
+}
+
+impl<R: Unpin> TrackEof<R> {
+    pub fn new(inner: R) -> Self {
+        Self { inner, eof: false }
+    }
+
+    pub fn project(self: Pin<&mut Self>) -> (Pin<&mut R>, &mut bool) {
+        let Self { inner, eof } = Pin::into_inner(self);
+        (Pin::new(inner), eof)
+    }
+}
+
+#[cfg(feature = "futures-io")]
+impl<R: futures::io::AsyncRead + Unpin> futures::io::AsyncRead for TrackEof<R> {
+    fn poll_read(self: Pin<&mut Self>, cx: &mut Context, buf: &mut [u8]) -> Poll<Result<usize>> {
+        let (inner, eof) = self.project();
+        assert!(!*eof);
+        match inner.poll_read(cx, buf) {
+            Poll::Ready(Ok(0)) => {
+                if !buf.is_empty() {
+                    *eof = true;
+                }
+                Poll::Ready(Ok(0))
+            }
+            other => other,
+        }
+    }
+}
+
+#[cfg(feature = "futures-io")]
+impl<R: futures::io::AsyncBufRead + Unpin> futures::io::AsyncBufRead for TrackEof<R> {
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<&[u8]>> {
+        let (inner, eof) = self.project();
+        assert!(!*eof);
+        match inner.poll_fill_buf(cx) {
+            Poll::Ready(Ok(buf)) => {
+                if buf.is_empty() {
+                    *eof = true;
+                }
+                Poll::Ready(Ok(buf))
+            }
+            other => other,
+        }
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        self.project().0.consume(amt)
+    }
+}
+
+#[cfg(feature = "tokio-02")]
+impl<R: tokio_02::io::AsyncRead + Unpin> tokio_02::io::AsyncRead for TrackEof<R> {
+    fn poll_read(self: Pin<&mut Self>, cx: &mut Context, buf: &mut [u8]) -> Poll<Result<usize>> {
+        let (inner, eof) = self.project();
+        assert!(!*eof);
+        match inner.poll_read(cx, buf) {
+            Poll::Ready(Ok(0)) => {
+                if !buf.is_empty() {
+                    *eof = true;
+                }
+                Poll::Ready(Ok(0))
+            }
+            other => other,
+        }
+    }
+}
+
+#[cfg(feature = "tokio-02")]
+impl<R: tokio_02::io::AsyncBufRead + Unpin> tokio_02::io::AsyncBufRead for TrackEof<R> {
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<&[u8]>> {
+        let (inner, eof) = self.project();
+        assert!(!*eof);
+        match inner.poll_fill_buf(cx) {
+            Poll::Ready(Ok(buf)) => {
+                if buf.is_empty() {
+                    *eof = true;
+                }
+                Poll::Ready(Ok(buf))
+            }
+            other => other,
+        }
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        self.project().0.consume(amt)
+    }
+}
+
+#[cfg(feature = "tokio-03")]
+impl<R: tokio_03::io::AsyncRead + Unpin> tokio_03::io::AsyncRead for TrackEof<R> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut tokio_03::io::ReadBuf,
+    ) -> Poll<Result<()>> {
+        let (inner, eof) = self.project();
+        assert!(!*eof);
+        let len = buf.filled().len();
+        match inner.poll_read(cx, buf) {
+            Poll::Ready(Ok(())) => {
+                if buf.filled().len() == len && buf.remaining() > 0 {
+                    *eof = true;
+                }
+                Poll::Ready(Ok(()))
+            }
+            other => other,
+        }
+    }
+}
+
+#[cfg(feature = "tokio-03")]
+impl<R: tokio_03::io::AsyncBufRead + Unpin> tokio_03::io::AsyncBufRead for TrackEof<R> {
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<&[u8]>> {
+        let (inner, eof) = self.project();
+        assert!(!*eof);
+        match inner.poll_fill_buf(cx) {
+            Poll::Ready(Ok(buf)) => {
+                if buf.is_empty() {
+                    *eof = true;
+                }
+                Poll::Ready(Ok(buf))
+            }
+            other => other,
+        }
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        self.project().0.consume(amt)
+    }
+}
+
+#[cfg(feature = "tokio")]
+impl<R: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for TrackEof<R> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context,
+        buf: &mut tokio::io::ReadBuf,
+    ) -> Poll<Result<()>> {
+        let (inner, eof) = self.project();
+        assert!(!*eof);
+        let len = buf.filled().len();
+        match inner.poll_read(cx, buf) {
+            Poll::Ready(Ok(())) => {
+                if buf.filled().len() == len && buf.remaining() > 0 {
+                    *eof = true;
+                }
+                Poll::Ready(Ok(()))
+            }
+            other => other,
+        }
+    }
+}
+
+#[cfg(feature = "tokio")]
+impl<R: tokio::io::AsyncBufRead + Unpin> tokio::io::AsyncBufRead for TrackEof<R> {
+    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<&[u8]>> {
+        let (inner, eof) = self.project();
+        assert!(!*eof);
+        match inner.poll_fill_buf(cx) {
+            Poll::Ready(Ok(buf)) => {
+                if buf.is_empty() {
+                    *eof = true;
+                }
+                Poll::Ready(Ok(buf))
+            }
+            other => other,
+        }
+    }
+
+    fn consume(self: Pin<&mut Self>, amt: usize) {
+        self.project().0.consume(amt)
+    }
+}