perf(pacquet): size rayon's pool in the Node.js addon like the CLI does (#16759)

The Node.js addon (`pnpm-napi`) embeds the Rust engine as a cdylib loaded
dynamically by Node.js. Rayon is statically linked into the cdylib and uses
Rayon's global thread pool for parallel operations (such as workspace
discovery, link phase clonefile/hardlinks, and store operations).

Previously, the Rayon global pool sizing policy was defined only in
`pnpm-cli`, meaning the Node.js addon defaulted to Rayon's unconstrained
core-count pool sizing (which can spin up excessive worker threads on
high-core machines and ignores container quota limits).

This change:
1. Moves the sizing policy and initialization logic (`configure_rayon_pool`,
   `rayon_pool_size`, `RAYON_THREADS_PER_CORE`, `MIN_RAYON_THREADS`,
   `MAX_RAYON_THREADS`) into `pnpm-package-manager`.
2. Uses `std::sync::Once` for idempotent initialization across multiple
   calls and tests within the same process.
3. Configures the pool lazily when Node.js addon entry points (`install()`,
   `rebuild()`, `get_peer_dependency_issues()`, etc.) are called (deliberately
   avoiding `module_init` to prevent deadlocking with dynamic loader locks
   during `dlopen`).
4. Adds unit tests validating the sizing bounds and idempotence.

Closes https://github.com/pnpm/tasks/issues/54
This commit is contained in:
Zoltan Kochan authored and GitHub committed 2026-10-08 23:45:48 +02:00
1 parent 6940e6d788
commit faa2606a8b
18 files changed
+228 -113

No files matched your search

