perf(pm): full PR #2818 perf bundle on top of worker-pool

crates/pm/src/helper/lock.rs

@@ -481,17 +481,35 @@ pub async fn is_pkg_lock_outdated(root_path: &Path) -> Result<bool> {
 
 /// Save PackageLock to disk synchronously
 pub async fn save_package_lock(path: &Path, package_lock: &PackageLock) -> Result<()> {
+    let t_total = std::time::Instant::now();
     let temp_path = path.join("package-lock.json.tmp");
     let target_path = path.join("package-lock.json");
 
-    // PackageLock now has all required fields (name, version, lockfile_version, requires, packages)
+    let t_serialize = std::time::Instant::now();
     let content = serde_json::to_string_pretty(package_lock)?;
+    let serialize_ms = t_serialize.elapsed().as_millis();
+    let bytes = content.len();
+
+    let t_write = std::time::Instant::now();
     fs::write(&temp_path, content)
         .await
         .context("Failed to write temporary package-lock.json")?;
+    let write_ms = t_write.elapsed().as_millis();
+
+    let t_rename = std::time::Instant::now();
     fs::rename(temp_path, target_path)
         .await
         .context("Failed to rename package-lock.json")?;
+    let rename_ms = t_rename.elapsed().as_millis();
+
+    tracing::info!(
+        "Save package-lock.json: total={}ms (serialize={}ms write={}ms rename={}ms, size={}KB)",
+        t_total.elapsed().as_millis(),
+        serialize_ms,
+        write_ms,
+        rename_ms,
+        bytes / 1024,
+    );
 
     Ok(())
 }

crates/pm/src/main.rs

@@ -352,6 +352,15 @@ fn main() {
     let result = tokio::runtime::Builder::new_multi_thread()
         .enable_all()
         .worker_threads(worker_threads)
+        // Cap the blocking pool to reduce thread-creation thrash on
+        // runs that burst-dispatch ~128 spawn_blocking calls (JSON
+        // parse for manifest fetches). Default 512 leaves room for
+        // unbounded growth; ~32 threads is enough to saturate a 4-10
+        // core machine on short CPU-bound tasks while keeping the
+        // pool warm across runs. Observed locally: default pool gave
+        // bimodal walls (2.7s fast / 6.9s thrash on M2); capping at
+        // 32 eliminates the thrash peak.
+        .max_blocking_threads(worker_threads)
         .build()
         .expect("failed to build tokio runtime")
         .block_on(async_main());

crates/pm/src/service/install.rs

@@ -287,6 +287,18 @@ impl InstallService {
         let added = clone_count().saturating_sub(clone_baseline);
         let download_delta = download_stats() - download_baseline;
         print_install_counts(added, download_delta.reused, download_delta.downloaded);
+
+        // Per-phase cumulative time (summed across concurrent tasks, not wall).
+        // Ratio to wall clock gives effective concurrency for that phase.
+        let phases = crate::util::downloader::phase_totals();
+        let clone_us = crate::util::cloner::clone_micros();
+        println!(
+            "  download {:.2}s · extract {:.2}s · clone {:.2}s · {:.1} MB fetched",
+            phases.download_micros as f64 / 1e6,
+            phases.extract_micros as f64 / 1e6,
+            clone_us as f64 / 1e6,
+            phases.downloaded_bytes as f64 / (1024.0 * 1024.0),
+        );
         Ok(())
     }
 

crates/pm/src/util/cloner.rs

@@ -1,5 +1,6 @@
 use std::path::{Path, PathBuf};
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::time::Instant;
 
 use anyhow::{Context, Result};
 use once_cell::sync::Lazy;
@@ -28,11 +29,19 @@ static CLONE_CACHE: Lazy<OnceMap<PathBuf, ()>> = Lazy::new(OnceMap::new);
 /// Mirrors pnpm's "added" semantic.
 static CLONE_COUNT: AtomicUsize = AtomicUsize::new(0);
 
+/// Wall microseconds summed across every per-package clone invocation.
+static CLONE_MICROS: AtomicU64 = AtomicU64::new(0);
+
 /// Returns the number of fresh clones performed (pnpm "added" equivalent).
 pub fn clone_count() -> usize {
     CLONE_COUNT.load(Ordering::Relaxed)
 }
 
+/// Cumulative wall microseconds spent in `clone_package` across the process.
+pub fn clone_micros() -> u64 {
+    CLONE_MICROS.load(Ordering::Relaxed)
+}
+
 /// Normalize a target path into the canonical key used by `CLONE_CACHE`.
 #[cfg(windows)]
 fn cache_key(target_path: &Path) -> PathBuf {
@@ -78,6 +87,7 @@ pub async fn clone_package_once(
     CLONE_CACHE
         .get_or_init(key, || async move {
             let cache_path = resolve_cache_path(&name, &version, &tarball_url).await?;
+            let t0 = Instant::now();
             let fresh = clone_package(&cache_path, &target_path, &name, &version, !is_git)
                 .await
                 .inspect_err(|e| {
@@ -90,6 +100,7 @@ pub async fn clone_package_once(
                     )
                 })
                 .ok()?;
+            CLONE_MICROS.fetch_add(t0.elapsed().as_micros() as u64, Ordering::Relaxed);
 
             if fresh {
                 CLONE_COUNT.fetch_add(1, Ordering::Relaxed);

crates/pm/src/util/downloader.rs

@@ -1,6 +1,7 @@
 use std::path::PathBuf;
 use std::sync::OnceLock;
-use std::sync::atomic::{AtomicU32, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicU32, AtomicU64, AtomicUsize, Ordering};
+use std::time::Instant;
 
 use anyhow::{Context, Result};
 use bytes::Bytes;
@@ -30,6 +31,28 @@ static DOWNLOAD_SEMAPHORE: OnceLock<Semaphore> = OnceLock::new();
 static DOWNLOAD_COUNT: AtomicUsize = AtomicUsize::new(0);
 static REUSE_COUNT: AtomicUsize = AtomicUsize::new(0);
 
+// Per-operation wall time summed across all packages. Sum is non-exclusive
+// across cores — dividing by wall clock gives average concurrency for the phase.
+static DOWNLOAD_MICROS: AtomicU64 = AtomicU64::new(0);
+static EXTRACT_MICROS: AtomicU64 = AtomicU64::new(0);
+static DOWNLOAD_BYTES: AtomicU64 = AtomicU64::new(0);
+
+/// Snapshot of per-phase totals accumulated during install.
+#[derive(Debug, Clone, Copy, Default)]
+pub struct PhaseTotals {
+    pub download_micros: u64,
+    pub extract_micros: u64,
+    pub downloaded_bytes: u64,
+}
+
+pub fn phase_totals() -> PhaseTotals {
+    PhaseTotals {
+        download_micros: DOWNLOAD_MICROS.load(Ordering::Relaxed),
+        extract_micros: EXTRACT_MICROS.load(Ordering::Relaxed),
+        downloaded_bytes: DOWNLOAD_BYTES.load(Ordering::Relaxed),
+    }
+}
+
 /// Process-global counters for tarball outcomes, matching pnpm's
 /// vocabulary. Each unique `(name, version)` pair lands in exactly one
 /// bucket thanks to `DOWNLOAD_CACHE`'s `OnceMap` dedup; git/file/link
@@ -181,16 +204,21 @@ pub async fn download_to_cache(name: &str, version: &str, tarball_url: &str) ->
             let semaphore = DOWNLOAD_SEMAPHORE
                 .get_or_init(|| Semaphore::new(get_manifests_concurrency_limit_sync()));
             let _permit = semaphore.acquire().await.ok()?;
+            let t0 = Instant::now();
             let bytes = download_bytes(&tarball_url)
                 .await
                 .inspect_err(|e| tracing::warn!("Download failed: {}@{}: {}", name, version, e))
                 .ok()?;
+            DOWNLOAD_MICROS.fetch_add(t0.elapsed().as_micros() as u64, Ordering::Relaxed);
+            DOWNLOAD_BYTES.fetch_add(bytes.len() as u64, Ordering::Relaxed);
 
             // Extract
+            let t1 = Instant::now();
             extract_and_write(bytes, &cache_path)
                 .await
                 .inspect_err(|e| tracing::warn!("Extract failed: {}@{}: {}", name, version, e))
                 .ok()?;
+            EXTRACT_MICROS.fetch_add(t1.elapsed().as_micros() as u64, Ordering::Relaxed);
 
             DOWNLOAD_COUNT.fetch_add(1, Ordering::Relaxed);
             tracing::debug!("Downloaded: {}@{}", name, version);

crates/pm/src/util/extractor.rs

@@ -65,13 +65,24 @@ async fn extract_tarball(gzip_bytes: Bytes, dest: &Path) -> Result<()> {
     rx.await.with_context(|| "Extract task panicked")?
 }
 
-/// Synchronous extraction: decompress + parse + parallel write, all on rayon.
+/// Synchronous extraction: decompress + parse + chunked parallel write, all on rayon.
+///
+/// Writes are batched into fixed-size chunks: each rayon task writes a contiguous
+/// run of `WRITE_CHUNK_SIZE` files sequentially. This keeps multi-core parallelism
+/// for IO-overlap while cutting the rayon task count (and its work-stealing
+/// futex park/unpark pressure) by the chunk factor. Cross-package parallelism is
+/// preserved by the outer `rayon::spawn` in `extract_tarball`.
 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};
 
+    /// Files per rayon task when writing extracted tar entries. Sized to amortize
+    /// work-stealing overhead while leaving enough tasks to keep cores busy on
+    /// large packages (e.g. 500-file packages → ~16 tasks with WRITE_CHUNK_SIZE=32).
+    const WRITE_CHUNK_SIZE: usize = 32;
+
     // Decompress gzip using libdeflate
     let mut output = vec![0u8; estimated_size];
 
@@ -152,21 +163,27 @@ fn extract_tarball_sync(gzip_bytes: Bytes, estimated_size: usize, dest: &Path) -
         fs::create_dir(dir).ok();
     }
 
-    // Write files in parallel using rayon
-    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)
-            .with_context(|| format!("Failed to write: {}", entry.path.display()))?;
-
-        // Skip chmod for 0o644 (most files) — File::create() already produces
-        // this via umask (0o666 & ~0o022 = 0o644).
-        #[cfg(unix)]
-        if entry.mode != 0o644 {
-            fs::set_permissions(&entry.path, fs::Permissions::from_mode(entry.mode)).ok();
-        }
-        Ok(())
-    })?;
+    // Chunked parallel writes: N files per rayon task. Retains IO-overlap
+    // parallelism across cores while cutting the rayon task count (and the
+    // associated work-stealing futex traffic) relative to per-file par_iter.
+    entries
+        .par_chunks(WRITE_CHUNK_SIZE)
+        .try_for_each(|chunk| -> Result<()> {
+            for entry in chunk {
+                let mut file = fs::File::create(&entry.path)
+                    .with_context(|| format!("Failed to create: {}", entry.path.display()))?;
+                file.write_all(&entry.content)
+                    .with_context(|| format!("Failed to write: {}", entry.path.display()))?;
+
+                // Skip chmod for 0o644 (most files) — File::create() already
+                // produces this via umask (0o666 & ~0o022 = 0o644).
+                #[cfg(unix)]
+                if entry.mode != 0o644 {
+                    fs::set_permissions(&entry.path, fs::Permissions::from_mode(entry.mode)).ok();
+                }
+            }
+            Ok(())
+        })?;
 
     // Set directory permissions
     #[cfg(unix)]

crates/pm/src/util/logger.rs

@@ -36,8 +36,23 @@ pub fn init_tracing(verbose: bool) -> Result<(PathBuf, WorkerGuard)> {
     let console_filter = EnvFilter::try_from_default_env()
         .unwrap_or_else(|_| EnvFilter::new(format!("utoo={console_level}")));
 
-    // File filter: always capture debug+ for troubleshooting
-    let file_filter = EnvFilter::new("utoo=debug");
+    // File filter: capture info+ by default. Hot-path debug!() calls
+    // (cache hits, per-package preload events, BFS dispatch) emit
+    // ~5-10 events per resolved manifest. With 2730+ manifests during
+    // a cold preload that's 15-30k events that — even routed through
+    // the non_blocking appender's channel — pay format/serialise CPU
+    // on the resolving thread. CI standalone manifest-bench reaches
+    // avg_conc=92 at cap=128 with the same reqwest stack; ruborist
+    // sat at avg_conc=56 after every other Mutex/clone hot-path was
+    // eliminated. Tracing was the last remaining shared cost.
+    //
+    // Override via `UTOO_FILE_LOG=debug` (or any RUST_LOG-style spec)
+    // to bring the troubleshooting captures back when actually
+    // diagnosing.
+    let file_filter = match env::var("UTOO_FILE_LOG") {
+        Ok(spec) if !spec.is_empty() => EnvFilter::new(spec),
+        _ => EnvFilter::new("utoo=info"),
+    };
 
     // 2. Create log file
     let timestamp = SystemTime::now()
@@ -188,15 +203,23 @@ impl utoo_ruborist::progress::EventReceiver for ProgressReceiver {
     fn on_event(&self, event: utoo_ruborist::progress::BuildEvent) {
         use utoo_ruborist::progress::BuildEvent;
         match event {
-            BuildEvent::PreloadStart { count } | BuildEvent::PreloadQueued { count } => {
-                PROGRESS_BAR.inc_length(count as u64);
-            }
-            BuildEvent::PreloadFetching { name } => {
-                log_progress(&format!("fetching {}", name));
-            }
-            BuildEvent::PreloadProgress { name, .. } => {
-                PROGRESS_BAR.inc(1);
-                log_progress(&format!("resolved {}", name));
+            BuildEvent::PreloadStart { .. }
+            | BuildEvent::PreloadQueued { .. }
+            | BuildEvent::PreloadFetching { .. }
+            | BuildEvent::PreloadProgress { .. } => {
+                // Hot path: 4571 dispatches + 4571 completions = ~9000
+                // events landing on the main task during a 3-4 s phase.
+                // Every `PROGRESS_BAR.inc[_length]` takes indicatif's
+                // internal ProgressBar `Mutex`, contending with the
+                // steady_tick draw thread. Standalone manifest-bench
+                // sustains avg_conc=95 at cap=128; ruborist with these
+                // events stuck at 56 — the indicatif lock acquisitions
+                // cap the main loop's fill-and-drain rate.
+                //
+                // Drop all per-event progress bar updates during preload.
+                // The spinner still ticks via steady_tick so the user
+                // sees the phase is alive; the final summary prints
+                // success/fail counts at PreloadComplete.
             }
             BuildEvent::PreloadComplete { success, failed } => {
                 PROGRESS_BAR.set_position(0);

crates/pm/src/util/user_config.rs

@@ -132,9 +132,22 @@ pub fn get_install_scope() -> InstallScope {
     INSTALL_SCOPE.get().copied().unwrap_or_default()
 }
 
-// Manifest fetch concurrency configuration
+// Manifest fetch concurrency configuration.
+//
+// Sweep history under the worker-pool architecture:
+//   cap=96  spawn_blocking parse: wall=2.23s avg_conc=66
+//   cap=128 spawn_blocking parse: wall=2.15s avg_conc=84
+//   cap=160 spawn_blocking parse: wall=1.90s avg_conc=81 (parse queue p95=200ms)
+//   cap=160 inline parse        : wall=2.14s avg_conc=119 (parse queue=0ms)
+//
+// Inline parse eliminated blocking-pool queue but cap=160 lifted
+// concurrency past the per-source Cloudflare throttle, inflating
+// per-req from 55 ms → 93 ms. Net wall flat. cap=128 + inline parse
+// should land near the sweet spot: enough concurrency to saturate
+// independent connections, low enough to stay under Cloudflare's
+// per-source throttle threshold.
 static MANIFESTS_CONCURRENCY_LIMIT: LazyLock<ConfigValue<usize>> =
-    LazyLock::new(|| ConfigValue::new("manifests-concurrency-limit", 64));
+    LazyLock::new(|| ConfigValue::new("manifests-concurrency-limit", 128));
 
 pub fn set_manifests_concurrency_limit(value: Option<usize>) {
     MANIFESTS_CONCURRENCY_LIMIT.set(value);