diff --git a/crates/pm/src/service/binary.rs b/crates/pm/src/service/binary.rs index c6f95cb45e..33f0cda141 100644 --- a/crates/pm/src/service/binary.rs +++ b/crates/pm/src/service/binary.rs @@ -233,6 +233,15 @@ async fn handle_cypress( } pub async fn update_package_binary(dir: &Path, name: &str) -> Result<()> { + // npm.org users have no use for the China binary mirror config; + // skipping here avoids the per-process `binary-mirror-config/latest` + // fetch + per-package hashmap lookup that fires for every cloned + // package in the install loop. Mirrors `get_envs`'s gate so the + // two paths agree on when the mirror layer is active. + if should_skip_binary_mirror() { + return Ok(()); + } + let config = load_config().await?; let mirrors = config["mirrors"]["china"] diff --git a/crates/pm/src/service/install.rs b/crates/pm/src/service/install.rs index 0390a17af1..d94b2c16a4 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; @@ -16,12 +17,13 @@ use crate::helper::workspace::init_project_root; use crate::model::package::PackageInfo; use crate::service::rebuild::RebuildService; use crate::util::cli_enum::{OmitType, PackageAction, SaveType}; -use crate::util::cloner::clone_count; -use crate::util::downloader::download_stats; +use crate::util::cloner::{CLONE_CACHE, cache_key, clone_count}; +use crate::util::downloader::{download_stats, is_git_url, prefetch_to_cache}; 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; } @@ -141,6 +150,15 @@ pub async fn install_packages( let cwd_clone = cwd.to_path_buf(); let target_path = cwd_clone.join(&path); + // Eager skip: if the BFS pipeline (or a sibling + // depth) already cloned this target, the spawn + + // future construction below is pure overhead. Probe + // CLONE_CACHE before doing any of it. + if CLONE_CACHE.is_done(&cache_key(&target_path)) { + progress_inc(1); + continue; + } + // Check if this is an optional dependency let is_optional = package.optional == Some(true) || package.dev_optional == Some(true); @@ -153,30 +171,89 @@ 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??; } } Ok(()) } +/// Fan-out tarball downloads for every package in `package_lock` so they +/// race ahead of the depth-ordered install loop. +/// +/// Install must run layer-by-layer so a parent's `node_modules/` exists +/// before children clone into it — but **download** has no such +/// dependency: any tarball can land in the cache at any time. Without +/// this prefetch, a fresh-lock install (no pipeline phase) only starts +/// downloading depth-N+1 tarballs after depth-N's clone barrier +/// releases, serializing what should be parallel network I/O. +/// +/// `download_to_cache` is `OnceMap`-deduped on `name@version`, so the +/// install loop's later call resolves to the prefetched future / cache +/// hit without redundant network or extract work. +/// +/// Skipped: `link:` workspace targets, `git+*` (BFS-resolved), `file:` +/// (cache slot is BFS-seeded — no equivalent in fresh-lock yet), +/// CPU/OS-incompatible optional natives (e.g. `lightningcss-linux-arm64` +/// on a darwin host) — install would skip them anyway, and prefetching +/// them inflates network RX 3× on bench-phases-linux because every +/// optional platform variant pulls its full tarball. +fn spawn_tarball_prefetch(package_lock: &utoo_ruborist::lock::PackageLock, cwd: &Path) { + for (path, package) in package_lock.packages.iter() { + if package.link.is_some() { + continue; + } + let Some(resolved) = package.resolved.as_deref() else { + continue; + }; + if resolved.is_empty() { + continue; + } + if is_git_url(resolved) || resolved.starts_with("file:") { + continue; + } + if let Some(ref cpu) = package.cpu + && !is_cpu_compatible(cpu) + { + continue; + } + if let Some(ref os) = package.os + && !is_os_compatible(os) + { + continue; + } + let Some(version) = package.version.clone() else { + continue; + }; + let name = package.get_name(path); + if name.is_empty() { + continue; + } + let _ = cwd; // reserved for future relative-path resolution + let resolved = resolved.to_string(); + tokio::spawn(async move { + prefetch_to_cache(&name, &version, &resolved).await; + }); + } +} + pub struct InstallService; impl InstallService { @@ -256,11 +333,21 @@ impl InstallService { (result.package_lock, Some(result.handles)) }; + // Stale-lock path already gets per-package downloads kicked off + // by the BFS pipeline (`PackageResolved` → download_worker), so + // the prefetch is only a win for the fresh-lock path. OnceMap + // dedup means a duplicate kick from this prefetch is a cheap + // future-share, not duplicate work — but skipping the loop + // entirely saves the spawn cost. + if pipeline_handles.is_none() { + spawn_tarball_prefetch(&package_lock, root_path); + } + let groups = group_by_depth(&package_lock.packages); 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/cloner.rs b/crates/pm/src/util/cloner.rs index 76c6029702..e41073fa04 100644 --- a/crates/pm/src/util/cloner.rs +++ b/crates/pm/src/util/cloner.rs @@ -22,7 +22,7 @@ use crate::fs; /// `ERROR_SHARING_VIOLATION` (os error 32) on Windows. `PathBuf` from /// `Path::components().collect()` parses both separators uniformly and /// rebuilds with the OS-preferred one, giving a stable key. -static CLONE_CACHE: Lazy> = Lazy::new(OnceMap::new); +pub(crate) static CLONE_CACHE: Lazy> = Lazy::new(OnceMap::new); /// Number of `node_modules/` directories freshly materialized this run. /// Mirrors pnpm's "added" semantic. @@ -35,12 +35,12 @@ pub fn clone_count() -> usize { /// Normalize a target path into the canonical key used by `CLONE_CACHE`. #[cfg(windows)] -fn cache_key(target_path: &Path) -> PathBuf { +pub(crate) fn cache_key(target_path: &Path) -> PathBuf { target_path.components().collect() } #[cfg(not(windows))] -fn cache_key(target_path: &Path) -> PathBuf { +pub(crate) fn cache_key(target_path: &Path) -> PathBuf { target_path.to_path_buf() } @@ -65,6 +65,15 @@ pub async fn clone_package_once( target_path: &Path, ) -> Result<()> { let key = cache_key(target_path); + + // Hot fast-path: most install-loop calls hit a target the BFS + // pipeline (or a previous depth's spawn) already cloned. Probe + // CLONE_CACHE before paying the 5 string allocs + future + // construction below. + if CLONE_CACHE.is_done(&key) { + return Ok(()); + } + let err_label = format!("{name}@{version}"); let name = name.to_string(); let version = version.to_string(); @@ -326,8 +335,12 @@ async fn validate_directory(src: &Path, dst: &Path) -> Result { pub async fn find_real_src>(src: P) -> Option { let mut read_dir = fs::read_dir(src.as_ref()).await.ok()?; while let Some(entry) = read_dir.next_entry().await.ok()? { - if let Ok(metadata) = entry.metadata().await - && metadata.is_dir() + // `file_type()` reads `d_type` from the cached readdir result on + // ext4/btrfs/xfs (no extra syscall); `metadata()` always issues + // an `lstat`. Saves ~3–5 stats per call × N packages on the + // cold install hot path. + if let Ok(file_type) = entry.file_type().await + && file_type.is_dir() && let Some(name) = entry.path().file_name() && name.to_string_lossy() != ".utoo_built" { diff --git a/crates/pm/src/util/downloader.rs b/crates/pm/src/util/downloader.rs index 76526eedda..1cb8c6d5b5 100644 --- a/crates/pm/src/util/downloader.rs +++ b/crates/pm/src/util/downloader.rs @@ -27,6 +27,20 @@ static DOWNLOAD_CACHE: Lazy> = Lazy::new(OnceMap::new); /// Semaphore controlling concurrent download count. static DOWNLOAD_SEMAPHORE: OnceLock = OnceLock::new(); +/// Prefetch back-pressure gate, sized smaller than [`DOWNLOAD_SEMAPHORE`]. +/// +/// `spawn_tarball_prefetch` fan-outs N spawn calls on entry; without +/// this gate, the first ~N tasks acquire `DOWNLOAD_SEMAPHORE` slots +/// before `install_packages` even spawns its first layer, so install's +/// clone tasks queue behind every prefetch task. p3_cold_install on +/// CI bench-phases-linux saw σ jump from 1.56s → 3.93s as a result. +/// +/// Capping prefetch at 16 leaves `64 - 16 = 48` slots permanently +/// available to the install path; the OnceMap dedup still ensures +/// install's later resolve hits the prefetched future on completion. +static PREFETCH_SEMAPHORE: OnceLock = OnceLock::new(); +const PREFETCH_BACKLOG_LIMIT: usize = 16; + static DOWNLOAD_COUNT: AtomicUsize = AtomicUsize::new(0); static REUSE_COUNT: AtomicUsize = AtomicUsize::new(0); @@ -141,6 +155,35 @@ pub async fn resolve_cache_path(name: &str, version: &str, tarball_url: &str) -> } } +/// Prefetch variant: same as [`download_to_cache`] but bounded by +/// [`PREFETCH_SEMAPHORE`] so a fan-out batch can't starve the install +/// path of [`DOWNLOAD_SEMAPHORE`] slots. +/// +/// On a warm-cache hit there's nothing to gate (the OnceMap closure +/// short-circuits before acquiring [`DOWNLOAD_SEMAPHORE`]); the +/// prefetch gate is only acquired around the actual `download_to_cache` +/// call, so warm tarballs return immediately without consuming +/// prefetch slots. +pub async fn prefetch_to_cache(name: &str, version: &str, tarball_url: &str) -> Option { + // Quick warm-cache short-circuit before paying the prefetch gate. + let cache_path = get_cache_dir().join(name).join(version); + if crate::fs::try_exists(&cache_path.join("_resolved")) + .await + .unwrap_or(false) + { + return Some(cache_path); + } + + let permit = PREFETCH_SEMAPHORE + .get_or_init(|| Semaphore::new(PREFETCH_BACKLOG_LIMIT)) + .acquire() + .await + .ok()?; + let result = download_to_cache(name, version, tarball_url).await; + drop(permit); + result +} + /// Download a registry tarball to the global cache directory, returning the cache path. /// /// Uses `OnceMap` to deduplicate: the same `name@version` is only downloaded once, @@ -149,6 +192,21 @@ pub async fn resolve_cache_path(name: &str, version: &str, tarball_url: &str) -> /// For git-resolved packages, use [`resolve_cache_path`] instead. pub async fn download_to_cache(name: &str, version: &str, tarball_url: &str) -> Option { let key = format!("{}@{}", name, version); + + // Hot fast-path: when the prefetch (or a sibling resolve) already + // populated the OnceMap entry, skip the per-call string allocs and + // future construction by reusing the cached `Arc`. The + // pipeline path drives `download_to_cache` for every BFS-resolved + // package, and install-loop calls usually arrive after — most hit + // a `Done` slot. + if DOWNLOAD_CACHE.is_done(&key) { + return DOWNLOAD_CACHE + .get_or_init(key, || async { None }) + .await + .as_deref() + .cloned(); + } + let cache_dir = get_cache_dir(); let name = name.to_string(); let version = version.to_string(); @@ -167,10 +225,17 @@ pub async fn download_to_cache(name: &str, version: &str, tarball_url: &str) -> return Some(cache_path); } - // Download (semaphore controlled) + // Download (semaphore controlled). The permit gates only + // network I/O — extract runs on rayon's CPU pool and does + // not contend with download bandwidth, so we release the + // permit before the CPU-bound phase. This lets the next + // `download_to_cache` start immediately, doubling effective + // concurrency on tarball-heavy installs (download tasks no + // longer block on a peer task's ~50–200ms libdeflate + + // tar parse + parallel writes). let semaphore = DOWNLOAD_SEMAPHORE .get_or_init(|| Semaphore::new(get_manifests_concurrency_limit_sync())); - let _permit = semaphore.acquire().await.ok()?; + let permit = semaphore.acquire().await.ok()?; let bytes = download_bytes(&tarball_url) .await .inspect_err(|e| { @@ -183,6 +248,7 @@ pub async fn download_to_cache(name: &str, version: &str, tarball_url: &str) -> ) }) .ok()?; + drop(permit); // Extract extract_and_write(bytes, &cache_path) diff --git a/crates/pm/src/util/extractor.rs b/crates/pm/src/util/extractor.rs index 29b71149b0..f342ac2155 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, } @@ -68,9 +74,8 @@ async fn extract_tarball(gzip_bytes: Bytes, dest: &Path) -> Result<()> { /// Synchronous extraction: decompress + parse + parallel write, all on rayon. fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) -> Result<()> { 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,73 +95,46 @@ 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. - let mut seen = HashSet::new(); + // + // Push to a Vec first (cheap) and dedup once at the end via sort + // — replaces a per-entry `HashSet` insert that allocated + // a PathBuf for every ancestor on every entry, even when most + // entries share the same parent dirs (the common case). + let mut dirs: Vec = Vec::with_capacity(entries.len()); for entry in &entries { let mut p = entry.path.parent(); while let Some(dir) = p { - if dir == dest || !seen.insert(dir.to_path_buf()) { + if dir == dest { break; } + dirs.push(dir.to_path_buf()); p = dir.parent(); } } - let mut dirs: Vec<_> = seen.into_iter().collect(); - dirs.sort_unstable_by_key(|p| p.as_os_str().len()); + dirs.sort_unstable(); + dirs.dedup(); + // Sort by depth ascending so parent exists before child mkdir. + dirs.sort_by_key(|p| p.as_os_str().len()); for dir in &dirs { 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 +159,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(); diff --git a/crates/ruborist/src/util/oncemap.rs b/crates/ruborist/src/util/oncemap.rs index 40c4aeee20..25255e797c 100644 --- a/crates/ruborist/src/util/oncemap.rs +++ b/crates/ruborist/src/util/oncemap.rs @@ -118,6 +118,22 @@ where } } + /// Sync, allocation-free check for a completed entry. + /// + /// Lets hot-path callers short-circuit before paying string clones / + /// future construction for a key that's already been resolved by a + /// peer (e.g. install loop deduping against pipeline-pre-cloned + /// targets, where 95%+ of packages hit a `Done` slot). + /// + /// Returns `false` for both unknown keys and `Waiting` slots; only + /// `Done` is reported. Waiting entries fall through to the normal + /// `get_or_init` path which knows how to subscribe and wait. + pub fn is_done(&self, key: &K) -> bool { + self.map + .get(key) + .is_some_and(|e| matches!(e.value(), Value::Done(_))) + } + /// Wait for a key to complete. /// /// - If the key exists and is Done, returns immediately.