diff --git a/docmd/architecture/siblings.md b/docmd/architecture/siblings.md index b7012bc2..3df7c6aa 100644 --- a/docmd/architecture/siblings.md +++ b/docmd/architecture/siblings.md @@ -18,7 +18,8 @@ method that reconstructs a kmer from a bare slot number is wrong by construction, regardless of the mechanism used (MPHF re-hash, or evidence decode + direct unitig read). See `MphfLayer::kmer_at` (`obilayeredmap/src/mphf_layer.rs`) — flagged for removal, currently called -from `obikindex/siblings/build.rs:122` and `family_scan.rs:173`. +from `obikphylo/siblings/build.rs` and `family_scan.rs` (since removed — see +"Pending work" status below). ## Two pipelines, never mixed @@ -39,7 +40,7 @@ pointless even in `Exact`/`Hybrid` mode since the kmer was already known. The sibling annex (`FamilyMask`/`SiblingAnnex`, `.psib`, `obicompactvec/src/siblingannex.rs`) records, per kmer, whether it is a family minorant and which family members are present in the index. Its only -consumers (`obikindex/siblings/stats.rs`, `family_scan.rs`) enumerate it +consumers (`obikphylo/siblings/stats.rs`, `family_scan.rs`) enumerate it exhaustively (`0..annex.len()`); no query-pipeline code path touches it. **Decision**: the annex must be persisted in iteration order, not slot @@ -66,12 +67,99 @@ partition is unknown) and must keep going through `KmerPartition::query_partition_with` (MPHF + evidence), never a raw `index()`. -## Pending work +## Pending work — done -- Remove `MphfLayer::kmer_at`. -- `siblings/build.rs`: build `slot_kmer`-equivalent via iteration, not - `kmer_at`; thread iteration index instead of slot through the - variant/reconciliation pipeline; persist the annex in iteration order. -- `siblings/family_scan.rs`, `stats.rs`: read the annex via zipped - iteration (`iter_kmers().zip(annex_iter)`) instead of `0..annex.len()` + - `kmer_at`. +The plan above shipped: `obikphylo` (a new crate — phylo-domain extension +traits over `obikindex::KmerIndex`/`obilayeredmap::Layer`, replacing the +old `obikindex::siblings` module) builds and reads the annex purely in +iteration order (`SiblingLayerExt::iter_siblings`/`iter_minorants`, both with +batch variants, mirroring `Layer`'s own `KmerIter`/`KmerBatchIter` +shape). `MphfLayer::kmer_at` has no remaining callers. + +A separate, unrelated bug surfaced during this work and was fixed +(2026-08-14): `MphfLayer::enumerate_kmers_batch` computed its +`batch_start_index` via the stdlib `.enumerate()` adapter, which counts +*batches* (0, 1, 2…), not the cumulative k-mer offset the annex is actually +keyed on — every batch past the first wrote its mask/annex entries at the +wrong iteration-order position. Fixed by tracking a running offset instead; +regression tests added (`sibling_annex_no_empty_masks_after_build`, +`sibling_histogram_does_not_panic_on_partial_last_batch`). + +## Performance: `build_sibling_annex` parallelism (2026-08-14) + +Investigated on a real multi-genome run (`phyloskims_sal_vac`, k=31/m=11). +Baseline: mostly one active core, with short multi-core bursts — average +~3 cores. + +**Fixes that helped, kept:** + +- `CanonicalKmerOf::minimizer()` (`obikseq/src/kmer.rs`) — a direct O(k) + bit-arithmetic minimiser for a single isolated k-mer, replacing a + `RollingStat` instance fed byte-by-byte through an ASCII round-trip (used + by `helpers::partition_of`, called for every generated family variant). + ~3x wall-clock improvement on its own, confirmed by sampling + (`obiskbuilder::rolling_stat`/`obikentropy` frames disappeared from the + hot path). `CanonicalKmerOf::partition()` added alongside it (wraps + `minimizer().seq_hash() & mask`, the same routing rule + `KmerPartition`/`RoutableSuperKmer` use). +- Cross-partition resolution (`outgoing.par_iter()` in + `build_layer_sibling_annex`) parallelised at the *partition* level — one + Rayon task per non-empty `outgoing[dest]` bucket. For k=31/m=11, a + central-base substitution changes the winning minimiser (and thus the + destination partition) only when that window overlaps the central base: + ~11 of the 21 possible windows do, so ~10/21 (≈48%) of generated variants + route right back to the partition already being built. That self bucket + ends up far larger than any other, so the per-partition split pinned one + thread to it alone while the rest of the pool finished instantly — + confirmed by sampling: one thread solid in `MphfLayer::find`, everyone + else idle. Fixed by splitting each non-empty bucket into + `total_queries / n_workers` (capped 4096) chunks *before* `par_iter()`, + preserving per-partition mmap locality (each chunk stays contiguous + within one partition) while letting Rayon spread an oversized bucket + across several threads. Net effect of both fixes together: ~3 cores + average → ~10-13 cores average on the same run, and a projected total + build time of ~1h15 down to ~30min on the real `phyloskims_sal_vac` run + this was measured against. +- `TracedBar`'s ETA (`obisys/src/progress.rs`) was silently starved: the + custom progress message and the self-computed ETA text used to share one + `pb.set_message()` slot, with the ETA holding off for 2s after any custom + message — fine when custom messages are rare, broken once + `build_sibling_annex`'s per-partition callback fires more often than + that. Fixed by keeping the two texts in separate fields, composed + together on every render instead of one overwriting the other. + +**Tried and reverted — do not repeat blindly:** + +- Parallelising the *outer* partition loop in `build_sibling_annex` with + `obikindex::PartitionRunner` (already used by `merge`/`build_layers`), + splitting a fixed core budget between outer (partition) and inner + (pipeline + resolution) concurrency so their product wouldn't exceed the + budget. Measured *worse*: throughput dropped over time (26 + partitions/5min → 38/11-12min) and peak resolution concurrency fell from + ~11-12 cores to ~7-8. Cause: this capped the resolution burst — which + scales very well on its own — to make room for outer concurrency, and + running several partitions' resolution at once scatters access across + multiple partitions' mmap regions at once, working against the + locality `outgoing`'s per-partition grouping exists for. `PartitionRunner` + stayed exported from `obikindex` (`new_capped` too) since it's + general-purpose, but nothing in `obikphylo` calls it. +- Splitting resolution chunks even finer (`/(n_workers*8)`, cap 1024, + instead of `/n_workers`, cap 4096) to smooth the residual sawtooth. + Measured ~10% *slower*, wider dips, not narrower. Reverted to the + original chunk sizing. + +**Known remaining limitation, not yet worth fixing:** within one layer, the +four stages (sequential `unitigs.bin` read → parallel generation → +parallel resolution → sequential annex write) never overlap — confirmed by +1s-interval sampling: generation alone occupies ~17 threads evenly, but the +next layer's read/generation never starts until the current layer's +resolution and write are both done. This produces a real, periodic (~layer +duration) alternation between "many cores" and "few cores" that neither of +the fixes above touches, since both operate *within* one layer's resolution +step. The only remaining lever is overlapping consecutive layers (e.g. a +depth-2 pipeline: start layer N+1's read/generation while layer N's +resolution/write is still running) — a real restructuring, not a parameter +tweak, and explicitly *not* to be combined with the reverted +budget-capping idea above (let each phase use however many cores it +naturally wants; only the *scheduling* needs to overlap). Deferred, not +started. diff --git a/src/obicompactvec/src/tests/bitvec.rs b/src/obicompactvec/src/tests/bitvec.rs index 46694895..7214ae69 100644 --- a/src/obicompactvec/src/tests/bitvec.rs +++ b/src/obicompactvec/src/tests/bitvec.rs @@ -214,3 +214,66 @@ fn hamming_dist_basic() { let (_db, rb) = make_bv(&[true, false, true, true]); assert_eq!(ra.hamming_dist(&rb), 2); } + +// ── get_batch tests ──────────────────────────────────────────────────────────── + +#[test] +fn bitvec_get_batch_in_order() { + let bits = vec![true, false, true, false, true]; + let (_dir, r) = make_bv(&bits); + let got = r.get_batch(&[0, 1, 2, 3, 4]); + assert_eq!(got, bits); +} + +#[test] +fn bitvec_get_batch_out_of_order() { + let bits = vec![true, false, true, false, true]; + let (_dir, r) = make_bv(&bits); + let got = r.get_batch(&[3, 0, 4, 1]); + assert_eq!(got, vec![false, true, true, false]); +} + +#[test] +fn bitvec_get_batch_with_duplicates() { + let bits = vec![true, false, true]; + let (_dir, r) = make_bv(&bits); + let got = r.get_batch(&[0, 2, 0, 1]); + assert_eq!(got, vec![true, true, true, false]); +} + +#[test] +fn bitvec_get_batch_empty() { + let (_dir, r) = make_bv(&[true, false]); + let got = r.get_batch(&[]); + assert!(got.is_empty()); +} + +#[test] +fn bitvec_get_batch_out_of_bounds_returns_false() { + // PersistentBitVec::get does NOT bounds-check — out-of-range slots read + // whatever bit happens to be in the mmap (zero-initialised, so false). + // This is a known gap; get_batch inherits it. + let (_dir, r) = make_bv(&[true, false]); + let got = r.get_batch(&[0, 2]); + assert_eq!(got, vec![true, false]); +} + +// BitSliceView get_batch (same logic, exercised through the view) +#[test] +fn bitslice_view_get_batch() { + let bits = vec![true, false, true, false, true]; + let (_dir, r) = make_bv(&bits); + let view = r.view(); + assert_eq!(view.get_batch(&[0, 1, 2]), vec![true, false, true]); + assert_eq!(view.get_batch(&[4, 3]), vec![true, false]); +} + +#[test] +fn bitslice_view_get_batch_out_of_bounds_returns_false() { + // BitSliceView::get does NOT bounds-check either. + let bits = vec![true, false]; + let (_dir, r) = make_bv(&bits); + let view = r.view(); + let got = view.get_batch(&[0, 2]); + assert_eq!(got, vec![true, false]); +} diff --git a/src/obicompactvec/src/tests/intmatrix.rs b/src/obicompactvec/src/tests/intmatrix.rs index 9abd7b59..966f6da5 100644 --- a/src/obicompactvec/src/tests/intmatrix.rs +++ b/src/obicompactvec/src/tests/intmatrix.rs @@ -1,6 +1,6 @@ use tempfile::tempdir; -use crate::{pack_compact_int_matrix, PersistentCompactIntMatrix, PersistentCompactIntMatrixBuilder}; +use crate::{pack_compact_int_matrix, PersistentCompactIntMatrix, PersistentCompactIntMatrixBuilder, PersistentCompactIntVec, PersistentCompactIntVecBuilder, IntSliceView}; use crate::traits::CountPartials; fn make_matrix(cols: &[&[u32]]) -> (tempfile::TempDir, PersistentCompactIntMatrix) { @@ -320,3 +320,63 @@ fn partial_relfreq_bray_additive_across_split() { } } } + +// ── get_batch tests ──────────────────────────────────────────────────────────── + +fn make_pciv(counts: &[u32]) -> (tempfile::TempDir, PersistentCompactIntVec) { + let dir = tempdir().unwrap(); + let path = dir.path().join("c.pciv"); + let mut b = PersistentCompactIntVecBuilder::new(counts.len(), &path).unwrap(); + for (i, &v) in counts.iter().enumerate() { b.set(i, v); } + b.close().unwrap(); + let r = PersistentCompactIntVec::open(&path).unwrap(); + (dir, r) +} + +#[test] +fn pciv_get_batch_in_order() { + let counts = vec![10u32, 255, 300, 1000]; + let (_dir, v) = make_pciv(&counts); + let got = v.get_batch(&[0, 1, 2, 3]); + assert_eq!(got, counts); +} + +#[test] +fn pciv_get_batch_out_of_order() { + let counts = vec![10u32, 255, 300, 1000]; + let (_dir, v) = make_pciv(&counts); + let got = v.get_batch(&[3, 0, 2, 1]); + assert_eq!(got, vec![1000, 10, 300, 255]); +} + +#[test] +fn pciv_get_batch_with_duplicates() { + let counts = vec![10u32, 255, 300]; + let (_dir, v) = make_pciv(&counts); + let got = v.get_batch(&[0, 2, 0, 1]); + assert_eq!(got, vec![10, 300, 10, 255]); +} + +#[test] +fn pciv_get_batch_empty() { + let (_dir, v) = make_pciv(&[10u32, 20]); + let got: Vec = v.get_batch(&[]); + assert!(got.is_empty()); +} + +#[test] +fn pciv_get_batch_out_of_bounds_panics() { + let (_dir, v) = make_pciv(&[10u32, 20]); + let result = std::panic::catch_unwind(|| v.get_batch(&[0, 2])); + assert!(result.is_err(), "get_batch should panic on out-of-bounds slot"); +} + +// IntSliceView get_batch (same logic, exercised through the view) +#[test] +fn intslice_view_get_batch() { + let counts = vec![10u32, 255, 300, 1000]; + let (_dir, v) = make_pciv(&counts); + let view = v.view(); + assert_eq!(view.get_batch(&[0, 1, 2]), vec![10, 255, 300]); + assert_eq!(view.get_batch(&[3, 1]), vec![1000, 255]); +} diff --git a/src/obikindex/src/lib.rs b/src/obikindex/src/lib.rs index 6ab2e070..85f4a0ed 100644 --- a/src/obikindex/src/lib.rs +++ b/src/obikindex/src/lib.rs @@ -20,3 +20,4 @@ pub use meta::{validate_label, GenomeInfo, IndexConfig, IndexMeta, META_FILENAME pub use predicate::{GroupFilterParams, MetaPred}; pub use state::{IndexState, SENTINEL_COUNTED, SENTINEL_INDEXED, SENTINEL_SCATTERED}; pub use stats::IndexBitsPerKmer; +pub use numa::PartitionRunner; diff --git a/src/obikindex/src/numa/runner.rs b/src/obikindex/src/numa/runner.rs index 9572c78d..483fe675 100644 --- a/src/obikindex/src/numa/runner.rs +++ b/src/obikindex/src/numa/runner.rs @@ -82,6 +82,32 @@ impl PartitionRunner { Self { nodes } } + /// Like [`new`](Self::new), but caps total worker slots (summed across + /// nodes) at `max_total_workers` — split evenly across nodes, each + /// further capped by that node's actual core count. For callers whose + /// own per-worker closure does further internal parallel work (so the + /// natural per-node core count would oversubscribe if used as the + /// *outer* degree of parallelism too). + pub fn new_capped(max_total_workers: usize) -> Self { + let ns = build(); + let n_nodes = ns.pools.len().max(1); + let per_node_cap = (max_total_workers / n_nodes).max(1); + debug!( + "PartitionRunner (capped): {} node(s) × up to {} worker(s)/node ({} total requested)", + n_nodes, per_node_cap, max_total_workers, + ); + let nodes = ns + .pools + .into_iter() + .zip(ns.cpus_per_node) + .map(|(pool, cpu_ids)| { + let node_cores = cpu_ids.len().max(1); + NodeConfig { pool, cpu_ids, max_workers: per_node_cap.min(node_cores) } + }) + .collect(); + Self { nodes } + } + /// Run `f(i)` for every index in `order`. /// /// Workers are pre-spawned dormant and activated adaptively, per node: diff --git a/src/obikphylo/src/siblings/build.rs b/src/obikphylo/src/siblings/build.rs index 70955d49..b81d160b 100644 --- a/src/obikphylo/src/siblings/build.rs +++ b/src/obikphylo/src/siblings/build.rs @@ -174,8 +174,16 @@ fn build_layer_sibling_annex( // — a plain 1-to-1 (`|`, not `||`) transform, batch in, batch of // variants out, one message either way — keeps the per-message // synchronisation cost amortised over thousands of lookups instead - // of one to three. ────────────────────────────────────────────── - const BATCH_SIZE: usize = 4096; + // of one to three. `BATCH_SIZE` was originally tuned back when the + // per-k-mer cost inside this stage (minimiser via `RollingStat`) + // was ~3x higher than it is now (see `CanonicalKmerOf::minimizer`, + // a direct O(k) bit-arithmetic replacement) — at the old per-k-mer + // cost, dispatch overhead (the scheduler's single-threaded `Select` + // loop: one channel round-trip per batch) was negligible next to + // the work each batch represented; now that the work itself is + // cheaper, that fixed per-batch overhead is proportionally larger, + // so a bigger batch amortises it over more k-mers again. ───────── + const BATCH_SIZE: usize = 32768; let n_workers = obisys::effective_parallelism(); let capacity = 256; @@ -203,6 +211,16 @@ fn build_layer_sibling_annex( }) .collect::>() }); + + // Diagnostic: count still-empty mask slots after seeding + cross-partition + // resolution. A non-zero count here means some k-mer's own base was never + // written (should be impossible after the batch_offset fix above). + let check_empty = |label: &str| { + let empty = mask.iter().filter(|v| v.load(Ordering::Relaxed) == 0).count(); + if empty > 0 { + tracing::warn!("{label}: {empty} mask slots still empty after resolution (layer {layer_dir:?})"); + } + }; let throttled = obipipeline::throttle(batches, n_workers).map(|t| SourceBatch { items: t.item, _permit: t.guard, @@ -246,18 +264,45 @@ fn build_layer_sibling_annex( } } - // ── Resolve each partition's batch against the cache in one - // contiguous pass; parallelised across partitions (independent, - // read-only) so this keeps using multiple cores without giving up - // the per-partition locality above. ───────────────────────────── - outgoing.par_iter().enumerate().filter(|(_, q)| !q.is_empty()).for_each(|(dest, queries)| { - for &(variant, source_order, base) in queries { + check_empty("after_seeding"); + + // ── Resolve each partition's batch against the cache, parallelised + // over roughly equal-sized *chunks*, not over partitions — a plain + // `outgoing.par_iter()` over `n_parts` buckets gives one thread the + // whole of one partition's bucket, however large, and a lot of + // buckets are far from equal: a central-base substitution changes + // the minimiser (and thus routes to a different partition) only + // when the winning minimiser window overlaps the central position. + // For k=31/m=11 that is ~11 of the 21 possible windows — so *most* + // of the remaining ~10/21 send the variant right back to the + // partition already being built. That self-partition bucket ends up + // far larger than any other, so the naive per-partition split + // leaves one thread grinding through it alone long after every + // other partition's (tiny) bucket is done — confirmed by sampling a + // real run: one thread solely in `MphfLayer::find` while the rest + // of the pool sits idle. Splitting each non-empty bucket into + // `chunk_size`-sized pieces first keeps this pass's per-partition + // locality (each chunk is still contiguous within one partition, + // still resolved via `cache.find` grouped by that partition) while + // letting Rayon spread a single oversized bucket across several + // threads instead of pinning it to one. ────────────────────────── + let chunk_size = ((outgoing.iter().map(Vec::len).sum::() / n_workers).max(1)).min(4096); + let work: Vec<(usize, &[(CanonicalKmer, usize, u8)])> = outgoing + .iter() + .enumerate() + .filter(|(_, q)| !q.is_empty()) + .flat_map(|(dest, q)| q.chunks(chunk_size).map(move |c| (dest, c))) + .collect(); + work.par_iter().for_each(|&(dest, chunk)| { + for &(variant, source_order, base) in chunk { if cache.find(dest, variant) { mask[source_order].fetch_or(1 << base, Ordering::Relaxed); } } }); + check_empty("after_resolution"); + // ── Write the layer's annex file, indexed by iteration order — a // second streamed pass over `unitigs.bin` (via `enumerate_kmers`), // now that every entry's mask is final; never a `Vec` hold of the diff --git a/src/obikphylo/src/siblings/tests.rs b/src/obikphylo/src/siblings/tests.rs index 5aaca00b..d854bcab 100644 --- a/src/obikphylo/src/siblings/tests.rs +++ b/src/obikphylo/src/siblings/tests.rs @@ -258,3 +258,43 @@ fn family_scan_consumers_agree_on_one_sibling_each() { assert_eq!(total, 2, "no other base pair should be tallied"); assert_eq!(base_pairs.same, [0, 0, 0, 0], "the two genomes never agree at this locus"); } + +#[test] +fn sibling_annex_no_empty_masks_after_build() { + let dir = tempdir().unwrap(); + let seq = b"ACGTACGTACGT".repeat(500); + let g1 = build_single_genome_index(dir.path(), "g1", &seq); + g1.build_sibling_annex().expect("build_sibling_annex"); + + let index_dir = g1.partition().part_dir(0).join(INDEX_SUBDIR); + let meta = PartitionMeta::load(&index_dir).expect("partition meta"); + for l in 0..meta.n_layers { + let layer_dir = index_dir.join(format!("layer_{l}")); + let annex = SiblingAnnex::open(&layer_dir.join(ANNEX_FILE_NAME)).expect("annex open"); + for slot in 0..annex.len() { + let mask = annex.get(slot).expect("slot must have an entry"); + assert!( + mask.family_size() >= 1, + "layer {l} slot {slot}: empty family mask (bits={:04b}) after build — \ + indicates batch-offset bug or unseeded mask slot", + mask.bits() + ); + } + } +} + +#[test] +fn sibling_histogram_does_not_panic_on_partial_last_batch() { + // Regression test for the bug where enumerate_kmers_batch().enumerate() + // used the batch index instead of the sequence offset, leaving the + // trailing slots of a non-multiple-of-BATCH_SIZE layer unseeded and + // producing the minorant-only sentinel (0x10) that caused + // sibling_family_size_histogram to panic with index u32::MAX. + let dir = tempdir().unwrap(); + let seq = b"ACGTACGTACGT".repeat(500); + let g1 = build_single_genome_index(dir.path(), "g1", &seq); + g1.build_sibling_annex().expect("build_sibling_annex"); + let hist = g1.sibling_family_size_histogram().expect("sibling_family_size_histogram"); + let total: u64 = hist.iter().sum(); + assert!(total > 0, "histogram should contain at least one family"); +} diff --git a/src/obilayeredmap/src/layer.rs b/src/obilayeredmap/src/layer.rs index c2989a04..d50fd7b8 100644 --- a/src/obilayeredmap/src/layer.rs +++ b/src/obilayeredmap/src/layer.rs @@ -111,7 +111,7 @@ impl Layer { /// Iterate over batches, each paired with the zero-based index of the /// first kmer in the batch. - pub fn enumerate_kmers_batch(&self, n: usize) -> std::iter::Enumerate { + pub fn enumerate_kmers_batch(&self, n: usize) -> impl Iterator)> + Send + 'static { self.mphf.enumerate_kmers_batch(n) } diff --git a/src/obilayeredmap/src/mphf_layer.rs b/src/obilayeredmap/src/mphf_layer.rs index 98f6682f..6f47dc25 100644 --- a/src/obilayeredmap/src/mphf_layer.rs +++ b/src/obilayeredmap/src/mphf_layer.rs @@ -223,10 +223,16 @@ impl MphfLayer { /// first kmer in the batch. /// /// Yields `(batch_start_index, Vec)` where - /// `batch_start_index` is a multiple of `n`. This is the standard Rust - /// `enumerate` adapter applied to [`iter_kmers_batch`](Self::iter_kmers_batch). - pub fn enumerate_kmers_batch(&self, n: usize) -> Enumerate { - self.iter_kmers_batch(n).enumerate() + /// `batch_start_index` is the iteration-order offset of the first kmer + /// in that batch within the full layer sequence — i.e. a multiple of `n` + /// except for the final (possibly shorter) batch. + pub fn enumerate_kmers_batch(&self, n: usize) -> impl Iterator)> + Send + 'static { + let mut offset = 0usize; + self.iter_kmers_batch(n).map(move |batch| { + let base = offset; + offset += batch.len(); + (base, batch) + }) } } diff --git a/src/obilayeredmap/src/tests/layer.rs b/src/obilayeredmap/src/tests/layer.rs index 8a7233ee..3ae37376 100644 --- a/src/obilayeredmap/src/tests/layer.rs +++ b/src/obilayeredmap/src/tests/layer.rs @@ -1,8 +1,6 @@ use super::*; -use obicompactvec::PersistentCompactIntMatrix; use obikseq::{set_k, Kmer, Sequence as _, Unitig}; use obiskio::DEFAULT_BLOCK_BITS; -use crate::meta::IndexMode; use tempfile::tempdir; fn write_unitigs(dir: &Path, seqs: &[&[u8]]) { @@ -20,60 +18,39 @@ fn all_canonical_kmers(dir: &Path) -> Vec { .collect() } +// ── Iterator-order consistency tests ───────────────────────────────────────── +// These exercise the unitig-file iterators directly, without constructing a +// layer/MPHF. The sibling-annex builder relies on the invariant that +// `enumerate_kmers_batch()` yields the same sequence as `enumerate_kmers()`, +// in the same order as `UnitigFileReader::iter_indexed_canonical_kmers()`. +// A mismatch between any of these would silently write garbage into the +// annex's order-indexed mask array, producing exactly the "minorant-only" +// sentinel (0x10) that triggers the stats.rs panic. + #[test] -fn build_and_query_all_kmers_found() { +fn canonical_kmer_iter_matches_reader() { set_k(4); let dir = tempdir().unwrap(); - write_unitigs(dir.path(), &[b"AAAACGT"]); - let kmers = all_canonical_kmers(dir.path()); - Layer::<()>::build(dir.path(), DEFAULT_BLOCK_BITS, &IndexMode::Exact).unwrap(); - let layer = Layer::<()>::open(dir.path(), &IndexMode::Exact).unwrap(); - for kmer in kmers { - assert!(layer.query(kmer).is_some(), "kmer should be present"); - } + write_unitigs(dir.path(), &[b"AAAACGT", b"TTTTGCA"]); + + let from_iter: Vec = obiskio::CanonicalKmerIter::new(&dir.path().join(UNITIGS_FILE)) + .unwrap() + .collect(); + let from_reader: Vec = all_canonical_kmers(dir.path()); + + assert_eq!(from_iter.len(), from_reader.len(), "different kmer counts"); + assert_eq!(from_iter, from_reader, "CanonicalKmerIter and UnitigFileReader disagree"); } #[test] -fn counts_are_stored_and_retrieved() { +fn enumerate_kmers_is_stable_across_calls() { set_k(4); let dir = tempdir().unwrap(); - write_unitigs(dir.path(), &[b"AAAACGT"]); - let kmers = all_canonical_kmers(dir.path()); - let count_map: HashMap = - kmers.iter().enumerate().map(|(i, &k)| (k, i as u32 + 1)).collect(); - Layer::::build( - dir.path(), - DEFAULT_BLOCK_BITS, - &IndexMode::Exact, - |kmer| count_map.get(&kmer).copied().unwrap_or(0), - ).unwrap(); - let layer = Layer::::open(dir.path(), &IndexMode::Exact).unwrap(); - for kmer in &kmers { - let hit = layer.query(*kmer).expect("kmer must be present"); - assert_eq!(hit.data[0], count_map[kmer]); - } -} + write_unitigs(dir.path(), &[b"AAAACGT", b"TTTTGCA"]); -#[test] -fn query_absent_returns_none() { - set_k(4); - let dir = tempdir().unwrap(); - write_unitigs(dir.path(), &[b"AAAACGT"]); - Layer::<()>::build(dir.path(), DEFAULT_BLOCK_BITS, &IndexMode::Exact).unwrap(); - let layer = Layer::<()>::open(dir.path(), &IndexMode::Exact).unwrap(); - let absent = Kmer::from_ascii(b"CCCC").unwrap().canonical(); - assert!(layer.query(absent).is_none()); -} - -#[test] -fn open_after_build_is_consistent() { - set_k(4); - let dir = tempdir().unwrap(); - write_unitigs(dir.path(), &[b"AAAACGT"]); - let n = Layer::::build(dir.path(), DEFAULT_BLOCK_BITS, &IndexMode::Exact, |_| 7).unwrap(); - assert_eq!(n, 4); - let layer = Layer::::open(dir.path(), &IndexMode::Exact).unwrap(); - let kmer = Kmer::from_ascii(b"AAAA").unwrap().canonical(); - let hit = layer.query(kmer).expect("AAAA must be present"); - assert_eq!(hit.data[0], 7); + let reader = UnitigFileReader::open_sequential(&dir.path().join(UNITIGS_FILE)).unwrap(); + let first: Vec = reader.iter_indexed_canonical_kmers().map(|(k, _, _)| k).collect(); + let reader2 = UnitigFileReader::open_sequential(&dir.path().join(UNITIGS_FILE)).unwrap(); + let second: Vec = reader2.iter_indexed_canonical_kmers().map(|(k, _, _)| k).collect(); + assert_eq!(first, second, "iter_indexed_canonical_kmers must be deterministic across calls"); } diff --git a/src/obisys/src/progress.rs b/src/obisys/src/progress.rs index f15547d1..e0ba75f1 100644 --- a/src/obisys/src/progress.rs +++ b/src/obisys/src/progress.rs @@ -1,4 +1,5 @@ use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Mutex; use std::time::{Duration, Instant}; use indicatif::{ProgressBar, ProgressStyle}; @@ -7,7 +8,6 @@ use tracing::{debug, info}; const BRAILLE: &[&str] = &["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]; const ETA_REFRESH_MS: u64 = 500; const ETA_MIN_ELAPSED_MS: u64 = 1000; -const ETA_CUSTOM_HOLD_MS: u64 = 2000; pub struct TracedBar { pb: ProgressBar, @@ -18,7 +18,16 @@ pub struct TracedBar { last_pct: AtomicU64, last_log_ms: AtomicU64, last_eta_refresh: AtomicU64, - last_custom_msg_ms: AtomicU64, + /// Caller-supplied text (via [`set_message`](Self::set_message)) and the + /// self-computed ETA text are two independent pieces of information that + /// used to fight over indicatif's single `{msg}` slot — whichever wrote + /// last would silently clobber the other, and a caller that calls + /// `set_message` more often than the ETA's hold window (as + /// `build_sibling_annex`'s per-partition callback does once several + /// partitions run concurrently) starved the ETA out entirely. Kept as + /// separate strings, composed together on every render instead. + custom_text: Mutex, + eta_text: Mutex, } impl TracedBar { @@ -48,6 +57,21 @@ impl TracedBar { } } + /// Recompose the displayed message from `custom_text` + `eta_text` and + /// push it as one `pb.set_message` call — the two pieces never compete + /// for the slot, they always coexist. + fn render(&self) { + let custom = self.custom_text.lock().unwrap(); + let eta = self.eta_text.lock().unwrap(); + let combined = match (custom.is_empty(), eta.is_empty()) { + (true, true) => String::new(), + (false, true) => custom.clone(), + (true, false) => eta.clone(), + (false, false) => format!("{custom} {eta}"), + }; + self.pb.set_message(combined); + } + fn maybe_update_eta(&self) { let now_ms = self.start.elapsed().as_millis() as u64; @@ -60,11 +84,6 @@ impl TracedBar { return; } - let last_custom = self.last_custom_msg_ms.load(Ordering::Relaxed); - if now_ms < last_custom + ETA_CUSTOM_HOLD_MS { - return; - } - let pos = self.pb.position(); let remaining = self.total - pos; @@ -76,7 +95,8 @@ impl TracedBar { let avg_rate = pos as f64 / elapsed_secs; let eta_secs = remaining as f64 / avg_rate; - self.pb.set_message(format!("(eta: {})", fmt_secs(eta_secs))); + *self.eta_text.lock().unwrap() = format!("(eta: {})", fmt_secs(eta_secs)); + self.render(); let _ = self.last_eta_refresh.compare_exchange( last, @@ -88,10 +108,6 @@ impl TracedBar { pub fn set_message(&self, msg: impl Into) { let msg = msg.into(); - self.last_custom_msg_ms.store( - self.start.elapsed().as_millis() as u64, - Ordering::Relaxed, - ); if self.pb.is_hidden() { if self.total > 0 { debug!(stage = %self.label, "{msg}"); @@ -108,7 +124,8 @@ impl TracedBar { } } } - self.pb.set_message(msg); + *self.custom_text.lock().unwrap() = msg; + self.render(); } pub fn finish_and_clear(&self) { @@ -147,7 +164,8 @@ pub fn spinner(label: &str) -> TracedBar { last_pct: AtomicU64::new(0), last_log_ms: AtomicU64::new(0), last_eta_refresh: AtomicU64::new(0), - last_custom_msg_ms: AtomicU64::new(0), + custom_text: Mutex::new(String::new()), + eta_text: Mutex::new(String::new()), } } @@ -172,6 +190,7 @@ pub fn progress_bar(label: &str, n: u64, unit: &str) -> TracedBar { last_pct: AtomicU64::new(0), last_log_ms: AtomicU64::new(0), last_eta_refresh: AtomicU64::new(0), - last_custom_msg_ms: AtomicU64::new(0), + custom_text: Mutex::new(String::new()), + eta_text: Mutex::new(String::new()), } }