+5
View File
@@ -0,0 +1,5 @@
---
"@pnpm/napi": patch
---
The Node.js addon now sizes rayon's thread pool according to the engine thread-pool sizing policy [pnpm/tasks#54](https://github.com/pnpm/tasks/issues/54).
Generated
+1
View File
@@ -5628,6 +5628,7 @@ dependencies = [
"pnpm-testing-utils",
"pnpm-workspace",
"pretty_assertions",
"rayon",
"serde",
"serde_json",
"tempfile",
+1 -77
View File
@@ -57,6 +57,7 @@ use config_overrides::ConfigOverrides;
use flag_relocation::relocate_pre_subcommand_flags;
use miette::set_panic_hook;
use pnpm_diagnostics::{enable_tracing_by_env, install_report_handler};
use pnpm_package_manager::configure_rayon_pool;
use state::State;
use std::{ffi::OsString, future::Future, path::Path, process::ExitCode};
@@ -376,83 +377,6 @@ fn inject_alias_subcommand(exe_name: Option<&str>, mut argv: Vec<OsString>) -> V
argv
}
/// Size rayon's global pool with [`rayon_pool_size`].
///
/// Must run before anything touches rayon. The first parallel iterator
/// builds the global pool at rayon's default size, after which this
/// call can no longer size it, so debug builds assert that it did. The
/// repeat-install fast path uses rayon for workspace discovery.
///
/// Deliberately NOT communicated via the `RAYON_NUM_THREADS`
/// environment variable: a process-env write would leak into every
/// child the install spawns (lifecycle scripts, `node --version`,
/// git), and pnpm exposes no such variable to scripts. An explicit
/// `RAYON_NUM_THREADS` from the caller is honoured by skipping the
/// override.
///
/// Use [`std::thread::available_parallelism`] rather than the
/// workspace's existing `num_cpus::get()` so cgroup / CPU-quota
/// limits in containers and CI runners are respected — `num_cpus`
/// reports the host's logical CPU count, which on a quota-limited
/// runner can spin up far more rayon threads than the kernel will
/// actually schedule onto our cores.
fn configure_rayon_pool() {
if std::env::var_os("RAYON_NUM_THREADS").is_some() {
return;
}
let parallelism = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get);
let built = rayon::ThreadPoolBuilder::new()
.num_threads(rayon_pool_size(parallelism, RAYON_THREADS_PER_CORE))
.build_global();
let configured = built.is_ok();
debug_assert!(configured, "rayon's global pool was built before it was configured: {built:?}");
}
/// `threads_per_core × parallelism`, kept between [`MIN_RAYON_THREADS`]
/// and [`MAX_RAYON_THREADS`]. See [`RAYON_THREADS_PER_CORE`] for the
/// multiplier. The link phase is dominated by clonefile /
/// hardlink syscalls that block the calling thread on the kernel's
/// metadata journal, not by CPU work, so oversubscribing CPUs gives
/// more in-flight syscalls and a higher effective throughput.
/// Empirically sweeping 4-200 threads on a 1352-package warm install
/// on macOS APFS, 2× was the knee — fewer threads underutilize the
/// journal, way more (100+) loses to context switching and per-thread
/// fixed costs (`user` time scales linearly past 50 without any
/// wall-time payoff). That knee holds up to the ceiling below. Past 16
/// threads, 2× no longer pays off.
///
/// **Floor of 4 threads is intentional.** A 1-2-CPU CI runner left
/// at `2 × parallelism` would be capped to 2-4 rayon threads, and
/// at that point we go back to the original "one rayon thread is
/// blocked on a `clonefile` while the next fully-ready snapshot
/// can't even start" pattern that the 2× tuning is trying to
/// avoid. The kernel metadata journal is the bottleneck even on
/// small hosts, so a small intentional oversubscription
/// (`max(4, 2 × parallelism)`) is a better trade than respecting the
/// quota literally.
///
/// **Ceiling of 16 threads.** Past 16, the extra workers add system
/// time without shortening the install. A warm-install sweep (nuxt, next, nitro
/// fixtures) found no machine where 2× beat 16 threads: a 32-vCPU
/// Linux runner kept its wall time and halved its system time, a
/// 16-vCPU Windows runner got 10% faster, and a 10-core M1 Max, the
/// one host where 2× beat 8 threads, was unchanged at 16
/// (pnpm/tasks#51). A ceiling of 8 cut Linux system time further and
/// sped up Windows, but cost that Mac 19%.
fn rayon_pool_size(parallelism: usize, threads_per_core: usize) -> usize {
parallelism.saturating_mul(threads_per_core).clamp(MIN_RAYON_THREADS, MAX_RAYON_THREADS)
}
/// Two threads per core, except on Windows, where the sweeps mostly
/// favoured one (pnpm/tasks#52). Warm frozen installs were 4-5% faster
/// at 1× on 4- and 8-vCPU runners. A fresh install of the 1352-package
/// benchmark fixture took 3.9 s at 1× and 4.5 s at 2× on a 4-vCPU
/// runner, and was even on an 8-vCPU one. A 16-vCPU runner was fastest
/// at 4 threads, below what this multiplier gives it.
const RAYON_THREADS_PER_CORE: usize = if cfg!(windows) { 1 } else { 2 };
const MIN_RAYON_THREADS: usize = 4;
const MAX_RAYON_THREADS: usize = 16;
#[cfg(test)]
mod tests;
+2 -1
View File
@@ -1,4 +1,5 @@
use super::{inject_alias_subcommand, parse_cli_args, prepare_cli_argv, rayon_pool_size};
use super::{inject_alias_subcommand, parse_cli_args, prepare_cli_argv};
use pnpm_package_manager::rayon_pool_size;
use std::ffi::OsString;
fn argv(parts: &[&str]) -> Vec<OsString> {
+1
View File
@@ -63,6 +63,7 @@ napi-build = { workspace = true }
mockito = { workspace = true }
pretty_assertions = { workspace = true }
pnpm-testing-utils = { workspace = true }
rayon = { workspace = true }
tempfile = { workspace = true }
[target.'cfg(not(target_family = "wasm"))'.dependencies]
+1
View File
@@ -102,6 +102,7 @@ pub struct RenderDependentsInput {
/// which is an answer, not an error.
#[napi]
pub async fn get_dependents(options: DependentsOptions) -> napi::Result<serde_json::Value> {
pnpm_package_manager::configure_rayon_pool();
tokio::task::spawn_blocking(move || build_trees(&options)).await
.map_err(|join_error| {
napi::Error::from_reason(format!(
+1
View File
@@ -99,6 +99,7 @@ pub async fn install(
read_package_batch_hook: Option<BatchHookSink>,
on_output: Option<OutputSink>,
) -> napi::Result<InstallResult> {
pnpm_package_manager::configure_rayon_pool();
let _guard = engine_call_lock().lock().await;
let renderer = build_renderer(&options, on_output);
let (tx, rx) = tokio::sync::oneshot::channel();
@@ -7,6 +7,7 @@ use super::{
pub async fn get_peer_dependency_issues(
options: PeerIssuesOptions,
) -> napi::Result<serde_json::Value> {
pnpm_package_manager::configure_rayon_pool();
let _guard = engine_call_lock().lock().await;
let (tx, rx) = tokio::sync::oneshot::channel();
std::thread::Builder::new()
+1
View File
@@ -11,6 +11,7 @@ pub async fn rebuild(
selected_names: Option<Vec<String>>,
on_output: Option<OutputSink>,
) -> napi::Result<()> {
pnpm_package_manager::configure_rayon_pool();
let _guard = engine_call_lock().lock().await;
let renderer = build_renderer(&options, on_output);
let (tx, rx) = tokio::sync::oneshot::channel();
+8
View File
@@ -66,6 +66,12 @@ pub fn engine_version() -> &'static str {
/// Honor the same `TRACE` env var the pacquet CLI honors: an addon
/// embedded in a Node host has no `main` of its own, so the subscriber
/// is installed when the module loads.
///
/// Rayon's global thread pool is deliberately NOT initialized here:
/// `module_init` runs during `dlopen` while the dynamic linker lock is held.
/// Spawning rayon worker threads here would block waiting for the dynamic
/// linker lock, deadlocking the process. The pool is configured upon the first
/// entry point call instead.
#[napi_derive::module_init]
fn init_tracing() {
pnpm_diagnostics::enable_tracing_by_env();
@@ -76,3 +82,5 @@ fn init_tracing() {
/// self-contained. See the module for the full rationale.
#[cfg(test)]
mod napi_runtime_test_stubs;
#[cfg(test)]
mod tests;
+32 -35
View File
@@ -77,6 +77,7 @@ pub struct WriteLockfileOptions {
/// Inputs for [`filter_lockfile_by_importers`].
#[napi(object)]
#[derive(Default)]
pub struct FilterLockfileOptions {
/// Whether the listed importers keep their `dependencies`. Defaults to
/// `true`.
@@ -105,6 +106,7 @@ pub struct FilterLockfileOptions {
pub async fn read_lockfile(
options: ReadLockfileOptions,
) -> napi::Result<Option<serde_json::Value>> {
pnpm_package_manager::configure_rayon_pool();
let kind = LockfileKind::parse(options.kind.as_deref())?;
let path = lockfile_path(&options.dir, options.modules_dir.as_deref(), &kind);
let loaded = tokio::task::spawn_blocking(move || Lockfile::load_from_path(&path))
@@ -124,6 +126,7 @@ pub async fn read_lockfile(
/// Write the lockfile, formatted exactly as an install writes it.
#[napi]
pub async fn write_lockfile(options: WriteLockfileOptions) -> napi::Result<()> {
pnpm_package_manager::configure_rayon_pool();
let kind = LockfileKind::parse(options.kind.as_deref())?;
let path = lockfile_path(&options.dir, options.modules_dir.as_deref(), &kind);
let lockfile: Lockfile = serde_json::from_value(options.lockfile)
@@ -137,6 +140,32 @@ pub async fn write_lockfile(options: WriteLockfileOptions) -> napi::Result<()> {
.map_err(|error| to_napi_error(&error))
}
fn into_filter_options(options: Option<FilterLockfileOptions>) -> FilterByImportersOptions {
let options = options.unwrap_or_default();
let skipped: HashSet<PackageKey> = options.skipped
.unwrap_or_default()
.iter()
// An unparsable dep path matches no snapshot key, so skipping it
// is a no-op either way; dropping it keeps a stale entry in a
// host's skip list from failing the whole call.
.filter_map(|dep_path| dep_path.parse().ok())
.collect();
FilterByImportersOptions {
include: IncludedDependencies {
dependencies: options.include_dependencies.unwrap_or(true),
dev_dependencies: options.include_dev_dependencies.unwrap_or(true),
optional_dependencies: options.include_optional_dependencies.unwrap_or(true),
},
skipped,
fail_on_missing_dependencies: options.fail_on_missing_dependencies.unwrap_or(false),
peer_edges: PeerEdgeOptions {
resolve_peers_from_workspace_root: options
.resolve_peers_from_workspace_root
.unwrap_or(false),
},
}
}
/// The lockfile narrowed to what `importerIds` reaches: those importers
/// keep only the dependency groups asked for, and `packages` / `snapshots`
/// are pruned to the transitive closure of what they still depend on.
@@ -149,46 +178,14 @@ pub fn filter_lockfile_by_importers(
importer_ids: Vec<String>,
options: Option<FilterLockfileOptions>,
) -> napi::Result<serde_json::Value> {
pnpm_package_manager::configure_rayon_pool();
let lockfile: Lockfile = serde_json::from_value(lockfile)
.map_err(|err| {
napi::Error::from_reason(format!("the lockfile argument is not a lockfile: {err}"))
})?;
let options = options.unwrap_or(FilterLockfileOptions {
include_dependencies: None,
include_dev_dependencies: None,
include_optional_dependencies: None,
skipped: None,
fail_on_missing_dependencies: None,
resolve_peers_from_workspace_root: None,
});
let skipped: HashSet<PackageKey> = options.skipped
.unwrap_or_default()
.iter()
// An unparsable dep path matches no snapshot key, so skipping it
// is a no-op either way; dropping it keeps a stale entry in a
// host's skip list from failing the whole call.
.filter_map(|dep_path| dep_path.parse().ok())
.collect();
let filter_options = into_filter_options(options);
let filtered = lockfile
.filter_by_importers(
importer_ids,
&FilterByImportersOptions {
include: IncludedDependencies {
dependencies: options.include_dependencies.unwrap_or(true),
dev_dependencies: options.include_dev_dependencies.unwrap_or(true),
optional_dependencies: options.include_optional_dependencies.unwrap_or(true),
},
skipped,
fail_on_missing_dependencies: options
.fail_on_missing_dependencies
.unwrap_or(false),
peer_edges: PeerEdgeOptions {
resolve_peers_from_workspace_root: options
.resolve_peers_from_workspace_root
.unwrap_or(false),
},
},
)
.filter_by_importers(importer_ids, &filter_options)
.map_err(|error| to_napi_error(&error))?;
serde_json::to_value(filtered)
.map_err(|err| napi::Error::from_reason(format!("serializing the lockfile: {err}")))
+1
View File
@@ -55,6 +55,7 @@ pub struct PackResult {
#[napi]
pub async fn pack(options: PackOptions, on_log: Option<LogSink>) -> napi::Result<PackResult> {
pnpm_package_manager::configure_rayon_pool();
// Share the serialization lock with install/rebuild: they all drive the
// same process-global log sink, so overlapping calls would misroute events.
let _guard = engine_call_lock().lock().await;
+1
View File
@@ -108,6 +108,7 @@ pub struct ResolvedConfig {
#[napi(js_name = "readConfig")]
pub fn read_config(options: ReadConfigOptions) -> napi::Result<ResolvedConfig> {
pnpm_package_manager::configure_rayon_pool();
let dir = std::path::PathBuf::from(options.dir);
let config =
resolve_config(&dir, &ConfigOverlay::default()).map_err(|error| to_napi_error(&error))?;
+1
View File
@@ -65,6 +65,7 @@ pub async fn resolve_dependency(
wanted: WantedDependencyInput,
options: ResolveDependencyOptions,
) -> napi::Result<ResolveDependencyResult> {
pnpm_package_manager::configure_rayon_pool();
let (tx, rx) = tokio::sync::oneshot::channel();
std::thread::Builder::new()
.name("pnpm-napi-resolve".to_string())
+39
View File
@@ -0,0 +1,39 @@
use crate::read_config::{ReadConfigOptions, read_config};
#[test]
fn addon_entry_point_configures_rayon_pool() {
const CHILD: &str = "PNPM_NAPI_RAYON_POOL_TEST_CHILD";
if std::env::var_os(CHILD).is_none() {
let output = std::process::Command::new(std::env::current_exe().unwrap())
.args(["--exact", "tests::addon_entry_point_configures_rayon_pool"])
.env(CHILD, "1")
.output()
.unwrap();
assert!(output.status.success(), "{output:?}");
return;
}
let temp = tempfile::tempdir().unwrap();
let config = read_config(ReadConfigOptions {
dir: temp
.path()
.to_str()
.unwrap()
.to_string(),
})
.expect("read_config must succeed");
let _ = config.registries;
let current_threads = rayon::current_num_threads();
let parallelism = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get);
let override_threads = std::env::var("RAYON_NUM_THREADS")
.ok()
.and_then(|value| value.parse::<usize>().ok());
let expected = match override_threads {
Some(0) => parallelism,
Some(threads) => threads,
None => pnpm_package_manager::rayon_pool_size(
parallelism,
pnpm_package_manager::RAYON_THREADS_PER_CORE,
),
};
assert_eq!(current_threads, expected);
}
+2
View File
@@ -29,6 +29,7 @@ pub use pnpm_patching::{
PatchCommitError, PkgFilesForDiff, diff_folders, prepare_pkg_files_for_diff,
};
pub use prefetching_resolver::*;
pub use rayon_pool::*;
pub use remove::*;
pub use resolution_observer::*;
pub use resolution_policy::{PickPolicy, create_configured_registry_resolver};
@@ -78,6 +79,7 @@ mod peer_dependency_issues;
mod prefetching_resolver;
mod prune_merged_branch_lockfile;
mod prune_virtual_store;
mod rayon_pool;
mod remove;
mod resolution_observer;
mod resolution_policy;
@@ -0,0 +1,94 @@
//! Rayon thread pool sizing policy and configuration.
//!
//! Sizes rayon's global pool for installation workloads across the CLI
//! and the Node.js addon.
use std::sync::Once;
/// Size rayon's global pool with [`rayon_pool_size`].
///
/// Must run before anything touches rayon. The first parallel iterator
/// builds the global pool at rayon's default size, after which this
/// call can no longer size it, so debug builds assert that it did. The
/// repeat-install fast path uses rayon for workspace discovery.
///
/// Deliberately NOT communicated via the `RAYON_NUM_THREADS`
/// environment variable: a process-env write would leak into every
/// child the install spawns (lifecycle scripts, `node --version`,
/// git), and pnpm exposes no such variable to scripts. An explicit
/// `RAYON_NUM_THREADS` from the caller is honoured by skipping the
/// override.
///
/// Use [`std::thread::available_parallelism`] rather than the
/// workspace's existing `num_cpus::get()` so cgroup / CPU-quota
/// limits in containers and CI runners are respected — `num_cpus`
/// reports the host's logical CPU count, which on a quota-limited
/// runner can spin up far more rayon threads than the kernel will
/// actually schedule onto our cores.
pub fn configure_rayon_pool() {
static INITIALIZE: Once = Once::new();
INITIALIZE.call_once(|| {
if std::env::var_os("RAYON_NUM_THREADS").is_some() {
return;
}
let parallelism =
std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get);
let built = rayon::ThreadPoolBuilder::new()
.num_threads(rayon_pool_size(parallelism, RAYON_THREADS_PER_CORE))
.build_global();
let configured = built.is_ok();
debug_assert!(
configured,
"rayon's global pool was built before it was configured: {built:?}",
);
});
}
/// `threads_per_core × parallelism`, kept between [`MIN_RAYON_THREADS`]
/// and [`MAX_RAYON_THREADS`]. See [`RAYON_THREADS_PER_CORE`] for the
/// multiplier. The link phase is dominated by clonefile /
/// hardlink syscalls that block the calling thread on the kernel's
/// metadata journal, not by CPU work, so oversubscribing CPUs gives
/// more in-flight syscalls and a higher effective throughput.
/// Empirically sweeping 4-200 threads on a 1352-package warm install
/// on macOS APFS, 2× was the knee — fewer threads underutilize the
/// journal, way more (100+) loses to context switching and per-thread
/// fixed costs (`user` time scales linearly past 50 without any
/// wall-time payoff). That knee holds up to the ceiling below. Past 16
/// threads, 2× no longer pays off.
///
/// **Floor of 4 threads is intentional.** A 1-2-CPU CI runner left
/// at `2 × parallelism` would be capped to 2-4 rayon threads, and
/// at that point we go back to the original "one rayon thread is
/// blocked on a `clonefile` while the next fully-ready snapshot
/// can't even start" pattern that the 2× tuning is trying to
/// avoid. The kernel metadata journal is the bottleneck even on
/// small hosts, so a small intentional oversubscription
/// (`max(4, 2 × parallelism)`) is a better trade than respecting the
/// quota literally.
///
/// **Ceiling of 16 threads.** Past 16, the extra workers add system
/// time without shortening the install. A warm-install sweep (nuxt, next, nitro
/// fixtures) found no machine where 2× beat 16 threads: a 32-vCPU
/// Linux runner kept its wall time and halved its system time, a
/// 16-vCPU Windows runner got 10% faster, and a 10-core M1 Max, the
/// one host where 2× beat 8 threads, was unchanged at 16
/// (pnpm/tasks#51). A ceiling of 8 cut Linux system time further and
/// sped up Windows, but cost that Mac 19%.
#[must_use]
pub fn rayon_pool_size(parallelism: usize, threads_per_core: usize) -> usize {
parallelism.saturating_mul(threads_per_core).clamp(MIN_RAYON_THREADS, MAX_RAYON_THREADS)
}
/// Two threads per core, except on Windows, where the sweeps mostly
/// favoured one (pnpm/tasks#52). Warm frozen installs were 4-5% faster
/// at 1× on 4- and 8-vCPU runners. A fresh install of the 1352-package
/// benchmark fixture took 3.9 s at 1× and 4.5 s at 2× on a 4-vCPU
/// runner, and was even on an 8-vCPU one. A 16-vCPU runner was fastest
/// at 4 threads, below what this multiplier gives it.
pub const RAYON_THREADS_PER_CORE: usize = if cfg!(windows) { 1 } else { 2 };
pub const MIN_RAYON_THREADS: usize = 4;
pub const MAX_RAYON_THREADS: usize = 16;
#[cfg(test)]
mod tests;
@@ -0,0 +1,36 @@
use super::{configure_rayon_pool, rayon_pool_size};
#[test]
fn rayon_pool_is_the_scaled_parallelism_between_4_and_16_threads() {
let parallelism = [1, 2, 3, 4, 8, 9, 32, 256];
assert_eq!(parallelism.map(|cores| rayon_pool_size(cores, 2)), [4, 4, 6, 8, 16, 16, 16, 16]);
assert_eq!(parallelism.map(|cores| rayon_pool_size(cores, 1)), [4, 4, 4, 4, 8, 9, 16, 16]);
}
#[test]
fn configure_rayon_pool_is_idempotent() {
const CHILD: &str = "PNPM_RAYON_POOL_TEST_CHILD";
if std::env::var_os(CHILD).is_none() {
let output = std::process::Command::new(std::env::current_exe().unwrap())
.args(["--exact", "rayon_pool::tests::configure_rayon_pool_is_idempotent"])
.env(CHILD, "1")
.output()
.unwrap();
assert!(output.status.success(), "{output:?}");
return;
}
configure_rayon_pool();
configure_rayon_pool();
let current_threads = rayon::current_num_threads();
let parallelism = std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get);
let override_threads = std::env::var("RAYON_NUM_THREADS")
.ok()
.and_then(|value| value.parse::<usize>().ok());
let expected = match override_threads {
Some(0) => parallelism,
Some(threads) => threads,
None => rayon_pool_size(parallelism, super::RAYON_THREADS_PER_CORE),
};
assert_eq!(current_threads, expected);
}