perf(pm): partition install_packages into classify + 3-phase pipeline

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<OmitType>) -> 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<usize, Vec<(String, Package)>>,
     cwd: &Path,
     omit: &HashSet<OmitType>,
 ) -> 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<tokio::task::JoinHandle<Result<()>>> = 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<Plan> = packages
+            .iter()
+            .map(|(path, pkg)| classify(path, pkg, omit, cwd))
+            .collect::<Result<_>>()?;
+
+        // 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<OmitType>, cwd: &Path) -> Result<Plan> {
+    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();

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<ProgressState>`
+/// 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<Instant> = OnceCell::new();