util: update to bytes 0.6 (#3071)

Copies the implementation of poll_read_buf() from tokio::io::util::read_buf.

examples/Cargo.toml

@@ -11,7 +11,7 @@ tokio = { version = "0.3.0", path = "../tokio", features = ["full", "tracing"] }
 tracing = "0.1"
 tracing-subscriber = { version = "0.2.7", default-features = false, features = ["fmt", "ansi", "env-filter", "chrono", "tracing-log"] }
 tokio-util = { version = "0.4.0", path = "../tokio-util", features = ["full"] }
-bytes = "0.5"
+bytes = "0.6"
 futures = "0.3.0"
 http = "0.2"
 serde = "1.0"

tokio-util/Cargo.toml

@@ -35,7 +35,7 @@ rt = ["tokio/rt"]
 [dependencies]
 tokio = { version = "0.3.0", path = "../tokio" }
 
-bytes = "0.5.0"
+bytes = "0.6.0"
 futures-core = "0.3.0"
 futures-sink = "0.3.0"
 futures-io = { version = "0.3.0", optional = true }

tokio-util/src/lib.rs

@@ -65,6 +65,7 @@ mod util {
     use bytes::BufMut;
     use futures_core::ready;
     use std::io;
+    use std::mem::MaybeUninit;
     use std::pin::Pin;
     use std::task::{Context, Poll};
 
@@ -77,17 +78,24 @@ mod util {
             return Poll::Ready(Ok(0));
         }
 
-        let orig = buf.bytes_mut().as_ptr() as *const u8;
-        let mut b = ReadBuf::uninit(buf.bytes_mut());
+        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!(io.poll_read(cx, &mut buf)?);
 
-        ready!(io.poll_read(cx, &mut b))?;
-        let n = b.filled().len();
+            // Ensure the pointer does not change from under us
+            assert_eq!(ptr, buf.filled().as_ptr());
+            buf.filled().len()
+        };
 
-        // Safety: we can assume `n` bytes were read, since they are in`filled`.
-        assert_eq!(orig, b.filled().as_ptr());
+        // 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);
         }
+
         Poll::Ready(Ok(n))
     }
 }