From b92fc85a256a4b13a6a4b3e0dbaf1a9956509d0b Mon Sep 17 00:00:00 2001 From: elrrrrrrr Date: Sat, 2 May 2026 23:09:21 +0800 Subject: [PATCH] =?UTF-8?q?perf(pm):=20install=20hot-path=20=E2=80=94=20TT?= =?UTF-8?q?Y-gate=20progress,=20stream=20layer=20joins,=20zero-copy=20tar?= =?UTF-8?q?=20entries?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three independent install-path changes that compound on cold installs. 1. **TTY-gate `PROGRESS_BAR.inc`/`set_length`** — `install.rs` had 8 raw `PROGRESS_BAR.inc(1)` plus a `set_length` that bypassed the `IS_TTY` short-circuit already present in `ProgressReceiver`. indicatif always takes the internal `Mutex` write-lock even with a hidden draw target, so non-TTY runs (CI, piped output) were paying ~9k Mutex acquisitions per install. Wrapped them in `progress_inc` / `progress_set_length` helpers that match the receiver's gating pattern. 2. **`install_packages` per-layer streaming** — replaced `for task in clone_tasks { task.await?? }` with `FuturesUnordered`. The previous serial join walked tasks in spawn order, so a single slow tarball blocked the layer's `await` loop while later finishers piled up unobserved — pure idle latency between layers on deep trees. Now the layer drains in completion order and propagates the first error early. 3. **Zero-copy tar entries in `extract_tarball_sync`** — `ExtractedEntry` previously held an owned `Vec` copied via `read_to_end` for every file. Each tarball with N files paid N allocs + N memcpys before the parallel-write phase even started. Switched to `(data_offset, data_len)` indices into the decompressed buffer; the rayon write phase slices into the shared `&[u8]` directly. Tar Slip + path-traversal guards untouched, plus a buffer-overrun check on the recorded offsets. Built on top of #2874 to compound with the rayon offload + h1+pool+dns network layer. CI bench-phases-linux will validate the deltas. --- crates/pm/src/service/install.rs | 35 ++++++--- crates/pm/src/util/extractor.rs | 131 ++++++++++++++++++++----------- crates/pm/src/util/logger.rs | 21 +++++ 3 files changed, 127 insertions(+), 60 deletions(-) diff --git a/crates/pm/src/service/install.rs b/crates/pm/src/service/install.rs index 0390a17af1..d1c28fc431 100644 --- a/crates/pm/src/service/install.rs +++ b/crates/pm/src/service/install.rs @@ -1,6 +1,7 @@ use crate::util::cli_enum::ScriptPolicy; use anyhow::Context; use anyhow::Result; +use futures::stream::{FuturesUnordered, StreamExt}; use std::collections::{HashMap, HashSet}; use std::path::Path; use std::time::Instant; @@ -21,7 +22,8 @@ use crate::util::downloader::download_stats; use crate::util::json::load_package_lock_json_from_path; use crate::util::linker::link; use crate::util::logger::{ - PROGRESS_BAR, finish_progress_bar, log_progress, print_install_counts, start_progress_bar, + finish_progress_bar, log_progress, print_install_counts, progress_inc, progress_set_length, + start_progress_bar, }; use utoo_ruborist::compat::{is_cpu_compatible, is_os_compatible}; @@ -83,13 +85,20 @@ pub async fn install_packages( depths.sort_unstable(); for depth in depths.iter() { - let mut clone_tasks: Vec>> = Vec::new(); + // Stream task completion within a layer instead of joining serially. + // The previous `for task in clone_tasks { task.await?? }` walks tasks + // in spawn order — slow tasks block the pop of already-finished + // ones, leaving idle latency between layers on deep trees. With + // FuturesUnordered we drain in completion order, propagating the + // first error early. + let mut clone_tasks: FuturesUnordered>> = + FuturesUnordered::new(); if let Some(packages) = groups.get(depth) { for (path, package) in packages.iter() { // Skip packages based on omit config if should_omit_package(package, omit) { - PROGRESS_BAR.inc(1); + progress_inc(1); continue; } let path = path.clone(); @@ -108,13 +117,13 @@ pub async fn install_packages( if package.link.is_some() { let link_name = extract_package_name(&path); if link_name.is_empty() { - PROGRESS_BAR.inc(1); + progress_inc(1); continue; } link(Path::new(&resolved), Path::new(&path)) .await .with_context(|| format!("Link failed: {resolved} -> {path}"))?; - PROGRESS_BAR.inc(1); + progress_inc(1); continue; } @@ -122,14 +131,14 @@ pub async fn install_packages( if let Some(ref cpu) = package.cpu && !is_cpu_compatible(cpu) { - PROGRESS_BAR.inc(1); + progress_inc(1); continue; } if let Some(ref os) = package.os && !is_os_compatible(os) { - PROGRESS_BAR.inc(1); + progress_inc(1); continue; } @@ -153,24 +162,24 @@ pub async fn install_packages( tracing::warn!( "Optional dependency {name} failed (ignored): {e:#}" ); - PROGRESS_BAR.inc(1); + progress_inc(1); return Ok(()); } return Err(e); } - PROGRESS_BAR.inc(1); + progress_inc(1); log_progress(&format!("{name} resolved")); update_package_binary(&target_path, &name).await }); clone_tasks.push(task); } else { - PROGRESS_BAR.inc(1); + progress_inc(1); } } } - for task in clone_tasks { - task.await??; + while let Some(joined) = clone_tasks.next().await { + joined??; } } @@ -260,7 +269,7 @@ impl InstallService { if !package_lock.packages.is_empty() { start_progress_bar(); - PROGRESS_BAR.set_length(package_lock.packages.len() as u64); + progress_set_length(package_lock.packages.len() as u64); } let link_start = Instant::now(); diff --git a/crates/pm/src/util/extractor.rs b/crates/pm/src/util/extractor.rs index 29b71149b0..a77bf5afc1 100644 --- a/crates/pm/src/util/extractor.rs +++ b/crates/pm/src/util/extractor.rs @@ -41,9 +41,15 @@ fn estimate_uncompressed_size(gzip_data: &[u8]) -> usize { } } +/// Tarball entry recorded during the parse pass, deferred until parallel +/// write. `data_offset`/`data_len` index into the decompressed buffer +/// (lifetime-erased from the buffer for reuse during `par_iter`); avoids +/// the per-file `Vec` allocation+copy that `read_to_end` previously +/// did for every file in the archive. struct ExtractedEntry { path: PathBuf, - content: Vec, + data_offset: usize, + data_len: usize, mode: u32, } @@ -70,7 +76,7 @@ fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) - use rayon::prelude::*; use std::collections::HashSet; use std::fs; - use std::io::{Cursor, Read, Write}; + use std::io::Write; // Decompress gzip using libdeflate let mut output = vec![0u8; estimated_size]; @@ -90,49 +96,11 @@ fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) - }; output.truncate(actual_size); - // Parse tar entries - let cursor = Cursor::new(&output[..]); - let mut archive = tar::Archive::new(cursor); - let mut entries = Vec::new(); - - for entry_result in archive.entries()? { - let mut entry = entry_result.with_context(|| "Failed to read tar entry")?; - let path = entry - .path() - .with_context(|| "Failed to get entry path")? - .into_owned(); - - // Guard against path traversal (Tar Slip): reject absolute paths - // and entries containing ".." components. - if path.is_absolute() - || path - .components() - .any(|c| matches!(c, std::path::Component::ParentDir)) - { - tracing::warn!("Skipping tar entry with unsafe path: {}", path.display()); - continue; - } - - let full_path = dest.join(&path); - - if !entry.header().entry_type().is_dir() { - let mut content = Vec::new(); - entry - .read_to_end(&mut content) - .with_context(|| format!("Failed to read tar entry: {}", path.display()))?; - - // Normalize to npm/pnpm convention: 0o755 if any exec bit is set, - // else 0o644. Preserving raw tar modes (e.g. 0o640 in google-protobuf) - // breaks world-readability in containers and multi-user setups. - let raw_mode = entry.header().mode().unwrap_or(0o644); - let mode = if raw_mode & 0o111 != 0 { 0o755 } else { 0o644 }; - entries.push(ExtractedEntry { - path: full_path, - content, - mode, - }); - } - } + // Parse tar entries — record offsets into `output` instead of copying + // each file's bytes. The buffer outlives the parallel write phase and + // is read-only after the parse loop, so concurrent slice reads are + // safe. + let entries = parse_tar_entries(&output, dest)?; // Collect every ancestor directory (up to `dest`), then create them // shallowest-first so a single mkdir() per dir is sufficient. @@ -152,11 +120,14 @@ fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) - fs::create_dir(dir).ok(); } - // Write files in parallel using rayon + // Write files in parallel using rayon. Each task slices into the + // shared `output` buffer (immutable borrow) — no per-file copy. + let buf: &[u8] = &output; entries.par_iter().try_for_each(|entry| -> Result<()> { let mut file = fs::File::create(&entry.path) .with_context(|| format!("Failed to create: {}", entry.path.display()))?; - file.write_all(&entry.content) + let data = &buf[entry.data_offset..entry.data_offset + entry.data_len]; + file.write_all(data) .with_context(|| format!("Failed to write: {}", entry.path.display()))?; // Skip chmod for 0o644 (most files) — File::create() already produces @@ -181,3 +152,69 @@ fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) - Ok(()) } + +/// Walk the tar stream and collect file entries as `(path, offset, len, mode)` +/// tuples into the decompressed `buf`. Skips directories, ignores any +/// path-traversal entries. +fn parse_tar_entries(buf: &[u8], dest: &Path) -> Result> { + use std::io::Cursor; + + let cursor = Cursor::new(buf); + let mut archive = tar::Archive::new(cursor); + let mut entries = Vec::new(); + + for entry_result in archive.entries()? { + let entry = entry_result.with_context(|| "Failed to read tar entry")?; + let path = entry + .path() + .with_context(|| "Failed to get entry path")? + .into_owned(); + + // Guard against path traversal (Tar Slip): reject absolute paths + // and entries containing ".." components. + if path.is_absolute() + || path + .components() + .any(|c| matches!(c, std::path::Component::ParentDir)) + { + tracing::warn!("Skipping tar entry with unsafe path: {}", path.display()); + continue; + } + + if entry.header().entry_type().is_dir() { + continue; + } + + // tar::Entry exposes the byte offset and size inside the parsed + // stream — that's the same `buf` we'll later slice in `par_iter`. + let data_offset = entry.raw_file_position() as usize; + let data_len = entry.size() as usize; + if data_offset + .checked_add(data_len) + .is_none_or(|end| end > buf.len()) + { + anyhow::bail!( + "Tar entry {} extends past decompressed buffer ({}..+{} > {})", + path.display(), + data_offset, + data_len, + buf.len() + ); + } + + let full_path = dest.join(&path); + // Normalize to npm/pnpm convention: 0o755 if any exec bit is set, + // else 0o644. Preserving raw tar modes (e.g. 0o640 in google-protobuf) + // breaks world-readability in containers and multi-user setups. + let raw_mode = entry.header().mode().unwrap_or(0o644); + let mode = if raw_mode & 0o111 != 0 { 0o755 } else { 0o644 }; + entries.push(ExtractedEntry { + path: full_path, + data_offset, + data_len, + mode, + }); + } + + Ok(entries) +} diff --git a/crates/pm/src/util/logger.rs b/crates/pm/src/util/logger.rs index 6ba124bb24..c8570b2eca 100644 --- a/crates/pm/src/util/logger.rs +++ b/crates/pm/src/util/logger.rs @@ -161,6 +161,27 @@ pub fn log_progress(text: &str) { PROGRESS_BAR.set_message(text.to_string()); } +/// TTY-gated `PROGRESS_BAR.inc(1)`. Non-TTY callers pay no Mutex cost — +/// indicatif's `inc` always takes the internal `Mutex` +/// write-lock even with a hidden draw target, which contends across the +/// install loop's spawned tasks. +#[inline] +pub fn progress_inc(n: u64) { + if !*IS_TTY { + return; + } + PROGRESS_BAR.inc(n); +} + +/// TTY-gated `PROGRESS_BAR.set_length`. Same rationale as [`progress_inc`]. +#[inline] +pub fn progress_set_length(len: u64) { + if !*IS_TTY { + return; + } + PROGRESS_BAR.set_length(len); +} + // Global timer for log_time/log_time_end static START_TIME: OnceCell = OnceCell::new();