diff --git a/crates/pm/src/service/install.rs b/crates/pm/src/service/install.rs index 0390a17af1..cecf48354f 100644 --- a/crates/pm/src/service/install.rs +++ b/crates/pm/src/service/install.rs @@ -1,8 +1,9 @@ use crate::util::cli_enum::ScriptPolicy; use anyhow::Context; use anyhow::Result; +use futures::stream::{self, TryStreamExt}; use std::collections::{HashMap, HashSet}; -use std::path::Path; +use std::path::{Path, PathBuf}; use std::time::Instant; use crate::cmd::deps::build_deps; @@ -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}; @@ -62,13 +64,44 @@ fn should_omit_package(package: &Package, omit: &HashSet) -> bool { false } +/// Maximum in-flight clone operations per depth layer (Phase 3 below). +/// +/// Mirrors `manifests-concurrency-limit` default and `par_chunks(64)` in +/// the extractor so the install pipeline has one consistent throttle. +const SPAWN_BUDGET: usize = 64; + +/// What `install_packages` decided to do with one lockfile entry, after +/// all the cheap "should we even touch this?" checks. Splitting this from +/// the actual execution lets us classify the whole layer with pure CPU, +/// then dispatch each variant on the schedule that fits it: skip is free, +/// link is sequential I/O, clone is bounded-concurrent. +enum Plan { + /// Omit / no resolved URL / cpu-or-os incompat / empty workspace name. + /// Still consumes one progress slot so the bar stays accurate. + Skip, + /// Workspace `file:` link — cheap symlink syscall, kept sequential + /// because typical projects have a handful of these. + Link { resolved: String, path: String }, + /// Real download + extract work — only this variant joins the + /// concurrent stream below. + Clone(CloneWork), +} + +/// All the data the clone path needs, precomputed during classification +/// so the concurrent phase doesn't re-walk the `Package` struct. +struct CloneWork { + name: String, + version: String, + resolved: String, + target_path: PathBuf, + is_optional: bool, +} + pub async fn install_packages( groups: &HashMap>, cwd: &Path, omit: &HashSet, ) -> Result<()> { - use crate::util::cloner::clone_package_once; - // Surface the clean step in the spinner — it doesn't move `pos`, so // without a message the bar looks frozen on large trees. log_progress("validating node_modules"); @@ -83,100 +116,155 @@ pub async fn install_packages( depths.sort_unstable(); for depth in depths.iter() { - let mut clone_tasks: Vec>> = Vec::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); - continue; - } - let path = path.clone(); - let package = package.clone(); - if let Some(ref resolved) = package.resolved { - // Lockfile stores `file:` URLs root-relative (npm format). - // Cloner only understands absolute URLs — re-absolutize - // here so the cloner/downloader stays unaware of project - // root plumbing. - let resolved = match resolved.strip_prefix("file:") { - Some(rel) if !Path::new(rel).is_absolute() => { - format!("file:{}", cwd.join(rel).display()) - } - _ => resolved.clone(), - }; - if package.link.is_some() { - let link_name = extract_package_name(&path); - if link_name.is_empty() { - PROGRESS_BAR.inc(1); - continue; - } - link(Path::new(&resolved), Path::new(&path)) - .await - .with_context(|| format!("Link failed: {resolved} -> {path}"))?; - PROGRESS_BAR.inc(1); - continue; - } - - // skip when cpu or os is not compatible - if let Some(ref cpu) = package.cpu - && !is_cpu_compatible(cpu) - { - PROGRESS_BAR.inc(1); - continue; - } - - if let Some(ref os) = package.os - && !is_os_compatible(os) - { - PROGRESS_BAR.inc(1); - continue; - } - - let name = package.get_name(&path); - let version = package - .version - .clone() - .ok_or_else(|| anyhow::anyhow!("package {name} missing version"))?; - let cwd_clone = cwd.to_path_buf(); - let target_path = cwd_clone.join(&path); - - // Check if this is an optional dependency - let is_optional = - package.optional == Some(true) || package.dev_optional == Some(true); - - let task = tokio::spawn(async move { - if let Err(e) = - clone_package_once(&name, &version, &resolved, &target_path).await - { - if is_optional { - tracing::warn!( - "Optional dependency {name} failed (ignored): {e:#}" - ); - PROGRESS_BAR.inc(1); - return Ok(()); - } - return Err(e); - } - PROGRESS_BAR.inc(1); - log_progress(&format!("{name} resolved")); - update_package_binary(&target_path, &name).await - }); - clone_tasks.push(task); - } else { - PROGRESS_BAR.inc(1); + let Some(packages) = groups.get(depth) else { + continue; + }; + + // Phase 1 — pure-CPU classification. ~µs per entry; does not + // touch the runtime, does not occupy any semaphore. + let plans: Vec = packages + .iter() + .map(|(path, pkg)| classify(path, pkg, omit, cwd)) + .collect::>()?; + + // Phase 2 — drain the cheap variants. Skip is a counter bump, + // link is one syscall awaited inline. Both finish before any + // download starts, so they cannot starve in-flight socket reads + // (there are none yet at this depth). + let mut clones = Vec::new(); + for plan in plans { + match plan { + Plan::Skip => progress_inc(1), + Plan::Link { resolved, path } => { + link(Path::new(&resolved), Path::new(&path)) + .await + .with_context(|| format!("Link failed: {resolved} -> {path}"))?; + progress_inc(1); } + Plan::Clone(work) => clones.push(work), } } - for task in clone_tasks { - task.await??; - } + // Phase 3 — bounded-concurrent stream of real download+extract + // work. `try_for_each_concurrent` short-circuits on the first + // error. + // + // `consume_budget` per stream item is load-bearing: pcap on the + // partition design without it (run 25551591284) measured + // `zwin=14` vs baseline `0`. The pattern + // `tokio::spawn(...).await` does not give the runtime a yield + // window between consecutive spawns when JoinHandle's await + // resolves immediately (cache hit, fast tarball). Hinting one + // tick of work per stream-poll item lets the runtime service + // in-flight socket reads on the global download semaphore's + // tarball connections; otherwise the kernel recv buffer fills + // before tokio gets back to drain it. + // + // Compared to `yield_now()` (which always yields), `consume_budget` + // only yields when the per-task tick budget is exhausted — + // ~5ns when it doesn't, ~100ns when it does. For the typical + // mixed pattern (some items resolve instantly via cache hit, + // most await network) it pays approximately nothing on the + // slow path while still preventing the tight-spawn starvation. + stream::iter(clones.into_iter().map(Ok::<_, anyhow::Error>)) + .try_for_each_concurrent(SPAWN_BUDGET, |work| async move { + tokio::task::consume_budget().await; + tokio::spawn(execute_clone(work)).await? + }) + .await?; } Ok(()) } +/// Decide what to do with a single lockfile entry. Pure CPU — no I/O, +/// no spawn — so the per-layer Phase 1 sweep finishes in <5ms even on +/// deep trees and never holds up the runtime. +fn classify(path: &str, package: &Package, omit: &HashSet, cwd: &Path) -> Result { + if should_omit_package(package, omit) { + return Ok(Plan::Skip); + } + + let Some(resolved) = package.resolved.as_ref() else { + return Ok(Plan::Skip); + }; + + // Lockfile stores `file:` URLs root-relative (npm format). Cloner only + // understands absolute URLs — re-absolutize here so the cloner / + // downloader stays unaware of project root plumbing. + let resolved = match resolved.strip_prefix("file:") { + Some(rel) if !Path::new(rel).is_absolute() => { + format!("file:{}", cwd.join(rel).display()) + } + _ => resolved.clone(), + }; + + if package.link.is_some() { + let link_name = extract_package_name(path); + if link_name.is_empty() { + return Ok(Plan::Skip); + } + return Ok(Plan::Link { + resolved, + path: path.to_string(), + }); + } + + if let Some(ref cpu) = package.cpu + && !is_cpu_compatible(cpu) + { + return Ok(Plan::Skip); + } + if let Some(ref os) = package.os + && !is_os_compatible(os) + { + return Ok(Plan::Skip); + } + + let name = package.get_name(path); + let version = package + .version + .clone() + .ok_or_else(|| anyhow::anyhow!("package {name} missing version"))?; + let target_path = cwd.join(path); + let is_optional = package.optional == Some(true) || package.dev_optional == Some(true); + + Ok(Plan::Clone(CloneWork { + name, + version, + resolved, + target_path, + is_optional, + })) +} + +/// Phase 3 worker: download + extract one package, then update its +/// binary symlinks. Optional-dep failures are demoted to a warn so a +/// missing platform binary doesn't fail the whole install. +async fn execute_clone(work: CloneWork) -> Result<()> { + use crate::util::cloner::clone_package_once; + + let CloneWork { + name, + version, + resolved, + target_path, + is_optional, + } = work; + + if let Err(e) = clone_package_once(&name, &version, &resolved, &target_path).await { + if is_optional { + tracing::warn!("Optional dependency {name} failed (ignored): {e:#}"); + progress_inc(1); + return Ok(()); + } + return Err(e); + } + progress_inc(1); + log_progress(&format!("{name} resolved")); + update_package_binary(&target_path, &name).await +} + pub struct InstallService; impl InstallService { @@ -260,7 +348,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/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();