Turbopack: perf: Avoid clones in RopeReader (#86708)

We (unsurprisingly) spend a lot of time encoding `Rope`s. Most of this is in `memcpy`ing the data out of the rope (not easy to optimize), but there's some cloning of refcounted `Bytes` containers here that can be easily avoided.

Yes, `Bytes`​ is refcounted, but updating refcounts isn't free, and it matters because this is a hot codepath.

This is too small to show up on any top-line measurement (<1%) of total time, but should still be an improvement.

Author: bgw

Cargo.lock

@@ -373,10 +373,6 @@ allsorts = { version = "0.14.0", default-features = false, features = [
   "flate2_rust",
 ] }
 anyhow = "1.0.100"
-async-compression = { version = "0.3.13", default-features = false, features = [
-  "gzip",
-  "tokio",
-] }
 async-trait = "0.1.64"
 bitfield = "0.18.0"
 byteorder = "1.5.0"

Cargo.toml

@@ -1912,7 +1912,7 @@ impl File {
     }
 
     /// Returns a Read/AsyncRead/Stream/Iterator to access the File's contents.
-    pub fn read(&self) -> RopeReader {
+    pub fn read(&self) -> RopeReader<'_> {
         self.content.read()
     }
 }

turbopack/crates/turbo-tasks-fs/src/lib.rs

