Merge pull request 'Implement NUMA topology detection and CPU budget capping' (#74) from push-tnkxqmwztnmm into main

Reviewed-on: #74

.kilo/plans/1778319532254-eager-panda.md

@@ -1,87 +0,0 @@
-# Plan d'amélioration technique - obiskio
-
-## 1. Contexte et objectifs
-- **Objectif** : Renforcer la robustesse, la maintenabilité et les performances de la crate `obiskio`.
-- **Priorités** : 
-  1. Gestion des erreurs
-  2. Optimisation de la mémoire du pool
-  3. Robustesse concurrente
-  4. Couverture de tests
-  5. Documentation
-
----
-
-## 2. Axes d'amélioration détaillés
-
-### 2.1 Gestion des erreurs
-- **Problème** : `SKError` ne couvre pas tous les cas (format invalide, taille maximale, CRC)
-- **Actions** :
-  - Ajouter variante `ParseError(String)` dans `src/error.rs`
-  - Valider les tailles de SuperKmer avant parsing
-  - Remplacer `expect()` par `unwrap_or_else` avec messages explicites
-  - Documenter chaque variante d’erreur dans le README
-
-### 2.2 Optimisation du pool de fichiers
-- **Problème** : `SKFilePool` utilise un `Vec<WriteEntry>` non contraint et n’effectue pas de nettoyage en cas d’erreur
-- **Actions** :
-  - Implémenter un `LimitedVec` avec limite stricte à `MAX_POOL_SIZE`
-  - Créer `clear_memory()` qui supprime les entrées orphelines
-  - Ajouter `evict_lru_threshold()` pour éviction proactive
-  - Introduire un `RwLock` pour les opérations de lecture massives
-
-### 2.3 Robustesse concurrente
-- **Problème** : Risque de deadlocks dans `SKFileWriter::write_batch()` et `SKFileReader::reopen_and_seek()`
-- **Actions** :
-  - Remplacer `Mutex` par `RwLock` pour les accès en lecture
-  - Ajouter un compteur de blocage et logs de timeout
-  - Utiliser `std::thread::park_timeout` pour débloquer
-  - Insérer `debug_assert!` sur les états invariants
-
-### 2.4 Couverture de tests
-- **Problème** : Absence de benchmarks, de tests de migration, de résilience de fichiers corrompus
-- **Actions** :
-  - Benchmarks I/O sur 10k+ SuperKmer avec `criterion`
-  - Tests de migration de version de fichier `.meta` → `.v2.meta`
-  - Tests de corruption volontaire (truncature, inversion de bits)
-  - Tests de stress sur pool saturation (100 threads)
-
-### 2.5 Documentation & exemples
-- **Actions** :
-  - Ajouter des examples dans chaque module (`# Examples`)
-  - Documenter la logique LRU avec diagrammes Mermaid
-  - Créer un guide « How to recover from eviction »
-  - Mettre à jour le `README.md` avec tableau des variantes d’erreur
-
----
-
-## 3. Plan d'exécution (Roadmap)
-
-| Sprint | Durée | Livrables clés |
-|--------|-------|----------------|
-| **S1** | 2 jours | Refactorisation `SKError`, ajout de tests unitaires |
-| **S2** | 3 jours | Implémentation `clear_memory()` + `LimitedVec` |
-| **S3** | 2 jours | Passage à `RwLock`, ajout de compteurs de blocage |
-| **S4** | 2 jours | Benchmarks + tests de migration |
-| **S5** | 1 jour | Documentation finale & mise à jour du README |
-
----
-
-## 4. Dépendances externes
-- Mettre à jour `niffler` vers la version 2.0 (performance compression)
-- Évaluer `bincode` vs `serde_json` pour les métas (I/O)
-- Ajouter dépendance `criterion` (dev‑dependencies)
-
----
-
-## 5. KPI de suivi
-- **Couverture de tests** : ≥85 % des chemins critiques
-- **Latence moyenne d’écriture** : ↓15 % après optimisation du pool
-- **Taux d’erreurs résolues** : 100 % des nouvelles variantes couvertes
-- **Temps de build CI** : ≤5 min pour l’ensemble des benchmarks
-
----
-
-## 6. Validation finale
-- Revue de code avec `cargo clippy -- -D warnings`
-- Analyse de toxicité avec `cargo deny open-source-licenses`
-- Vérification de la conformité aux standards de naming du projet

.serena/.gitignore

@@ -1,2 +0,0 @@
-/cache
-/project.local.yml

.serena/project.yml

@@ -1,169 +0,0 @@
-# the name by which the project can be referenced within Serena/when chatting with the LLM.
-project_name: "obikmer"
-
-# the encoding used by text files in the project
-# For a list of possible encodings, see https://docs.python.org/3.11/library/codecs.html#standard-encodings
-encoding: "utf-8"
-
-# line ending convention to use when writing source files.
-# Possible values: unset (use global setting), "lf", "crlf", or "native" (platform default)
-# This does not affect Serena's own files (e.g. memories and configuration files), which always use native line endings.
-line_ending:
-
-# The language backend to use for this project.
-# If not set, the global setting from serena_config.yml is used.
-# Valid values: LSP, JetBrains
-# Note: the backend is fixed at startup. If a project with a different backend
-# is activated post-init, an error will be returned.
-language_backend:
-
-# whether to use project's .gitignore files to ignore files
-ignore_all_files_in_gitignore: true
-
-# advanced configuration option allowing to configure language server-specific options.
-# Maps the language key to the options.
-# The settings are considered only if the project is trusted (see global configuration to define trusted projects).
-# See https://oraios.github.io/serena/02-usage/050_configuration.html#language-server-specific-settings
-ls_specific_settings: {}
-
-# list of additional paths to ignore in this project.
-# Same syntax as gitignore, so you can use * and **.
-# Important: quote patterns that start with `*`, otherwise YAML treats them as aliases.
-# Example:
-#   ignored_paths:
-#     - "examples/**"
-#     - ".worktrees/**"
-#     - "**/bin/**"
-#     - "**/obj/**"
-# Note: global ignored_paths from serena_config.yml are also applied additively.
-ignored_paths: []
-
-# whether the project is in read-only mode
-# If set to true, all editing tools will be disabled and attempts to use them will result in an error
-# Added on 2025-04-18
-read_only: false
-
-# list of tool names to exclude.
-# This extends the existing exclusions (e.g. from the global configuration)
-# Find the list of tools here: https://oraios.github.io/serena/01-about/035_tools.html
-excluded_tools: []
-
-# list of tools to include that would otherwise be disabled (particularly optional tools that are disabled by default).
-# This extends the existing inclusions (e.g. from the global configuration).
-# Find the list of tools here: https://oraios.github.io/serena/01-about/035_tools.html
-included_optional_tools: []
-
-# fixed set of tools to use as the base tool set (if non-empty), replacing Serena's default set of tools.
-# This cannot be combined with non-empty excluded_tools or included_optional_tools.
-# Find the list of tools here: https://oraios.github.io/serena/01-about/035_tools.html
-fixed_tools: []
-
-# list of mode names that are to be activated by default, overriding the setting in the global configuration.
-# The full set of modes to be activated is base_modes (from global config) + default_modes + added_modes.
-# If the setting is undefined/empty, the default_modes from the global configuration (serena_config.yml) apply.
-# Otherwise, this overrides the setting from the global configuration (serena_config.yml).
-# Therefore, you can set this to [] if you do not want the default modes defined in the global config to apply
-# for this project.
-# This setting can, in turn, be overridden by CLI parameters (--mode).
-# See https://oraios.github.io/serena/02-usage/050_configuration.html#modes
-default_modes:
-
-# list of mode names to be activated additionally for this project, e.g. ["query-projects"]
-# The full set of modes to be activated is base_modes (from global config) + default_modes + added_modes.
-# See https://oraios.github.io/serena/02-usage/050_configuration.html#modes
-added_modes:
-
-# initial prompt for the project. It will always be given to the LLM upon activating the project
-# (contrary to the memories, which are loaded on demand).
-initial_prompt: ""
-
-# time budget (seconds) per tool call for the retrieval of additional symbol information
-# such as docstrings or parameter information.
-# This overrides the corresponding setting in the global configuration; see the documentation there.
-# If null or missing, use the setting from the global configuration.
-symbol_info_budget:
-
-# list of regex patterns which, when matched, mark a memory entry as read‑only.
-# Extends the list from the global configuration, merging the two lists.
-read_only_memory_patterns: []
-
-# list of regex patterns for memories to completely ignore.
-# Matching memories will not appear in list_memories or activate_project output
-# and cannot be accessed via read_memory or write_memory.
-# To access ignored memory files, use the read_file tool on the raw file path.
-# Extends the list from the global configuration, merging the two lists.
-# Example: ["_archive/.*", "_episodes/.*"]
-ignored_memory_patterns: []
-
-# list of additional workspace folder paths for cross-package reference support.
-# Paths can be absolute or relative to the project root.
-# Each folder is registered as an LSP workspace folder, enabling language servers to discover
-# symbols and references across package boundaries, but these folders are not indexed by Serena,
-# i.e. the respective symbols will not be found using Serena's symbol search tools.
-# Example:
-#   additional_workspace_folders:
-#     - ../sibling-package
-#     - ../shared-lib
-ls_additional_workspace_folders: []
-
-# list of language servers to start when using the LSP backend; choose from:
-#   ada                    al                     angular                ansible                bash
-#   bsl                    clojure                cpp                    cpp_ccls               crystal
-#   csharp                 csharp_omnisharp       cue                    dart                   deno
-#   elixir                 elm                    erlang                 fortran                fsharp
-#   gdscript               gleam                  go                     groovy                 haskell
-#   haxe                   hlsl                   html                   java                   json
-#   julia                  kotlin                 latex                  lean4                  lua
-#   luau                   markdown               matlab                 msl                    nextflow
-#   nix                    ocaml                  pascal                 perl                   php
-#   php_phpactor           php_phpantom           powershell             python                 python_basedpyright
-#   python_jedi            python_pyrefly         python_ty              qml                    r
-#   rego                   ruby                   ruby_solargraph        rust                   scala
-#   scss                   solidity               svelte                 swift                  systemverilog
-#   terraform              toml                   typescript             typescript_vts         vue
-#   wolfram                yaml                   zig
-#   (This list may be outdated; generated with scripts/print_language_list.py;
-#   For the current list, see values of the LanguageServerId enum here:
-#   https://github.com/oraios/serena/blob/main/src/solidlsp/ls_config.py)
-# For some languages, there are several alternative language servers, e.g. csharp_omnisharp, ruby_solargraph.)
-# Note:
-#   - For C, use cpp
-#   - For JavaScript, use typescript
-#   - For Angular projects, use angular (subsumes typescript+html; requires `npm install` in the project root)
-#   - For Svelte projects, use svelte (subsumes typescript/javascript for .svelte projects; requires npm)
-#   - For Deno projects, use deno (serves the same .ts/.js files as typescript; requires the deno CLI on PATH)
-#   - For SCSS / Sass / plain CSS, use scss (some-sass-language-server handles all three)
-#   - For Free Pascal/Lazarus, use pascal
-# Special requirements:
-#   Some language servers require additional setup/installations.
-#   See here for details: https://oraios.github.io/serena/01-about/020_programming-languages.html#language-servers
-# When using multiple language servers, the first language server that supports a given file will be used for that file.
-# The first language server is the default language and the respective language server will be used as a fallback.
-# Note that when using the JetBrains backend, language servers are not used and this list is correspondingly ignored.
-language_servers:
-- rust
-
-# list of workspace folder paths (LSP backend only).
-# These folders will be used to build up Serena's symbol index.
-# Paths must be within the project root and should thus be relative to the project root.
-# Furthermore, the paths should not be filtered by ignore settings.
-# Default setting: The entire project root folder (".") is considered.
-# In (large) monorepos, this can be used to index only subfolders of the project root, e.g.
-#   ls_workspace_folders:
-#     - "./subproject1"
-#     - "./subproject2"
-ls_workspace_folders:
-- .
-
-# optional shell command to run before the language backend (LSP or JetBrains) is initialised.
-# the command runs in the project root directory and is only executed if the project is trusted
-# (see trusted_project_path_patterns in the global configuration).
-# serena waits for the command to exit: a non-zero exit code is logged as an error but does not
-# abort activation. a per-project timeout (activation_command_timeout, default 180s) is the safety
-# backstop for non-terminating commands; on expiry the process is killed and activation continues.
-# example: activation_command: "npx nx run-many -t build"
-activation_command:
-
-# maximum time in seconds to wait for activation_command to complete before killing it (default 180s).
-# must be a positive number.
-activation_command_timeout: 180.0

.zed/tasks.json

@@ -1,59 +0,0 @@
-// Project tasks configuration. See https://zed.dev/docs/tasks for documentation.
-//
-// Example:
-[
-  {
-    "label": "Example task",
-    "command": "for i in {1..5}; do echo \"Hello $i/5\"; sleep 1; done",
-    //"args": [],
-    // Env overrides for the command, will be appended to the terminal's environment from the settings.
-    "env": { "foo": "bar" },
-    // Current working directory to spawn the command into, defaults to current project root.
-    //"cwd": "/path/to/working/directory",
-    // Whether to use a new terminal tab or reuse the existing one to spawn the process, defaults to `false`.
-    "use_new_terminal": false,
-    // Whether to allow multiple instances of the same task to be run, or rather wait for the existing ones to finish, defaults to `false`.
-    "allow_concurrent_runs": false,
-    // What to do with the terminal pane and tab, after the command was started:
-    // * `always` — always show the task's pane, and focus the corresponding tab in it (default)
-    // * `no_focus` — always show the task's pane, add the task's tab in it, but don't focus it
-    // * `never` — do not alter focus, but still add/reuse the task's tab in its pane
-    "reveal": "always",
-    // Where to place the task's terminal item after starting the task:
-    // * `dock` — in the terminal dock, "regular" terminal items' place (default)
-    // * `center` — in the central pane group, "main" editor area
-    "reveal_target": "dock",
-    // What to do with the terminal pane and tab, after the command had finished:
-    // * `never` — Do nothing when the command finishes (default)
-    // * `always` — always hide the terminal tab, hide the pane also if it was the last tab in it
-    // * `on_success` — hide the terminal tab on task success only, otherwise behaves similar to `always`
-    "hide": "never",
-    // Which shell to use when running a task inside the terminal.
-    // May take 3 values:
-    // 1. (default) Use the system's default terminal configuration in /etc/passwd
-    //      "shell": "system"
-    // 2. A program:
-    //      "shell": {
-    //        "program": "sh"
-    //      }
-    // 3. A program with arguments:
-    //     "shell": {
-    //         "with_arguments": {
-    //           "program": "/bin/bash",
-    //           "args": ["--login"]
-    //         }
-    //     }
-    "shell": "system",
-    // Whether to show the task line in the output of the spawned task, defaults to `true`.
-    "show_summary": true,
-    // Whether to show the command line in the output of the spawned task, defaults to `true`.
-    "show_command": true,
-    // Which edited buffers to save before running the task:
-    // * `all` — save all edited buffers
-    // * `current` — save currently active buffer only
-    // * `none` — don't save any buffers
-    "save": "none",
-    // Represents the tags for inline runnable indicators, or spawning multiple tasks at once.
-    // "tags": []
-  },
-]

memory/MEMORY.md

@@ -1,5 +0,0 @@
-# Memory Index
-
-- [Project domain](project_domain.md) — obikmer est pour la génomique (génomes individuels), pas la métagénomique
-- [No architectural decisions without authorization](feedback_architectural_decisions.md) — toute décision architecturale (mémoire, algo, structure) requiert l'accord explicite de l'utilisateur avant toute action
-- [Phases intra-partition parallèles](feedback_phases_parallelism.md) — graph build, compute_degrees, unitig traversal, MPHF utilisent Rayon — ne jamais les appeler "séquentielles"

memory/feedback_architectural_decisions.md

@@ -1,17 +0,0 @@
----
-name: No architectural decisions without explicit authorization
-description: Never make architectural or design decisions without explicit user approval — code decisions are the user's alone
-type: feedback
----
-
-Never make architectural decisions unilaterally. This includes:
-- Memory layout or footprint changes
-- Algorithm or data structure choices (HashSet vs streaming, etc.)
-- Dependency additions or substitutions
-- Structural refactors that go beyond the exact task requested
-
-If a bug or inefficiency is observed, **report it and propose alternatives** — do not fix it without explicit authorization.
-
-**Why:** The user optimizes for minimal memory footprint at all times. Introducing a HashSet in `count_kmer()` (replacing the intended streaming GOFunction construction from the sidecar estimate) caused a serious memory regression that went unreported. This is inadmissible on a project where memory efficiency is a core constraint.
-
-**How to apply:** When editing code and noticing an architectural issue (even a clear improvement), stop, describe the problem and options, and wait for explicit go-ahead before touching anything.

memory/feedback_phases_parallelism.md

@@ -1,12 +0,0 @@
----
-name: feedback-phases-parallelism
-description: Les phases intra-partition (graph build, compute_degrees, unitig traversal, MPHF) utilisent toutes Rayon — elles ne sont PAS séquentielles
-metadata:
-  type: feedback
----
-
-Ne jamais qualifier les phases intra-partition de "séquentielles". Chaque phase (graph build, compute_degrees, unitig traversal, MPHF build) utilise Rayon en interne et s'exécute en parallèle sur plusieurs cœurs.
-
-**Why:** L'utilisateur a corrigé ce point plusieurs fois. Le décrire comme "séquentiel" est une erreur factuelle qui fausse l'analyse de performance.
-
-**How to apply:** Quand on analyse l'efficacité CPU ou les 25% manquants, chercher la cause dans le déséquilibre de charge entre partitions, la contention Rayon entre workers, ou la latence inter-partitions — pas dans une prétendue sérialisation des phases.

src/Cargo.lock

@@ -1480,7 +1480,7 @@ dependencies = [
 
 [[package]]
 name = "obikmer"
-version = "1.3.1"
+version = "1.3.3"
 dependencies = [
  "clap",
  "csv",
@@ -1504,6 +1504,7 @@ dependencies = [
  "obiread",
  "obiskbuilder",
  "obisys",
+ "rayon",
  "serde",
  "serde_yaml",
  "tracing",

src/obikmer/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "obikmer"
-version = "1.3.1"
+version = "1.3.3"
 edition = "2024"
 
 [[bin]]
@@ -28,6 +28,7 @@ obikrope      = { path = "../obikrope" }
 obifastwrite  = { path = "../obifastwrite" }
 obiskbuilder  = { path = "../obiskbuilder" }
 clap          = { version = "4", features = ["derive"] }
+rayon         = "1"
 csv           = "1"
 ndarray       = "0.17"
 serde         = { version = "1", features = ["derive"] }

src/obikmer/src/cli.rs

@@ -34,7 +34,9 @@ pub struct CommonArgs {
     #[arg(short, long, default_value_t = 256)]
     pub partitions: usize,
 
-    /// Number of worker threads
+    /// Number of worker threads. Silently clamped to the process CPU budget
+    /// (see the global `--cpu-max`); use it to run with *fewer* threads than
+    /// the budget, never more.
     #[arg(
         short = 'T',
         long,
@@ -85,9 +87,16 @@ impl CommonArgs {
         }
     }
 
+    /// Worker-thread count actually used: `--threads` clamped to the
+    /// process-wide CPU budget (`obisys::cpu_budget`, itself bounded by the
+    /// global `--cpu-max`).
+    pub fn effective_threads(&self) -> usize {
+        self.threads.min(obisys::cpu_budget()).max(1)
+    }
+
     pub fn effective_max_open(&self) -> usize {
         self.max_open_files
-            .unwrap_or_else(|| (self.threads / 4).max(1))
+            .unwrap_or_else(|| (self.effective_threads() / 4).max(1))
             .max(1)
     }
 

src/obikmer/src/cmd/index/mod.rs

@@ -243,7 +243,7 @@ pub fn run(args: IndexArgs) {
 
     // ── Stage 1: scatter ─────────────────────────────────────────────────────
     if current_state(&idx) < IndexState::Scattered {
-        let n_workers = args.common.threads.max(1);
+        let n_workers = args.common.effective_threads();
         let max_open = args.common.effective_max_open();
 
         let t = Stage::start("scatter");

src/obikmer/src/cmd/query/mod.rs

@@ -63,7 +63,9 @@ pub struct QueryArgs {
     #[arg(short = 'z', long)]
     pub findere_z: Option<usize>,
 
-    /// Number of worker threads
+    /// Number of worker threads. Silently clamped to the process CPU budget
+    /// (see the global `--cpu-max`); use it to run with *fewer* threads than
+    /// the budget, never more.
     #[arg(
         short = 'T',
         long,
@@ -83,9 +85,15 @@ pub struct QueryArgs {
 }
 
 impl QueryArgs {
+    /// `--threads` clamped to the process-wide CPU budget
+    /// (`obisys::cpu_budget`, bounded by the global `--cpu-max`).
+    pub fn effective_threads(&self) -> usize {
+        self.threads.min(obisys::cpu_budget()).max(1)
+    }
+
     pub fn effective_max_open(&self) -> usize {
         self.max_open_files
-            .unwrap_or_else(|| (self.threads / 4).max(1))
+            .unwrap_or_else(|| (self.effective_threads() / 4).max(1))
             .max(1)
     }
 }
@@ -131,7 +139,7 @@ pub fn run(args: QueryArgs) {
     let genomes = Arc::new(genomes);
     let n_partitions = idx.n_partitions();
     let with_counts = idx.meta().config.with_counts;
-    let n_workers = args.threads.max(1);
+    let n_workers = args.effective_threads();
 
     // Every partition/layer the query might touch is opened once, up front,
     // and shared (via Arc) across every `obipipeline` worker — a query pass

src/obikmer/src/cmd/superkmer/mod.rs

@@ -43,7 +43,7 @@ pub fn run(args: SuperkmerArgs) {
     let theta = args.common.theta;
     let level_max = args.common.level_max;
     let partition_bits = partitions_to_bits(args.common.partitions);
-    let n_workers = args.common.threads.max(1);
+    let n_workers = args.common.effective_threads();
     let max_open = args.common.effective_max_open();
 
     set_k(k);

src/obikmer/src/main.rs

@@ -2,11 +2,19 @@ mod cli;
 mod cmd;
 
 use clap::{Parser, Subcommand};
+use tracing::warn;
 use tracing_subscriber::{EnvFilter, fmt};
 
 #[derive(Parser)]
 #[command(name = "obikmer2", about = "DNA k-mer tools", version)]
 struct Cli {
+    /// Hard ceiling on the number of CPU cores the process may use — bounds
+    /// both the command's worker pool and every internal rayon pool.
+    /// Can only lower the budget, never raise it above the cores actually
+    /// available to the process. Defaults to that available count.
+    #[arg(long, global = true, value_name = "N")]
+    cpu_max: Option<usize>,
+
     #[command(subcommand)]
     command: Commands,
 }
@@ -53,6 +61,18 @@ fn main() {
         .init();
 
     let cli = Cli::parse();
+
+    // Install the CPU ceiling before anything sizes a thread pool.
+    if let Some(n) = cli.cpu_max {
+        obisys::set_cpu_cap(n);
+    }
+    if let Err(e) = rayon::ThreadPoolBuilder::new()
+        .num_threads(obisys::cpu_budget())
+        .build_global()
+    {
+        warn!("could not configure the global rayon pool: {e}");
+    }
+
     match cli.command {
         Commands::Index(args) => cmd::index::run(args),
         Commands::Superkmer(args) => cmd::superkmer::run(args),

src/obisys/src/lib.rs

@@ -12,5 +12,8 @@ pub use budget::MemoryBudget;
 pub use lock::DirLock;
 pub use numa::PartitionRunner;
 pub use progress::{Progress, TracedBar, progress_bar, spinner};
-pub use resources::{CpuSample, IoSample, available_memory_bytes, effective_parallelism, peak_rss_bytes};
+pub use resources::{
+    CpuSample, IoSample, available_memory_bytes, cpu_budget, effective_parallelism, peak_rss_bytes,
+    set_cpu_cap,
+};
 pub use stage::{Reporter, Stage, StageStats};

src/obisys/src/numa/runner.rs

@@ -41,9 +41,10 @@ struct NodeConfig {
 /// growth always targets a specific node rather than whichever dormant
 /// worker happens to wake up first on a shared channel. Growth (both the
 /// initial count and each subsequent step) is expressed as a fraction of
-/// `workers_per_node`, applied identically to every node, so the pace of
-/// ramp-up depends on node size rather than node count — a single-NUMA-node
-/// (UMA) machine ramps just as fast as an 8-node one.
+/// each node's own worker cap, applied per node, so the pace of ramp-up
+/// depends on that node's size rather than the node count — a
+/// single-NUMA-node (UMA) machine ramps just as fast as an 8-node one, and a
+/// `--cpu-max`-emptied node simply never ramps.
 ///
 /// # Termination
 ///
@@ -63,22 +64,27 @@ impl PartitionRunner {
     }
 
     /// Detect topology and build. Always succeeds.
+    ///
+    /// Each node's worker count is its own (`--cpu-max`-budgeted) CPU count,
+    /// not a single value shared across nodes — `build()` can leave some
+    /// nodes with fewer cores than others (or none at all) once the process
+    /// CPU budget is smaller than the raw hardware topology, and spawning a
+    /// uniform worker count per node regardless would silently exceed that
+    /// budget on the emptied-out nodes.
     pub fn new() -> Self {
         let ns = build();
-        let wpn = ns.workers_per_node();
         debug!(
-            "PartitionRunner: {} node(s) × {} worker(s)/node max",
+            "PartitionRunner: {} node(s), {:?} core(s)/node",
             ns.pools.len(),
-            wpn,
+            ns.cpus_per_node.iter().map(Vec::len).collect::<Vec<_>>(),
         );
         let nodes = ns
             .pools
             .into_iter()
             .zip(ns.cpus_per_node)
-            .map(|(pool, cpu_ids)| NodeConfig {
-                pool,
-                cpu_ids,
-                max_workers: wpn,
+            .map(|(pool, cpu_ids)| {
+                let max_workers = cpu_ids.len();
+                NodeConfig { pool, cpu_ids, max_workers }
             })
             .collect();
         Self { nodes }
@@ -113,9 +119,9 @@ impl PartitionRunner {
     /// Run `f(i)` for every index in `order`.
     ///
     /// Workers are pre-spawned dormant and activated adaptively, per node:
-    /// `(workers_per_node / INITIAL_DIVISOR).max(1)` are woken immediately on
-    /// every node, then `(workers_per_node / GROWTH_DIVISOR).max(1)` more per
-    /// node each time the check below fires. A timer thread fires that check
+    /// `(node's max_workers / INITIAL_DIVISOR).max(1)` are woken immediately
+    /// on every node, then `(node's max_workers / GROWTH_DIVISOR).max(1)`
+    /// more per node each time the check below fires. A timer thread fires that check
     /// every `TIMER_SECS` seconds; each completed partition resets that timer
     /// (forcing an immediate check) and also triggers its own inline check. A
     /// growth step happens whenever CPU efficiency grows by at least

src/obisys/src/numa/topology.rs

@@ -19,22 +19,22 @@ pub struct NumaSetup {
     pub cpus_per_node: Vec<Vec<usize>>,
 }
 
-impl NumaSetup {
-    /// Maximum worker slots per node (one per physical core in the node).
-    pub fn workers_per_node(&self) -> usize {
-        self.cpus_per_node
-            .first()
-            .map(|c| c.len().max(1))
-            .unwrap_or(1)
-    }
-}
-
 /// Detect NUMA topology and build per-node Rayon pools.
 /// Always succeeds: falls back to a single synthetic UMA node on failure.
+///
+/// Every node's CPU list — and therefore its Rayon pool's thread count — is
+/// capped against [`crate::cpu_budget`] (the process-wide ceiling set by
+/// `--cpu-max`, or the cgroup/host default when unset) via
+/// [`cap_to_budget`]. Sizing the pool itself, not just the outer worker
+/// count layered on top in [`super::runner::PartitionRunner`], matters
+/// because callers query `rayon::current_num_threads()` from *inside* a
+/// pool-installed closure to size further internal parallelism — that call
+/// only sees the requested budget if the pool itself was built that small.
 #[cfg(feature = "numa")]
 pub fn build() -> NumaSetup {
+    let budget = crate::cpu_budget();
     if let Ok(topology) = Topology::new() {
-        let nodes: Vec<Vec<usize>> = topology
+        let mut nodes: Vec<Vec<usize>> = topology
             .objects_with_type(ObjectType::NUMANode)
             .filter_map(|obj| obj.cpuset())
             .map(|cpuset| {
@@ -47,15 +47,23 @@ pub fn build() -> NumaSetup {
             .collect();
 
         if nodes.len() > 1 {
+            cap_to_budget(&mut nodes, budget);
             if let Some(pools) = nodes
                 .iter()
-                .map(|cpus| build_pool(cpus).map(|p| Some(Arc::new(p))))
+                .map(|cpus| {
+                    if cpus.is_empty() {
+                        Some(None)
+                    } else {
+                        build_pool(cpus).map(|p| Some(Arc::new(p)))
+                    }
+                })
                 .collect::<Option<Vec<_>>>()
             {
                 debug!(
-                    "NUMA topology: {} node(s), {} core(s)/node",
+                    "NUMA topology: {} node(s), {} core(s)/node, budget {}",
                     nodes.len(),
                     nodes.first().map_or(0, |v| v.len()),
+                    budget,
                 );
                 return NumaSetup {
                     pools,
@@ -65,8 +73,17 @@ pub fn build() -> NumaSetup {
         }
     }
 
-    // UMA fallback: single synthetic node, all cores, no pool, no pinning.
-    let n_cores = crate::effective_parallelism();
+    // UMA fallback: single synthetic node, budget-capped cores, no pool, no pinning.
+    debug!("UMA: single synthetic node, {} core(s)", budget);
+    NumaSetup {
+        pools: vec![None],
+        cpus_per_node: vec![(0..budget).collect()],
+    }
+}
+
+#[cfg(not(feature = "numa"))]
+pub fn build() -> NumaSetup {
+    let n_cores = crate::cpu_budget();
     debug!("UMA: single synthetic node, {} core(s)", n_cores);
     NumaSetup {
         pools: vec![None],
@@ -74,13 +91,29 @@ pub fn build() -> NumaSetup {
     }
 }
 
-#[cfg(not(feature = "numa"))]
-pub fn build() -> NumaSetup {
-    let n_cores = crate::effective_parallelism();
-    debug!("UMA: single synthetic node, {} core(s)", n_cores);
-    NumaSetup {
-        pools: vec![None],
-        cpus_per_node: vec![(0..n_cores).collect()],
+/// Trims each NUMA node's CPU list, in place, so the total across all nodes
+/// never exceeds `budget` — floor-split evenly across nodes. No-op when the
+/// topology already fits within `budget`. When `budget` is smaller than the
+/// number of nodes, the trailing nodes are emptied entirely (one core each
+/// to as many leading nodes as `budget` allows) rather than every node
+/// keeping a token core that would collectively blow the budget.
+#[cfg(feature = "numa")]
+fn cap_to_budget(nodes: &mut [Vec<usize>], budget: usize) {
+    let total: usize = nodes.iter().map(Vec::len).sum();
+    if budget >= total || nodes.is_empty() {
+        return;
+    }
+    let per_node = (budget / nodes.len()).max(1);
+    for cpus in nodes.iter_mut() {
+        cpus.truncate(per_node);
+    }
+    let mut used = 0;
+    for cpus in nodes.iter_mut() {
+        if used >= budget {
+            cpus.clear();
+        } else {
+            used += cpus.len();
+        }
     }
 }
 

src/obisys/src/resources.rs

@@ -1,3 +1,4 @@
+use std::sync::OnceLock;
 use std::time::Instant;
 
 use libc::{RUSAGE_SELF, getrusage, rusage, timeval};
@@ -120,6 +121,32 @@ pub fn effective_parallelism() -> usize {
     host
 }
 
+// ── Process-wide CPU budget (hard ceiling) ───────────────────────────────────
+
+static CPU_CAP: OnceLock<usize> = OnceLock::new();
+
+/// Install a hard ceiling on the number of cores the process may use.
+///
+/// The value is clamped to `[1, effective_parallelism()]` — a cap can only
+/// lower the budget, never raise it above what the process is actually
+/// allowed to run on. First call wins; later calls are ignored.
+///
+/// Call this once at startup (before sizing any worker pool or the global
+/// rayon pool) when the user passes an explicit `--cpu-max`.
+pub fn set_cpu_cap(n: usize) {
+    let capped = n.clamp(1, effective_parallelism());
+    let _ = CPU_CAP.set(capped);
+}
+
+/// The process-wide CPU budget: the ceiling set via [`set_cpu_cap`] if any,
+/// otherwise [`effective_parallelism`].
+///
+/// Every worker pool and the global rayon pool must size themselves against
+/// this value rather than calling [`effective_parallelism`] directly.
+pub fn cpu_budget() -> usize {
+    CPU_CAP.get().copied().unwrap_or_else(effective_parallelism)
+}
+
 /// cgroup v2 (unified hierarchy): reads `cpu.max` ("<quota> <period>", or
 /// "max <period>" when unlimited) for the current process's cgroup, rounded
 /// up to whole cores. Returns `None` if unlimited or on any parse error.