diff --git a/Cargo.lock b/Cargo.lock index c228da3e01..d67969e7de 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11805,6 +11805,7 @@ dependencies = [ "parking_lot", "pathdiff", "petgraph 0.6.5", + "rayon", "reqwest 0.12.24", "rustls", "rustls-native-certs", diff --git a/crates/pm/src/cmd/view.rs b/crates/pm/src/cmd/view.rs index d8355e9922..d5632a69af 100644 --- a/crates/pm/src/cmd/view.rs +++ b/crates/pm/src/cmd/view.rs @@ -27,10 +27,9 @@ 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 resolved_version = utoo_ruborist::resolver::version::resolve_target_version( &full_manifest.dist_tags, - &version_list, + &full_manifest.versions, version_spec, ) .map_err(|e| { diff --git a/crates/ruborist/Cargo.toml b/crates/ruborist/Cargo.toml index ffdbcfae77..58f2b26372 100644 --- a/crates/ruborist/Cargo.toml +++ b/crates/ruborist/Cargo.toml @@ -54,7 +54,11 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls- # Native-only dependencies (not compiled for WASM) [target.'cfg(not(target_arch = "wasm32"))'.dependencies] -libc = "0.2" +libc = "0.2" +# Dedicated CPU thread pool for simd_json manifest parsing — keeps the +# tokio runtime free of CPU-bound work so in-flight network IO is not +# blocked by sibling parses. WASM falls back to inline parse. +rayon = "1.10" # Browser handles TLS via the fetch API, so reqwest needs no rustls # feature on this target. diff --git a/crates/ruborist/src/model/manifest.rs b/crates/ruborist/src/model/manifest.rs index dea63dc7bc..c3a10fb53d 100644 --- a/crates/ruborist/src/model/manifest.rs +++ b/crates/ruborist/src/model/manifest.rs @@ -111,6 +111,33 @@ impl FullManifest { } } +/// Re-parse the raw manifest bytes on rayon's CPU thread pool to extract +/// a single version. Each resolved package triggers one full re-parse via +/// `simd_json::to_borrowed_value`, which on hot paths (300+ resolves per +/// install) accumulates into the dominant CPU bucket. +/// +/// Returns `Arc` so callers (e.g. UnifiedRegistry's +/// version-manifest cache) can store and clone the result without +/// deep-copying the underlying `HashMap`s. Wasm32 falls back to inline +/// parse — no rayon, no thread pool. +pub async fn extract_core_version_off_runtime( + full: std::sync::Arc, + version: String, +) -> Option> { + #[cfg(not(target_arch = "wasm32"))] + { + let (tx, rx) = tokio::sync::oneshot::channel(); + rayon::spawn(move || { + let _ = tx.send(full.get_core_version(&version).map(std::sync::Arc::new)); + }); + rx.await.ok().flatten() + } + #[cfg(target_arch = "wasm32")] + { + full.get_core_version(&version).map(std::sync::Arc::new) + } +} + /// Deserialize a versions map by extracting only the keys, skipping all values. /// /// Uses `IgnoredAny` to skip over version manifest JSON objects without allocating, diff --git a/crates/ruborist/src/resolver/preload.rs b/crates/ruborist/src/resolver/preload.rs index 9fb5aff67b..b667d6e3d3 100644 --- a/crates/ruborist/src/resolver/preload.rs +++ b/crates/ruborist/src/resolver/preload.rs @@ -15,7 +15,7 @@ use crate::traits::progress::{BuildEvent, EventReceiver}; use crate::traits::registry::RegistryClient; /// Default concurrency limit for manifest fetching -pub const DEFAULT_CONCURRENCY: usize = 64; +pub const DEFAULT_CONCURRENCY: usize = 256; /// A dependency spec: (name, version_spec) pub type Dep = (String, String); diff --git a/crates/ruborist/src/resolver/registry.rs b/crates/ruborist/src/resolver/registry.rs index 6edeb9069c..91e2e47310 100644 --- a/crates/ruborist/src/resolver/registry.rs +++ b/crates/ruborist/src/resolver/registry.rs @@ -137,14 +137,12 @@ pub fn resolve_from_manifest( manifest: &FullManifest, spec: &str, ) -> Result> { - let version_list: Vec = manifest.versions.clone(); - - if version_list.is_empty() { + if manifest.versions.is_empty() { return Err(ResolveError::NoVersions(manifest.name.clone())); } // Resolve version using shared logic - let resolved_version = resolve_target_version(&manifest.dist_tags, &version_list, spec) + let resolved_version = resolve_target_version(&manifest.dist_tags, &manifest.versions, spec) .map_err(|e| ResolveError::Version(format!("{}@{}: {}", manifest.name, spec, e)))?; // Get manifest for resolved version (lazy: parse from Value on demand) diff --git a/crates/ruborist/src/service/manifest.rs b/crates/ruborist/src/service/manifest.rs index b9dbff7c8d..695cab58fd 100644 --- a/crates/ruborist/src/service/manifest.rs +++ b/crates/ruborist/src/service/manifest.rs @@ -13,6 +13,35 @@ use super::fetch::{ use super::http::get_client; use crate::model::manifest::{CoreVersionManifest, FullManifest}; +/// Parse JSON bytes on rayon's CPU thread pool (native) or inline +/// (wasm32). Keeps the tokio runtime free of `simd_json` CPU work so +/// other in-flight manifest fetches can keep driving network IO while +/// this one is parsing. Rayon's global pool work-steals across +/// `num_cpus` threads, so concurrent manifest parses parallelise. +async fn parse_json_off_runtime(bytes: Vec) -> Result +where + T: serde::de::DeserializeOwned + Send + 'static, +{ + #[cfg(not(target_arch = "wasm32"))] + { + let (tx, rx) = tokio::sync::oneshot::channel(); + rayon::spawn(move || { + let mut buf = bytes; + let result = simd_json::serde::from_slice::(&mut buf) + .map_err(|e| anyhow!("JSON parse error: {e}")); + // Receiver dropped means the awaiting future was cancelled. + let _ = tx.send(result); + }); + rx.await + .map_err(|e| anyhow!("rayon parse channel closed: {e}"))? + } + #[cfg(target_arch = "wasm32")] + { + let mut buf = bytes; + simd_json::serde::from_slice::(&mut buf).map_err(|e| anyhow!("JSON parse error: {e}")) + } +} + /// Result of a full manifest fetch with ETag support. /// Transient return value, immediately destructured — Box not needed. #[allow(clippy::large_enum_variant)] @@ -91,11 +120,13 @@ pub async fn fetch_full_manifest(opts: FetchManifestOptions<'_>) -> Result(bytes) + .await + .map_err(FetchError::Permanent) } else { Err(classify_status(response.status(), &url)) } diff --git a/crates/ruborist/src/service/registry.rs b/crates/ruborist/src/service/registry.rs index 0022a72e5e..7eed5ccaf0 100644 --- a/crates/ruborist/src/service/registry.rs +++ b/crates/ruborist/src/service/registry.rs @@ -244,11 +244,19 @@ impl UnifiedRegistry { .await .map_err(RegistryError)? { - manifest::FetchManifestResult::Ok(manifest, new_etag) => { + manifest::FetchManifestResult::Ok(mut manifest, new_etag) => { + // Move version_list / dist_tags out of `manifest` + // into `VersionsInfo` instead of cloning. The cached + // `Arc` keeps `raw` + scalar metadata; + // the canonical version_list / dist_tags now live in + // the `versions_info` cache slot only. Readers that + // need them go through `cache.get_versions(name)`. + let version_list = std::mem::take(&mut manifest.versions); + let dist_tags = std::mem::take(&mut manifest.dist_tags); let versions_info = Arc::new(VersionsInfo { versions: Versions { - version_list: manifest.versions.clone(), - dist_tags: manifest.dist_tags.clone(), + version_list, + dist_tags, }, etag: new_etag, last_updated: current_timestamp_secs(), @@ -269,7 +277,7 @@ impl UnifiedRegistry { self.cache.set_versions(name.to_string(), versions_info); } else { // Persistent store corrupted/missing, fetch fresh (without etag). - let (manifest, new_etag) = manifest::fetch_full_manifest_fresh( + let (mut manifest, new_etag) = manifest::fetch_full_manifest_fresh( &self.registry_url, name, manifest::MetadataFormat::Abbreviated, @@ -277,10 +285,12 @@ impl UnifiedRegistry { .await .map_err(RegistryError)?; + let version_list = std::mem::take(&mut manifest.versions); + let dist_tags = std::mem::take(&mut manifest.dist_tags); let versions_info = Arc::new(VersionsInfo { versions: Versions { - version_list: manifest.versions.clone(), - dist_tags: manifest.dist_tags.clone(), + version_list, + dist_tags, }, etag: new_etag, last_updated: current_timestamp_secs(), @@ -362,9 +372,8 @@ impl UnifiedRegistry { // version from the full manifest. `resolve_full_manifest` // is itself inflight-gated, so concurrent specs for the // same name share one full-manifest fetch. - let (resolved_version, manifest) = + let (resolved_version, arc) = self.resolve_via_full_manifest(name, spec).await?; - let arc = Arc::new(manifest); self.cache.set_version_manifest( name.to_string(), spec.to_string(), @@ -426,16 +435,39 @@ impl UnifiedRegistry { &self, name: &str, spec: &str, - ) -> Result<(String, CoreVersionManifest), RegistryError> { + ) -> Result<(String, Arc), RegistryError> { match self.resolve_full_manifest(name).await? { FullManifestResult::Full(full) => { - if full.versions.is_empty() { + // Canonical `version_list` / `dist_tags` live in the + // `versions_info` cache slot; the parallel slot is filled by + // the same code path that filled `full_manifests`. We don't + // duplicate them on the cached `Arc`. + let versions_info = self.cache.get_versions(name).ok_or_else(|| { + RegistryError(anyhow!("Versions cache not found for {}", name)) + })?; + if versions_info.versions.version_list.is_empty() { return Err(RegistryError(anyhow!("No versions available for {}", name))); } - let resolved_version = - resolve_target_version(&full.dist_tags, &full.versions, spec) - .map_err(|e| RegistryError(anyhow!("{}@{}: {}", name, spec, e)))?; - let core = full.get_core_version(&resolved_version).ok_or_else(|| { + let resolved_version = resolve_target_version( + &versions_info.versions.dist_tags, + &versions_info.versions.version_list, + spec, + ) + .map_err(|e| RegistryError(anyhow!("{}@{}: {}", name, spec, e)))?; + // Short-circuit: a sibling spec may have already resolved to + // this same version and warmed the cache. Skip the rayon + // reparse (the dominant cost on the resolve hot path) and + // share the existing `Arc` instead of + // deep-cloning its HashMaps. + if let Some(cached) = self.cache.get_version_manifest(name, &resolved_version) { + return Ok((resolved_version, cached)); + } + let core = crate::model::manifest::extract_core_version_off_runtime( + Arc::clone(&full), + resolved_version.clone(), + ) + .await + .ok_or_else(|| { RegistryError(anyhow!( "Version {} not found in manifest for {}", resolved_version, @@ -469,7 +501,7 @@ impl UnifiedRegistry { }) .await .map_err(RegistryError)?; - Ok((resolved_version, manifest)) + Ok((resolved_version, Arc::new(manifest))) } } } diff --git a/crates/ruborist/src/traits/registry.rs b/crates/ruborist/src/traits/registry.rs index df8adac00f..c6ad7d532a 100644 --- a/crates/ruborist/src/traits/registry.rs +++ b/crates/ruborist/src/traits/registry.rs @@ -157,11 +157,11 @@ pub trait RegistryClient { ) -> impl Future, Self::Error>> { async move { let manifest = self.fetch_full_manifest(name).await?; - let version_list: Vec = manifest.versions.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)))?; + let resolved_version = + resolve_target_version(&manifest.dist_tags, &manifest.versions, spec) + .map_err(|e| RegistryError(anyhow::anyhow!("{}@{}: {}", name, spec, e)))?; manifest .get_core_version(&resolved_version) @@ -227,9 +227,8 @@ pub trait RegistryClient { fetch_spec ); let full_manifest = self.fetch_full_manifest(&fetch_name).await?; - let version_list: Vec = full_manifest.versions.clone(); - if version_list.is_empty() { + if full_manifest.versions.is_empty() { return Err(RegistryError(anyhow::anyhow!( "No versions available for {}", fetch_name @@ -237,9 +236,12 @@ pub trait RegistryClient { .into()); } - let resolved_version = - resolve_target_version(&full_manifest.dist_tags, &version_list, &fetch_spec) - .map_err(|e| RegistryError(anyhow::anyhow!("{}@{}: {}", name, spec, e)))?; + let resolved_version = resolve_target_version( + &full_manifest.dist_tags, + &full_manifest.versions, + &fetch_spec, + ) + .map_err(|e| RegistryError(anyhow::anyhow!("{}@{}: {}", name, spec, e)))?; let version_manifest = full_manifest .get_core_version(&resolved_version)