@@ -11,7 +11,7 @@ use std::{
 
 use RopeElem::{Local, Shared};
 use anyhow::{Context, Result};
-use bytes::{Buf, Bytes};
+use bytes::Bytes;
 use futures::Stream;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use serde_bytes::ByteBuf;
@@ -103,7 +103,7 @@ impl Rope {
     }
 
     /// Returns a [Read]/[AsyncRead]/[Iterator] instance over all bytes.
-    pub fn read(&self) -> RopeReader {
+    pub fn read(&self) -> RopeReader<'_> {
         RopeReader::new(&self.data, 0)
     }
 
@@ -688,28 +688,30 @@ impl DeterministicHash for RopeElem {
 
 #[derive(Debug, Default)]
 /// Implements the [Read]/[AsyncRead]/[Iterator] trait over a [Rope].
-pub struct RopeReader {
-    /// The Rope's tree is kept as a cloned stack, allowing us to accomplish
-    /// incremental yielding.
-    stack: Vec<StackElem>,
+pub struct RopeReader<'a> {
+    /// The Rope's tree is kept as a stack, allowing us to accomplish incremental yielding.
+    stack: Vec<StackElem<'a>>,
+    /// An offset in the current buffer, used by the `read` implementation.
+    offset: usize,
 }
 
 /// A StackElem holds the current index into either a Bytes or a shared Rope.
 /// When the index reaches the end of the associated data, it is removed and we
 /// continue onto the next item in the stack.
 #[derive(Debug)]
-enum StackElem {
-    Local(Bytes),
-    Shared(InnerRope, usize),
+enum StackElem<'a> {
+    Local(&'a Bytes),
+    Shared(&'a InnerRope, usize),
 }
 
-impl RopeReader {
-    fn new(inner: &InnerRope, index: usize) -> Self {
+impl<'a> RopeReader<'a> {
+    fn new(inner: &'a InnerRope, index: usize) -> Self {
         if index >= inner.len() {
             Default::default()
         } else {
             RopeReader {
-                stack: vec![StackElem::Shared(inner.clone(), index)],
+                stack: vec![StackElem::Shared(inner, index)],
+                offset: 0,
             }
         }
     }
@@ -720,30 +722,30 @@ impl RopeReader {
         let mut remaining = want;
 
         while remaining > 0 {
-            let mut bytes = match self.next() {
+            let bytes = match self.next_internal() {
                 None => break,
                 Some(b) => b,
             };
 
-            let amount = min(bytes.len(), remaining);
+            let lower = self.offset;
+            let upper = min(bytes.len(), lower + remaining);
 
-            buf.put_slice(&bytes[0..amount]);
+            buf.put_slice(&bytes[self.offset..upper]);
 
-            if amount < bytes.len() {
-                bytes.advance(amount);
+            if upper < bytes.len() {
+                self.offset = upper;
                 self.stack.push(StackElem::Local(bytes))
+            } else {
+                self.offset = 0;
             }
-            remaining -= amount;
+            remaining -= upper - lower;
         }
 
         want - remaining
     }
-}
 
-impl Iterator for RopeReader {
-    type Item = Bytes;
-
-    fn next(&mut self) -> Option<Self::Item> {
+    /// Returns the next item in the iterator without modifying `self.offset`.
+    fn next_internal(&mut self) -> Option<&'a Bytes> {
         // Iterates the rope's elements recursively until we find the next Local
         // section, returning its Bytes.
         loop {
@@ -756,7 +758,7 @@ impl Iterator for RopeReader {
                 Some(StackElem::Shared(r, i)) => (r, i),
             };
 
-            let el = inner[index].clone();
+            let el = &inner[index];
             index += 1;
             if index < inner.len() {
                 self.stack.push(StackElem::Shared(inner, index));
@@ -767,13 +769,22 @@ impl Iterator for RopeReader {
     }
 }
 
-impl Read for RopeReader {
+impl<'a> Iterator for RopeReader<'a> {
+    type Item = &'a Bytes;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.offset = 0;
+        self.next_internal()
+    }
+}
+
+impl Read for RopeReader<'_> {
     fn read(&mut self, buf: &mut [u8]) -> IoResult<usize> {
         Ok(self.read_internal(buf.len(), &mut ReadBuf::new(buf)))
     }
 }
 
-impl AsyncRead for RopeReader {
+impl AsyncRead for RopeReader<'_> {
     fn poll_read(
         self: Pin<&mut Self>,
         _cx: &mut TaskContext<'_>,
@@ -785,12 +796,12 @@ impl AsyncRead for RopeReader {
     }
 }
 
-impl BufRead for RopeReader {
+impl BufRead for RopeReader<'_> {
     /// Never returns an error.
     fn fill_buf(&mut self) -> IoResult<&[u8]> {
         // Returns the full buffer without coping any data. The same bytes will
         // continue to be returned until [consume] is called.
-        let bytes = match self.next() {
+        let bytes = match self.next_internal() {
             None => return Ok(EMPTY_BUF),
             Some(b) => b,
         };
@@ -803,37 +814,44 @@ impl BufRead for RopeReader {
             unreachable!()
         };
 
-        Ok(bytes)
+        Ok(&bytes[self.offset..])
     }
 
     fn consume(&mut self, amt: usize) {
         if let Some(StackElem::Local(b)) = self.stack.last_mut() {
-            if amt == b.len() {
+            // https://doc.rust-lang.org/std/io/trait.BufRead.html#tymethod.consume
+            debug_assert!(
+                self.offset + amt <= b.len(),
+                "It is a logic error if `amount` exceeds the number of unread bytes in the \
+                 internal buffer, which is returned by `fill_buf`."
+            );
+            // Consume some amount of bytes from the current Bytes instance, ensuring those bytes
+            // are not returned on the next call to `fill_buf`.
+            self.offset += amt;
+            if self.offset == b.len() {
+                // whole Bytes instance was consumed
                 self.stack.pop();
-            } else {
-                // Consume some amount of bytes from the current Bytes instance, ensuring
-                // those bytes are not returned on the next call to [fill_buf].
-                b.advance(amt);
+                self.offset = 0;
             }
         }
     }
 }
 
-impl Stream for RopeReader {
-    // The Result<Bytes> item type is required for this to be streamable into a
-    // [Hyper::Body].
-    type Item = Result<Bytes>;
+impl<'a> Stream for RopeReader<'a> {
+    /// This is efficiently streamable into a [`Hyper::Body`] if each item is cloned into an owned
+    /// `Bytes` instance.
+    type Item = Result<&'a Bytes>;
 
-    // Returns a "result" of reading the next shared bytes reference. This
-    // differs from [Read::read] by not copying any memory.
+    /// Returns a "result" of reading the next shared bytes reference. This
+    /// differs from [`Read::read`] by not copying any memory.
     fn poll_next(self: Pin<&mut Self>, _cx: &mut TaskContext<'_>) -> Poll<Option<Self::Item>> {
         let this = self.get_mut();
         Poll::Ready(this.next().map(Ok))
     }
 }
 
-impl From<RopeElem> for StackElem {
-    fn from(el: RopeElem) -> Self {
+impl<'a> From<&'a RopeElem> for StackElem<'a> {
+    fn from(el: &'a RopeElem) -> Self {
         match el {
             Local(bytes) => Self::Local(bytes),
             Shared(inner) => Self::Shared(inner, 0),

turbopack/crates/turbo-tasks-fs/src/rope.rs

@@ -17,8 +17,8 @@ workspace = true
 
 [dependencies]
 anyhow = { workspace = true }
-async-compression = { workspace = true }
 auto-hash-map = { workspace = true }
+flate2 = { workspace = true }
 futures = { workspace = true }
 hyper = { version = "0.14", features = ["full"] }
 hyper-tungstenite = "0.9.0"

turbopack/crates/turbopack-dev-server/Cargo.toml

@@ -1,13 +1,15 @@
+use std::{io::Read, iter};
+
 use anyhow::{Result, anyhow};
 use auto_hash_map::AutoSet;
-use futures::{StreamExt, TryStreamExt};
+use flate2::{Compression, bufread::GzEncoder};
+use futures::{StreamExt, TryStreamExt, stream};
 use hyper::{
     Request, Response,
     header::{CONTENT_ENCODING, CONTENT_LENGTH, HeaderName},
     http::HeaderValue,
 };
 use mime::Mime;
-use tokio_util::io::{ReaderStream, StreamReader};
 use turbo_tasks::{
     CollectiblesSource, OperationVc, ReadRef, ResolvedVc, TransientInstance, Vc, apply_effects,
     util::SharedError,
@@ -175,22 +177,30 @@ pub async fn process_request_with_content_source(
                 let response = if should_compress {
                     header_map.insert(CONTENT_ENCODING, HeaderValue::from_static("gzip"));
 
-                    // Grab ropereader stream, coerce anyhow::Error to std::io::Error
-                    let stream_ext = content.read().into_stream().map_err(std::io::Error::other);
-
-                    let gzipped_stream =
-                        ReaderStream::new(async_compression::tokio::bufread::GzipEncoder::new(
-                            StreamReader::new(stream_ext),
-                        ));
-
-                    response.body(hyper::Body::wrap_stream(gzipped_stream))?
+                    // Hyper requires an owned reader... We could do this with streaming by cloning
+                    // each `Bytes` and implementing `BufRead` for `Iterator<bytes::Bytes>`, but
+                    // it's not really worth it, just compressing the whole thing up-front is fine.
+                    //
+                    // Use fast compression, since we're likely just tranferring data over
+                    // localhost.
+                    let mut gz_bytes = Vec::new();
+                    GzEncoder::new(content.read(), Compression::fast())
+                        .read_to_end(&mut gz_bytes)
+                        .expect("read of Rope should never fail");
+                    response.body(hyper::Body::wrap_stream(stream::iter(iter::once(
+                        hyper::Result::Ok(gz_bytes),
+                    ))))?
                 } else {
+                    // hyper requires an owned stream, so we must clone the iterator items
+                    // this is relatively cheap: each chunk is a `Bytes`, so `Clone` updates a
+                    // refcount
+                    let owned_chunks: Vec<_> =
+                        content.read().cloned().map(hyper::Result::Ok).collect();
                     header_map.insert(
                         CONTENT_LENGTH,
                         hyper::header::HeaderValue::try_from(content.len().to_string())?,
                     );
-
-                    response.body(hyper::Body::wrap_stream(content.read()))?
+                    response.body(hyper::Body::wrap_stream(stream::iter(owned_chunks)))?
                 };
 
                 return Ok((response, side_effects));