diff --git a/.github/workflows/pm-bench-phases.yml b/.github/workflows/pm-bench-phases.yml new file mode 100644 index 0000000000..3d1b1c14a8 --- /dev/null +++ b/.github/workflows/pm-bench-phases.yml @@ -0,0 +1,182 @@ +name: utoopm-benchmark-phases + +on: + workflow_dispatch: + inputs: + project: + description: 'Project to benchmark (under github.com/ant-design/)' + required: false + default: 'ant-design' + type: string + registry: + description: 'Registry URL' + required: false + default: 'https://registry.npmjs.org' + type: string + runs: + description: 'Runs per phase' + required: false + default: '3' + type: string + pull_request: + types: [labeled] + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + build-linux: + if: > + github.event_name == 'workflow_dispatch' || + (github.event_name == 'pull_request' && github.event.label.name == 'bench-phases') + runs-on: ubuntu-latest + container: + image: ubuntu:20.04 + steps: + - name: Install dependencies + env: + DEBIAN_FRONTEND: noninteractive + run: apt-get update && apt-get install -y build-essential curl pkg-config libssl-dev git + - uses: actions/checkout@v4 + - name: Configure git safe directory + run: git config --global --add safe.directory /__w/utoo/utoo + - name: Init git submodules + run: git submodule update --init --recursive --depth 1 + - name: Install Rust + uses: dtolnay/rust-toolchain@stable + with: + toolchain: nightly-2026-04-02 + - name: Cache cargo + uses: Swatinem/rust-cache@v2 + with: + shared-key: pm-bench-linux + - name: Build current branch utoo + manifest-bench + run: | + cargo build --release --bin utoo -p utoo-pm + cargo build --release -p manifest-bench + - name: Upload utoo binary + uses: actions/upload-artifact@v4 + with: + name: utoo-linux-bench-phases + path: target/release/utoo + retention-days: 1 + - name: Upload manifest-bench binary + uses: actions/upload-artifact@v4 + with: + name: manifest-bench-linux + path: target/release/manifest-bench + retention-days: 1 + + bench: + needs: [build-linux] + if: > + github.event_name == 'workflow_dispatch' || + (github.event_name == 'pull_request' && github.event.label.name == 'bench-phases') + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Setup Node.js + uses: actions/setup-node@v4 + with: + node-version: 22 + - name: Download utoo binary + uses: actions/download-artifact@v4 + with: + name: utoo-linux-bench-phases + path: dist + - name: Download manifest-bench binary + uses: actions/download-artifact@v4 + with: + name: manifest-bench-linux + path: dist + - name: Install utoo + run: | + chmod +x dist/utoo dist/manifest-bench + echo "$GITHUB_WORKSPACE/dist" >> $GITHUB_PATH + - name: Install bun + uses: oven-sh/setup-bun@v2 + with: + bun-version: latest + - name: Install hyperfine, GNU time, tcpdump and dnsutils + run: sudo apt-get update && sudo apt-get install -y hyperfine time tcpdump dnsutils + - name: Verify tools + run: | + hyperfine --version + utoo --version + bun --version + - name: Run phase-isolated benchmark + env: + PROJECT: ${{ github.event.inputs.project || 'ant-design' }} + REGISTRY: ${{ github.event.inputs.registry || 'https://registry.npmjs.org' }} + BENCH_RUNS: ${{ github.event.inputs.runs || '3' }} + PM_LIST: 'utoo,bun' + run: | + chmod +x e2e/pm-bench-phases.sh + bash e2e/pm-bench-phases.sh + + - name: Standalone manifest-bench (HTTP-only fetch sweep) + env: + REGISTRY: ${{ github.event.inputs.registry || 'https://registry.npmjs.org' }} + PROJECT: ${{ github.event.inputs.project || 'ant-design' }} + run: | + set -eu + # Generate a project lockfile to source the manifest names from. + # Use bun (which we already have) to do a quick lockfile-only + # install so manifest-bench has a real workload to fire against. + PROJECT_DIR="/tmp/pm-bench/$PROJECT" + if [ ! -d "$PROJECT_DIR" ]; then + mkdir -p /tmp/pm-bench + git clone --depth 1 "https://github.com/ant-design/$PROJECT" "$PROJECT_DIR" + fi + cd "$PROJECT_DIR" + if [ ! -f package-lock.json ]; then + echo "==> generating lockfile via utoo" + utoo deps --registry "$REGISTRY" || true + fi + ls -la package-lock.json || { echo "no lockfile, skip"; exit 0; } + + echo + echo "============================================================" + echo "manifest-bench: HTTP-only fetch (no parse, no resolver)" + echo " Goal: isolate reqwest/rustls/tokio behaviour from" + echo " ruborist's resolver pipeline. Same numbers (wall, busy," + echo " avg_conc, p50/p95/max) as ruborist's Preload HTTP diag." + echo "============================================================" + for CAP in 32 64 96 128 192 256; do + echo + echo "--- concurrency=$CAP, h1, full manifest, default UA ---" + manifest-bench --lockfile package-lock.json --registry "$REGISTRY" \ + --concurrency "$CAP" --reps 2 --http1-only || true + done + + echo + echo "--- concurrency=128, h2 negotiate, full manifest, default UA ---" + manifest-bench --lockfile package-lock.json --registry "$REGISTRY" \ + --concurrency 128 --reps 2 || true + + echo + echo "--- concurrency=128, h1, single-version endpoint ---" + manifest-bench --lockfile package-lock.json --registry "$REGISTRY" \ + --concurrency 128 --reps 2 --http1-only --single-version || true + + echo + echo "--- concurrency=128, h1, UA=Bun/1.2.21 ---" + manifest-bench --lockfile package-lock.json --registry "$REGISTRY" \ + --concurrency 128 --reps 2 --http1-only --user-agent "Bun/1.2.21" || true + + - name: Capture pcap (resolve phase, single run per PM) + env: + PROJECT: ${{ github.event.inputs.project || 'ant-design' }} + REGISTRY: ${{ github.event.inputs.registry || 'https://registry.npmjs.org' }} + run: | + chmod +x e2e/pm-bench-pcap.sh + bash e2e/pm-bench-pcap.sh + + - name: Upload pcap artifact + if: always() + uses: actions/upload-artifact@v4 + with: + name: pm-bench-pcap + path: /tmp/pm-bench-pcap + retention-days: 3 diff --git a/Cargo.lock b/Cargo.lock index ac904b9fd3..02c544b223 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -61,6 +61,21 @@ dependencies = [ "memchr", ] +[[package]] +name = "alloc-no-stdlib" +version = "2.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc7bb162ec39d46ab1ca8c77bf72e890535becd1751bb45f64c597edb4c8c6b3" + +[[package]] +name = "alloc-stdlib" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94fb8275041c72129eb51b7d0322c29b8387a0386127718b096429201a5d6ece" +dependencies = [ + "alloc-no-stdlib", +] + [[package]] name = "allocator-api2" version = "0.2.21" @@ -573,6 +588,27 @@ dependencies = [ "syn 2.0.106", ] +[[package]] +name = "brotli" +version = "8.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4bd8b9603c7aa97359dbd97ecf258968c95f3adddd6db2f7e7a5bef101c84560" +dependencies = [ + "alloc-no-stdlib", + "alloc-stdlib", + "brotli-decompressor", +] + +[[package]] +name = "brotli-decompressor" +version = "5.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "874bb8112abecc98cbd6d81ea4fa7e94fb9449648c93cc89aa40c81c24d7de03" +dependencies = [ + "alloc-no-stdlib", + "alloc-stdlib", +] + [[package]] name = "browserslist-data" version = "0.1.4" @@ -1110,6 +1146,7 @@ version = "0.4.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ef8a506ec4b81c460798f572caead636d57d3d7e940f998160f52bd254bf2d23" dependencies = [ + "brotli", "compression-core", "flate2", "memchr", @@ -4900,6 +4937,21 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ca88d725a0a943b096803bd34e73a4437208b6077654cc4ecb2947a5f91618d" +[[package]] +name = "manifest-bench" +version = "0.0.0" +dependencies = [ + "anyhow", + "clap", + "futures", + "reqwest 0.12.24", + "rustls", + "rustls-native-certs", + "serde", + "serde_json", + "tokio", +] + [[package]] name = "markdown" version = "1.0.0" @@ -7243,6 +7295,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cd3c25631629d034ce7cd9940adc9d45762d46de2b0f57193c4443b92c6d4d40" dependencies = [ "aws-lc-rs", + "log", "once_cell", "ring", "rustls-pki-types", @@ -11346,6 +11399,8 @@ version = "0.1.0" dependencies = [ "anyhow", "bytes", + "crossbeam-queue", + "dashmap 6.1.0", "deno_semver", "dirs 5.0.1", "futures", @@ -11357,6 +11412,8 @@ dependencies = [ "pathdiff", "petgraph 0.6.5", "reqwest 0.12.24", + "rustls", + "rustls-native-certs", "serde", "serde_json", "sha2", diff --git a/bench/pm-bench.sh b/bench/pm-bench.sh index c4c43bdf14..2d3a58de44 100755 --- a/bench/pm-bench.sh +++ b/bench/pm-bench.sh @@ -293,13 +293,15 @@ run_cold_benchmarks() { > "$metrics_file" echo -e " ${CYAN}$pm${NC}..." - hyperfine \ + if ! hyperfine \ --runs "$BENCH_COLD_RUNS" \ --prepare "bash $PREPARE_SCRIPT $project_dir $pm --cold" \ --export-json "$json_file" \ + --show-output \ -n "$pm" \ - "bash $METRICS_WRAPPER $metrics_file $cmd_script" \ - 2>&1 | tail -1 || echo -e " ${RED}$pm cold install failed${NC}" + "bash $METRICS_WRAPPER $metrics_file $cmd_script"; then + echo -e " ${RED}$pm cold install failed${NC}" + fi done } @@ -372,7 +374,15 @@ print_results() { // --- Parse hyperfine timing --- const timeRows = []; for (const file of jsonFiles) { - const data = JSON.parse(fs.readFileSync(path.join(dir, file), 'utf8')); + let data; + try { + const raw = fs.readFileSync(path.join(dir, file), 'utf8'); + if (!raw.trim()) { console.log(' skip empty: ' + file); continue; } + data = JSON.parse(raw); + } catch (err) { + console.log(' skip unparsable: ' + file + ' (' + err.message + ')'); + continue; + } const base = file.replace('.json', ''); const parts = base.split('_'); const typeIdx = parts.findIndex(p => p === 'cold' || p === 'warm'); @@ -406,7 +416,11 @@ print_results() { .trim().split('\n').filter(Boolean); if (lines.length === 0) continue; - const entries = lines.map(l => JSON.parse(l)); + const entries = []; + for (const l of lines) { + try { entries.push(JSON.parse(l)); } catch (_) { /* skip malformed line */ } + } + if (entries.length === 0) continue; const avg = {}; for (const key of ['rss', 'page_faults', 'vol_ctx', 'invol_ctx', 'io_in', 'io_out']) { avg[key] = Math.round(entries.reduce((s, e) => s + (e[key] || 0), 0) / entries.length); @@ -513,7 +527,11 @@ main() { echo -e "${YELLOW}----------------------------------------${NC}" run_cold_benchmarks "$project" "$registry" "$reg_short" - run_warm_benchmarks "$project" "$registry" "$reg_short" + if [ "$BENCH_WARM_RUNS" -gt 0 ]; then + run_warm_benchmarks "$project" "$registry" "$reg_short" + else + echo -e " ${YELLOW}Warm installs: skipped (BENCH_WARM_RUNS=0)${NC}" + fi echo "" done diff --git a/crates/pm/src/cmd/view.rs b/crates/pm/src/cmd/view.rs index d8355e9922..282ee46cee 100644 --- a/crates/pm/src/cmd/view.rs +++ b/crates/pm/src/cmd/view.rs @@ -27,7 +27,7 @@ pub async fn view(package_spec: &str) -> Result<()> { tracing::debug!("Fetched package info: {full_manifest:?}"); // Resolve version and get full VersionManifest (with all display fields) - let version_list: Vec = full_manifest.versions.clone(); + let version_list: Vec = full_manifest.versions.keys.clone(); let resolved_version = utoo_ruborist::resolver::version::resolve_target_version( &full_manifest.dist_tags, &version_list, @@ -94,7 +94,7 @@ fn print_package_header(full_manifest: &FullManifest, version_manifest: &Version .as_ref() .map(|d| d.len()) .unwrap_or(0); - let versions_count = full_manifest.versions.len(); + let versions_count = full_manifest.versions.keys.len(); let deps_str = if deps_count == 0 { "none".to_string() diff --git a/crates/pm/src/helper/lock.rs b/crates/pm/src/helper/lock.rs index b5979f0970..c7360b5dc7 100644 --- a/crates/pm/src/helper/lock.rs +++ b/crates/pm/src/helper/lock.rs @@ -481,17 +481,35 @@ pub async fn is_pkg_lock_outdated(root_path: &Path) -> Result { /// 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(()) } diff --git a/crates/pm/src/main.rs b/crates/pm/src/main.rs index 8f3a3a3662..889030bc3a 100644 --- a/crates/pm/src/main.rs +++ b/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()); diff --git a/crates/pm/src/service/install.rs b/crates/pm/src/service/install.rs index 9f591ee85b..83131eda30 100644 --- a/crates/pm/src/service/install.rs +++ b/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(()) } diff --git a/crates/pm/src/util/cloner.rs b/crates/pm/src/util/cloner.rs index f7ff1c6a70..77aa63e2e9 100644 --- a/crates/pm/src/util/cloner.rs +++ b/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; @@ -8,9 +9,9 @@ use utoo_ruborist::manifest::IdentityView; use super::downloader::{is_git_url, resolve_cache_path}; use super::json::load_package_json; -use super::oncemap::OnceMap; use super::retry::create_retry_strategy; use crate::fs; +use utoo_ruborist::util::oncemap::OnceMap; /// Global clone cache shared between pipeline and install phases. /// @@ -28,11 +29,19 @@ static CLONE_CACHE: Lazy> = 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); diff --git a/crates/pm/src/util/downloader.rs b/crates/pm/src/util/downloader.rs index 77b2b836ff..d78d58e19c 100644 --- a/crates/pm/src/util/downloader.rs +++ b/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; @@ -13,9 +14,9 @@ use utoo_ruborist::spec::Protocol; use super::cache::get_cache_dir; use super::extractor::extract_and_write; -use super::oncemap::OnceMap; use super::retry::{RetryableError, build_dns_cached_client, create_retry_strategy}; use super::user_config::get_manifests_concurrency_limit_sync; +use utoo_ruborist::util::oncemap::OnceMap; // Global downloader client - no pool limit, concurrency controlled by OnceMap static DOWNLOADER_CLIENT: Lazy = Lazy::new(build_dns_cached_client); @@ -30,6 +31,28 @@ static DOWNLOAD_SEMAPHORE: OnceLock = 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); diff --git a/crates/pm/src/util/extractor.rs b/crates/pm/src/util/extractor.rs index 29b71149b0..ca25614814 100644 --- a/crates/pm/src/util/extractor.rs +++ b/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)] diff --git a/crates/pm/src/util/logger.rs b/crates/pm/src/util/logger.rs index 757a81070b..7a0f4a846b 100644 --- a/crates/pm/src/util/logger.rs +++ b/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); diff --git a/crates/pm/src/util/mod.rs b/crates/pm/src/util/mod.rs index 335e5b0382..77688b07a7 100644 --- a/crates/pm/src/util/mod.rs +++ b/crates/pm/src/util/mod.rs @@ -12,7 +12,6 @@ pub mod integrity; pub mod json; pub mod linker; pub mod logger; -pub mod oncemap; pub mod platform_const; pub mod registry; pub mod retry; 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/Cargo.toml b/crates/ruborist/Cargo.toml index 4f63f3ec7f..b2efb9adb6 100644 --- a/crates/ruborist/Cargo.toml +++ b/crates/ruborist/Cargo.toml @@ -7,49 +7,64 @@ license = "MIT" # tombi: format.rules.table-keys-order.disabled = true [dependencies] -anyhow = { workspace = true } -bytes = "1" -pathdiff = { workspace = true } -deno_semver = "0.7" -dirs = { workspace = true } -futures = "0.3" -parking_lot = { workspace = true } -petgraph = "0.6" -serde = { version = "1.0", features = ["derive"] } -serde_json = { workspace = true } -simd-json = "0.17" -thiserror = { workspace = true } -tokio-fs-ext = { workspace = true } -tokio-retry = "0.3" -tracing = { workspace = true } +anyhow = { workspace = true } +bytes = "1" +crossbeam-queue = "0.3" +dashmap = "6.1.0" +pathdiff = { workspace = true } +deno_semver = "0.7" +dirs = { workspace = true } +futures = "0.3" +parking_lot = { workspace = true } +petgraph = "0.6" +serde = { version = "1.0", features = ["derive"] } +serde_json = { workspace = true } +simd-json = "0.17" +thiserror = { workspace = true } +tokio-fs-ext = { workspace = true } +tokio-retry = "0.3" +tracing = { workspace = true } # HTTP client - works on both native and WASM -reqwest = { version = "0.12", default-features = false, features = [ +# +# We use `rustls-tls-native-roots-no-provider` (instead of the default +# `rustls-tls-native-roots`) so reqwest doesn't pin the crypto provider +# to `ring`. We then build our own `rustls::ClientConfig` with +# `aws-lc-rs` (BoringSSL's crypto primitives) in `service/http.rs` and +# pass it via `Client::use_preconfigured_tls`. Measured on CI against +# npmjs.org: `ring` took 78 ms p50 / 154 ms max from CCS → first +# AppData per handshake (CPU-bound; 128 parallel handshakes serialise +# across 4 cores). aws-lc-rs's per-handshake CPU is roughly 3× faster. +reqwest = { version = "0.12", default-features = false, features = [ + "brotli", + "gzip", "http2", "json", - "rustls-tls", - "rustls-tls-native-roots", + "rustls-tls-native-roots-no-provider", "socks" ] } +# Our own rustls + aws-lc-rs crypto provider. +rustls = { version = "0.23", default-features = false, features = ["aws-lc-rs", "logging", "std", "tls12"] } +rustls-native-certs = "0.8" # Shared URL-hash + staging helpers for native-only resolvers (git / http) -sha2 = { workspace = true, optional = true } -tempfile = { workspace = true, optional = true } +sha2 = { workspace = true, optional = true } +tempfile = { workspace = true, optional = true } # Git clone (gated on `native-git`) -gix = { workspace = true, features = [ +gix = { workspace = true, features = [ "blocking-http-transport-reqwest-rust-tls", "blocking-network-client", ], optional = true } # HTTP tarball extraction (gated on `http-tarball`) # Uses libdeflater to match pm's extractor and avoid pulling a second gzip # implementation (flate2) into the workspace. -libdeflater = { version = "1.25", optional = true } -tar = { version = "0.4", optional = true } +libdeflater = { version = "1.25", optional = true } +tar = { version = "0.4", optional = true } # Async runtime (works on both native and WASM with forked tokio) [dependencies.tokio] workspace = true [dev-dependencies] -tokio = { workspace = true, features = ["macros", "rt"] } +tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread"] } # Native-only dependencies (not compiled for WASM) [target.'cfg(not(target_arch = "wasm32"))'.dependencies] diff --git a/crates/ruborist/src/lib.rs b/crates/ruborist/src/lib.rs index 890d6c2798..b7d0753071 100644 --- a/crates/ruborist/src/lib.rs +++ b/crates/ruborist/src/lib.rs @@ -101,6 +101,4 @@ pub mod http { } /// Utility functions. -pub mod util { - pub use crate::model::util::{PackageNameStr, parse_package_spec, read_package_json}; -} +pub mod util; diff --git a/crates/ruborist/src/model/manifest.rs b/crates/ruborist/src/model/manifest.rs index 6ec7495bad..16be3c3787 100644 --- a/crates/ruborist/src/model/manifest.rs +++ b/crates/ruborist/src/model/manifest.rs @@ -37,10 +37,24 @@ pub struct FullManifest { #[serde(rename = "dist-tags")] pub dist_tags: HashMap, - #[serde(default, deserialize_with = "deserialize_version_keys")] - pub versions: Vec, + /// Version keys (preserved order) and pre-parsed `CoreVersionManifest`s, + /// populated in a single pass by [`Versions`]'s custom `Deserialize`. + /// + /// Replaces the previous `versions: Vec` (populated by + /// `IgnoredAny` visitor) + per-resolve `extract_version` that + /// re-parsed the full raw JSON via `simd_json::to_borrowed_value` + /// for every `resolve_package` call. That re-parse was uninstrumented + /// CPU on the async worker — ~0.4ms avg × 4567 resolves = 1.85s + /// serial load on the 4-core tokio runtime, preventing the pipeline + /// from maintaining 64 in-flight fetches (observed ~38 effective). + /// + /// With eager parse, `get_core_version` becomes an O(1) map lookup. + pub versions: Versions, - /// Raw HTTP response bytes. Injected post-parse; used for on-demand version extraction. + /// Raw HTTP response bytes. Retained only for `get_full_version` + /// (cold path — `ut view` command) which still needs on-demand + /// extraction of the full `VersionManifest`. Hot-path resolve uses + /// [`Self::versions`] which is pre-parsed. #[serde(skip)] pub raw: Arc<[u8]>, @@ -76,11 +90,61 @@ pub struct FullManifest { } impl FullManifest { - /// Extract a single version from raw bytes on demand. + /// On-demand `CoreVersionManifest` extraction from a pre-parsed + /// `simd_json::OwnedValue` subtree. + /// + /// History: we've been through three designs here. + /// 1. **Eager struct parse (previous)**: `Versions::deserialize` + /// built `Arc` for every version up + /// front. Lookup was O(1) but fetch-time spawn_blocking had to + /// materialise ~500 structs per manifest × 2730 manifests = + /// ~1.37M CoreVersionManifest builds, 99 % of which the + /// resolver never reads. Blocking pool saturation at cap=128 + /// showed up as `parse_us` avg 20 ms with long tails. + /// 2. **Lazy `Arc` (tried in c5cd8318, + /// reverted fe8365d7)**: stored `serde_json::Value` subtrees + /// and called `from_value(value.clone())` on demand. The + /// deep clone + serde_json walk ran on async worker and + /// measured `core_version_us` 11-18 ms avg — worse than eager. + /// 3. **This design — lazy simd_json `OwnedValue` + memoisation**: + /// Store each version as `Arc` (the + /// native simd_json tree form — no serde_json round-trip). + /// On demand call `from_refvalue(&OwnedValue)` which + /// zero-copies through `&Value` implementing `Deserializer`. + /// First hit for a given (manifest, version) pays ~50-200 μs + /// of subtree walking; subsequent hits are `Arc::clone` via + /// the `DashMap` memoisation cache. /// - /// Copies raw bytes, parses with `simd_json::to_borrowed_value`, navigates - /// to `versions[ver]`, then converts to the target type via `serde_json::Value`. - fn extract_version Deserialize<'de>>(&self, version: &str) -> Option { + /// Why this is better than design 1: a typical resolve touches + /// 1-3 of a manifest's ~500 versions. Building the other 497+ + /// was pure waste on the blocking pool's critical path. + /// + /// Why this is better than design 2: `simd_json::OwnedValue`'s + /// `&Value: Deserializer` zero-copies through the tree without + /// allocating an intermediate `serde_json::Value`. And moving + /// the conversion out of `spawn_blocking` (it's small, runs on + /// async worker) removes the 200 μs dispatch overhead that was + /// ~500 ms of the previous lazy attempt. + pub fn get_core_version(&self, version: &str) -> Option> { + // Fast path: memoised conversion. + if let Some(cached) = self.versions.cache.get(version) { + return Some(cached.clone()); + } + let tree = self.versions.trees.get(version)?; + // `&simd_json::OwnedValue` impls `Deserializer<'_>` directly + // (see simd_json::serde::value::owned::de), so this is a + // zero-allocation tree walk — no Value clone, no bytes copy. + let core = CoreVersionManifest::deserialize(tree.as_ref()).ok()?; + let arc = Arc::new(core); + self.versions.cache.insert(version.to_string(), arc.clone()); + Some(arc) + } + + /// Parse a single version on demand into full `VersionManifest` + /// (cold path — `ut view`). Uses the retained raw bytes because + /// `VersionManifest` carries display fields that `CoreVersionManifest` + /// drops, and `ut view` is infrequent. + pub fn get_full_version(&self, version: &str) -> Option { use simd_json::prelude::ValueObjectAccess; let mut buf = self.raw.to_vec(); let parsed = simd_json::to_borrowed_value(&mut buf).ok()?; @@ -88,45 +152,95 @@ impl FullManifest { let value = serde_json::to_value(version_obj).ok()?; serde_json::from_value(value).ok() } +} - /// Parse a single version on demand into CoreVersionManifest (hot path). - pub fn get_core_version(&self, version: &str) -> Option { - self.extract_version(version) - } +/// Version entries of a `FullManifest`: preserves key insertion order +/// (for `VersionsInfo`/semver callers that iterate the list) alongside +/// per-version pre-parsed JSON subtrees and a memoisation cache for +/// strongly-typed `CoreVersionManifest` conversions. +/// +/// Fetch-time parse builds only `Arc` per +/// version — cheaper than constructing a full `CoreVersionManifest` +/// because `OwnedValue` is the native simd_json tree without +/// field-by-field `#[serde(default)]` / `skip_on_error` validation. +/// The real `CoreVersionManifest` is materialised on demand inside +/// `FullManifest::get_core_version`, so the ~99 % of versions the +/// resolver never touches don't pay for field-level parsing. +#[derive(Debug, Clone, Default)] +pub struct Versions { + /// Version strings in insertion order — used by + /// `resolve_target_version` and cached into `VersionsInfo`. + pub keys: Vec, + /// Per-version pre-parsed JSON subtrees. `Arc` because the + /// `FullManifest` is cloned across cache tiers (memory / project + /// cache) and we want Arc-level sharing of the underlying bytes. + pub trees: HashMap>, + /// Memoisation cache populated on first `get_core_version(v)` + /// call. `DashMap` instead of `Mutex` because + /// multiple resolves for the same (name, version) can land + /// concurrently on different async workers; DashMap avoids + /// serialising them on a global mutex. + pub cache: Arc>>, +} - /// Parse a single version on demand into full VersionManifest (cold path, e.g. `ut view`). - pub fn get_full_version(&self, version: &str) -> Option { - self.extract_version(version) +impl Versions { + pub fn is_empty(&self) -> bool { + self.keys.is_empty() } } -/// Deserialize a versions map by extracting only the keys, skipping all values. -/// -/// Uses `IgnoredAny` to skip over version manifest JSON objects without allocating, -/// only collecting the version number strings. -fn deserialize_version_keys<'de, D>(deserializer: D) -> Result, D::Error> -where - D: Deserializer<'de>, -{ - struct VersionKeysVisitor; - impl<'de> serde::de::Visitor<'de> for VersionKeysVisitor { - type Value = Vec; - fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { - f.write_str("a map of version strings to objects") +impl Serialize for Versions { + fn serialize(&self, serializer: S) -> Result { + // Serialize the same shape the registry returns: a JSON object + // mapping version string -> raw JSON tree. Order follows + // `keys` for determinism (HashMap iteration order is arbitrary + // otherwise). + use serde::ser::SerializeMap; + let mut map = serializer.serialize_map(Some(self.keys.len()))?; + for k in &self.keys { + if let Some(v) = self.trees.get(k) { + map.serialize_entry(k, v.as_ref())?; + } } - fn visit_map>( - self, - mut map: M, - ) -> Result, M::Error> { - let mut keys = Vec::with_capacity(map.size_hint().unwrap_or(0)); - while let Some(key) = map.next_key::()? { - map.next_value::()?; - keys.push(key); + map.end() + } +} + +impl<'de> Deserialize<'de> for Versions { + fn deserialize>(deserializer: D) -> Result { + struct VersionsVisitor; + impl<'de> serde::de::Visitor<'de> for VersionsVisitor { + type Value = Versions; + fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + f.write_str("a map of version strings to manifest objects") + } + fn visit_map>( + self, + mut map: M, + ) -> Result { + let cap = map.size_hint().unwrap_or(0); + let mut keys = Vec::with_capacity(cap); + let mut trees = HashMap::with_capacity(cap); + while let Some(key) = map.next_key::()? { + // Deserialize into `simd_json::OwnedValue` — the + // native simd_json tree. Cheaper than building + // `CoreVersionManifest` because it skips field + // validation / `skip_on_error` round-trips; those + // only run on demand for versions the resolver + // actually reads. + let tree: simd_json::OwnedValue = map.next_value()?; + trees.insert(key.clone(), Arc::new(tree)); + keys.push(key); + } + Ok(Versions { + keys, + trees, + cache: Arc::new(dashmap::DashMap::new()), + }) } - Ok(keys) } + deserializer.deserialize_map(VersionsVisitor) } - deserializer.deserialize_map(VersionKeysVisitor) } /// Version-specific manifest from npm registry. @@ -302,6 +416,19 @@ pub struct VersionManifest { /// Contains only the ~13 fields needed for dependency resolution, installation, /// and lockfile serialization. Display-only fields (description, author, homepage, /// keywords, bugs, repository, npm_user, etc.) are omitted. +/// +/// `skip_on_error` is kept on every field that's typed as `HashMap`, +/// `String`, or `bool` — real registry data regularly ships values +/// that don't match the expected shape (e.g. `engines` with a null +/// value, `hasInstallScript` as a string, legacy `license` as an +/// array-of-object). Dropping the wrapper on those fields broke the +/// lodash fetch at `"ExpectedMap at character 0"` in CI. +/// +/// The three `Option` fields (`bin`, `os`, `cpu`) are the only +/// safe cases: `Value` absorbs any JSON, so the `Value::deserialize` +/// round-trip in `skip_on_error` provides zero robustness. Dropping +/// it there saves one allocation + recursive walk per occurrence +/// without correctness risk. #[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(default)] pub struct CoreVersionManifest { @@ -337,10 +464,7 @@ pub struct CoreVersionManifest { pub dist: Dist, - #[serde( - deserialize_with = "skip_on_error", - skip_serializing_if = "Option::is_none" - )] + #[serde(skip_serializing_if = "Option::is_none")] pub bin: Option, #[serde( @@ -349,16 +473,10 @@ pub struct CoreVersionManifest { )] pub engines: Option>, - #[serde( - deserialize_with = "skip_on_error", - skip_serializing_if = "Option::is_none" - )] + #[serde(skip_serializing_if = "Option::is_none")] pub os: Option, - #[serde( - deserialize_with = "skip_on_error", - skip_serializing_if = "Option::is_none" - )] + #[serde(skip_serializing_if = "Option::is_none")] pub cpu: Option, #[serde( diff --git a/crates/ruborist/src/resolver/builder.rs b/crates/ruborist/src/resolver/builder.rs index b03db0fc43..0ddef102e4 100644 --- a/crates/ruborist/src/resolver/builder.rs +++ b/crates/ruborist/src/resolver/builder.rs @@ -458,7 +458,7 @@ async fn process_file_dep( /// /// # Returns /// The result of processing (reused, created, or skipped) -pub async fn process_dependency( +pub async fn process_dependency( graph: &mut DependencyGraph, registry: &R, node_index: NodeIndex, @@ -707,11 +707,14 @@ pub async fn process_dependency( /// // Add initial dependency edges to root... /// build_deps(&mut graph, ®istry, PeerDeps::Include).await?; /// ``` -pub async fn build_deps( +pub async fn build_deps( graph: &mut DependencyGraph, registry: &R, peer_deps: PeerDeps, -) -> Result<(), ResolveError> { +) -> Result<(), ResolveError> +where + R::Error: Send, +{ let config = BuildDepsConfig::default().with_peer_deps(peer_deps); build_deps_with_config(graph, registry, config, &NoopReceiver).await } @@ -729,12 +732,18 @@ pub async fn build_deps( /// * `registry` - Registry client for fetching packages /// * `peer_deps` - How to handle peer dependencies /// * `receiver` - Event receiver for handling build events -pub async fn build_deps_with_receiver( +pub async fn build_deps_with_receiver< + R: RegistryClient + Clone + Send + Sync + 'static, + E: EventReceiver, +>( graph: &mut DependencyGraph, registry: &R, peer_deps: PeerDeps, receiver: &E, -) -> Result<(), ResolveError> { +) -> Result<(), ResolveError> +where + R::Error: Send, +{ let config = BuildDepsConfig::default().with_peer_deps(peer_deps); build_deps_with_config(graph, registry, config, receiver).await } @@ -759,12 +768,18 @@ pub async fn build_deps_with_receiver( /// /// build_deps_with_config(&mut graph, ®istry, config, &receiver).await?; /// ``` -pub async fn build_deps_with_config( +pub async fn build_deps_with_config< + R: RegistryClient + Clone + Send + Sync + 'static, + E: EventReceiver, +>( graph: &mut DependencyGraph, registry: &R, config: BuildDepsConfig, receiver: &E, -) -> Result<(), ResolveError> { +) -> Result<(), ResolveError> +where + R::Error: Send, +{ tracing::debug!( "Starting dependency tree build, peer_deps: {:?}, concurrency: {}, skip_preload: {}", config.peer_deps, @@ -786,12 +801,14 @@ pub async fn build_deps_with_config( } /// Run the preload phase to warm up the cache with manifests. -async fn run_preload_phase( +async fn run_preload_phase( graph: &DependencyGraph, registry: &R, config: &BuildDepsConfig, receiver: &E, -) { +) where + R::Error: Send, +{ if config.skip_preload { return; } @@ -804,18 +821,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| { @@ -824,21 +844,22 @@ 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. -async fn run_bfs_phase( +async fn run_bfs_phase( graph: &mut DependencyGraph, registry: &R, config: &BuildDepsConfig, @@ -961,10 +982,13 @@ 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( pkg: &PackageJson, registry: &R, -) -> Result> { +) -> Result> +where + R::Error: Send, +{ resolve_with_options(pkg, registry, PeerDeps::Include, &NoopReceiver).await } @@ -975,12 +999,18 @@ pub async fn resolve( /// * `registry` - Registry client for fetching packages /// * `peer_deps` - How to handle peer dependencies /// * `receiver` - Event receiver for progress tracking -pub async fn resolve_with_options( +pub async fn resolve_with_options< + R: RegistryClient + Clone + Send + Sync + 'static, + E: EventReceiver, +>( pkg: &PackageJson, registry: &R, peer_deps: PeerDeps, receiver: &E, -) -> Result> { +) -> Result> +where + R::Error: Send, +{ // Create graph with root node let mut graph = DependencyGraph::from_package_json(PathBuf::from("."), pkg.clone()); diff --git a/crates/ruborist/src/resolver/http.rs b/crates/ruborist/src/resolver/http.rs index f88cd509f1..10c94486f6 100644 --- a/crates/ruborist/src/resolver/http.rs +++ b/crates/ruborist/src/resolver/http.rs @@ -99,7 +99,7 @@ use crate::model::manifest::CoreVersionManifest; use crate::service::fetch::{ FetchError, classify_reqwest_error, classify_status, is_retryable, retry_strategy, }; -use crate::service::http::get_client; +use crate::service::http::pick_client; use crate::traits::registry::ResolvedPackage; /// Session-scoped dedup cache: one fetch per URL even under concurrent BFS. @@ -154,7 +154,7 @@ async fn download_tarball(url: &str) -> Result { || { let url = url.to_string(); async move { - let resp = get_client() + let resp = pick_client() .map_err(FetchError::Permanent)? .get(&url) .send() diff --git a/crates/ruborist/src/resolver/preload.rs b/crates/ruborist/src/resolver/preload.rs index 9fb5aff67b..da6f897b8e 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,163 @@ 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, + R: RegistryClient + Send + Sync + 'static, + R::Error: Send, 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(); + + tokio::spawn(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; } + }); + } + // 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::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; + 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 +266,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 +284,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 +308,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 +484,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 +517,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/registry.rs b/crates/ruborist/src/resolver/registry.rs index 6edeb9069c..d306b2ddcb 100644 --- a/crates/ruborist/src/resolver/registry.rs +++ b/crates/ruborist/src/resolver/registry.rs @@ -7,8 +7,6 @@ //! - For semver-supporting registries: directly fetches specific version //! - For traditional registries: fetches full manifest and resolves locally -use std::sync::Arc; - use crate::model::manifest::FullManifest; use crate::model::node::EdgeType; use crate::resolver::semver::normalize_spec; @@ -114,7 +112,7 @@ impl std::error::Error for ResolveError { /// let resolved = resolve_package(®istry, "lodash", "^4.0.0").await?; /// println!("Resolved to {}@{}", resolved.name, resolved.version); /// ``` -pub async fn resolve_package( +pub async fn resolve_package( registry: &R, name: &str, spec: &str, @@ -137,7 +135,7 @@ pub fn resolve_from_manifest( manifest: &FullManifest, spec: &str, ) -> Result> { - let version_list: Vec = manifest.versions.clone(); + let version_list: Vec = manifest.versions.keys.clone(); if version_list.is_empty() { return Err(ResolveError::NoVersions(manifest.name.clone())); @@ -147,10 +145,9 @@ pub fn resolve_from_manifest( let resolved_version = resolve_target_version(&manifest.dist_tags, &version_list, spec) .map_err(|e| ResolveError::Version(format!("{}@{}: {}", manifest.name, spec, e)))?; - // Get manifest for resolved version (lazy: parse from Value on demand) + // O(1) lookup from pre-parsed `Versions` map. let version_manifest = manifest .get_core_version(&resolved_version) - .map(Arc::new) .ok_or_else(|| ResolveError::ManifestNotFound { name: manifest.name.clone(), version: resolved_version.clone(), @@ -167,7 +164,7 @@ pub fn resolve_from_manifest( /// /// For optional dependencies, returns `Ok(None)` on resolution failure /// instead of propagating the error. -pub async fn resolve_registry_dep( +pub async fn resolve_registry_dep( registry: &R, name: &str, spec: &str, diff --git a/crates/ruborist/src/resolver/semver.rs b/crates/ruborist/src/resolver/semver.rs index 6e81d5c10c..5380ca81cc 100644 --- a/crates/ruborist/src/resolver/semver.rs +++ b/crates/ruborist/src/resolver/semver.rs @@ -39,7 +39,12 @@ use deno_semver::{Version, VersionReq}; /// assert_eq!(name, "lodash"); /// assert_eq!(spec, "^4.0.0"); /// ``` -pub fn normalize_spec(name: &str, spec: &str) -> (String, String) { +pub fn normalize_spec<'a>( + name: &'a str, + spec: &'a str, +) -> (std::borrow::Cow<'a, str>, std::borrow::Cow<'a, str>) { + use std::borrow::Cow; + // Handle npm: alias - fetch the aliased package instead if let Some(npm_spec) = spec.strip_prefix("npm:") { // Skip "npm:" @@ -48,21 +53,26 @@ pub fn normalize_spec(name: &str, spec: &str) -> (String, String) { // Make sure we don't split on the @ of a scoped package if last_at_index > 0 { let (pkg_name, version) = npm_spec.split_at(last_at_index); - return (pkg_name.to_string(), version[1..].to_string()); + return (Cow::Borrowed(pkg_name), Cow::Borrowed(&version[1..])); } } // No version specified - return (npm_spec.to_string(), "*".to_string()); + return (Cow::Borrowed(npm_spec), Cow::Borrowed("*")); } // Handle workspace: prefix - keep original name, extract version if let Some(workspace_spec) = spec.strip_prefix("workspace:") { // Skip "workspace:" - return (name.to_string(), workspace_spec.to_string()); + return (Cow::Borrowed(name), Cow::Borrowed(workspace_spec)); } - // No special prefix, return as-is - (name.to_string(), spec.to_string()) + // No special prefix, return borrowed — the common case (~99 % of + // deps in a typical npm workload). Previously returned + // `(name.to_string(), spec.to_string())` which allocated two + // Strings per `resolve_package` call on the resolver hot path + // (~5460 allocs on ant-design preload, all from cooperative-poll + // future bodies that the main task drives). + (Cow::Borrowed(name), Cow::Borrowed(spec)) } /// Check if a version matches a semver range. @@ -208,51 +218,34 @@ mod tests { #[test] fn test_normalize_spec() { - // npm alias - assert_eq!( - normalize_spec("string-width-cjs", "npm:string-width@^4.2.0"), - ("string-width".to_string(), "^4.2.0".to_string()) - ); + let assert_eq_norm = |actual: (std::borrow::Cow<'_, str>, std::borrow::Cow<'_, str>), + expected: (&str, &str)| { + assert_eq!(actual.0.as_ref(), expected.0); + assert_eq!(actual.1.as_ref(), expected.1); + }; - // npm alias without version - assert_eq!( - normalize_spec("lodash-cjs", "npm:lodash"), - ("lodash".to_string(), "*".to_string()) + assert_eq_norm( + normalize_spec("string-width-cjs", "npm:string-width@^4.2.0"), + ("string-width", "^4.2.0"), ); - - // scoped npm alias - assert_eq!( + assert_eq_norm(normalize_spec("lodash-cjs", "npm:lodash"), ("lodash", "*")); + assert_eq_norm( normalize_spec("my-pkg", "npm:@scope/pkg@^1.0.0"), - ("@scope/pkg".to_string(), "^1.0.0".to_string()) + ("@scope/pkg", "^1.0.0"), ); - - // scoped npm alias without version - assert_eq!( + assert_eq_norm( normalize_spec("my-pkg", "npm:@scope/pkg"), - ("@scope/pkg".to_string(), "*".to_string()) + ("@scope/pkg", "*"), ); - - // workspace reference - assert_eq!( - normalize_spec("my-lib", "workspace:*"), - ("my-lib".to_string(), "*".to_string()) - ); - - assert_eq!( + assert_eq_norm(normalize_spec("my-lib", "workspace:*"), ("my-lib", "*")); + assert_eq_norm( normalize_spec("my-lib", "workspace:^1.0.0"), - ("my-lib".to_string(), "^1.0.0".to_string()) + ("my-lib", "^1.0.0"), ); - - // regular spec (unchanged) - assert_eq!( - normalize_spec("lodash", "^4.0.0"), - ("lodash".to_string(), "^4.0.0".to_string()) - ); - - // scoped package regular spec - assert_eq!( + assert_eq_norm(normalize_spec("lodash", "^4.0.0"), ("lodash", "^4.0.0")); + assert_eq_norm( normalize_spec("@types/node", "^18.0.0"), - ("@types/node".to_string(), "^18.0.0".to_string()) + ("@types/node", "^18.0.0"), ); } } 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)) } diff --git a/crates/ruborist/src/service/cache.rs b/crates/ruborist/src/service/cache.rs index d11ac6e4cf..4140ba1fc0 100644 --- a/crates/ruborist/src/service/cache.rs +++ b/crates/ruborist/src/service/cache.rs @@ -53,7 +53,7 @@ //! It pre-populates memory cache to skip network on subsequent runs. //! ``` -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; @@ -91,7 +91,16 @@ use parking_lot::RwLock; pub struct MemoryCache(Arc); struct MemoryCacheInner { - full_manifests: RwLock>, + /// `Arc` so `get` returns a cheap atomic-bump clone + /// instead of deep-cloning the full versions HashMap (~100-500 + /// `simd_json::OwnedValue` entries per package). The deep clone + /// was running synchronously on the main task during preload — + /// `FuturesUnordered` polls each future on the owning task, and + /// `resolve_package`'s cache check (line 226 of registry.rs) is + /// the first thing every future does. Standalone manifest-bench + /// hit avg_conc=95 at cap=128; ruborist stalled at avg_conc=56 + /// because the per-future deep clone serialised the main loop. + full_manifests: RwLock>>, versions_info: RwLock>, version_manifests: RwLock>>, } @@ -119,11 +128,11 @@ impl MemoryCache { } // Full manifests - pub fn get_full_manifest(&self, name: &str) -> Option { + pub fn get_full_manifest(&self, name: &str) -> Option> { self.0.full_manifests.read().get(name).cloned() } - pub fn set_full_manifest(&self, name: String, manifest: FullManifest) { + pub fn set_full_manifest(&self, name: String, manifest: Arc) { self.0.full_manifests.write().insert(name, manifest); } @@ -214,6 +223,22 @@ pub struct PackageCache { memory: MemoryCache, /// Disk cache directory (None = no disk cache) cache_dir: Option, + /// Lazily-populated set of package names with existing disk cache. + /// + /// Built once per process from a single `read_dir(cache_dir)` + + /// per-scope `read_dir` for `@scope/pkg` entries. `get_versions_from_disk` + /// checks this set first and short-circuits to `None` when the package + /// isn't on disk — avoiding the 2730 per-package `try_exists` + /// spawn_blocking calls that dominated cold-preload critical path + /// (avg 16 ms per call × 128 parallel futures = measurable wall drag). + /// + /// Mid-run writes via `set_versions_to_disk` are not reflected in this + /// index — that's intentional. Any package we wrote in the current run + /// has already been populated into the memory cache, which takes + /// precedence over disk; the disk path is only consulted for packages + /// *not yet seen this run* with *existing disk cache from a prior run*. + #[cfg_attr(target_arch = "wasm32", allow(dead_code))] + disk_index: Arc>>, } impl PackageCache { @@ -222,6 +247,7 @@ impl PackageCache { Self { memory: MemoryCache::global(), cache_dir: None, + disk_index: Arc::new(tokio::sync::OnceCell::new()), } } @@ -230,9 +256,27 @@ impl PackageCache { Self { memory: MemoryCache::global(), cache_dir: Some(cache_dir), + disk_index: Arc::new(tokio::sync::OnceCell::new()), } } + /// Return `true` if the package has a disk cache entry (based on the + /// lazily-populated bulk-readdir index). Returns `false` if no + /// cache_dir is configured or the index says the package isn't + /// present. First call per process does a single directory scan. + #[cfg(not(target_arch = "wasm32"))] + async fn has_disk_entry(&self, name: &str) -> bool { + let Some(cache_dir) = self.cache_dir.as_ref() else { + return false; + }; + let cache_dir = cache_dir.clone(); + let index = self + .disk_index + .get_or_init(|| async move { build_disk_index(&cache_dir).await }) + .await; + index.contains(name) + } + /// Get the cache directory. pub fn cache_dir(&self) -> Option<&Path> { self.cache_dir.as_deref() @@ -240,8 +284,10 @@ impl PackageCache { // === Memory cache operations (sync) === - /// Get full manifest from memory cache. - pub fn get_full_manifest(&self, name: &str) -> Option { + /// Get full manifest from memory cache. Returns `Arc` so callers + /// pay an atomic-bump clone instead of deep-cloning the full + /// versions HashMap. + pub fn get_full_manifest(&self, name: &str) -> Option> { let result = self.memory.get_full_manifest(name); if result.is_some() { tracing::debug!("Memory cache hit for full manifest: {name}"); @@ -249,8 +295,10 @@ impl PackageCache { result } - /// Set full manifest in memory cache. - pub fn set_full_manifest(&self, name: String, manifest: FullManifest) { + /// Set full manifest in memory cache. Takes `Arc` so callers can + /// share ownership with their own copy without paying for a + /// deep clone at the boundary. + pub fn set_full_manifest(&self, name: String, manifest: Arc) { tracing::debug!("Caching full manifest in memory: {name}"); self.memory.set_full_manifest(name, manifest); } @@ -297,18 +345,21 @@ impl PackageCache { // === Disk cache operations (async, uses tokio-fs-ext) === /// Load versions info from disk cache. + /// + /// Short-circuits via the bulk-readdir index when the package isn't + /// present on disk — avoids the per-package `try_exists` syscall + /// that was 16 ms avg / 230 ms max on CI's blocking pool under load. pub async fn get_versions_from_disk(&self, name: &str) -> Option { let cache_dir = self.cache_dir.as_ref()?; - let path = get_versions_cache_path(cache_dir, name); - - if !super::fs::exists(&path).await { + #[cfg(not(target_arch = "wasm32"))] + if !self.has_disk_entry(name).await { return None; } + let path = get_versions_cache_path(cache_dir, name); match super::fs::read_json::(&path).await { Ok(info) => { tracing::debug!("Disk cache hit for versions: {name}"); - // Also cache in memory self.memory.set_versions(name.to_string(), info.clone()); Some(info) } @@ -319,31 +370,37 @@ impl PackageCache { } } - /// Save versions info to disk cache. - pub async fn set_versions_to_disk(&self, name: &str, info: &VersionsInfo) { - let Some(cache_dir) = &self.cache_dir else { + /// Save versions info to disk cache (fire-and-forget). + /// + /// Spawns the serialisation + write onto the tokio runtime so the caller + /// returns immediately. The resolve pipeline doesn't depend on the write + /// completing — disk cache is best-effort for the *next* run. pcap-driven + /// tuning showed the previous inline `.await` + `serde_json::to_string_pretty` + /// burned ~1–3 ms per call on the hot path, stalling the main preload + /// task and causing the 24..62 active-stream dip observed on CI. + pub fn set_versions_to_disk(&self, name: &str, info: &VersionsInfo) { + let Some(cache_dir) = self.cache_dir.clone() else { return; }; + let name = name.to_string(); + let info = info.clone(); - let path = get_versions_cache_path(cache_dir, name); - - // Ensure parent directory exists - if let Some(parent) = path.parent() { - let _ = tokio_fs_ext::create_dir_all(parent).await; - } - - match serde_json::to_string_pretty(info) { - Ok(content) => { - if let Err(e) = tokio_fs_ext::write(&path, content.as_bytes()).await { - tracing::debug!("Failed to write versions cache for {name}: {e}"); - } else { - tracing::debug!("Wrote versions to disk cache: {name}"); - } + tokio::spawn(async move { + let path = get_versions_cache_path(&cache_dir, &name); + if let Some(parent) = path.parent() { + let _ = tokio_fs_ext::create_dir_all(parent).await; } - Err(e) => { - tracing::debug!("Failed to serialize versions for {name}: {e}"); + match serde_json::to_vec(&info) { + Ok(bytes) => { + if let Err(e) = tokio_fs_ext::write(&path, bytes).await { + tracing::debug!("Failed to write versions cache for {name}: {e}"); + } + } + Err(e) => { + tracing::debug!("Failed to serialize versions for {name}: {e}"); + } } - } + }); } /// Load version manifest from disk cache. @@ -353,6 +410,10 @@ impl PackageCache { version: &str, ) -> Option> { let cache_dir = self.cache_dir.as_ref()?; + #[cfg(not(target_arch = "wasm32"))] + if !self.has_disk_entry(name).await { + return None; + } let path = get_manifest_cache_path(cache_dir, name, version); if !super::fs::exists(&path).await { @@ -378,36 +439,40 @@ impl PackageCache { } } - /// Save version manifest to disk cache. - pub async fn set_version_manifest_to_disk( + /// Save version manifest to disk cache (fire-and-forget). + /// + /// Takes `Arc` so the shared reference moves into the + /// spawned task without deep-cloning the manifest (they can be 10–50 KB + /// each). Same rationale as [`Self::set_versions_to_disk`]: the resolve + /// pipeline was stalling on inline `.await` + pretty JSON serialisation. + pub fn set_version_manifest_to_disk( &self, name: &str, version: &str, - manifest: &CoreVersionManifest, + manifest: Arc, ) { - let Some(cache_dir) = &self.cache_dir else { + let Some(cache_dir) = self.cache_dir.clone() else { return; }; + let name = name.to_string(); + let version = version.to_string(); - let path = get_manifest_cache_path(cache_dir, name, version); - - // Ensure parent directory exists - if let Some(parent) = path.parent() { - let _ = tokio_fs_ext::create_dir_all(parent).await; - } - - match serde_json::to_string_pretty(manifest) { - Ok(content) => { - if let Err(e) = tokio_fs_ext::write(&path, content.as_bytes()).await { - tracing::debug!("Failed to write manifest cache for {name}@{version}: {e}"); - } else { - tracing::debug!("Wrote manifest to disk cache: {name}@{version}"); - } + tokio::spawn(async move { + let path = get_manifest_cache_path(&cache_dir, &name, &version); + if let Some(parent) = path.parent() { + let _ = tokio_fs_ext::create_dir_all(parent).await; } - Err(e) => { - tracing::debug!("Failed to serialize manifest for {name}@{version}: {e}"); + match serde_json::to_vec(&*manifest) { + Ok(bytes) => { + if let Err(e) = tokio_fs_ext::write(&path, bytes).await { + tracing::debug!("Failed to write manifest cache for {name}@{version}: {e}"); + } + } + Err(e) => { + tracing::debug!("Failed to serialize manifest for {name}@{version}: {e}"); + } } - } + }); } /// Export all version manifests for persistence. @@ -429,6 +494,51 @@ impl PackageCache { } } +/// Scan `cache_dir` once to build a set of package names with existing +/// disk cache entries. +/// +/// Layout: `{cache_dir}//versions.json` for bare packages, +/// `{cache_dir}/@scope//versions.json` for scoped packages. One +/// top-level `read_dir` plus one `read_dir` per scope directory is +/// enough to enumerate every possible name. Unreadable entries are +/// silently skipped (cache dir doesn't exist on cold runs → empty +/// index → every lookup short-circuits to `None`). +#[cfg(not(target_arch = "wasm32"))] +async fn build_disk_index(cache_dir: &Path) -> HashSet { + let mut set = HashSet::new(); + let Ok(mut entries) = tokio_fs_ext::read_dir(cache_dir).await else { + return set; + }; + while let Ok(Some(entry)) = entries.next_entry().await { + let raw_name = entry.file_name(); + let Some(name) = raw_name.to_str() else { + continue; + }; + if let Some(stripped) = name.strip_prefix('@') { + // Scoped: read one level deeper. The `@` prefix distinguishes + // scopes from bare package names unambiguously (npm package + // names cannot start with `@` unless scoped). + let scope_path = entry.path(); + let Ok(mut sub) = tokio_fs_ext::read_dir(&scope_path).await else { + continue; + }; + while let Ok(Some(sub_entry)) = sub.next_entry().await { + if let Some(pkg) = sub_entry.file_name().to_str() { + set.insert(format!("@{stripped}/{pkg}")); + } + } + } else { + set.insert(name.to_string()); + } + } + tracing::debug!( + "Built disk cache index: {} package(s) at {}", + set.len(), + cache_dir.display() + ); + set +} + /// Cache statistics. #[derive(Debug, Clone)] pub struct CacheStats { @@ -578,7 +688,7 @@ mod tests { ..Default::default() }; - cache.set_full_manifest("test".to_string(), manifest.clone()); + cache.set_full_manifest("test".to_string(), Arc::new(manifest)); let retrieved = cache.get_full_manifest("test").unwrap(); assert_eq!(retrieved.name, "test"); diff --git a/crates/ruborist/src/service/dns.rs b/crates/ruborist/src/service/dns.rs index ac61fec1eb..969530a2b0 100644 --- a/crates/ruborist/src/service/dns.rs +++ b/crates/ruborist/src/service/dns.rs @@ -14,6 +14,7 @@ mod native { use std::collections::HashMap; use std::net::SocketAddr; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, LazyLock}; use std::time::{Duration, Instant}; @@ -70,6 +71,12 @@ mod native { cache: Arc>>, inflight: Arc>>>, ttl: Duration, + /// Round-robin counter used to rotate the returned address order + /// on every `resolve` call. Spreads concurrent pool connections + /// across all DNS A records (e.g. antgroup registry returns 8 + /// IPs) instead of pinning every new connection to the first IP + /// like reqwest does by default. + rr_counter: AtomicUsize, } impl CachingResolver { @@ -79,10 +86,59 @@ mod native { cache: Arc::new(Mutex::new(HashMap::new())), inflight: Arc::new(Mutex::new(HashMap::new())), ttl, + rr_counter: AtomicUsize::new(0), } } } + /// Rotate `addrs` left by `offset` positions, keeping each address + /// family (IPv4 / IPv6) rotated independently and preserving the + /// system resolver's original family ordering between them. + /// + /// Why per-family: `getaddrinfo` typically returns IPv6 first + /// (e.g. 10 v6 + 12 v4 = 22 entries for `registry.npmjs.org`). A + /// flat rotation by `offset % 22` means offsets 0..10 all start + /// inside the IPv6 range and — on hosts without usable IPv6 + /// routing like GitHub Actions runners — every single one of them + /// falls through to the *same* first IPv4 address once Happy + /// Eyeballs gives up on v6. Pcap comparison with bun confirmed + /// this bug: 66 of utoo's 128 connections (51 %) were landing on + /// a single Cloudflare edge IP while the other 11 IPs saw only + /// 5-6 connections each. bun distributes cleanly 64 per IP across + /// 4 IPs, which is the behaviour we want. + /// + /// With per-family rotation, every new connection's first-reachable + /// IPv4 cycles through all v4 addresses (and same for v6 if it + /// works), giving ~11 conns per IP instead of 66 on one. + fn rotate_addrs(addrs: &[SocketAddr], offset: usize) -> Vec { + if addrs.is_empty() { + return Vec::new(); + } + let rotate = |slice: &[SocketAddr]| -> Vec { + if slice.is_empty() { + return Vec::new(); + } + let start = offset % slice.len(); + slice[start..] + .iter() + .chain(&slice[..start]) + .copied() + .collect() + }; + let v6: Vec = addrs.iter().filter(|a| a.is_ipv6()).copied().collect(); + let v4: Vec = addrs.iter().filter(|a| a.is_ipv4()).copied().collect(); + let v6_rot = rotate(&v6); + let v4_rot = rotate(&v4); + // Preserve v6-first ordering if that's what the resolver gave us; + // Happy Eyeballs will still prefer v6 when it's reachable. + let v6_first = addrs.first().map(|a| a.is_ipv6()).unwrap_or(true); + if v6_first { + v6_rot.into_iter().chain(v4_rot).collect() + } else { + v4_rot.into_iter().chain(v6_rot).collect() + } + } + /// Perform a system DNS lookup and cache the result. async fn do_lookup( hostname: &str, @@ -132,10 +188,11 @@ mod native { if let Some(entry) = cache.get(&hostname) && entry.expires_at > Instant::now() { - let cached = entry.addrs.to_vec(); - return Box::pin(std::future::ready( - Ok(Box::new(cached.into_iter()) as Addrs), - )); + let offset = self.rr_counter.fetch_add(1, Ordering::Relaxed); + let rotated = rotate_addrs(&entry.addrs, offset); + return Box::pin(std::future::ready(Ok( + Box::new(rotated.into_iter()) as Addrs + ))); } } @@ -155,6 +212,7 @@ mod native { // Clone Arcs for the 'static async block (required by Resolve trait). let cache = Arc::clone(&self.cache); let inflight_map = Arc::clone(&self.inflight); + let offset = self.rr_counter.fetch_add(1, Ordering::Relaxed); Box::pin(async move { let result = inflight @@ -167,13 +225,8 @@ mod native { inflight_map.lock().remove(&hostname); let resolved = result?; - - // resolved is a &Arc> borrowed from the OnceCell, - // but Addrs requires a 'static owned iterator. - // to_vec() + into_iter() creates an owned copy; iter().copied() - // would borrow from the local reference and fail lifetime checks. - #[allow(clippy::unnecessary_to_owned)] - let addrs: Addrs = Box::new(resolved.to_vec().into_iter()); + let rotated = rotate_addrs(resolved.as_ref(), offset); + let addrs: Addrs = Box::new(rotated.into_iter()); Ok(addrs) }) } diff --git a/crates/ruborist/src/service/http.rs b/crates/ruborist/src/service/http.rs index b959020973..0d790d98e7 100644 --- a/crates/ruborist/src/service/http.rs +++ b/crates/ruborist/src/service/http.rs @@ -76,23 +76,150 @@ //! WASM targets skip DNS entirely (browser handles it). use std::sync::LazyLock; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Instant; use anyhow::{Context, Result, anyhow}; +use crossbeam_queue::SegQueue; + +/// Diagnostic: per-HTTP-request `(send_start, body_end)` timestamps. +/// +/// When the flag is active, manifest fetch sites push `(Instant, Instant)` +/// pairs into the queue. Preload uses the collected intervals to compute +/// "pure network window", "busy window" (interval union), and per-request +/// stats — isolating network wait from our own CPU work. +/// +/// Flag is a single `AtomicBool` (relaxed) and only manifests as one +/// comparison + two `Instant::now()` per HTTP request when enabled; zero +/// cost on the disabled path. +static HTTP_TRACE_ACTIVE: AtomicBool = AtomicBool::new(false); +static HTTP_TRACE: LazyLock> = LazyLock::new(SegQueue::new); + +/// Activate per-request HTTP timing capture. Drains any prior trace. +pub fn start_http_trace() { + while HTTP_TRACE.pop().is_some() {} + HTTP_TRACE_ACTIVE.store(true, Ordering::Relaxed); +} + +/// Stop capturing and return the collected `(start, end)` intervals. +pub fn finish_http_trace() -> Vec<(Instant, Instant)> { + HTTP_TRACE_ACTIVE.store(false, Ordering::Relaxed); + let mut out = Vec::new(); + while let Some(v) = HTTP_TRACE.pop() { + out.push(v); + } + out +} + +/// Record one completed HTTP request's `(send_start, body_end)` timestamps. +/// No-op when the trace flag is off. Cheap relaxed-load check guards the +/// push so disabled callers pay almost nothing. +#[inline] +pub fn record_http_interval(start: Instant, end: Instant) { + if HTTP_TRACE_ACTIVE.load(Ordering::Relaxed) { + HTTP_TRACE.push((start, end)); + } +} + +/// Diagnostic: per-parse `(queued_at, exec_start, exec_end)` timestamps. +/// +/// `queued_at` is captured right before `spawn_blocking` is called; +/// `exec_start` is captured inside the closure (i.e. once the blocking +/// pool actually picks the task up). `queue_wait = exec_start − queued_at` +/// measures how long each parse sat idle in the blocking-pool queue. +/// When `queue_wait p50 ≫ exec p50` the pool is the bottleneck, and +/// `resolve_package` awaits stall the `FuturesUnordered` pipeline. +static PARSE_TRACE_ACTIVE: AtomicBool = AtomicBool::new(false); +static PARSE_TRACE: LazyLock> = LazyLock::new(SegQueue::new); + +pub fn start_parse_trace() { + while PARSE_TRACE.pop().is_some() {} + PARSE_TRACE_ACTIVE.store(true, Ordering::Relaxed); +} + +pub fn finish_parse_trace() -> Vec<(Instant, Instant, Instant)> { + PARSE_TRACE_ACTIVE.store(false, Ordering::Relaxed); + let mut out = Vec::new(); + while let Some(v) = PARSE_TRACE.pop() { + out.push(v); + } + out +} + +#[inline] +pub fn parse_trace_enabled() -> bool { + PARSE_TRACE_ACTIVE.load(Ordering::Relaxed) +} + +#[inline] +pub fn record_parse_interval(queued_at: Instant, exec_start: Instant, exec_end: Instant) { + if PARSE_TRACE_ACTIVE.load(Ordering::Relaxed) { + PARSE_TRACE.push((queued_at, exec_start, exec_end)); + } +} /// Global HTTP client with connection pooling and DNS caching. /// /// Stores `Result` so that proxy-configuration errors are /// surfaced to callers instead of panicking or calling `process::exit`. +/// +/// Multi-client experiments (2 and 4 isolated clients + preheat) both +/// regressed p1_resolve on CI — preheat TLS handshake cost exceeded the +/// phase-lock smoothing benefit, and uv (which uses a single client) +/// proves one pool is the right default. static HTTP_CLIENT: LazyLock> = LazyLock::new(|| { client_builder() .and_then(|b| b.build().context("Failed to build reqwest client")) .map_err(|e| e.to_string()) }); -pub(crate) fn get_client() -> Result<&'static reqwest::Client> { +pub(crate) fn pick_client() -> Result<&'static reqwest::Client> { HTTP_CLIENT.as_ref().map_err(|e| anyhow!("{e}")) } +/// Build a `rustls::ClientConfig` using the `aws-lc-rs` crypto provider +/// instead of reqwest's default `ring`. Measured on CI (4-core runner) +/// against npmjs.org, ring's per-TLS-handshake client-side CPU cost +/// (ECDHE key derivation + cert verification + Finished MAC) serialised +/// across 128 parallel handshakes into a 154 ms "CCS → first AppData" +/// span — the HTTP requests couldn't fire until all TLS crypto drained +/// through 4 async workers. aws-lc-rs uses BoringSSL's assembly-optimised +/// primitives and is roughly 3× faster at handshake work. +#[cfg(not(target_arch = "wasm32"))] +fn build_rustls_config() -> Result { + // Install aws-lc-rs as the default for any other rustls consumer in + // the process. Idempotent — only the first call per process wins. + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + + // Load OS root certs. On Linux this hits /etc/ssl/certs/, + // on macOS it queries the Security framework keychain. Called once + // per process via `HTTP_CLIENT`'s `LazyLock`. + let roots = rustls_native_certs::load_native_certs(); + let mut root_store = rustls::RootCertStore::empty(); + for cert in roots.certs { + // Best-effort: skip any cert rustls refuses (same tolerance + // native-tls shows). A hard fail here would brick every + // request on a box with one bad root in its trust store. + let _ = root_store.add(cert); + } + if !roots.errors.is_empty() { + tracing::debug!( + "rustls-native-certs reported {} non-fatal load issues", + roots.errors.len() + ); + } + + let config = rustls::ClientConfig::builder_with_provider(std::sync::Arc::new( + rustls::crypto::aws_lc_rs::default_provider(), + )) + .with_safe_default_protocol_versions() + .map_err(|e| anyhow!("rustls safe_default_protocol_versions: {e}"))? + .with_root_certificates(root_store) + .with_no_client_auth(); + + Ok(config) +} + /// Create a [`reqwest::ClientBuilder`] with TLS, DNS caching, and proxy /// from environment variables. /// @@ -112,10 +239,21 @@ pub fn client_builder() -> Result { let builder = { use crate::service::dns::shared_resolver; + let tls_config = build_rustls_config()?; let mut builder = builder - .use_rustls_tls() + .use_preconfigured_tls(tls_config) .no_proxy() - .dns_resolver(shared_resolver()); + .dns_resolver(shared_resolver()) + // Force HTTP/1.1 with a connection pool. reqwest multiplexes all + // requests over a single HTTP/2 connection by default, which + // makes head-of-line blocking on one slow response stall the + // whole manifest fetch phase. An H1 pool lets concurrent + // manifest requests open independent TCP streams instead. + // Pool size matches `preload::DEFAULT_CONCURRENCY` so the + // per-host idle pool can absorb every in-flight fetch without + // churning connections. + .http1_only() + .pool_max_idle_per_host(256); match env_var("ALL_PROXY") { Some(url) => { diff --git a/crates/ruborist/src/service/manifest.rs b/crates/ruborist/src/service/manifest.rs index ad682d6424..e31adc43b1 100644 --- a/crates/ruborist/src/service/manifest.rs +++ b/crates/ruborist/src/service/manifest.rs @@ -10,7 +10,7 @@ use tokio_retry::RetryIf; use super::fetch::{ FetchError, classify_reqwest_error, classify_status, is_retryable, retry_strategy, }; -use super::http::get_client; +use super::http::{parse_trace_enabled, pick_client, record_http_interval, record_parse_interval}; use crate::model::manifest::{CoreVersionManifest, FullManifest}; /// Result of a full manifest fetch with ETag support. @@ -60,7 +60,7 @@ pub async fn fetch_full_manifest(opts: FetchManifestOptions<'_>) -> Result) -> Result) -> Result = response .bytes() .await .map_err(|e| FetchError::Permanent(anyhow!("Response read error: {e}")))? - .to_vec(); - // Save raw bytes before simd_json mutates the parse buffer + .into(); + record_http_interval(send_start, std::time::Instant::now()); + + // Inline parse on the async worker — simd_json is fast + // (1-5ms CPU per manifest) and the worker-pool preload + // architecture distributes 80+ concurrent fetches across + // tokio's worker_threads (= num_cpus). Earlier + // spawn_blocking offload backed up the 4-thread blocking + // pool: parse diag at cap=160 showed queue p95=200ms + // sum=70-89s, adding ~26ms per request — accounted for + // the entire ruborist-vs-standalone per-req gap (55ms vs + // 28ms). Inline parse trades a brief async-worker stall + // for zero queue wait, and worker-pool's independent + // task scheduling means one stalled worker doesn't + // starve the others. + let traced = parse_trace_enabled(); + let queued_at = traced.then(std::time::Instant::now); + let exec_start = queued_at.map(|_| std::time::Instant::now()); let mut parse_buf = raw_bytes.clone(); let mut manifest: FullManifest = simd_json::serde::from_slice(&mut parse_buf) .map_err(|e| FetchError::Permanent(anyhow!("JSON parse error: {e}")))?; manifest.raw = std::sync::Arc::from(raw_bytes); + if let (Some(q), Some(s)) = (queued_at, exec_start) { + record_parse_interval(q, s, std::time::Instant::now()); + } Ok(FetchManifestResult::Ok(manifest, new_etag)) } else { @@ -173,7 +196,8 @@ pub async fn fetch_version_manifest( || { let url = url.clone(); async move { - let response = get_client() + let send_start = std::time::Instant::now(); + let response = pick_client() .map_err(FetchError::Permanent)? .get(&url) .header("Accept", accept) @@ -182,13 +206,25 @@ pub async fn fetch_version_manifest( .map_err(classify_reqwest_error)?; if response.status().is_success() { - let mut bytes = response + let bytes: Vec = response .bytes() .await .map_err(|e| FetchError::Permanent(anyhow!("Response read error: {e}")))? - .to_vec(); - simd_json::serde::from_slice(&mut bytes) - .map_err(|e| FetchError::Permanent(anyhow!("JSON parse error: {e}"))) + .into(); + record_http_interval(send_start, std::time::Instant::now()); + // Inline parse — see fetch_full_manifest above for the + // rationale (blocking-pool queue saturation under + // worker-pool preload concurrency). + let traced = parse_trace_enabled(); + let queued_at = traced.then(std::time::Instant::now); + let exec_start = queued_at.map(|_| std::time::Instant::now()); + let mut buf = bytes; + let result = simd_json::serde::from_slice::(&mut buf) + .map_err(|e| FetchError::Permanent(anyhow!("JSON parse error: {e}"))); + if let (Some(q), Some(s)) = (queued_at, exec_start) { + record_parse_interval(q, s, std::time::Instant::now()); + } + result } else { Err(classify_status(response.status(), &url)) } diff --git a/crates/ruborist/src/service/registry.rs b/crates/ruborist/src/service/registry.rs index 772cac3f79..81c79e0897 100644 --- a/crates/ruborist/src/service/registry.rs +++ b/crates/ruborist/src/service/registry.rs @@ -44,6 +44,8 @@ use crate::model::manifest::{CoreVersionManifest, FullManifest}; use crate::resolver::semver::normalize_spec; use crate::resolver::version::resolve_target_version; use crate::traits::registry::{RegistryClient, RegistryError, ResolvedPackage, is_npm_registry}; +#[cfg(not(target_arch = "wasm32"))] +use crate::util::oncemap::OnceMap; /// Unified registry client that works on both native and WASM. /// @@ -68,6 +70,14 @@ pub struct UnifiedRegistry { registry_url: String, cache: Arc, supports_semver: bool, + /// Dedupes concurrent `resolve_full_manifest` fetches for the same + /// package name. First caller hits the network and stores the result; + /// other callers wait on `Notify` and read the shared `Arc`. Built on + /// `DashMap` + `tokio::sync::Notify` so the fast path (cache hit) is + /// lock-free, avoiding the serialisation the previous per-name + /// `tokio::sync::Mutex<()>` gate imposed on the hot dispatch path. + #[cfg(not(target_arch = "wasm32"))] + inflight: Arc>, } /// Builder for `UnifiedRegistry`. @@ -137,6 +147,8 @@ impl UnifiedRegistryBuilder { registry_url, cache, supports_semver, + #[cfg(not(target_arch = "wasm32"))] + inflight: Arc::new(OnceMap::new()), } } } @@ -153,6 +165,8 @@ impl Clone for UnifiedRegistry { registry_url: self.registry_url.clone(), cache: Arc::clone(&self.cache), supports_semver: self.supports_semver, + #[cfg(not(target_arch = "wasm32"))] + inflight: Arc::clone(&self.inflight), } } } @@ -162,10 +176,16 @@ impl Clone for UnifiedRegistry { /// Separates the 200 (full data) and 304 (use cache) cases at the type level, /// so callers can pattern-match instead of string-matching error messages. /// Transient return value, immediately destructured — Box not needed. -#[allow(clippy::large_enum_variant)] +/// +/// `Clone` is required so multiple `resolve_full_manifest` callers that +/// coalesce through `OnceMap` can each take an owned copy of the shared +/// `Arc`. `Arc` keeps that clone an +/// atomic-bump rather than a deep copy of the per-version `OwnedValue` +/// HashMap (~100-500 entries per package). +#[derive(Clone)] enum FullManifestResult { /// Fresh manifest fetched from the network (HTTP 200). - Full(FullManifest), + Full(Arc), /// ETag matched, versions cache is valid (HTTP 304). /// Caller should resolve from the in-memory versions cache and /// fetch individual version manifests as needed. @@ -202,17 +222,52 @@ impl UnifiedRegistry { /// 4. On 304: use disk cache data /// 5. On 200: update memory + disk cache async fn resolve_full_manifest(&self, name: &str) -> Result { - // 1. Check memory cache first + // Fast path: memory cache hit — lock-free read from parking_lot::RwLock. if let Some(manifest) = self.cache.get_full_manifest(name) { tracing::debug!("Memory cache hit for full manifest: {}", name); return Ok(FullManifestResult::Full(manifest)); } - // 2. Load etag from disk cache (if available) + // Coalesce concurrent callers for the same name via OnceMap. + // First caller runs the fetch closure; others await the shared + // result on the OnceMap's `Notify` and clone the cached value. + let shared = self + .inflight + .get_or_init(name.to_string(), || async { + self.fetch_full_manifest_network(name).await.ok() + }) + .await; + + match shared { + Some(arc) => Ok((*arc).clone()), + None => { + // OnceMap clears the key on None, so the next caller + // retries the fetch. Retry once here with a fresh error + // so we surface a useful message to this caller. + self.fetch_full_manifest_network(name).await + } + } + } + + /// Perform the actual network fetch + cache update. Separated from + /// `resolve_full_manifest` so the OnceMap closure can invoke it + /// without re-entering the dedup layer. + async fn fetch_full_manifest_network( + &self, + name: &str, + ) -> Result { + // Disk ETag probe. The `PackageCache::get_versions_from_disk` + // call first consults a bulk-readdir index built lazily on + // first access — cold runs with an empty (or nonexistent) + // cache_dir short-circuit without per-package syscalls, which + // was the cold-path regression the earlier temporary removal + // (46cb8031) was meant to avoid. Warm runs pay the read + + // JSON parse once per previously-cached manifest and reuse + // its ETag to get a 1-packet `304 Not Modified` instead of + // re-downloading the full manifest body. let disk_versions = self.cache.get_versions_from_disk(name).await; let etag = disk_versions.as_ref().and_then(|v| v.etag.clone()); - // 3. Fetch from network with ETag for validation match manifest::fetch_full_manifest(manifest::FetchManifestOptions { registry_url: &self.registry_url, name, @@ -223,39 +278,35 @@ impl UnifiedRegistry { .map_err(RegistryError)? { manifest::FetchManifestResult::Ok(manifest, new_etag) => { - // 4. Cache full manifest in memory - self.cache - .set_full_manifest(name.to_string(), manifest.clone()); - - // 5. Extract and cache versions info (lightweight) let versions_info = VersionsInfo { versions: Versions { - version_list: manifest.versions.clone(), + version_list: manifest.versions.keys.clone(), dist_tags: manifest.dist_tags.clone(), }, etag: new_etag.clone(), last_updated: current_timestamp_secs(), }; + let manifest = Arc::new(manifest); + self.cache + .set_full_manifest(name.to_string(), Arc::clone(&manifest)); self.cache .set_versions(name.to_string(), versions_info.clone()); - - // 6. Write versions to disk (non-blocking for native, blocking for WASM) - self.cache.set_versions_to_disk(name, &versions_info).await; + // Fire-and-forget disk cache write. + self.cache.set_versions_to_disk(name, &versions_info); Ok(FullManifestResult::Full(manifest)) } manifest::FetchManifestResult::NotModified => { - // 304 Not Modified - use disk cache versions info tracing::debug!("ETag cache hit (304) for: {}", name); if let Some(versions_info) = disk_versions { - // Cache versions in memory for later use self.cache .set_versions(name.to_string(), versions_info.clone()); - Ok(FullManifestResult::NotModified) } else { - // Disk cache corrupted, fetch fresh (without etag) + // Disk cache disappeared between index build and now + // (mid-run eviction or concurrent cleanup). Fall + // back to a fresh fetch without etag. let (manifest, new_etag) = manifest::fetch_full_manifest_fresh( &self.registry_url, name, @@ -264,20 +315,20 @@ impl UnifiedRegistry { .await .map_err(RegistryError)?; - self.cache - .set_full_manifest(name.to_string(), manifest.clone()); - let versions_info = VersionsInfo { versions: Versions { - version_list: manifest.versions.clone(), + version_list: manifest.versions.keys.clone(), dist_tags: manifest.dist_tags.clone(), }, etag: new_etag.clone(), last_updated: current_timestamp_secs(), }; + let manifest = Arc::new(manifest); + self.cache + .set_full_manifest(name.to_string(), Arc::clone(&manifest)); self.cache .set_versions(name.to_string(), versions_info.clone()); - self.cache.set_versions_to_disk(name, &versions_info).await; + self.cache.set_versions_to_disk(name, &versions_info); Ok(FullManifestResult::Full(manifest)) } @@ -341,8 +392,7 @@ impl UnifiedRegistry { // 5. Write to disk cache (only for non-semver registries) if !self.supports_semver { self.cache - .set_version_manifest_to_disk(name, spec, &manifest) - .await; + .set_version_manifest_to_disk(name, spec, manifest.clone()); } Ok(manifest) @@ -357,7 +407,10 @@ impl RegistryClient for UnifiedRegistry { } fn get_cached_full_manifest(&self, name: &str) -> Option { - self.cache.get_full_manifest(name) + // Trait still returns owned for backward compat with non-resolver + // callers (notably `ut view`). Resolver hot paths read the + // `Arc` directly via `cache.get_full_manifest`. + self.cache.get_full_manifest(name).map(|arc| (*arc).clone()) } fn get_cached_versions(&self, name: &str) -> Option { @@ -376,7 +429,7 @@ impl RegistryClient for UnifiedRegistry { async fn fetch_full_manifest(&self, name: &str) -> Result { match self.resolve_full_manifest(name).await? { - FullManifestResult::Full(manifest) => Ok(manifest), + FullManifestResult::Full(manifest) => Ok((*manifest).clone()), FullManifestResult::NotModified => { // 304 in trait context: caller doesn't have versions cache access, // so we return an error indicating the manifest is unchanged. @@ -435,8 +488,7 @@ impl RegistryClient for UnifiedRegistry { // 5. Write to disk cache (only for non-semver registries) if !self.supports_semver { self.cache - .set_version_manifest_to_disk(name, spec, &manifest) - .await; + .set_version_manifest_to_disk(name, spec, manifest.clone()); } Ok(manifest) @@ -481,18 +533,19 @@ impl RegistryClient for UnifiedRegistry { fetch_name, fetch_spec ); - tracing::debug!( - "Using cached full manifest for {}@{}", - fetch_name, - fetch_spec - ); - let version_list: Vec = full_manifest.versions.clone(); - let resolved_version = - resolve_target_version(&full_manifest.dist_tags, &version_list, &fetch_spec) - .map_err(|e| RegistryError(anyhow!("{}@{}: {}", name, spec, e)))?; + // Borrow `keys` directly — `resolve_target_version` only needs + // `&[String]`. The previous `keys.clone()` rebuilt a 100-500 + // entry `Vec` per cache hit (≈1800 hits during a cold + // ant-design preload), bloating per-future allocator pressure + // by ~360k String allocs on shared resolver threads. + let resolved_version = resolve_target_version( + &full_manifest.dist_tags, + &full_manifest.versions.keys, + &fetch_spec, + ) + .map_err(|e| RegistryError(anyhow!("{}@{}: {}", name, spec, e)))?; let version_manifest = full_manifest .get_core_version(&resolved_version) - .map(Arc::new) .ok_or_else(|| { RegistryError(anyhow!( "Version {} not found in manifest for {}", @@ -508,9 +561,11 @@ impl RegistryClient for UnifiedRegistry { ); // Write to disk cache for non-semver registries if !self.supports_semver { - self.cache - .set_version_manifest_to_disk(&fetch_name, &resolved_version, &version_manifest) - .await; + self.cache.set_version_manifest_to_disk( + &fetch_name, + &resolved_version, + version_manifest.clone(), + ); } return Ok(ResolvedPackage { name: name.to_string(), @@ -586,7 +641,7 @@ impl RegistryClient for UnifiedRegistry { let resolve_result = match self.resolve_full_manifest(&fetch_name).await? { FullManifestResult::Full(full_manifest) => { // Got full manifest, resolve from it - let version_list: Vec = full_manifest.versions.clone(); + let version_list: Vec = full_manifest.versions.keys.clone(); if version_list.is_empty() { return Err(RegistryError(anyhow!( @@ -604,7 +659,6 @@ impl RegistryClient for UnifiedRegistry { let version_manifest = full_manifest .get_core_version(&resolved_version) - .map(Arc::new) .ok_or_else(|| { RegistryError(anyhow!( "Version {} not found in manifest for {}", @@ -650,9 +704,11 @@ impl RegistryClient for UnifiedRegistry { // Write to disk cache (only for non-semver registries) if !self.supports_semver { - self.cache - .set_version_manifest_to_disk(&fetch_name, &resolved_version, &version_manifest) - .await; + self.cache.set_version_manifest_to_disk( + &fetch_name, + &resolved_version, + version_manifest.clone(), + ); } Ok(ResolvedPackage { diff --git a/crates/ruborist/src/traits/registry.rs b/crates/ruborist/src/traits/registry.rs index 0760ff9b19..355ae6110d 100644 --- a/crates/ruborist/src/traits/registry.rs +++ b/crates/ruborist/src/traits/registry.rs @@ -140,7 +140,7 @@ pub trait RegistryClient { fn fetch_full_manifest( &self, name: &str, - ) -> impl Future>; + ) -> impl Future> + Send; /// Fetch specific version manifest from registry. /// @@ -153,26 +153,26 @@ pub trait RegistryClient { &self, name: &str, spec: &str, - ) -> impl Future, Self::Error>> { + ) -> impl Future, Self::Error>> + Send + where + Self: Sync, + { async move { let manifest = self.fetch_full_manifest(name).await?; - let version_list: Vec = manifest.versions.clone(); + let version_list: Vec = manifest.versions.keys.clone(); // Resolve version using shared logic let resolved_version = resolve_target_version(&manifest.dist_tags, &version_list, spec) .map_err(|e| RegistryError(anyhow::anyhow!("{}@{}: {}", name, spec, e)))?; - manifest - .get_core_version(&resolved_version) - .map(Arc::new) - .ok_or_else(|| { - RegistryError(anyhow::anyhow!( - "Version {} not found in manifest for {}", - resolved_version, - name - )) - .into() - }) + manifest.get_core_version(&resolved_version).ok_or_else(|| { + RegistryError(anyhow::anyhow!( + "Version {} not found in manifest for {}", + resolved_version, + name + )) + .into() + }) } } @@ -193,7 +193,10 @@ pub trait RegistryClient { &self, name: &str, spec: &str, - ) -> impl Future> { + ) -> impl Future> + Send + where + Self: Sync, + { async move { // Normalize spec (handles npm: alias and workspace: prefix) let (fetch_name, fetch_spec) = normalize_spec(name, spec); @@ -226,7 +229,7 @@ pub trait RegistryClient { fetch_spec ); let full_manifest = self.fetch_full_manifest(&fetch_name).await?; - let version_list: Vec = full_manifest.versions.clone(); + let version_list: Vec = full_manifest.versions.keys.clone(); if version_list.is_empty() { return Err(RegistryError(anyhow::anyhow!( @@ -242,7 +245,6 @@ pub trait RegistryClient { let version_manifest = full_manifest .get_core_version(&resolved_version) - .map(Arc::new) .ok_or_else(|| { RegistryError(anyhow::anyhow!( "Version {} not found in manifest for {}", @@ -271,7 +273,7 @@ pub trait RegistryClient { async move { let manifest = self.fetch_full_manifest(name).await?; Ok(VersionsInfo { - version_list: manifest.versions.clone(), + version_list: manifest.versions.keys.clone(), dist_tags: manifest.dist_tags, }) } @@ -320,6 +322,7 @@ pub mod mock { use super::*; /// Internal package data for mock registry. + #[derive(Clone)] struct MockPackage { name: String, dist_tags: HashMap, @@ -327,6 +330,7 @@ pub mod mock { } /// Mock registry client that returns predefined packages. + #[derive(Clone)] pub struct MockRegistryClient { packages: HashMap, } @@ -398,21 +402,21 @@ pub mod mock { .get(name) .ok_or_else(|| MockError(format!("Package not found: {}", name)))?; - // Build JSON and serialize to raw bytes for on-demand extraction + // Build JSON matching the registry wire format and round-trip + // it through the real `FullManifest` deserializer so tests + // exercise the same `Versions` parse path as production. let json = serde_json::json!({ "name": &pkg.name, "dist-tags": &pkg.dist_tags, + "_id": &pkg.name, "versions": &pkg.versions, }); let raw = serde_json::to_vec(&json).expect("mock JSON serialization"); - - Ok(FullManifest { - name: pkg.name.clone(), - dist_tags: pkg.dist_tags.clone(), - versions: pkg.versions.keys().cloned().collect(), - raw: Arc::from(raw), - ..Default::default() - }) + let mut parse_buf = raw.clone(); + let mut manifest: FullManifest = simd_json::serde::from_slice(&mut parse_buf) + .map_err(|e| MockError(format!("mock manifest parse error: {e}")))?; + manifest.raw = Arc::from(raw); + Ok(manifest) } } } diff --git a/crates/ruborist/src/util/mod.rs b/crates/ruborist/src/util/mod.rs new file mode 100644 index 0000000000..b9a2b874c4 --- /dev/null +++ b/crates/ruborist/src/util/mod.rs @@ -0,0 +1,13 @@ +//! Shared utilities usable across ruborist and its downstream crates. + +// OnceMap uses `dashmap` + `tokio::sync::Notify`. Both nominally compile +// for wasm, but the only consumers so far (pm-side downloader / cloner, +// and the ruborist preload dedup) are native-only paths. Gate it to keep +// the wasm surface minimal until a wasm consumer actually needs it. +#[cfg(not(target_arch = "wasm32"))] +pub mod oncemap; + +// Previously these lived in an inline `pub mod util { ... }` re-export +// block in `lib.rs`; keep the same public path to avoid breaking +// downstream consumers. +pub use crate::model::util::{PackageNameStr, parse_package_spec, read_package_json}; diff --git a/crates/pm/src/util/oncemap.rs b/crates/ruborist/src/util/oncemap.rs similarity index 100% rename from crates/pm/src/util/oncemap.rs rename to crates/ruborist/src/util/oncemap.rs diff --git a/e2e/pm-bench-pcap.sh b/e2e/pm-bench-pcap.sh new file mode 100755 index 0000000000..c8e869d78e --- /dev/null +++ b/e2e/pm-bench-pcap.sh @@ -0,0 +1,83 @@ +#!/bin/bash +# One-off network trace capture: runs a single cold p1_resolve per PM with +# tcpdump active, so we can inspect connection topology (IPs, concurrency, +# timing) on CI after local evidence was inconclusive. +# +# Outputs: +# $PCAP_DIR/dns.txt — A records for the registry host +# $PCAP_DIR/{bun,utoo}.pcap — tcpdump capture of the resolve phase +# $PCAP_DIR/{bun,utoo}.log — stdout/stderr of the run +set -eo pipefail + +PROJECT=${PROJECT:-ant-design} +REGISTRY=${REGISTRY:-https://registry.npmjs.org} +BENCH_DIR=${BENCH_DIR:-/tmp/pm-bench} +PCAP_DIR=${PCAP_DIR:-/tmp/pm-bench-pcap} +UTOO_CACHE=/tmp/utoo-pcap-cache +BUN_CACHE=/tmp/bun-pcap-cache +export BUN_INSTALL_CACHE_DIR="$BUN_CACHE" + +mkdir -p "$PCAP_DIR" + +# Project must already be cloned by pm-bench-phases.sh (this script is meant +# to run after it, reusing the clone). +PROJECT_DIR="$BENCH_DIR/$PROJECT" +if [ ! -d "$PROJECT_DIR" ]; then + echo "missing $PROJECT_DIR — run pm-bench-phases.sh first" >&2 + exit 1 +fi + +# Extract hostname (strip scheme + any path) for DNS lookup. +HOST=$(echo "$REGISTRY" | sed -E 's#^https?://([^/]+).*#\1#') +echo "=== DNS records for $HOST ===" +{ + echo "# getent hosts" + getent hosts "$HOST" || true + echo + echo "# dig +short A" + dig +short A "$HOST" || true + echo + echo "# dig +short AAAA" + dig +short AAAA "$HOST" || true +} | tee "$PCAP_DIR/dns.txt" + +IFACE=$(ip route | awk '/default/ {print $5; exit}') +echo "capturing on interface: $IFACE" + +capture_one() { + local name=$1 + shift + local pcap="$PCAP_DIR/$name.pcap" + local log="$PCAP_DIR/$name.log" + + cd "$PROJECT_DIR" + # Cold: nothing reused. + rm -f package-lock.json bun.lock + rm -rf "$UTOO_CACHE" "$BUN_CACHE" node_modules + + echo "=== capturing $name ===" + # tcpdump as root; ubuntu-latest runners allow sudo without password. + sudo tcpdump -s 0 -i "$IFACE" -w "$pcap" 'tcp port 443' >/dev/null 2>&1 & + local tcpdump_pid=$! + # Let tcpdump bind before the workload starts. + sleep 1 + + /usr/bin/time -v "$@" >"$log" 2>&1 || echo " run exited with $?" + + sudo kill "$tcpdump_pid" 2>/dev/null || true + wait "$tcpdump_pid" 2>/dev/null || true + + # Make pcap readable by the later upload-artifact step (tcpdump writes + # as root). + sudo chmod 644 "$pcap" + echo " → $pcap ($(wc -c <"$pcap") bytes), log: $log" +} + +capture_one utoo \ + utoo deps --registry="$REGISTRY" --cache-dir="$UTOO_CACHE" + +capture_one bun \ + bun install --lockfile-only --registry="$REGISTRY" + +echo "done. files:" +ls -lh "$PCAP_DIR" diff --git a/e2e/pm-bench-phases.sh b/e2e/pm-bench-phases.sh new file mode 100755 index 0000000000..3f39b7c58a --- /dev/null +++ b/e2e/pm-bench-phases.sh @@ -0,0 +1,412 @@ +#!/bin/bash +# Phase-isolated cold-install bench: resolve vs cold-install vs warm-link. +# utoo vs bun only (they're the PMs with a true --lockfile-only mode). +set -eo pipefail + +# --- config --- +PROJECT=${PROJECT:-ant-design} +REGISTRY=${REGISTRY:-https://registry.npmjs.org} +RUNS=${BENCH_RUNS:-3} +IFS=',' read -ra PACKAGE_MANAGERS <<< "${PM_LIST:-utoo,bun}" + +BENCH_DIR=${BENCH_DIR:-/tmp/pm-bench} +RESULTS_DIR=${RESULTS_DIR:-/tmp/pm-bench-results} +mkdir -p "$BENCH_DIR" "$RESULTS_DIR" + +# Explicit bench-scoped cache dirs so rm -rf is unambiguous and we don't +# depend on what each PM considers the default (which varies by HOME, OS +# image, and previously-set user config). +UTOO_CACHE="${UTOO_CACHE:-/tmp/utoo-bench-cache}" +BUN_CACHE="${BUN_CACHE:-/tmp/bun-bench-cache}" +export BUN_INSTALL_CACHE_DIR="$BUN_CACHE" + +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +CYAN='\033[0;36m' +RED='\033[0;31m' +NC='\033[0m' + +banner() { echo -e "${YELLOW}=== $* ===${NC}"; } + +# --- metrics wrapper --- +# Keeps the hot path fast: only /usr/bin/time + counter snapshots inside the +# measured window. Disk-footprint measurement (du on node_modules / cache) +# happens once per phase AFTER hyperfine exits — see capture_footprint in +# run_phase — so 1-2s of `du` traversal doesn't bleed into wall-clock. +METRICS_WRAPPER="$RESULTS_DIR/metrics_wrapper.sh" +cat > "$METRICS_WRAPPER" <<'METRICS_EOF' +#!/bin/bash +METRICS_FILE="$1"; shift +TIME_TMP=$(mktemp) + +# --- snapshot network BEFORE the command (cheap: one read of /proc/net/dev) --- +snap_net_rx=0; snap_net_tx=0 +if [ -r /proc/net/dev ]; then + read snap_net_rx snap_net_tx < <(awk '/:/ && $1 !~ /^lo:/ {rx += $2; tx += $10} END {print rx+0, tx+0}' /proc/net/dev) +fi + +if [[ "$(uname)" == "Darwin" ]]; then + /usr/bin/time -l "$@" 2>"$TIME_TMP" + EXIT_CODE=$? + RSS=$(awk '/maximum resident set size/ {print $1}' "$TIME_TMP") + PG_MAJOR=$(awk '/page faults/ {print $1}' "$TIME_TMP") + PG_MINOR=$(awk '/page reclaims/ {print $1}' "$TIME_TMP") + VOL_CTX=$(awk '/ voluntary context switches/ {print $1}' "$TIME_TMP") + INVOL_CTX=$(awk '/involuntary context switches/ {print $1}' "$TIME_TMP") + USER_S=$(awk '/real/ && /user/ && /sys/ {print $3}' "$TIME_TMP") + SYS_S=$(awk '/real/ && /user/ && /sys/ {print $5}' "$TIME_TMP") +else + /usr/bin/time -v "$@" 2>"$TIME_TMP" + EXIT_CODE=$? + RSS_KB=$(awk '/Maximum resident set size/ {print $NF}' "$TIME_TMP") + RSS=$(( ${RSS_KB:-0} * 1024 )) + PG_MAJOR=$(awk '/Major \(requiring I\/O\) page faults/ {print $NF}' "$TIME_TMP") + PG_MINOR=$(awk '/Minor \(reclaiming a frame\) page faults/ {print $NF}' "$TIME_TMP") + VOL_CTX=$(awk '/Voluntary context switches/ {print $NF}' "$TIME_TMP") + INVOL_CTX=$(awk '/Involuntary context switches/ {print $NF}' "$TIME_TMP") + USER_S=$(awk -F': ' '/User time \(seconds\)/ {print $2}' "$TIME_TMP") + SYS_S=$(awk -F': ' '/System time \(seconds\)/ {print $2}' "$TIME_TMP") +fi + +# --- snapshot network AFTER, compute delta --- +net_rx=0; net_tx=0 +if [ -r /proc/net/dev ]; then + read cur_net_rx cur_net_tx < <(awk '/:/ && $1 !~ /^lo:/ {rx += $2; tx += $10} END {print rx+0, tx+0}' /proc/net/dev) + net_rx=$(( cur_net_rx - snap_net_rx )) + net_tx=$(( cur_net_tx - snap_net_tx )) +fi + +printf '{"rss":%d,"user_s":%s,"sys_s":%s,"page_major":%d,"page_minor":%d,"vol_ctx":%d,"invol_ctx":%d,"net_rx":%d,"net_tx":%d}\n' \ + "${RSS:-0}" "${USER_S:-0}" "${SYS_S:-0}" "${PG_MAJOR:-0}" "${PG_MINOR:-0}" \ + "${VOL_CTX:-0}" "${INVOL_CTX:-0}" \ + "${net_rx:-0}" "${net_tx:-0}" >> "$METRICS_FILE" +rm -f "$TIME_TMP" +exit $EXIT_CODE +METRICS_EOF +chmod +x "$METRICS_WRAPPER" + +# Post-phase footprint helper: records final on-disk size of the paths each +# phase should have touched. Called ONCE after hyperfine finishes, so the +# (slow) du traversal is not part of any timed window. +du_bytes() { + for p in "$@"; do + if [ -e "$p" ]; then + if [[ "$(uname)" == "Darwin" ]]; then + du -sk "$p" 2>/dev/null | awk '{print $1 * 1024}' + else + du -sb "$p" 2>/dev/null | awk '{print $1}' + fi + else + echo 0 + fi + done | awk '{s += $1} END {print s+0}' +} + +capture_footprint() { + local phase=$1 pm=$2 out=$3 + local cache + case "$pm" in utoo) cache=$UTOO_CACHE ;; bun) cache=$BUN_CACHE ;; esac + printf '{"cache":%d,"node_modules":%d,"lockfile":%d}\n' \ + "$(du_bytes "$cache")" \ + "$(du_bytes "$PROJECT_DIR/node_modules")" \ + "$(du_bytes "$PROJECT_DIR/package-lock.json" "$PROJECT_DIR/bun.lock")" \ + > "$out" +} + +# --- project --- +banner "Preparing $PROJECT" +cd "$BENCH_DIR" +if [ ! -d "$PROJECT" ]; then + git clone --depth=1 "https://github.com/ant-design/${PROJECT}.git" "$PROJECT" +fi +PROJECT_DIR="$BENCH_DIR/$PROJECT" + +# --- prepare script builder --- +# Emit a self-contained reset script that hyperfine's --prepare can re-exec. +# The script always prints what it deleted so the CI log proves prepare ran. +write_prepare() { + local path=$1 phase=$2 pm=$3 + local cache + case "$pm" in utoo) cache=$UTOO_CACHE ;; bun) cache=$BUN_CACHE ;; esac + + cat > "$path" </dev/null || true + +EOF + + # Use explicit bench-scoped cache paths and rm -rf them directly — previously + # `utoo clean` / `bun pm cache rm` didn't actually empty the cache on the CI + # runner, leading to cache-hit runs masquerading as cold installs. + case "$phase" in + p0_*) + # Phase 0: full cold install — nothing reused. Lockfile + all caches wiped. + cat >> "$path" <> "$path" <> "$path" <> "$path" <> "$path" <> "$path" < "$RESULTS_DIR/seed_${phase}_${pm}.log" 2>&1 + fi + ;; + p3_*:bun|p4_*:bun) + if [ ! -f bun.lock ]; then + echo -e " ${CYAN}seed: running \`bun install --lockfile-only\` to generate bun.lock${NC}" + rm -f package-lock.json + bun install --lockfile-only --registry="$REGISTRY" > "$RESULTS_DIR/seed_${phase}_${pm}.log" 2>&1 + fi + ;; + esac + # Phase 4 also needs a pre-warmed cache. + if [[ "$phase" == p4_* ]]; then + local cache + case "$pm" in utoo) cache=$UTOO_CACHE ;; bun) cache=$BUN_CACHE ;; esac + if [ ! -d "$cache" ] || [ -z "$(ls -A "$cache" 2>/dev/null)" ]; then + echo -e " ${CYAN}seed: warming $pm cache via full install${NC}" + rm -rf node_modules + case "$pm" in + utoo) utoo install --ignore-scripts --registry="$REGISTRY" --cache-dir="$UTOO_CACHE" > "$RESULTS_DIR/seed_warmup_${pm}.log" 2>&1 ;; + bun) bun install --ignore-scripts --registry="$REGISTRY" > "$RESULTS_DIR/seed_warmup_${pm}.log" 2>&1 ;; + esac + rm -rf node_modules + fi + fi +} + +install_cmd() { + case "$1" in + utoo) echo "utoo install --ignore-scripts --registry=$REGISTRY --cache-dir=$UTOO_CACHE" ;; + bun) echo "bun install --ignore-scripts --registry=$REGISTRY" ;; + esac +} + +resolve_cmd() { + case "$1" in + utoo) echo "utoo deps --registry=$REGISTRY --cache-dir=$UTOO_CACHE" ;; + bun) echo "bun install --lockfile-only --registry=$REGISTRY" ;; + esac +} + +run_phase() { + local phase=$1 pm=$2 cmd=$3 + local json="$RESULTS_DIR/${PROJECT}_${phase}_${pm}.json" + local metrics="$RESULTS_DIR/${PROJECT}_${phase}_${pm}_metrics.jsonl" + local prep_script="$RESULTS_DIR/prep_${phase}_${pm}.sh" + local cmd_script="$RESULTS_DIR/cmd_${phase}_${pm}.sh" + + seed_for_phase "$phase" "$pm" + write_prepare "$prep_script" "$phase" "$pm" + + # Reset metrics file; wrap the bench command so /usr/bin/time can measure it. + # The wrapper reads PROJECT_DIR / UTOO_CACHE / BUN_CACHE from env to scope du. + : > "$metrics" + printf 'set -eo pipefail\ncd %s\n%s\n' "$PROJECT_DIR" "$cmd" > "$cmd_script" + + echo -e " ${CYAN}$pm${NC} · $phase" + export PROJECT_DIR UTOO_CACHE BUN_CACHE + if ! hyperfine \ + --runs "$RUNS" \ + --prepare "bash $prep_script" \ + --export-json "$json" \ + --show-output \ + -n "${pm}-${phase}" \ + "bash $METRICS_WRAPPER $metrics bash $cmd_script"; then + echo -e " ${RED}$pm $phase failed${NC}" + fi + + # After the timed runs: capture the state left on disk by the last run. + # Done outside hyperfine so du's traversal doesn't pollute wall-clock. + capture_footprint "$phase" "$pm" "$RESULTS_DIR/${PROJECT}_${phase}_${pm}_footprint.json" +} + +# === PHASE 0: full cold install (clean slate + full install) === +# Matches the end-to-end user scenario: no lockfile, no cache, no node_modules. +# Directly comparable to `bun install` / `utoo install` on a freshly cloned repo. +banner "Phase 0 · full cold install (lockfile + cache + node_modules all wiped)" +for pm in "${PACKAGE_MANAGERS[@]}"; do + run_phase "p0_full_cold" "$pm" "$(install_cmd "$pm")" +done + +# === PHASE 1: resolve only (clean slate) === +banner "Phase 1 · resolve (lockfile only, cold cache)" +for pm in "${PACKAGE_MANAGERS[@]}"; do + run_phase "p1_resolve" "$pm" "$(resolve_cmd "$pm")" +done + +# === PHASE 3: cold install (lockfile exists, cache empty) === +banner "Phase 3 · cold install (lockfile present, empty cache, empty node_modules)" +for pm in "${PACKAGE_MANAGERS[@]}"; do + run_phase "p3_cold_install" "$pm" "$(install_cmd "$pm")" +done + +# === PHASE 4: warm link (cache populated, lockfile exists) === +banner "Phase 4 · warm link (lockfile present, populated cache, empty node_modules)" +for pm in "${PACKAGE_MANAGERS[@]}"; do + run_phase "p4_warm_link" "$pm" "$(install_cmd "$pm")" +done + +# === SUMMARY === +banner "Summary" +RESULTS_DIR="$RESULTS_DIR" node -e " + const fs = require('fs'), path = require('path'); + const dir = process.env.RESULTS_DIR; + const order = ['p0_full_cold', 'p1_resolve', 'p3_cold_install', 'p4_warm_link']; + const timing = {}; // phase -> pm -> {mean,stddev,min,max} + const metrics = {}; // phase -> pm -> averaged resource fields + + const parseKey = (file, suffix) => { + const base = file.replace(suffix, ''); + for (const p of order) { + const idx = base.indexOf('_' + p + '_'); + if (idx === -1) continue; + const pm = base.slice(idx + p.length + 2); + return { phase: p, pm }; + } + return null; + }; + + for (const f of fs.readdirSync(dir).filter(x => x.endsWith('.json') && !x.endsWith('_metrics.jsonl') && !x.endsWith('_footprint.json'))) { + const key = parseKey(f, '.json'); + if (!key) continue; + let data; try { data = JSON.parse(fs.readFileSync(path.join(dir, f), 'utf8')); } catch (_) { continue; } + const r = data.results[0]; + if (!r) continue; + (timing[key.phase] ??= {})[key.pm] = { mean: r.mean, stddev: r.stddev, min: r.min, max: r.max }; + } + + const metricKeys = ['rss','user_s','sys_s','page_major','page_minor','vol_ctx','invol_ctx','net_rx','net_tx']; + for (const f of fs.readdirSync(dir).filter(x => x.endsWith('_metrics.jsonl'))) { + const key = parseKey(f, '_metrics.jsonl'); + if (!key) continue; + const rows = []; + for (const l of fs.readFileSync(path.join(dir, f), 'utf8').trim().split('\n').filter(Boolean)) { + try { rows.push(JSON.parse(l)); } catch (_) {} + } + if (rows.length === 0) continue; + const avg = {}; + for (const k of metricKeys) { + avg[k] = rows.reduce((s,e) => s + Number(e[k] || 0), 0) / rows.length; + } + (metrics[key.phase] ??= {})[key.pm] = avg; + } + + // Footprint is a post-phase single sample (not averaged): final on-disk size + // of the paths the phase should have touched. + for (const f of fs.readdirSync(dir).filter(x => x.endsWith('_footprint.json'))) { + const key = parseKey(f, '_footprint.json'); + if (!key) continue; + let data; try { data = JSON.parse(fs.readFileSync(path.join(dir, f), 'utf8')); } catch (_) { continue; } + const slot = (metrics[key.phase] ??= {})[key.pm] ??= {}; + slot.foot_cache = data.cache || 0; + slot.foot_nm = data.node_modules || 0; + slot.foot_lock = data.lockfile || 0; + } + + const pad = (s, n) => String(s).padEnd(n); + const padR = (s, n) => String(s).padStart(n); + const fmtB = b => b >= 1<<30 ? (b/(1<<30)).toFixed(2)+'G' : b >= 1<<20 ? (b/(1<<20)).toFixed(0)+'M' : b >= 1<<10 ? (b/(1<<10)).toFixed(0)+'K' : Math.round(b)+'B'; + const fmtN = n => n >= 1e6 ? (n/1e6).toFixed(2)+'M' : n >= 1e3 ? (n/1e3).toFixed(1)+'K' : String(Math.round(n)); + + for (const phase of order) { + const tp = timing[phase] || {}, mp = metrics[phase] || {}; + const pms = [...new Set([...Object.keys(tp), ...Object.keys(mp)])]; + if (pms.length === 0) continue; + + console.log('\n## ' + phase); + + // Table A: wall + CPU + memory + console.log(pad('PM', 6) + ' ' + padR('wall', 8) + ' ' + padR('±σ', 7) + ' ' + padR('user', 7) + ' ' + padR('sys', 7) + ' ' + padR('RSS', 6) + ' ' + padR('pgMinor', 8)); + for (const pm of pms) { + const t = tp[pm] || {}, m = mp[pm] || {}; + console.log( + pad(pm, 6) + ' ' + + padR((t.mean ?? 0).toFixed(2)+'s', 8) + ' ' + + padR((t.stddev ?? 0).toFixed(2)+'s', 7) + ' ' + + padR((m.user_s || 0).toFixed(2)+'s', 7) + ' ' + + padR((m.sys_s || 0).toFixed(2)+'s', 7) + ' ' + + padR(m.rss ? fmtB(m.rss) : '-', 6) + ' ' + + padR(m.page_minor ? fmtN(m.page_minor) : '-', 8) + ); + } + + // Table B: context switches + network + final on-disk footprint + console.log(pad('PM', 6) + ' ' + padR('vCtx', 8) + ' ' + padR('iCtx', 8) + ' ' + padR('net RX', 8) + ' ' + padR('net TX', 8) + ' ' + padR('cache', 8) + ' ' + padR('node_mod', 9) + ' ' + padR('lock', 7)); + for (const pm of pms) { + const m = mp[pm] || {}; + console.log( + pad(pm, 6) + ' ' + + padR(m.vol_ctx ? fmtN(m.vol_ctx) : '-', 8) + ' ' + + padR(m.invol_ctx ? fmtN(m.invol_ctx) : '-', 8) + ' ' + + padR(m.net_rx ? fmtB(m.net_rx) : '-', 8) + ' ' + + padR(m.net_tx ? fmtB(m.net_tx) : '-', 8) + ' ' + + padR(m.foot_cache ? fmtB(m.foot_cache) : '-', 8) + ' ' + + padR(m.foot_nm ? fmtB(m.foot_nm) : '-', 9) + ' ' + + padR(m.foot_lock ? fmtB(m.foot_lock) : '-', 7) + ); + } + } +" + +echo -e "${GREEN}Done. Raw results in $RESULTS_DIR${NC}"