Repository: denoland/deno
Pull request: perf(install): move HTTP decompression for registry requests off the async event loop

Bypass the tower-http `Decompression` middleware for npm registry requests and instead decompress gzip on a blocking thread. This prevents large packument decompression from blocking other HTTP/2 streams on the event loop.

This right now doesn't have a huge impact but after a later PR to get more parallelism of downloads it makes a difference

Changed files:
filename: cli/http_util.rs
patch: @@ -240,7 +240,9 @@ impl HttpClient {
   }
 
   pub async fn download(&self, url: Url) -> Result<Vec<u8>, DownloadError> {
-    let response = self.download_inner(url, &Default::default(), None).await?;
+    let response = self
+      .download_inner(url, &Default::default(), None, true)
+      .await?;
     response.into_bytes()
   }
 
@@ -251,7 +253,29 @@ impl HttpClient {
     progress_guard: &UpdateGuard,
   ) -> Result<HttpClientResponse, DownloadError> {
     crate::util::retry::retry(
-      || self.download_inner(url.clone(), headers, Some(progress_guard)),
+      || self.download_inner(url.clone(), headers, Some(progress_guard), true),
+      |e| {
+        matches!(
+          e.as_kind(),
+          DownloadErrorKind::BadResponse(_) | DownloadErrorKind::Fetch(_)
+        )
+      },
+    )
+    .await
+  }
+
+  /// Like `download_with_progress_and_retries`, but bypasses the transparent
+  /// decompression middleware. The response body will contain raw bytes
+  /// (potentially gzip-compressed). The caller should check the
+  /// Content-Encoding header and decompress if needed.
+  pub async fn download_with_progress_and_retries_no_decompress(
+    &self,
+    url: Url,
+    headers: &HeaderMap,
+    progress_guard: &UpdateGuard,
+  ) -> Result<HttpClientResponse, DownloadError> {
+    crate::util::retry::retry(
+      || self.download_inner(url.clone(), headers, Some(progress_guard), false),
       |e| {
         matches!(
           e.as_kind(),
@@ -267,7 +291,7 @@ impl HttpClient {
     url: Url,
     headers: &HeaderMap<HeaderValue>,
   ) -> Result<Url, AnyError> {
-    let (_, url) = self.get_redirected_response(url, headers).await?;
+    let (_, url) = self.get_redirected_response(url, headers, true).await?;
     Ok(url)
   }
 
@@ -276,8 +300,11 @@ impl HttpClient {
     url: Url,
     headers: &HeaderMap<HeaderValue>,
     progress_guard: Option<&UpdateGuard>,
+    should_decompress: bool,
   ) -> Result<HttpClientResponse, DownloadError> {
-    let (response, _) = self.get_redirected_response(url, headers).await?;
+    let (response, _) = self
+      .get_redirected_response(url, headers, should_decompress)
+      .await?;
 
     if response.status() == 404 {
       return Ok(HttpClientResponse::NotFound);
@@ -307,15 +334,16 @@ impl HttpClient {
     &self,
     mut url: Url,
     headers: &HeaderMap<HeaderValue>,
+    should_decompress: bool,
   ) -> Result<(http::Response<deno_fetch::ResBody>, Url), DownloadError> {
     let mut req = self.get(url.clone())?.build();
     *req.headers_mut() = headers.clone();
-    let mut response = self
-      .client
-      .clone()
-      .send(req)
-      .await
-      .map_err(|e| DownloadErrorKind::Fetch(e).into_box())?;
+    let mut response = if should_decompress {
+      self.client.clone().send(req).await
+    } else {
+      self.client.clone().send_no_decompress(req).await
+    }
+    .map_err(|e| DownloadErrorKind::Fetch(e).into_box())?;
     let status = response.status();
     if status.is_redirection() && status != http::StatusCode::NOT_MODIFIED {
       for _ in 0..5 {
@@ -329,12 +357,12 @@ impl HttpClient {
         }
         *req.headers_mut() = headers;
 
-        let new_response = self
-          .client
-          .clone()
-          .send(req)
-          .await
-          .map_err(|e| DownloadErrorKind::Fetch(e).into_box())?;
+        let new_response = if should_decompress {
+          self.client.clone().send(req).await
+        } else {
+          self.client.clone().send_no_decompress(req).await
+        }
+        .map_err(|e| DownloadErrorKind::Fetch(e).into_box())?;
         let status = new_response.status();
         if status.is_redirection() {
           response = new_response;
@@ -374,7 +402,7 @@ pub async fn get_response_body_with_progress(
         let bytes = item?;
         current_size += bytes.len() as u64;
         progress_guard.set_position(current_size);
-        data.extend(bytes.into_iter());
+        data.extend_from_slice(&bytes);
       }
       return Ok((parts.headers, data));
     }

filename: cli/npm.rs
patch: @@ -131,25 +131,17 @@ impl deno_npm_cache::NpmCacheHttpClient for CliNpmCacheHttpClient {
         ),
       );
     }
-    client
-      .download_with_progress_and_retries(url, &headers, &guard)
+    // Request gzip and bypass the tower-http Decompression middleware so
+    // that gzip inflate happens on a blocking thread instead of inline on
+    // the async event loop. This prevents large packument decompression
+    // (~12% of CPU during resolution) from blocking other HTTP/2 streams.
+    headers.insert(
+      http::header::ACCEPT_ENCODING,
+      http::header::HeaderValue::from_static("gzip"),
+    );
+    let response = client
+      .download_with_progress_and_retries_no_decompress(url, &headers, &guard)
       .await
-      .map(|response| match response {
-        crate::http_util::HttpClientResponse::Success { headers, body } => {
-          NpmCacheHttpClientResponse::Bytes(NpmCacheHttpClientBytesResponse {
-            etag: headers
-              .get(http::header::ETAG)
-              .and_then(|e| e.to_str().map(|t| t.to_string()).ok()),
-            bytes: body,
-          })
-        }
-        crate::http_util::HttpClientResponse::NotFound => {
-          NpmCacheHttpClientResponse::NotFound
-        }
-        crate::http_util::HttpClientResponse::NotModified => {
-          NpmCacheHttpClientResponse::NotModified
-        }
-      })
       .map_err(|err| {
         use crate::http_util::DownloadErrorKind::*;
         let status_code = match err.as_kind() {
@@ -171,10 +163,59 @@ impl deno_npm_cache::NpmCacheHttpClient for CliNpmCacheHttpClient {
           status_code,
           error: JsErrorBox::from_err(err),
         }
-      })
+      })?;
+    match response {
+      crate::http_util::HttpClientResponse::Success { headers, body } => {
+        // Decompress gzip on a blocking thread to keep the event loop free
+        let body = if headers
+          .get(http::header::CONTENT_ENCODING)
+          .and_then(|v| v.to_str().ok())
+          .is_some_and(|v| v == "gzip")
+        {
+          tokio::task::spawn_blocking(move || decompress_gzip(body))
+            .await
+            .map_err(|e| deno_npm_cache::DownloadError {
+              status_code: None,
+              error: JsErrorBox::generic(e.to_string()),
+            })?
+            .map_err(|e| deno_npm_cache::DownloadError {
+              status_code: None,
+              error: e,
+            })?
+        } else {
+          body
+        };
+        Ok(NpmCacheHttpClientResponse::Bytes(
+          NpmCacheHttpClientBytesResponse {
+            etag: headers
+              .get(http::header::ETAG)
+              .and_then(|e| e.to_str().map(|t| t.to_string()).ok()),
+            bytes: body,
+          },
+        ))
+      }
+      crate::http_util::HttpClientResponse::NotFound => {
+        Ok(NpmCacheHttpClientResponse::NotFound)
+      }
+      crate::http_util::HttpClientResponse::NotModified => {
+        Ok(NpmCacheHttpClientResponse::NotModified)
+      }
+    }
   }
 }
 
+fn decompress_gzip(compressed: Vec<u8>) -> Result<Vec<u8>, JsErrorBox> {
+  use std::io::Read;
+
+  use flate2::read::GzDecoder;
+  let mut decoder = GzDecoder::new(compressed.as_slice());
+  let mut decompressed = Vec::new();
+  decoder.read_to_end(&mut decompressed).map_err(|e| {
+    JsErrorBox::generic(format!("gzip decompression failed: {e}"))
+  })?;
+  Ok(decompressed)
+}
+
 #[derive(Debug)]
 pub struct NpmFetchResolver {
   nv_by_req: DashMap<PackageReq, Option<PackageNv>>,

filename: ext/fetch/lib.rs
patch: @@ -1298,6 +1298,30 @@ impl Client {
       .map_err(|e| ClientSendError { uri, source: e })?;
     Ok(resp.map(|b| b.map_err(|e| JsErrorBox::generic(e.to_string())).boxed()))
   }
+
+  /// Sends a request bypassing the transparent decompression middleware.
+  /// The response body will contain raw bytes (potentially compressed).
+  /// The caller is responsible for checking Content-Encoding and
+  /// decompressing if needed.
+  pub async fn send_no_decompress(
+    self,
+    mut req: http::Request<ReqBody>,
+  ) -> Result<http::Response<ResBody>, ClientSendError> {
+    self.inject_common_headers(&mut req);
+
+    req.headers_mut().entry(ACCEPT).or_insert(STAR_STAR);
+
+    let uri = req.uri().clone();
+
+    // .into_inner() unwraps the Decompression middleware layer
+    let resp = self
+      .inner
+      .into_inner()
+      .oneshot(req)
+      .await
+      .map_err(|e| ClientSendError { uri, source: e })?;
+    Ok(resp.map(|b| b.map_err(|e| JsErrorBox::generic(e.to_string())).boxed()))
+  }
 }
 
 // This is a custom enum to allow the retry policy to clone the variants that could be retried.