perf(pm): TTY-gate + cooperative yield_now in install spawn loop

crates/pm/src/service/install.rs

@@ -21,7 +21,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};
 
@@ -82,14 +83,34 @@ pub async fn install_packages(
     let mut depths: Vec<_> = groups.keys().cloned().collect();
     depths.sort_unstable();
 
+    // Cooperative-yield cadence inside the synchronous spawn-loop body
+    // below. Before #2903 (TTY-gate), every iteration called
+    // `PROGRESS_BAR.inc(1)`, whose internal mutex+state update gave the
+    // tokio runtime roughly one micro-pause per package — enough for
+    // the runtime to interleave socket reads on in-flight tarball
+    // downloads. After TTY-gate the inc became a no-op on CI (non-TTY),
+    // and pcap+iostat A/B on the new disk-burst-bounded baseline showed
+    // utoo `tcp.analysis.zero_window` events climb from 2 → 16 vs
+    // baseline (no disk regression — util / w_await flat). Adding an
+    // explicit `yield_now` inside the spawn loop restores the implicit
+    // yield without the indicatif mutex cost. Every-N rather than
+    // per-iteration so the yield itself doesn't dominate cheap omit /
+    // skip iterations.
+    const YIELD_EVERY: usize = 32;
+
     for depth in depths.iter() {
         let mut clone_tasks: Vec<tokio::task::JoinHandle<Result<()>>> = Vec::new();
+        let mut iter_count: usize = 0;
 
         if let Some(packages) = groups.get(depth) {
             for (path, package) in packages.iter() {
+                iter_count += 1;
+                if iter_count.is_multiple_of(YIELD_EVERY) {
+                    tokio::task::yield_now().await;
+                }
                 // 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 +129,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;
                     }
 
@@ -153,18 +174,18 @@ 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);
                 }
             }
         }
@@ -260,7 +281,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();