Compare commits
8
Commits
v1.3.1
...
e846d35adb
| Author | SHA1 | Date | |
|---|---|---|---|
 coissac
|
e846d35adb | ||
 Eric Coissac
|
c0e1968654 | ||
|
coissac
|
e280b6d0a3 | ||
|
Eric Coissac
|
98dba1802d | ||
|
coissac
|
dd4285b269 | ||
|
Eric Coissac
|
b66a488277 | ||
|
Eric Coissac
|
b954032faa | ||
|
coissac
|
43d89d8f02 |
@@ -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
|
||||
@@ -1,2 +0,0 @@
|
||||
/cache
|
||||
/project.local.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
|
||||
@@ -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": []
|
||||
},
|
||||
]
|
||||
@@ -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"
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
Generated
+2
-1
@@ -1480,7 +1480,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "obikmer"
|
||||
version = "1.3.1"
|
||||
version = "1.3.4"
|
||||
dependencies = [
|
||||
"clap",
|
||||
"csv",
|
||||
@@ -1504,6 +1504,7 @@ dependencies = [
|
||||
"obiread",
|
||||
"obiskbuilder",
|
||||
"obisys",
|
||||
"rayon",
|
||||
"serde",
|
||||
"serde_yaml",
|
||||
"tracing",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "obikmer"
|
||||
version = "1.3.1"
|
||||
version = "1.3.4"
|
||||
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"] }
|
||||
|
||||
+11
-2
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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,22 @@ 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}");
|
||||
}
|
||||
|
||||
// Always-on, near-zero-cost when disabled: RUST_LOG=obisys::profiler=trace
|
||||
// to see periodic CPU/IO/memory snapshots on a production run.
|
||||
obisys::start_profiler(std::time::Duration::from_secs(5));
|
||||
|
||||
match cli.command {
|
||||
Commands::Index(args) => cmd::index::run(args),
|
||||
Commands::Superkmer(args) => cmd::superkmer::run(args),
|
||||
|
||||
@@ -4,6 +4,7 @@
|
||||
mod budget;
|
||||
mod lock;
|
||||
mod numa;
|
||||
mod profiler;
|
||||
mod progress;
|
||||
mod resources;
|
||||
mod stage;
|
||||
@@ -11,6 +12,10 @@ mod stage;
|
||||
pub use budget::MemoryBudget;
|
||||
pub use lock::DirLock;
|
||||
pub use numa::PartitionRunner;
|
||||
pub use profiler::start as start_profiler;
|
||||
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};
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
use std::thread;
|
||||
use std::time::Duration;
|
||||
|
||||
use tracing::{Level, trace};
|
||||
|
||||
use crate::{CpuSample, IoSample, available_memory_bytes, cpu_budget, peak_rss_bytes};
|
||||
|
||||
/// Spawn a detached background thread that logs a periodic snapshot of
|
||||
/// process-wide resource usage — CPU cores used (and as a % of the process's
|
||||
/// `--cpu-max` budget), I/O throughput, peak RSS, and remaining host/cgroup
|
||||
/// memory — at `trace!` level, every `interval`.
|
||||
///
|
||||
/// Meant to run for the whole process lifetime as always-on instrumentation
|
||||
/// installed once at startup (no stop handle). The sampling itself (a
|
||||
/// `getrusage` call and a `/proc/self/io` read) is cheap, but
|
||||
/// `available_memory_bytes()` builds a fresh `sysinfo::System` snapshot each
|
||||
/// time, so the whole body is skipped unless the `obisys::profiler` target
|
||||
/// is actually enabled at `TRACE`.
|
||||
///
|
||||
/// Enable with `RUST_LOG=obisys::profiler=trace` (combine with
|
||||
/// `RUST_LOG=debug,obisys::profiler=trace` to also get the per-partition
|
||||
/// `debug!` logs that already exist elsewhere) to profile a run directly on
|
||||
/// production hardware without attaching a separate profiler.
|
||||
pub fn start(interval: Duration) {
|
||||
let budget = cpu_budget().max(1) as f64;
|
||||
thread::Builder::new()
|
||||
.name("obisys-profiler".into())
|
||||
.spawn(move || {
|
||||
let mut cpu = CpuSample::now();
|
||||
let mut io = IoSample::now();
|
||||
loop {
|
||||
thread::sleep(interval);
|
||||
if !tracing::enabled!(target: "obisys::profiler", Level::TRACE) {
|
||||
continue;
|
||||
}
|
||||
let cores = cpu.sample_cores().unwrap_or(0.0);
|
||||
let io_mb_s = io.sample_rate().unwrap_or(0.0) / 1_000_000.0;
|
||||
trace!(
|
||||
target: "obisys::profiler",
|
||||
cpu_pct = format_args!("{:.0}%", cores / budget * 100.0),
|
||||
cpu_cores = format_args!("{cores:.1}"),
|
||||
io_mb_s = format_args!("{io_mb_s:.1}"),
|
||||
rss_peak_mb = peak_rss_bytes() / 1_000_000,
|
||||
mem_avail_mb = available_memory_bytes() / 1_000_000,
|
||||
"profiler snapshot",
|
||||
);
|
||||
}
|
||||
})
|
||||
.expect("failed to spawn profiler thread");
|
||||
}
|
||||
+65
-16
@@ -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.
|
||||
@@ -237,18 +264,31 @@ impl CpuSample {
|
||||
cpu / (wall * n_cores as f64)
|
||||
}
|
||||
|
||||
pub fn do_i_activate(&mut self, threshold: f64) -> bool {
|
||||
/// Cores of CPU time consumed per wall-clock second since the last call
|
||||
/// (or since [`now`](Self::now) for the first call) — an absolute core
|
||||
/// count, unlike [`cpu_efficiency`](Self::cpu_efficiency) which
|
||||
/// normalizes to `n_cores` and never resets its baseline. Resets the
|
||||
/// window on every call that returns `Some`; returns `None` (state left
|
||||
/// untouched, so the window keeps accumulating) when less than 100 ms
|
||||
/// have elapsed — too short to be meaningful.
|
||||
pub fn sample_cores(&mut self) -> Option<f64> {
|
||||
let delta_wall = self.wall.elapsed().as_secs_f64();
|
||||
if delta_wall < 0.1 {
|
||||
// Window too short to be meaningful — leave state untouched so it
|
||||
// keeps accumulating until a real sample can be taken.
|
||||
return false;
|
||||
return None;
|
||||
}
|
||||
|
||||
let n = CpuSample::now();
|
||||
let delta_ru = (n.user_secs - self.user_secs) + (n.sys_secs - self.sys_secs);
|
||||
self.user_secs = n.user_secs;
|
||||
self.sys_secs = n.sys_secs;
|
||||
self.wall = n.wall;
|
||||
Some(delta_ru / delta_wall)
|
||||
}
|
||||
|
||||
pub fn do_i_activate(&mut self, threshold: f64) -> bool {
|
||||
let Some(efficiency) = self.sample_cores() else {
|
||||
return false;
|
||||
};
|
||||
|
||||
let efficiency = delta_ru / delta_wall;
|
||||
let activate = 0f64.max(efficiency - self.previous) >= threshold;
|
||||
|
||||
debug!(
|
||||
@@ -259,9 +299,6 @@ impl CpuSample {
|
||||
activate
|
||||
);
|
||||
self.previous = efficiency;
|
||||
self.user_secs = n.user_secs;
|
||||
self.sys_secs = n.sys_secs;
|
||||
self.wall = n.wall;
|
||||
|
||||
activate
|
||||
}
|
||||
@@ -324,18 +361,32 @@ impl IoSample {
|
||||
0
|
||||
}
|
||||
|
||||
/// Bytes/second submitted to the block layer since the last call (or
|
||||
/// since [`now`](Self::now) for the first call). Resets the window on
|
||||
/// every call that returns `Some`; returns `None` (state left
|
||||
/// untouched) when less than 100 ms have elapsed — too short to be
|
||||
/// meaningful.
|
||||
pub fn sample_rate(&mut self) -> Option<f64> {
|
||||
let elapsed = self.wall.elapsed().as_secs_f64();
|
||||
if elapsed < 0.1 {
|
||||
return None;
|
||||
}
|
||||
let n = Self::read_bytes();
|
||||
let rate = n.saturating_sub(self.bytes) as f64 / elapsed;
|
||||
self.bytes = n;
|
||||
self.wall = Instant::now();
|
||||
Some(rate)
|
||||
}
|
||||
|
||||
/// Same protocol as [`CpuSample::do_i_activate`] (0.1 s minimum window,
|
||||
/// state untouched on early return), but growth is measured relative to
|
||||
/// the previous rate. `threshold` is a fraction, e.g. `0.2` for a 20 %
|
||||
/// increase in throughput since the last real sample.
|
||||
pub fn do_i_activate(&mut self, threshold: f64) -> bool {
|
||||
let elapsed = self.wall.elapsed().as_secs_f64();
|
||||
if elapsed < 0.1 {
|
||||
let Some(rate) = self.sample_rate() else {
|
||||
return false;
|
||||
}
|
||||
};
|
||||
|
||||
let n = Self::read_bytes();
|
||||
let rate = n.saturating_sub(self.bytes) as f64 / elapsed;
|
||||
let activate = if self.previous_rate == 0.0 {
|
||||
rate > 0.0 // bootstrap: any measured throughput is signal enough
|
||||
} else {
|
||||
@@ -347,8 +398,6 @@ impl IoSample {
|
||||
self.previous_rate, rate, activate
|
||||
);
|
||||
self.previous_rate = rate;
|
||||
self.bytes = n;
|
||||
self.wall = Instant::now();
|
||||
|
||||
activate
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user