From 99905425cec265cea47a33d1e4a5332b414e8915 Mon Sep 17 00:00:00 2001 From: elrrrrrrr Date: Sat, 25 Apr 2026 23:34:07 +0800 Subject: [PATCH] perf(pm): preload worker-pool replaces FuturesUnordered MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The headline architectural change of #2818 — preload phase shifts from a single-task `FuturesUnordered` cooperative poller to N long-lived `tokio::spawn` workers (or `wasm_bindgen_futures::spawn_local` on wasm32 where Send isn't satisfied). Stacks on top of #2826. ## Why Old design: main task owned `FuturesUnordered`, polled all preload futures cooperatively, and ran every per-future continuation (post-await body, completion handler, dispatch refill) on the same single task. The deeper await chain inside `resolve_package` (cache check + `OnceMap::get_or_init` + `RetryIf` + `request.send` + `bytes` + parse spawn_blocking) made each future yield 5+ times, and every yield round-tripped through main — saturating it. CI ant-design preload sustained avg_conc=55-61 even after Mutex / allocator hot-path eliminations, while the standalone manifest-bench (#2824) hit 92 on the same reqwest stack. ## How N long-lived `tokio::spawn` workers pulling from a shared lock-free `SegQueue` with `DashSet` dedup. Each worker owns an `Arc` clone and runs `resolve_package` on tokio's global executor — futures progress fully independently, no cooperative poll bottleneck. Main task only drains an `mpsc::unbounded_channel` of completions to fire receiver events + on_manifest callback. Termination: workers track `dispatched` / `completed: AtomicUsize` and park on a shared `Notify` when the queue is empty. When the last completion makes `completed == dispatched` and the queue is empty, the finishing worker raises a `shutdown` flag and wakes others; all workers drop their result_tx clones, the channel closes, and the main `recv().await` loop exits. ## Trait surface change - `MockRegistryClient` + `MockPackage` `derive(Clone)` so tests can wrap the mock in `Arc` for the new signature - `preload_manifests` takes `registry: Arc` (was `&R`); call site in `run_preload_phase` clones the borrowed registry into a fresh `Arc`. Bound at every public surface up the chain bumped to `R: RegistryClient + Clone + MaybeSend + MaybeSync + 'static`, `R::Error: MaybeSend`. The `MaybeSend` / `MaybeSync` shims (added in #2826) keep the trait surface wasm-compatible. ## Companion changes folded in - **Inline simd_json parse** — drop `tokio::task::spawn_blocking` in `service/manifest.rs`. Worker-pool surfaced parse blocking- pool queue saturation: `queue p95=200ms sum=70-89s` over 2730 manifests on cap=4 CI runners. Inline parse on the worker thread eliminates dispatch + queue overhead. - **Workspace package.json parallel reads** — switch the per-pattern `for path in matched_paths` serial loop to `FuturesUnordered` fan-out. ant-design has ~200 workspace packages; saved ~150ms. - **Setup phase + lockfile-write timing logs** — round out the per-phase wall account for the bench-comment infrastructure. - **Manifests concurrency cap 64 → 128** — worker-pool delivers the parallelism that justifies the cap raise. CI ant-design avg_conc 84 at cap=128 (up from 55 under the old architecture); preload wall 3.10s → 2.15s. ## Wasm CI cfg-gates `tokio::spawn` to `wasm_bindgen_futures::spawn_local` on wasm32 since wasm-bindgen's `JsFuture` is `!Send`. Workers still run independently — single-threaded under wasm but the queue + Notify + mpsc termination story is unchanged. `cargo check -p utoo-wasm --target wasm32-unknown-unknown` clean. Tests: 164 ruborist + 10 doctests pass. Co-Authored-By: Claude Opus 4.7 (1M context) --- crates/pm/src/util/user_config.rs | 17 +- crates/ruborist/src/resolver/builder.rs | 91 +++-- crates/ruborist/src/resolver/preload.rs | 426 +++++++++++++++++----- crates/ruborist/src/resolver/workspace.rs | 71 ++-- crates/ruborist/src/service/api.rs | 24 ++ 5 files changed, 492 insertions(+), 137 deletions(-) diff --git a/crates/pm/src/util/user_config.rs b/crates/pm/src/util/user_config.rs index 34ee45a340..b9f7d4b8d9 100644 --- a/crates/pm/src/util/user_config.rs +++ b/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> = - LazyLock::new(|| ConfigValue::new("manifests-concurrency-limit", 64)); + LazyLock::new(|| ConfigValue::new("manifests-concurrency-limit", 128)); pub fn set_manifests_concurrency_limit(value: Option) { MANIFESTS_CONCURRENCY_LIMIT.set(value); diff --git a/crates/ruborist/src/resolver/builder.rs b/crates/ruborist/src/resolver/builder.rs index b074dfaf4e..74b3e0df48 100644 --- a/crates/ruborist/src/resolver/builder.rs +++ b/crates/ruborist/src/resolver/builder.rs @@ -707,11 +707,20 @@ pub async fn process_dependency( +pub async fn build_deps< + R: RegistryClient + + Clone + + crate::util::maybe_send::MaybeSend + + crate::util::maybe_send::MaybeSync + + 'static, +>( graph: &mut DependencyGraph, registry: &R, peer_deps: PeerDeps, -) -> Result<(), ResolveError> { +) -> Result<(), ResolveError> +where + R::Error: crate::util::maybe_send::MaybeSend, +{ let config = BuildDepsConfig::default().with_peer_deps(peer_deps); build_deps_with_config(graph, registry, config, &NoopReceiver).await } @@ -730,14 +739,21 @@ pub async fn build_deps( /// * `peer_deps` - How to handle peer dependencies /// * `receiver` - Event receiver for handling build events pub async fn build_deps_with_receiver< - R: RegistryClient + crate::util::maybe_send::MaybeSync, + R: RegistryClient + + Clone + + crate::util::maybe_send::MaybeSend + + crate::util::maybe_send::MaybeSync + + 'static, E: EventReceiver, >( graph: &mut DependencyGraph, registry: &R, peer_deps: PeerDeps, receiver: &E, -) -> Result<(), ResolveError> { +) -> Result<(), ResolveError> +where + R::Error: crate::util::maybe_send::MaybeSend, +{ let config = BuildDepsConfig::default().with_peer_deps(peer_deps); build_deps_with_config(graph, registry, config, receiver).await } @@ -763,14 +779,21 @@ pub async fn build_deps_with_receiver< /// build_deps_with_config(&mut graph, ®istry, config, &receiver).await?; /// ``` pub async fn build_deps_with_config< - R: RegistryClient + crate::util::maybe_send::MaybeSync, + R: RegistryClient + + Clone + + crate::util::maybe_send::MaybeSend + + crate::util::maybe_send::MaybeSync + + 'static, E: EventReceiver, >( graph: &mut DependencyGraph, registry: &R, config: BuildDepsConfig, receiver: &E, -) -> Result<(), ResolveError> { +) -> Result<(), ResolveError> +where + R::Error: crate::util::maybe_send::MaybeSend, +{ tracing::debug!( "Starting dependency tree build, peer_deps: {:?}, concurrency: {}, skip_preload: {}", config.peer_deps, @@ -793,14 +816,20 @@ pub async fn build_deps_with_config< /// Run the preload phase to warm up the cache with manifests. async fn run_preload_phase< - R: RegistryClient + crate::util::maybe_send::MaybeSync, + R: RegistryClient + + Clone + + crate::util::maybe_send::MaybeSend + + crate::util::maybe_send::MaybeSync + + 'static, E: EventReceiver, >( graph: &DependencyGraph, registry: &R, config: &BuildDepsConfig, receiver: &E, -) { +) where + R::Error: crate::util::maybe_send::MaybeSend, +{ if config.skip_preload { return; } @@ -813,18 +842,21 @@ async fn run_preload_phase< } tracing::debug!("Preload phase: {} initial dependencies", initial_deps.len()); - receiver.on_event(BuildEvent::PreloadStart { - count: initial_deps.len(), - }); + // Note: PreloadStart is emitted inside `preload_manifests` itself now. let preload_config = PreloadConfig { peer_deps: config.peer_deps, concurrency: config.concurrency, }; + // Wrap registry in Arc for the worker pool. Each worker holds an + // Arc so resolution runs on tokio's global executor independently + // of the main task driving result aggregation. + let registry_arc = std::sync::Arc::new(registry.clone()); + let stats = preload_manifests( initial_deps, - registry, + registry_arc, preload_config, receiver, |_name, _manifest| { @@ -833,17 +865,18 @@ async fn run_preload_phase< ) .await; - tracing::debug!( - "Preload phase completed: {} success, {} failed", + let elapsed = start.elapsed(); + tracing::info!( + "Preload phase: {:.2?} ({} success, {} failed, {} initial deps)", + elapsed, stats.success_count, - stats.failed_count + stats.failed_count, + stats.total_processed, ); receiver.on_event(BuildEvent::PreloadComplete { success: stats.success_count, failed: stats.failed_count, }); - - tracing::debug!("Preload phase: {:?}", start.elapsed()); } /// Run the BFS traversal phase to build the dependency tree. @@ -970,10 +1003,19 @@ use std::path::Path; /// let pkg: PackageJson = serde_json::from_str(&pkg_content)?; /// let lock = resolve(&pkg, ®istry).await?; /// ``` -pub async fn resolve( +pub async fn resolve< + R: RegistryClient + + Clone + + crate::util::maybe_send::MaybeSend + + crate::util::maybe_send::MaybeSync + + 'static, +>( pkg: &PackageJson, registry: &R, -) -> Result> { +) -> Result> +where + R::Error: crate::util::maybe_send::MaybeSend, +{ resolve_with_options(pkg, registry, PeerDeps::Include, &NoopReceiver).await } @@ -985,14 +1027,21 @@ pub async fn resolve( /// * `peer_deps` - How to handle peer dependencies /// * `receiver` - Event receiver for progress tracking pub async fn resolve_with_options< - R: RegistryClient + crate::util::maybe_send::MaybeSync, + R: RegistryClient + + Clone + + crate::util::maybe_send::MaybeSend + + crate::util::maybe_send::MaybeSync + + 'static, E: EventReceiver, >( pkg: &PackageJson, registry: &R, peer_deps: PeerDeps, receiver: &E, -) -> Result> { +) -> Result> +where + R::Error: crate::util::maybe_send::MaybeSend, +{ // Create graph with root node let mut graph = DependencyGraph::from_package_json(PathBuf::from("."), pkg.clone()); diff --git a/crates/ruborist/src/resolver/preload.rs b/crates/ruborist/src/resolver/preload.rs index 7f7e2521da..4400496f0a 100644 --- a/crates/ruborist/src/resolver/preload.rs +++ b/crates/ruborist/src/resolver/preload.rs @@ -1,21 +1,40 @@ -//! Parallel manifest preloading for dependency resolution. +//! Parallel manifest preloading via worker pool. //! -//! Uses FuturesUnordered for true streaming concurrency: when a package resolves, -//! its transitive dependencies are immediately added to the queue. +//! Architecture: N long-lived `tokio::spawn` workers pulling work from a +//! shared `SegQueue`. Replaces the prior `FuturesUnordered` design that +//! had main task own the futures and poll them cooperatively, which +//! capped effective parallelism at ~55-60 even when standalone +//! manifest-bench (same reqwest stack, no resolver overhead) sustained +//! 90+ concurrent at the same cap. The deeper `await` chain inside +//! `resolve_package` (registry cache check + `OnceMap::get_or_init` + +//! `RetryIf` + `request.send()` + `bytes()` + parse-spawn_blocking) +//! made every yielded poll round-trip through the main task — +//! starving the dispatch refill. Worker tasks run on tokio's global +//! pool so each future progresses independently. -use std::collections::{HashSet, VecDeque}; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::time::Instant; -use futures::stream::{FuturesUnordered, StreamExt}; +use crossbeam_queue::SegQueue; +use dashmap::DashSet; +use tokio::sync::{Notify, mpsc}; use crate::model::manifest::CoreVersionManifest; use crate::model::node::PeerDeps; +use crate::resolver::registry::ResolveError; use crate::resolver::registry::resolve_package; +use crate::service::http::{ + finish_http_trace, finish_parse_trace, start_http_trace, start_parse_trace, +}; use crate::traits::progress::{BuildEvent, EventReceiver}; use crate::traits::registry::RegistryClient; -/// Default concurrency limit for manifest fetching -pub const DEFAULT_CONCURRENCY: usize = 64; +/// Default concurrency limit for manifest fetching. +/// +/// Preload now runs N long-lived worker tasks; this is N. Each worker +/// processes one resolve_package at a time on tokio's global pool. +pub const DEFAULT_CONCURRENCY: usize = 256; /// A dependency spec: (name, version_spec) pub type Dep = (String, String); @@ -71,79 +90,170 @@ fn extract_transitive_deps(manifest: &CoreVersionManifest, config: &PreloadConfi deps } -/// Preload all package manifests in parallel with streaming concurrency. +/// Result message sent from worker to main task. +type Completion = ( + String, + Result>, + u64, +); + +/// Preload all package manifests in parallel via a tokio worker pool. +/// +/// `registry` is moved in and shared across workers via `Arc`. Each +/// worker is a long-lived spawned task that pulls work items from a +/// shared lock-free `SegQueue` until both the queue and the in-flight +/// counter reach zero, signalling end-of-phase. +/// +/// `receiver` and `on_manifest` run on the main task only — they do +/// not need to be `Send`/`Sync`. pub async fn preload_manifests( initial_deps: Vec, - registry: &R, + registry: Arc, config: PreloadConfig, receiver: &E, mut on_manifest: F, ) -> PreloadStats where - R: RegistryClient + crate::util::maybe_send::MaybeSync, + R: RegistryClient + + crate::util::maybe_send::MaybeSend + + crate::util::maybe_send::MaybeSync + + 'static, + R::Error: crate::util::maybe_send::MaybeSend, E: EventReceiver, F: FnMut(&str, Arc), { let mut stats = PreloadStats::default(); - let mut processed: HashSet = HashSet::new(); - let mut pending: VecDeque = initial_deps.into(); - let concurrency = config.concurrency; + let preload_wall_start = Instant::now(); + start_http_trace(); + start_parse_trace(); + + // Shared work queue and dedup set. + let pending: Arc> = Arc::new(SegQueue::new()); + let processed: Arc> = Arc::new(DashSet::new()); + + // Counters for global termination — when `dispatched == completed` + // and the queue is empty, the phase is done. + let dispatched: Arc = Arc::new(AtomicUsize::new(0)); + let completed: Arc = Arc::new(AtomicUsize::new(0)); + let shutdown: Arc = Arc::new(AtomicBool::new(false)); + + // Wake-up signal for workers parked on an empty queue. + let notify: Arc = Arc::new(Notify::new()); + + // Result channel — workers send completions, main task drains. + let (result_tx, mut result_rx) = mpsc::unbounded_channel::>(); + + // Seed the queue with initial deps (deduped via DashSet). + for dep in initial_deps { + let key = format!("{}@{}", dep.0, dep.1); + if processed.insert(key) { + pending.push(dep); + dispatched.fetch_add(1, Ordering::Relaxed); + } + } + let initial_count = dispatched.load(Ordering::Relaxed); + let concurrency = config.concurrency.max(1); tracing::debug!( - "Preload: {} initial deps, concurrency={}", - pending.len(), + "Preload: {} initial deps, concurrency={}, mode=worker-pool", + initial_count, concurrency ); - let mut futures = FuturesUnordered::new(); - let mut in_flight = 0usize; - let mut started = false; - - loop { - // Fill up to concurrency limit - while in_flight < concurrency { - let item = loop { - let Some((name, spec)) = pending.pop_front() else { - break None; - }; - let key = format!("{}@{}", name, spec); - if !processed.contains(&key) { - processed.insert(key); - break Some((name, spec)); + // Spawn N long-lived workers. Each loops: pop -> resolve -> push transitives. + for _ in 0..concurrency { + let pending = Arc::clone(&pending); + let processed = Arc::clone(&processed); + let dispatched = Arc::clone(&dispatched); + let completed = Arc::clone(&completed); + let shutdown = Arc::clone(&shutdown); + let notify = Arc::clone(¬ify); + let result_tx = result_tx.clone(); + let registry = Arc::clone(®istry); + let config_for_worker = config.clone(); + + let worker = async move { + loop { + // Try fetching work first — fast path when queue is hot. + if let Some((name, spec)) = pending.pop() { + let start = Instant::now(); + let result = resolve_package(&*registry, &name, &spec).await; + let elapsed_ms = start.elapsed().as_millis() as u64; + + if let Ok(resolved) = &result { + let mut new_added = 0usize; + for tdep in extract_transitive_deps(&resolved.manifest, &config_for_worker) + { + let key = format!("{}@{}", tdep.0, tdep.1); + if processed.insert(key) { + pending.push(tdep); + new_added += 1; + } + } + if new_added > 0 { + dispatched.fetch_add(new_added, Ordering::Release); + // Wake parked workers so they pick up the new work + // before checking the termination condition. + notify.notify_waiters(); + } + } + + if result_tx.send((name, result, elapsed_ms)).is_err() { + // Main task dropped the receiver — done collecting. + break; + } + let done = completed.fetch_add(1, Ordering::AcqRel) + 1; + + // After completion, check global done condition. The + // `Acquire` on dispatched pairs with the `Release` in + // the producer add above, so we won't miss late + // transitives queued by a sibling worker. + if done == dispatched.load(Ordering::Acquire) && pending.is_empty() { + shutdown.store(true, Ordering::Release); + notify.notify_waiters(); + break; + } + continue; } - }; - - let Some((name, spec)) = item else { - break; - }; - if !started { - receiver.on_event(BuildEvent::PreloadStart { count: 1 }); - started = true; - } else { - receiver.on_event(BuildEvent::PreloadQueued { count: 1 }); + // Queue empty — register interest before re-checking, then + // park. The Notify+enable() pattern guarantees we won't + // miss a `notify_waiters()` racing with our park. + if shutdown.load(Ordering::Acquire) { + break; + } + let notified = notify.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + if !pending.is_empty() { + continue; + } + if shutdown.load(Ordering::Acquire) { + break; + } + if completed.load(Ordering::Acquire) == dispatched.load(Ordering::Acquire) { + shutdown.store(true, Ordering::Release); + notify.notify_waiters(); + break; + } + notified.await; } - - receiver.on_event(BuildEvent::PreloadFetching { name: &name }); - - futures.push(async move { - let start = tokio::time::Instant::now(); - let result = resolve_package(registry, &name, &spec).await; - let elapsed_ms = start.elapsed().as_millis() as u64; - (name, result, elapsed_ms) - }); - in_flight += 1; - } - - if in_flight == 0 { - break; - } - - let Some((name, result, elapsed_ms)) = futures.next().await else { - break; }; - in_flight -= 1; + #[cfg(not(target_arch = "wasm32"))] + tokio::spawn(worker); + #[cfg(target_arch = "wasm32")] + wasm_bindgen_futures::spawn_local(worker); + } + // Drop the original sender so when all worker clones drop on exit, the + // result channel closes and the main loop terminates. + drop(result_tx); + receiver.on_event(BuildEvent::PreloadStart { + count: initial_count, + }); + + // Main task: drain completions, run user callbacks. + while let Some((name, result, elapsed_ms)) = result_rx.recv().await { if stats.success_count == 0 && stats.failed_count == 0 { stats.min_request_ms = elapsed_ms; stats.max_request_ms = elapsed_ms; @@ -163,11 +273,8 @@ where version: &resolved.version, current: stats.success_count, }); - - // Send PackageResolved event for pipeline downloading receiver.on_event(BuildEvent::PackageResolved((&*resolved.manifest).into())); - pending.extend(extract_transitive_deps(&resolved.manifest, &config)); on_manifest(&name, resolved.manifest); } Err(e) => { @@ -184,6 +291,12 @@ where failed: stats.failed_count, }); + let preload_wall_ms = preload_wall_start.elapsed().as_millis(); + let intervals = finish_http_trace(); + log_http_diagnostics(&intervals, preload_wall_ms); + let parses = finish_parse_trace(); + log_parse_diagnostics(&parses); + let total = stats.success_count + stats.failed_count; let avg = if total > 0 { stats.total_request_ms / total as u64 @@ -202,15 +315,156 @@ where stats } +/// Summarise captured HTTP intervals from the preload phase and log one +/// info-level line. Splits total preload wall into: +/// +/// - `wall` — first HTTP send → last HTTP body end (pure network window) +/// - `busy` — interval union (time at least one request was in-flight) +/// - `sum` — Σ per-request `(end − start)` (total req-level time) +/// - `cpu_tail` = preload_total − wall (our CPU work after last body) +/// - `avg_conc` = sum / busy (effective parallelism over busy window) +/// - percentiles of per-request latency +fn log_http_diagnostics(intervals: &[(Instant, Instant)], preload_wall_ms: u128) { + if intervals.is_empty() { + tracing::info!( + "Preload HTTP diag: no requests captured (total wall {}ms)", + preload_wall_ms + ); + return; + } + + let mut spans: Vec<(Instant, Instant)> = intervals.to_vec(); + spans.sort_by_key(|(s, _)| *s); + + let first_start = spans.first().unwrap().0; + let last_end = spans.iter().map(|(_, e)| *e).max().unwrap(); + let wall = last_end.duration_since(first_start).as_millis(); + + let sum: u128 = spans + .iter() + .map(|(s, e)| e.duration_since(*s).as_micros()) + .sum(); + + // Interval union: sweep sorted spans, merging overlaps. + let mut busy_us: u128 = 0; + let (mut cur_s, mut cur_e) = spans[0]; + for &(s, e) in &spans[1..] { + if s <= cur_e { + if e > cur_e { + cur_e = e; + } + } else { + busy_us += cur_e.duration_since(cur_s).as_micros(); + cur_s = s; + cur_e = e; + } + } + busy_us += cur_e.duration_since(cur_s).as_micros(); + + let mut per_req_us: Vec = spans + .iter() + .map(|(s, e)| e.duration_since(*s).as_micros()) + .collect(); + per_req_us.sort_unstable(); + let n = per_req_us.len(); + let p50 = per_req_us[n / 2]; + let p95 = per_req_us[(n * 95).div_ceil(100).saturating_sub(1)]; + let max = *per_req_us.last().unwrap(); + + let cpu_tail_ms = preload_wall_ms.saturating_sub(wall); + let avg_conc = if busy_us > 0 { + sum as f64 / busy_us as f64 + } else { + 0.0 + }; + + tracing::info!( + "Preload HTTP diag: n={} wall={}ms busy={}ms ({:.0}% of wall) sum={}ms avg_conc={:.1} p50={}ms p95={}ms max={}ms cpu_tail={}ms", + n, + wall, + busy_us / 1000, + if wall > 0 { + 100.0 * (busy_us as f64 / 1000.0) / wall as f64 + } else { + 0.0 + }, + sum / 1000, + avg_conc, + p50 / 1000, + p95 / 1000, + max / 1000, + cpu_tail_ms, + ); +} + +/// Summarise parse timing. +fn log_parse_diagnostics(parses: &[(Instant, Instant, Instant)]) { + if parses.is_empty() { + return; + } + + let mut queue_us: Vec = parses + .iter() + .map(|(q, s, _)| s.duration_since(*q).as_micros()) + .collect(); + let mut exec_us: Vec = parses + .iter() + .map(|(_, s, e)| e.duration_since(*s).as_micros()) + .collect(); + queue_us.sort_unstable(); + exec_us.sort_unstable(); + + let n = parses.len(); + let pct = |v: &[u128], p: usize| v[(n * p).div_ceil(100).saturating_sub(1)]; + let sum_queue: u128 = queue_us.iter().sum(); + let sum_exec: u128 = exec_us.iter().sum(); + + let mut spans: Vec<(Instant, Instant)> = parses.iter().map(|(_, s, e)| (*s, *e)).collect(); + spans.sort_by_key(|(s, _)| *s); + let (mut cur_s, mut cur_e) = spans[0]; + let mut exec_busy_us: u128 = 0; + for &(s, e) in &spans[1..] { + if s <= cur_e { + if e > cur_e { + cur_e = e; + } + } else { + exec_busy_us += cur_e.duration_since(cur_s).as_micros(); + cur_s = s; + cur_e = e; + } + } + exec_busy_us += cur_e.duration_since(cur_s).as_micros(); + let avg_exec_parallelism = if exec_busy_us > 0 { + sum_exec as f64 / exec_busy_us as f64 + } else { + 0.0 + }; + + tracing::info!( + "Preload parse diag: n={} queue(p50={}ms p95={}ms max={}ms sum={}ms) exec(p50={}ms p95={}ms max={}ms sum={}ms) exec_busy={}ms avg_parallel={:.1}", + n, + queue_us[n / 2] / 1000, + pct(&queue_us, 95) / 1000, + queue_us.last().unwrap() / 1000, + sum_queue / 1000, + exec_us[n / 2] / 1000, + pct(&exec_us, 95) / 1000, + exec_us.last().unwrap() / 1000, + sum_exec / 1000, + exec_busy_us / 1000, + avg_exec_parallelism, + ); +} + #[cfg(test)] mod tests { use super::*; use crate::model::manifest::CoreVersionManifest; use crate::traits::progress::NoopReceiver; use crate::traits::registry::mock::MockRegistryClient; - use std::cell::RefCell; use std::collections::HashMap; - use std::rc::Rc; + use std::sync::Mutex; fn manifest(name: &str, version: &str) -> CoreVersionManifest { CoreVersionManifest { @@ -237,30 +491,30 @@ mod tests { } } - #[tokio::test] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_preload_single() { let mut registry = MockRegistryClient::new(); registry.add_package("lodash", "4.17.21", manifest("lodash", "4.17.21")); - let cache: Rc>>> = Default::default(); - let cache_clone = Rc::clone(&cache); + let cache: Arc>>> = Default::default(); + let cache_clone = Arc::clone(&cache); let stats = preload_manifests( vec![("lodash".into(), "^4.17.0".into())], - ®istry, + Arc::new(registry), PreloadConfig::default(), &NoopReceiver, - |name, m| { - cache_clone.borrow_mut().insert(name.into(), m); + move |name, m| { + cache_clone.lock().unwrap().insert(name.into(), m); }, ) .await; assert_eq!(stats.success_count, 1); - assert!(cache.borrow().contains_key("lodash")); + assert!(cache.lock().unwrap().contains_key("lodash")); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_preload_transitive() { let mut registry = MockRegistryClient::new(); registry.add_package( @@ -270,44 +524,44 @@ mod tests { ); registry.add_package("b", "1.0.0", manifest("b", "1.0.0")); - let cache: Rc>>> = Default::default(); - let cache_clone = Rc::clone(&cache); + let cache: Arc>>> = Default::default(); + let cache_clone = Arc::clone(&cache); let stats = preload_manifests( vec![("a".into(), "^1.0.0".into())], - ®istry, + Arc::new(registry), PreloadConfig::default(), &NoopReceiver, - |name, m| { - cache_clone.borrow_mut().insert(name.into(), m); + move |name, m| { + cache_clone.lock().unwrap().insert(name.into(), m); }, ) .await; assert_eq!(stats.success_count, 2); - assert!(cache.borrow().contains_key("a")); - assert!(cache.borrow().contains_key("b")); + assert!(cache.lock().unwrap().contains_key("a")); + assert!(cache.lock().unwrap().contains_key("b")); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_preload_missing() { let registry = MockRegistryClient::new(); - let cache: Rc>>> = Default::default(); - let cache_clone = Rc::clone(&cache); + let cache: Arc>>> = Default::default(); + let cache_clone = Arc::clone(&cache); let stats = preload_manifests( vec![("nonexistent".into(), "^1.0.0".into())], - ®istry, + Arc::new(registry), PreloadConfig::default(), &NoopReceiver, - |name, m| { - cache_clone.borrow_mut().insert(name.into(), m); + move |name, m| { + cache_clone.lock().unwrap().insert(name.into(), m); }, ) .await; assert_eq!(stats.failed_count, 1); - assert!(cache.borrow().is_empty()); + assert!(cache.lock().unwrap().is_empty()); } #[test] diff --git a/crates/ruborist/src/resolver/workspace.rs b/crates/ruborist/src/resolver/workspace.rs index 6b550255e9..3e9b2cc715 100644 --- a/crates/ruborist/src/resolver/workspace.rs +++ b/crates/ruborist/src/resolver/workspace.rs @@ -68,42 +68,57 @@ impl WorkspaceDiscovery { None => return Ok(workspaces), }; + // Collect all workspace directories from every pattern first, then + // read every package.json concurrently. Ant-design has ~200 + // workspace packages; the previous serial `for path in + // matched_paths { read_package_json(...).await }` paid 200× + // single-file FS round-trips on the resolver hot path + // (~150-200 ms of sequential I/O during p1_resolve, the largest + // unmeasured chunk between preload completion and lockfile write). + let mut workspace_paths: Vec = Vec::new(); for pattern_str in patterns { - // Prepare glob pattern for package.json files let package_json_pattern = root_path.join(pattern_str).join("package.json"); - - // Match workspace directories using Glob::glob let matched_paths = self .glob .glob(&package_json_pattern) .await .map_err(|e| anyhow::anyhow!("Failed to glob pattern: {}", e))?; - for path in matched_paths { - let workspace_path = match path.parent() { - Some(p) => p.to_path_buf(), - None => continue, - }; - - // Read workspace package.json - let workspace_pkg: PackageJson = match read_package_json(&workspace_path).await { - Ok(pkg) => pkg, - Err(e) => { - tracing::debug!("Failed to read workspace package.json: {}", e); - continue; - } - }; - - tracing::debug!( - "Found workspace: {} at {:?}", - workspace_pkg.name, - workspace_path - ); - workspaces.push(WorkspacePackage { - name: workspace_pkg.name.clone(), - path: workspace_path, - package_json: workspace_pkg, - }); + if let Some(p) = path.parent() { + workspace_paths.push(p.to_path_buf()); + } + } + } + + // Fan out: dispatch every package.json read in parallel via + // `FuturesUnordered`. Each read is small (typical workspace + // package.json < 4 KB) so completion order doesn't matter. + use futures::stream::{FuturesUnordered, StreamExt}; + let mut futs: FuturesUnordered<_> = workspace_paths + .into_iter() + .map(|workspace_path| async move { + let pkg = read_package_json(&workspace_path).await; + (workspace_path, pkg) + }) + .collect(); + + while let Some((workspace_path, pkg)) = futs.next().await { + match pkg { + Ok(workspace_pkg) => { + tracing::debug!( + "Found workspace: {} at {:?}", + workspace_pkg.name, + workspace_path + ); + workspaces.push(WorkspacePackage { + name: workspace_pkg.name.clone(), + path: workspace_path, + package_json: workspace_pkg, + }); + } + Err(e) => { + tracing::debug!("Failed to read workspace package.json: {}", e); + } } } diff --git a/crates/ruborist/src/service/api.rs b/crates/ruborist/src/service/api.rs index db95373655..67e0270c52 100644 --- a/crates/ruborist/src/service/api.rs +++ b/crates/ruborist/src/service/api.rs @@ -135,6 +135,8 @@ where catalogs, } = options; + let t_setup = std::time::Instant::now(); + // 1. Find root path (workspace root if applicable) let discovery = WorkspaceDiscovery::new(glob.clone()); let root_path = discovery.find_root_path(&cwd).await?; @@ -290,17 +292,34 @@ where ); } + tracing::info!( + "Setup phase (workspace + graph init): {:.2?}", + t_setup.elapsed() + ); + // Preserve the typed error via `Error::new` + `.context(...)` so CLI // renderers (e.g. pm's format_print) can downcast and pretty-print the // dependency chain carried by `ResolveError::WithChain`. + let t_build = std::time::Instant::now(); build_deps_with_config(&mut graph, ®istry, config, &receiver) .await .map_err(|e| anyhow::Error::new(e).context("Dependency resolution failed"))?; + tracing::info!( + "Build deps (preload + BFS resolve): {:.2?}", + t_build.elapsed() + ); // 11. Serialize to PackageLock + let t_serialize = std::time::Instant::now(); let (packages, _total) = graph.serialize_to_packages(&root_path); + tracing::info!( + "Serialize graph to lockfile: {:.2?} ({} packages)", + t_serialize.elapsed(), + packages.len(), + ); // 12. Save project cache (export from memory cache) + let t_cache_save = std::time::Instant::now(); let mut new_cache_data = super::cache::ProjectCacheData::default(); // Export version manifests from memory cache to project cache // Memory cache key format: "name@spec", manifest contains resolved version @@ -321,6 +340,11 @@ where { tracing::warn!("Failed to save project cache: {}", e); } + tracing::info!( + "Project cache export + save: {:.2?} ({} entries)", + t_cache_save.elapsed(), + new_cache_data.cache.len(), + ); Ok(PackageLock::new(&pkg.name, &pkg.version, packages)) }