perf(http1): read chunk framing bytes without split_to

src/proto/h1/decode.rs

@@ -251,9 +251,8 @@ impl fmt::Debug for Decoder {
 
 macro_rules! byte (
     ($rdr:ident, $cx:expr) => ({
-        let buf = ready!($rdr.read_mem($cx, 1))?;
-        if !buf.is_empty() {
-            buf[0]
+        if let Some(b) = ready!($rdr.read_byte($cx))? {
+            b
         } else {
             return Poll::Ready(Err(io::Error::new(io::ErrorKind::UnexpectedEof,
                                       "unexpected EOF during chunk size line")));
@@ -721,6 +720,60 @@ mod tests {
         }
     }
 
+    impl MemRead for BytesMut {
+        fn read_mem(&mut self, _: &mut Context<'_>, len: usize) -> Poll<io::Result<Bytes>> {
+            let n = std::cmp::min(len, self.len());
+            Poll::Ready(Ok(self.split_to(n).freeze()))
+        }
+        fn read_byte(&mut self, _: &mut Context<'_>) -> Poll<io::Result<Option<u8>>> {
+            if self.is_empty() {
+                return Poll::Ready(Ok(None));
+            }
+            let b = self[0];
+            bytes::Buf::advance(self, 1);
+            Poll::Ready(Ok(Some(b)))
+        }
+    }
+
+    #[cfg(all(feature = "nightly", not(miri)))]
+    fn bench_chunked_body(b: &mut test::Bencher, n_chunks: usize, chunk: usize) {
+        let rt = new_runtime();
+        let mut vec = Vec::new();
+        for _ in 0..n_chunks {
+            vec.extend(format!("{:x}\r\n", chunk).as_bytes());
+            vec.extend(std::iter::repeat(b'a').take(chunk));
+            vec.extend(b"\r\n");
+        }
+        vec.extend(b"0\r\n\r\n");
+        b.bytes = (n_chunks * chunk) as u64;
+        b.iter(|| {
+            let mut raw = BytesMut::from(&vec[..]);
+            let mut decoder = Decoder::chunked(None, None);
+            rt.block_on(async {
+                let mut total = 0;
+                loop {
+                    let d = decoder
+                        .decode_fut(&mut raw)
+                        .await
+                        .unwrap()
+                        .into_data()
+                        .unwrap();
+                    if d.is_empty() {
+                        break;
+                    }
+                    total += d.len();
+                }
+                assert_eq!(total, n_chunks * chunk);
+            });
+        });
+    }
+
+    #[cfg(all(feature = "nightly", not(miri)))]
+    #[bench]
+    fn bench_decode_chunked_small_chunks(b: &mut test::Bencher) {
+        bench_chunked_body(b, 100, 16)
+    }
+
     impl MemRead for Bytes {
         fn read_mem(&mut self, _: &mut Context<'_>, len: usize) -> Poll<io::Result<Bytes>> {
             let n = std::cmp::min(len, self.len());

src/proto/h1/io.rs

@@ -351,6 +351,11 @@ impl<T: Unpin, B> Unpin for Buffered<T, B> {}
 // TODO: This trait is old... at least rename to PollBytes or something...
 pub(crate) trait MemRead {
     fn read_mem(&mut self, cx: &mut Context<'_>, len: usize) -> Poll<io::Result<Bytes>>;
+
+    /// Reads one byte, `None` on EOF.
+    fn read_byte(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<Option<u8>>> {
+        self.read_mem(cx, 1).map_ok(|b| b.first().copied())
+    }
 }
 
 impl<T, B> MemRead for Buffered<T, B>
@@ -367,6 +372,16 @@ where
             Poll::Ready(Ok(self.read_buf.split_to(::std::cmp::min(len, n)).freeze()))
         }
     }
+
+    // Skips the split/freeze (and its refcount traffic) of `read_mem` for chunk framing bytes.
+    fn read_byte(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<Option<u8>>> {
+        if self.read_buf.is_empty() && ready!(self.poll_read_from_io(cx))? == 0 {
+            return Poll::Ready(Ok(None));
+        }
+        let b = self.read_buf[0];
+        self.read_buf.advance(1);
+        Poll::Ready(Ok(Some(b)))
+    }
 }
 
 #[derive(Clone, Copy, Debug)]