perf(pm): install hot-path — TTY-gate progress, stream layer joins, zero-copy tar entries

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"]

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<tokio::task::JoinHandle<Result<()>>> = 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<tokio::task::JoinHandle<Result<()>>> =
+            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,28 +117,28 @@ 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;
                     }
 
                     // skip when cpu or os is not compatible
                     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();

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<OnceMap<PathBuf, ()>> = Lazy::new(OnceMap::new);
+pub(crate) static CLONE_CACHE: Lazy<OnceMap<PathBuf, ()>> = 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<bool> {
 pub async fn find_real_src<P: AsRef<Path>>(src: P) -> Option<PathBuf> {
     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"
         {

crates/pm/src/util/downloader.rs

@@ -27,6 +27,20 @@ static DOWNLOAD_CACHE: Lazy<OnceMap<String, PathBuf>> = Lazy::new(OnceMap::new);
 /// Semaphore controlling concurrent download count.
 static DOWNLOAD_SEMAPHORE: OnceLock<Semaphore> = 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<Semaphore> = 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<PathBuf> {
+    // 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<PathBuf> {
     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<PathBuf>`. 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)