Revert "util: upgrade tokio-util to bytes 0.6 (#3052)" (#3060)

This reverts commit fe2b997.

We are avoiding adding poll_read_buf to tokio itself for now. The patch is
reverted now in order to not block the v0.3.2 release (#3059).

3 files changed, 24 insertions, 95 deletions

diff --git a/tokio/src/io/util/async_read_ext.rs b/tokio/src/io/util/async_read_ext.rs
index 96a5f70d..1f918f19 100644
--- a/tokio/src/io/util/async_read_ext.rs
+++ b/tokio/src/io/util/async_read_ext.rs
@@ -1,6 +1,6 @@
 use crate::io::util::chain::{chain, Chain};
 use crate::io::util::read::{read, Read};
-use crate::io::util::read_buf::{poll_read_buf, read_buf, ReadBuf};
+use crate::io::util::read_buf::{read_buf, ReadBuf};
 use crate::io::util::read_exact::{read_exact, ReadExact};
 use crate::io::util::read_int::{
     ReadI128, ReadI128Le, ReadI16, ReadI16Le, ReadI32, ReadI32Le, ReadI64, ReadI64Le, ReadI8,
@@ -14,8 +14,6 @@ use crate::io::util::take::{take, Take};
 use crate::io::AsyncRead;
 
 use bytes::BufMut;
-use std::io;
-use std::task::{Context, Poll};
 
 cfg_io_util! {
     /// Defines numeric reader
@@ -233,28 +231,6 @@ cfg_io_util! {
             read_buf(self, buf)
         }
 
-        /// Attempts to pull some bytes from this source into the specified buffer,
-        /// advancing the buffer's internal cursor if the underlying reader is ready.
-        ///
-        /// Usually, only a single `read` syscall is issued, even if there is
-        /// more space in the supplied buffer.
-        ///
-        /// # Return
-        ///
-        /// On a successful read, the number of read bytes is returned. If the
-        /// supplied buffer is not empty and the function returns `Ok(0)` then
-        /// the source has reached an "end-of-file" event.
-        ///
-        /// # Errors
-        ///
-        /// If this function encounters any form of I/O or other error, an error
-        /// variant will be returned. If an error is returned then it must be
-        /// guaranteed that no bytes were read.
-        /// ```
-        fn poll_read_buf<'a, B>(&'a mut self, buf: &'a mut B, cx: &mut Context<'_>) -> Poll<io::Result<usize>> where Self: Unpin + Sized, B: BufMut {
-            poll_read_buf(self, buf, cx)
-        }
-
         /// Reads the exact number of bytes required to fill `buf`.
         ///
         /// Equivalent to:
diff --git a/tokio/src/io/util/read_buf.rs b/tokio/src/io/util/read_buf.rs
index 7df429d7..696deefd 100644
--- a/tokio/src/io/util/read_buf.rs
+++ b/tokio/src/io/util/read_buf.rs
@@ -40,44 +40,33 @@ where
     type Output = io::Result<usize>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
-        let mut me = self.project();
-        poll_read_buf(&mut me.reader, &mut me.buf, cx)
-    }
-}
+        use crate::io::ReadBuf;
+        use std::mem::MaybeUninit;
 
-pub(crate) fn poll_read_buf<'a, R, B>(
-    reader: &'a mut R,
-    buf: &'a mut B,
-    cx: &mut Context<'_>,
-) -> Poll<io::Result<usize>>
-where
-    R: AsyncRead + Unpin,
-    B: BufMut,
-{
-    use crate::io::ReadBuf;
-    use std::mem::MaybeUninit;
+        let me = self.project();
 
-    if !buf.has_remaining_mut() {
-        return Poll::Ready(Ok(0));
-    }
+        if !me.buf.has_remaining_mut() {
+            return Poll::Ready(Ok(0));
+        }
 
-    let n = {
-        let dst = buf.bytes_mut();
-        let dst = unsafe { &mut *(dst as *mut _ as *mut [MaybeUninit<u8>]) };
-        let mut buf = ReadBuf::uninit(dst);
-        let ptr = buf.filled().as_ptr();
-        ready!(Pin::new(reader).poll_read(cx, &mut buf)?);
+        let n = {
+            let dst = me.buf.bytes_mut();
+            let dst = unsafe { &mut *(dst as *mut _ as *mut [MaybeUninit<u8>]) };
+            let mut buf = ReadBuf::uninit(dst);
+            let ptr = buf.filled().as_ptr();
+            ready!(Pin::new(me.reader).poll_read(cx, &mut buf)?);
 
-        // Ensure the pointer does not change from under us
-        assert_eq!(ptr, buf.filled().as_ptr());
-        buf.filled().len()
-    };
+            // Ensure the pointer does not change from under us
+            assert_eq!(ptr, buf.filled().as_ptr());
+            buf.filled().len()
+        };
 
-    // Safety: This is guaranteed to be the number of initialized (and read)
-    // bytes due to the invariants provided by `ReadBuf::filled`.
-    unsafe {
-        buf.advance_mut(n);
-    }
+        // Safety: This is guaranteed to be the number of initialized (and read)
+        // bytes due to the invariants provided by `ReadBuf::filled`.
+        unsafe {
+            me.buf.advance_mut(n);
+        }
 
-    Poll::Ready(Ok(n))
+        Poll::Ready(Ok(n))
+    }
 }
diff --git a/tokio/tests/io_read_buf.rs b/tokio/tests/io_read_buf.rs
index 35c12126..0328168d 100644
--- a/tokio/tests/io_read_buf.rs
+++ b/tokio/tests/io_read_buf.rs
@@ -4,7 +4,6 @@
 use tokio::io::{AsyncRead, AsyncReadExt, ReadBuf};
 use tokio_test::assert_ok;
 
-use futures::future::poll_fn;
 use std::io;
 use std::pin::Pin;
 use std::task::{Context, Poll};
@@ -35,38 +34,3 @@ async fn read_buf() {
     assert_eq!(n, 11);
     assert_eq!(buf[..], b"hello world"[..]);
 }
-
-#[tokio::test]
-async fn poll_read_buf() {
-    struct Rd {
-        cnt: usize,
-    }
-
-    impl AsyncRead for Rd {
-        fn poll_read(
-            mut self: Pin<&mut Self>,
-            _cx: &mut Context<'_>,
-            buf: &mut ReadBuf<'_>,
-        ) -> Poll<io::Result<()>> {
-            self.cnt += 1;
-            buf.put_slice(b"hello world");
-            Poll::Ready(Ok(()))
-        }
-    }
-
-    let mut buf = vec![];
-    let mut rd = Rd { cnt: 0 };
-
-    let res = tokio::spawn(async move {
-        poll_fn(|cx| {
-            let res = rd.poll_read_buf(&mut buf, cx);
-            assert_eq!(1, rd.cnt);
-            assert_eq!(buf[..], b"hello world"[..]);
-            res
-        })
-        .await
-    })
-    .await;
-
-    assert!(matches!(res, Ok(Ok(11usize))));
-}