diff --git a/benchmark/tmp/test2/.obikmer.lock b/benchmark/tmp/test2/.obikmer.lock new file mode 100644 index 00000000..e69de29b diff --git a/scripts/smoke_test_index2.sh b/scripts/smoke_test_index2.sh new file mode 100755 index 00000000..e2c90d49 --- /dev/null +++ b/scripts/smoke_test_index2.sh @@ -0,0 +1,126 @@ +#!/usr/bin/env bash +# smoke_test_index2.sh — end-to-end smoke test of `obikmer2 index` (this +# working tree) + `obikmer query` (the release binary already on PATH). +# +# obikmer2 is being rebuilt incrementally from obikmer (see DevDocMD) and +# for now only implements `index`. This script checks that an index it +# produces is readable by an already-released, PATH-installed `obikmer` +# binary — the cross-binary counterpart to scripts/smoke_test_index.sh, +# which builds and queries with the same (in-tree) `obikmer`. +# +# What it does: +# 1. builds `obikmer2` (debug, via `cargo run`) +# 2. generates a small deterministic random FASTA +# 3. runs `obikmer2 index` on it +# 4. picks a real k-mer from the source sequence, avoiding low-complexity +# substrings (see scripts/smoke_test_index.sh's own note on this) +# 5. runs `obikmer query` (PATH binary, not built from this tree) and +# checks the k-mer round-trips +# 6. prints total-kmers-indexed and a clear PASS/FAIL, exit code matches +# +# Usage: +# scripts/smoke_test_index2.sh [-k KMER_SIZE] [-m MINIMIZER_SIZE] [-p PARTITIONS] [--keep] +# +# --keep leaves the temp directory in place (path printed) instead of +# deleting it on exit, for manual inspection of a failure. + +set -euo pipefail + +K=11 +M=5 +PARTITIONS=4 +KEEP=0 + +while [ $# -gt 0 ]; do + case "$1" in + -k) K="$2"; shift 2 ;; + -m) M="$2"; shift 2 ;; + -p) PARTITIONS="$2"; shift 2 ;; + --keep) KEEP=1; shift ;; + *) echo "unknown argument: $1" >&2; exit 2 ;; + esac +done + +REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +WORK="$(mktemp -d -t obikmer2_smoke.XXXXXX)" + +cleanup() { + if [ "$KEEP" -eq 1 ]; then + echo "kept: $WORK" + else + rm -rf "$WORK" + fi +} +trap cleanup EXIT + +fail() { + echo "FAIL: $1" >&2 + exit 1 +} + +command -v obikmer >/dev/null 2>&1 || fail "obikmer not found on PATH" + +# ── 1. generate a small deterministic random FASTA ───────────────────────── +python3 - "$WORK/test.fasta" "$K" <<'EOF' +import random, sys +path, k = sys.argv[1], int(sys.argv[2]) +random.seed(1234) +bases = "ACGT" +with open(path, "w") as f: + for i in range(3): + seq = "".join(random.choice(bases) for _ in range(300)) + f.write(f">seq{i}\n{seq}\n") +EOF + +# ── 2. build + run index (obikmer2, in-tree) ──────────────────────────────── +cd "$REPO_ROOT/src" +INDEX_LOG="$WORK/index.log" +if ! cargo run -q -p obikmer2 --bin obikmer2 -- \ + index -k "$K" -m "$M" --theta 0 -p "$PARTITIONS" \ + -o "$WORK/out.idx" "$WORK/test.fasta" > "$INDEX_LOG" 2>&1 +then + cat "$INDEX_LOG" >&2 + fail "obikmer2 index exited non-zero" +fi + +N_KMERS="$(grep -o '[0-9]* total kmers indexed' "$INDEX_LOG" | grep -o '^[0-9]*' || true)" +[ -n "$N_KMERS" ] || { cat "$INDEX_LOG" >&2; fail "could not find 'N total kmers indexed' in index log"; } +[ "$N_KMERS" -gt 0 ] || fail "index reports 0 kmers indexed" + +# ── 3+4. try candidate k-mers spread across the source sequence until one +# round-trips — see scripts/smoke_test_index.sh's own note: query's +# entropy filter can reject a genuinely-indexed low-complexity +# window, that's not a bug, just try the next candidate. +readarray -t CANDIDATES < <(python3 - "$WORK/test.fasta" "$K" <<'EOF' +import sys +path, k = sys.argv[1], int(sys.argv[2]) +with open(path) as f: + seq = "".join(l.strip() for l in f if not l.startswith(">")) +for start in range(0, len(seq) - k, 17): + print(seq[start:start+k]) +EOF +) +[ "${#CANDIDATES[@]}" -gt 0 ] || fail "could not extract any candidate k-mer from the source FASTA" + +QUERY_LOG="$WORK/query.log" +FOUND=0 +for QUERY_KMER in "${CANDIDATES[@]}"; do + printf ">q1\n%s\n" "$QUERY_KMER" > "$WORK/query.fasta" + # PATH binary, deliberately not built from this tree. + if ! obikmer query "$WORK/out.idx" "$WORK/query.fasta" > "$QUERY_LOG" 2>&1 + then + cat "$QUERY_LOG" >&2 + fail "obikmer query exited non-zero" + fi + if grep -q '"kmer_count":1' "$QUERY_LOG"; then + FOUND=1 + break + fi +done + +if [ "$FOUND" -ne 1 ]; then + cat "$QUERY_LOG" >&2 + fail "no candidate k-mer round-tripped (tried ${#CANDIDATES[@]}) — likely a real regression, not a low-complexity fixture" +fi + +echo "PASS: obikmer2 index + PATH obikmer query round-trip OK — $N_KMERS kmers indexed, query k-mer '$QUERY_KMER' found" diff --git a/src/Cargo.lock b/src/Cargo.lock index 5a5915cd..eeb66157 100644 --- a/src/Cargo.lock +++ b/src/Cargo.lock @@ -1551,6 +1551,8 @@ version = "0.1.0" dependencies = [ "ndarray", "obicompactvec", + "obikindex", + "obikseq", "rayon", "tempfile", ] diff --git a/src/obikidxcache/Cargo.toml b/src/obikidxcache/Cargo.toml index 884e8c88..29a7f586 100644 --- a/src/obikidxcache/Cargo.toml +++ b/src/obikidxcache/Cargo.toml @@ -4,6 +4,8 @@ version = "0.1.0" edition = "2024" [dependencies] +obikindex = { path = "../obikindex"} +obikseq = { path = "../obikseq"} obicompactvec = { path = "../obicompactvec" } ndarray = "0.17" rayon = "1" diff --git a/src/obikidxcache/src/index_cache.rs b/src/obikidxcache/src/index_cache.rs new file mode 100644 index 00000000..53d9aaec --- /dev/null +++ b/src/obikidxcache/src/index_cache.rs @@ -0,0 +1,132 @@ +use obikindex::{KmerIndex, layer::KmerLayer}; +use obikseq::CanonicalKmer; + +use crate::meta_cache::MetaCache; + +pub struct IndexCache<'a> { + raw_index: &'a KmerIndex, + meta: MetaCache, + + n_partition: usize, + n_layer: usize, + layer_cache: Vec, +} + +impl<'a> IndexCache<'a> { + /// Opens every layer of every partition once, up front, and keeps them + /// alive for the cache's lifetime — `layer(p, l)` stored flat at + /// `p * n_layer + l` (`n_layer` uniform across partitions, guaranteed by + /// construction). + /// + /// Panics if the index isn't fully built: a cache can only be built on a + /// functional index, so any failure here (missing/incomplete layer) is a + /// caller error, not a case to design around with a `Result`. + pub fn new(index: &'a KmerIndex) -> Self { + let n_partition = index.n_partitions(); + let n_layer = index.n_layers_per_partition().unwrap_or_else(|e| { + panic!("IndexCache::new: index is not fully built (n_layers_per_partition failed): {e}") + }); + + let mut layer_cache = Vec::with_capacity(n_partition * n_layer); + for p in 0..n_partition { + for l in 0..n_layer { + let layer = index + .layer(p, l) + .and_then(KmerLayer::open) + .unwrap_or_else(|e| { + panic!( + "IndexCache::new: failed to open layer (partition {p}, layer {l}): {e} — index is not fully built" + ) + }); + layer_cache.push(layer); + } + } + + IndexCache { + raw_index: index, + meta: MetaCache::from_meta(&index.meta()), + n_partition, + n_layer, + layer_cache, + } + } + + /// The index's metadata, snapshotted once at construction — see + /// [`MetaCache`]. Read-only + #[inline] + pub fn meta(&self) -> &MetaCache { + &self.meta + } + + #[inline] + pub fn n_layer(&self) -> usize { + self.n_layer + } + + #[inline] + pub fn n_partition(&self) -> usize { + self.n_partition + } + + #[inline] + pub fn get_layer(&self, partition: usize, layer: usize) -> Option<&KmerLayer> { + self.layer_cache.get(partition * self.n_layer + layer) + } + + pub fn iter(&self) -> impl Iterator { + self.layer_cache.iter() + } + + pub fn iter_indexed(&self) -> impl Iterator { + let n_layer = self.n_layer; + self.layer_cache + .iter() + .enumerate() + .map(move |(i, layer)| (i / n_layer, i % n_layer, layer)) + } + + #[inline] + pub fn hash(&self, partition: usize, layer: usize, kmer: CanonicalKmer) -> Option { + Some(self.get_layer(partition, layer)?.hash(kmer)) + } + + pub fn n_kmer(&self, partition: usize, layer: usize) -> Option { + Some(self.get_layer(partition, layer)?.n()) + } + + /// Membership-checked lookup in one `(partition, layer)` — `None` if + /// that layer doesn't carry `kmer`. Unlike [`hash`](Self::hash), this + /// goes through `KmerLayer::find_slot`'s evidence check, not a blind + /// MPHF lookup. + #[inline] + pub fn find_in_layer(&self, partition: usize, layer: usize, kmer: CanonicalKmer) -> Option { + self.get_layer(partition, layer)?.find_slot(kmer) + } + + /// Membership-checked lookup across every layer of one partition — + /// first layer that carries `kmer` wins. `None` if `partition` is out + /// of range or no layer of it carries `kmer`. + pub fn find_in_partition( + &self, + partition: usize, + kmer: CanonicalKmer, + ) -> Option<(usize, usize, usize)> { + for l in 0..self.n_layer { + if let Some(slot) = self.find_in_layer(partition, l, kmer) { + return Some((partition, l, slot)); + } + } + None + } + + /// Membership-checked lookup across every partition and layer — first + /// one that carries `kmer` wins. + pub fn find(&self, kmer: CanonicalKmer) -> Option<(usize, usize, usize)> { + for p in 0..self.n_partition { + if let Some(hit) = self.find_in_partition(p, kmer) { + return Some(hit); + } + } + None + } +} diff --git a/src/obikidxcache/src/lib.rs b/src/obikidxcache/src/lib.rs index 15919976..ffd3e3ed 100644 --- a/src/obikidxcache/src/lib.rs +++ b/src/obikidxcache/src/lib.rs @@ -6,6 +6,5 @@ //! `CountPartials`/`BitPartials` aggregation across them. No dependency on //! `obikindex` — generic over any `S` implementing those traits. -mod layered_store; - -pub use layered_store::LayeredStore; +pub mod index_cache; +pub mod meta_cache; diff --git a/src/obikidxcache/src/meta_cache.rs b/src/obikidxcache/src/meta_cache.rs new file mode 100644 index 00000000..6f281c42 --- /dev/null +++ b/src/obikidxcache/src/meta_cache.rs @@ -0,0 +1,49 @@ +use obikindex::{GenomeInfo, IndexConfig, IndexMeta, IndexState}; + +/// A read-only, fully in-memory snapshot of an index's `IndexMeta`, taken +/// once at construction. `IndexMeta` itself only caches `config` — `genomes` +/// and `state` are re-read from disk on every call, by design, since they +/// can legitimately change during a run. This type is the opposite trade: +/// everything frozen at snapshot time, no disk access ever, at the cost of +/// going stale if the index is mutated after the cache is built — matching +/// the "cache can only be built on a functional index, read-only" contract +/// the rest of `IndexCache` follows. +pub struct MetaCache { + config: IndexConfig, + genomes: Vec, + state: IndexState, +} + +impl MetaCache { + /// Panics if reading `genomes`/`state` off `meta` fails — same + /// discipline as `IndexCache::new`: a cache can only be built on a + /// functional index, so a failure here means the index isn't one. + pub(crate) fn from_meta(meta: &IndexMeta) -> Self { + let genomes = meta + .genomes() + .unwrap_or_else(|e| panic!("MetaCache::from_meta: failed to read genomes: {e}")); + let state = meta + .state() + .unwrap_or_else(|e| panic!("MetaCache::from_meta: failed to read state: {e}")); + MetaCache { + config: meta.config().clone(), + genomes, + state, + } + } + + #[inline] + pub fn config(&self) -> &IndexConfig { + &self.config + } + + #[inline] + pub fn genomes(&self) -> &[GenomeInfo] { + &self.genomes + } + + #[inline] + pub fn state(&self) -> IndexState { + self.state + } +} diff --git a/src/obikindex/src/index/builder.rs b/src/obikindex/src/index/builder.rs index 958a94ca..2a24f7f7 100644 --- a/src/obikindex/src/index/builder.rs +++ b/src/obikindex/src/index/builder.rs @@ -86,10 +86,17 @@ impl IndexBuilder for KmerIndex { let output = output.as_ref(); fs::create_dir_all(output).map_err(OKIError::Io)?; let meta = IndexMeta::create_at(output, config, genomes).map_err(OKIError::Io)?; - Ok(KmerIndex { + let idx = KmerIndex { root_path: output.to_owned(), meta: Arc::new(meta), - }) + lock: std::sync::Mutex::new(None), + }; + if !idx.acquire_lock() { + return Err(OKIError::Io(std::io::Error::other( + "failed to acquire index lock right after create_skeleton", + ))); + } + Ok(idx) } fn finalize_indexed>(output: P, rep: &mut Reporter) -> OKIResult { diff --git a/src/obikindex/src/index/kmer_index.rs b/src/obikindex/src/index/kmer_index.rs index a4c56925..4956fab5 100644 --- a/src/obikindex/src/index/kmer_index.rs +++ b/src/obikindex/src/index/kmer_index.rs @@ -1,6 +1,6 @@ use std::fs; use std::path::{Path, PathBuf}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use crate::layer::KmerLayer; use crate::partition::KmerPartition; @@ -15,10 +15,14 @@ use crate::index::meta::{GenomeInfo, IndexConfig, IndexMeta}; pub struct KmerIndex { pub(crate) root_path: PathBuf, pub(crate) meta: Arc, + pub(crate) lock: Mutex>, } impl KmerIndex { - /// Create a new index at `path`. + /// Create a new index at `path`, holding its directory lock on return + /// (see [`acquire_lock`](Self::acquire_lock)) — a freshly created index + /// is definitionally about to be written to (scatter/dereplicate/ + /// count/build), so it is exclusive from the start, unlike `open`. /// /// If `genome_info` is `Some`, it is stored immediately. /// If `None`, the genome entry will be added when `mark_scattered` is called. @@ -33,12 +37,26 @@ impl KmerIndex { set_m(config.minimizer_size); let genomes = genome_info.into_iter().collect(); let meta = IndexMeta::create_at(&root_path, config, genomes).map_err(OKIError::Io)?; - Ok(Self { + let idx = Self { root_path, meta: Arc::new(meta), - }) + lock: Mutex::new(None), + }; + if !idx.acquire_lock() { + return Err(OKIError::Io(std::io::Error::other( + "failed to acquire index lock right after create", + ))); + } + Ok(idx) } + /// Open an existing index for reading — doesn't touch the directory + /// lock at all: concurrent readers (`query`, `dump`, a `merge` source, + /// ...) must never block on each other. A caller that needs exclusive + /// access (resuming an interrupted build) uses + /// [`open_lock`](Self::open_lock) instead; either way, `self.lock` + /// starts empty — [`test_lock`](Self::test_lock)/ + /// [`acquire_lock`](Self::acquire_lock) are opt-in from here. pub fn open>(path: P) -> OKIResult { let root_path = path.as_ref().to_owned(); let meta = IndexMeta::open_at(&root_path).map_err(OKIError::Io)?; @@ -47,9 +65,54 @@ impl KmerIndex { Ok(Self { root_path, meta: Arc::new(meta), + lock: Mutex::new(None), }) } + /// `open`, then hold the directory lock exclusively — for a caller + /// that resumes a write pipeline (e.g. `obikmer index` re-run on an + /// existing, not-yet-`Indexed` directory) rather than reading + /// concurrently with others. + pub fn open_lock>(path: P) -> OKIResult { + let idx = Self::open(path)?; + if !idx.acquire_lock() { + return Err(OKIError::Io(std::io::Error::other( + "failed to acquire index lock", + ))); + } + Ok(idx) + } + + /// Block until this index's directory lock is acquired, then hold it + /// (released on [`release_lock`](Self::release_lock) or when `self` is + /// dropped). Returns `false` only on a genuine I/O failure acquiring + /// the lock file itself — contention blocks rather than failing. + pub fn acquire_lock(&self) -> bool { + match obisys::DirLock::acquire(&self.root_path) { + Ok(lock) => { + *self.lock.lock().unwrap() = Some(lock); + true + } + Err(_) => false, + } + } + + /// Non-blocking probe of this index's directory lock — doesn't retain + /// it either way: `true` if the lock was free (briefly acquired, then + /// immediately released), `false` if another process currently holds + /// it. For a read-only caller that wants to warn about a build in + /// progress without itself becoming exclusive; use + /// [`acquire_lock`](Self::acquire_lock)/[`open_lock`](Self::open_lock) + /// to actually hold it. + pub fn test_lock(&self) -> bool { + matches!(obisys::DirLock::try_acquire(&self.root_path), Ok(Some(_))) + } + + /// Release this index's directory lock, if held. A no-op if it isn't. + pub fn release_lock(&self) { + *self.lock.lock().unwrap() = None; + } + /// Return `true` if `path` points to a valid obikmer index directory. /// /// A directory is considered an index only when it contains an `index.meta` diff --git a/src/obikindex/src/layer/tests/typed_layer.rs b/src/obikindex/src/layer/tests/typed_layer.rs index 722e1490..7c1faad9 100644 --- a/src/obikindex/src/layer/tests/typed_layer.rs +++ b/src/obikindex/src/layer/tests/typed_layer.rs @@ -68,7 +68,7 @@ fn presence_layer_generic_over_sparse_matches_dense() { (kmer.raw().wrapping_add(g as u64)) % 2 == 0 }).unwrap(); - let dense_layer = TypedLayer::::open(dir.path(), &mode).unwrap(); + let dense_layer = TypedLayer::::open(dir.path()).unwrap(); assert!(dense_layer.n_cols() >= 1); // Build the sparse form directly into the same `presence/` dir the @@ -81,7 +81,7 @@ fn presence_layer_generic_over_sparse_matches_dense() { .close() .unwrap(); - let sparse_layer = TypedLayer::::open(dir.path(), &mode).unwrap(); + let sparse_layer = TypedLayer::::open(dir.path()).unwrap(); // Same generic methods, same results, different concrete `D`. assert_eq!(dense_layer.n_cols(), sparse_layer.n_cols()); @@ -107,7 +107,7 @@ fn count_layer_reports_count_content_and_columnar_storage() { write_unitigs(dir.path(), &[b"AAAACGT"]); TypedLayer::::build(dir.path(), DEFAULT_BLOCK_BITS, &IndexMode::Exact, |_| 1) .unwrap(); - let layer = TypedLayer::::open(dir.path(), &IndexMode::Exact).unwrap(); + let layer = TypedLayer::::open(dir.path()).unwrap(); assert_eq!(layer.content(), LayerContent::Count); assert_eq!(layer.storage_kind(), obicompactvec::StorageKind::Columnar); @@ -122,7 +122,7 @@ fn presence_layer_reports_presence_content_and_columnar_storage() { TypedLayer::::build_presence(dir.path(), DEFAULT_BLOCK_BITS, &IndexMode::Exact, 2, |kmer, g| { (kmer.raw().wrapping_add(g as u64)) % 2 == 0 }).unwrap(); - let layer = TypedLayer::::open(dir.path(), &IndexMode::Exact).unwrap(); + let layer = TypedLayer::::open(dir.path()).unwrap(); assert_eq!(layer.content(), LayerContent::Presence); assert_eq!(layer.storage_kind(), obicompactvec::StorageKind::Columnar); diff --git a/src/obikindex/src/partition/mod.rs b/src/obikindex/src/partition/mod.rs index 87b31152..f134422e 100644 --- a/src/obikindex/src/partition/mod.rs +++ b/src/obikindex/src/partition/mod.rs @@ -111,5 +111,13 @@ impl KmerPartition { } } -#[cfg(test)] -mod tests; +// `tests.rs` exercises the old eager `KmerPartition::open`/`.find()`/ +// `.layers()` cross-layer lookup, removed when `KmerPartition` became the +// current lazy `new`/`layer(i)`/`n_layers()` model. That capability's +// intended future home is `obikphylo::siblings::cache::PartitionCache` (see +// this module's doc comment) — a separate, not-yet-started migration. +// Parked here, deliberately excluded from compilation, until that lands; +// `tests.rs` itself is left untouched so its assertions can be ported +// as-is once `PartitionCache` exists. +// #[cfg(test)] +// mod tests; diff --git a/src/obikindexer/src/algorithms/partitionner/tests.rs b/src/obikindexer/src/algorithms/partitionner/tests.rs index 6905a285..c1757388 100644 --- a/src/obikindexer/src/algorithms/partitionner/tests.rs +++ b/src/obikindexer/src/algorithms/partitionner/tests.rs @@ -10,6 +10,7 @@ use obikseq::SuperKmer; use obiskbuilder::build_superkmers; use super::PartitionRouter; +use crate::extensions::PrivateBuilder; const K: usize = 11; const M: usize = 5; @@ -68,8 +69,8 @@ fn pipeline_counts(seqs: &[&[u8]]) -> (u64, u64) { drop(kp); // ends the borrow of `index` early — `PartitionRouter`'s `Drop` impl would otherwise extend it to the end of scope Dereplicator::new(&index).run().unwrap(); - let part_dir = index.layer_dir(0, 0); - let dedup_path = obikindex::layer::dereplicated_superkmers_path(&part_dir); + let layer0 = index.layer0(0).unwrap(); + let dedup_path = layer0.dereplicated_superkmers_path(); if !dedup_path.exists() { return (0, 0); } diff --git a/src/obikmer2/src/cmd/index/mod.rs b/src/obikmer2/src/cmd/index/mod.rs index 642dc0b8..7bcace91 100644 --- a/src/obikmer2/src/cmd/index/mod.rs +++ b/src/obikmer2/src/cmd/index/mod.rs @@ -168,16 +168,6 @@ pub fn run(args: IndexArgs) { let output = args.output.clone(); let mut rep = Reporter::new(); - // Locked for the whole build (including a possible --force removal + - // recreation below): a second `index` run resuming/overwriting the same - // output directory concurrently would otherwise corrupt it. Unlinking - // the lock file via --force's remove_dir_all is safe — the held file - // descriptor keeps the lock regardless of the directory entry. - let _lock = obisys::DirLock::acquire(&output).unwrap_or_else(|e| { - eprintln!("error locking output directory {}: {e}", output.display()); - std::process::exit(1); - }); - // ── Resolve evidence kind ──────────────────────────────────────────────── let (evidence, effective_kmer_size) = if args.approx { let (z, b, fp) = resolve_approx_params(args.findere_z, args.evidence_bits, args.fp); diff --git a/src/obiskio/src/unitig_index/reader.rs b/src/obiskio/src/unitig_index/reader.rs index 781b8f44..54226d53 100644 --- a/src/obiskio/src/unitig_index/reader.rs +++ b/src/obiskio/src/unitig_index/reader.rs @@ -23,13 +23,13 @@ use super::{DEFAULT_BLOCK_BITS, MAGIC, idx_path}; /// both modes. Without `.idx` they fall back to an O(i) sequential scan — /// correct but slower. pub struct UnitigFileReader { - mmap: Mmap, + mmap: Mmap, block_offsets: Vec, - n_unitigs: usize, - n_kmers: usize, - k: usize, - block_bits: u8, - mask: usize, // (1 << block_bits) - 1 + n_unitigs: usize, + n_kmers: usize, + k: usize, + block_bits: u8, + mask: usize, // (1 << block_bits) - 1 } impl UnitigFileReader { @@ -52,13 +52,13 @@ impl UnitigFileReader { let mmap = unsafe { Mmap::map(&file).map_err(SKError::Io)? }; let k = obikseq::params::k(); - let mut offset = 0usize; + let mut offset = 0usize; let mut n_unitigs = 0usize; - let mut n_kmers = 0usize; + let mut n_kmers = 0usize; while offset < mmap.len() { let seql_minus_k = mmap[offset] as usize; - n_kmers += seql_minus_k + 1; - offset += 1 + (seql_minus_k + k + 3) / 4; + n_kmers += seql_minus_k + 1; + offset += 1 + (seql_minus_k + k + 3) / 4; n_unitigs += 1; } @@ -94,11 +94,22 @@ impl UnitigFileReader { }) } - pub fn len(&self) -> usize { self.n_unitigs } - pub fn is_empty(&self) -> bool { self.n_unitigs == 0 } - pub fn n_kmers(&self) -> usize { self.n_kmers } - pub fn block_bits(&self) -> u8 { self.block_bits } - pub fn has_direct_access(&self) -> bool { !self.block_offsets.is_empty() } + pub fn len(&self) -> usize { + self.n_unitigs + } + pub fn is_empty(&self) -> bool { + self.n_unitigs == 0 + } + pub fn n_kmers(&self) -> usize { + self.n_kmers + } + pub fn block_bits(&self) -> u8 { + self.block_bits + } + #[inline] + pub fn has_direct_access(&self) -> bool { + !self.block_offsets.is_empty() + } /// Byte offset of record `i` in the mmap. /// @@ -106,12 +117,12 @@ impl UnitigFileReader { /// sequential scan otherwise. #[inline] fn chunk_start(&self, i: usize) -> usize { - if !self.block_offsets.is_empty() { + if self.has_direct_access() { if self.block_bits == 0 { return self.block_offsets[i] as usize; } let block = i >> self.block_bits; - let rem = i & self.mask; + let rem = i & self.mask; let mut offset = self.block_offsets[block] as usize; for _ in 0..rem { let seql_minus_k = self.mmap[offset] as usize; @@ -136,10 +147,12 @@ impl UnitigFileReader { /// Reconstruct chunk `i` as a [`Unitig`]. pub fn unitig(&self, i: usize) -> Unitig { - let offset = self.chunk_start(i); - let seql = self.mmap[offset] as usize + self.k; + let offset = self.chunk_start(i); + let seql = self.mmap[offset] as usize + self.k; let byte_len = (seql + 3) / 4; - let bytes = self.mmap[offset + 1..offset + 1 + byte_len].to_vec().into_boxed_slice(); + let bytes = self.mmap[offset + 1..offset + 1 + byte_len] + .to_vec() + .into_boxed_slice(); Unitig::new((seql % 4) as u8, bytes) } @@ -171,20 +184,26 @@ impl UnitigFileReader { // ── Sequential iterators (O(n) running-offset cursor) ───────────────────── pub(crate) fn iter_chunks_sequential(&self) -> impl Iterator + '_ { - let k = self.k; + let k = self.k; let mmap = &*self.mmap; - let n = self.n_unitigs; + let n = self.n_unitigs; let mut offset = 0usize; (0..n).map(move |chunk_id| { - let seql = mmap[offset] as usize + k; + let seql = mmap[offset] as usize + k; let byte_len = (seql + 3) / 4; - let bytes = mmap[offset + 1..offset + 1 + byte_len].to_vec().into_boxed_slice(); + let bytes = mmap[offset + 1..offset + 1 + byte_len] + .to_vec() + .into_boxed_slice(); offset += 1 + byte_len; (chunk_id, Unitig::new((seql % 4) as u8, bytes)) }) } - /// Iterate all unitigs sequentially. Works without `.idx` (sequential open). + /// Iterate all unitigs sequentially, paired with their `chunk_id`. + /// Goes through [`iter_chunks_sequential`](Self::iter_chunks_sequential) — + /// never touches `chunk_start`/`block_offsets`, so it works identically + /// whether opened via `open_sequential` or `open_direct_access` (`.idx` + /// loaded or not). pub fn iter_unitigs(&self) -> impl Iterator + '_ { self.iter_chunks_sequential() } @@ -194,15 +213,27 @@ impl UnitigFileReader { .flat_map(|(_, u)| u.into_kmers()) } + pub fn iter_canonical_kmers(&self) -> impl Iterator + '_ { + self.iter_chunks_sequential() + .flat_map(|(_, u)| u.into_canonical_kmers()) + } + + /// Sequential scan, `.idx` not required — despite the name, "indexed" + /// describes what's returned (each k-mer paired with its `(chunk_id, + /// rank)` position), not a dependency on the `.idx` sidecar. Works + /// identically whether or not `.idx` exists, since it goes through + /// [`iter_chunks_sequential`](Self::iter_chunks_sequential) — never + /// `chunk_start`/`block_offsets`. This is what `build_exact_evidence`/ + /// `build()` use to fill `evidence.bin` from a freshly written + /// `unitigs.bin`, before `.idx` exists. pub fn iter_indexed_canonical_kmers( &self, ) -> impl Iterator + '_ { - self.iter_chunks_sequential() - .flat_map(|(chunk_id, u)| { - u.into_canonical_kmers() - .enumerate() - .map(move |(rank, kmer)| (kmer, chunk_id, rank)) - }) + self.iter_chunks_sequential().flat_map(|(chunk_id, u)| { + u.into_canonical_kmers() + .enumerate() + .map(move |(rank, kmer)| (kmer, chunk_id, rank)) + }) } /// Same streamed sequence as [`iter_indexed_canonical_kmers`](Self::iter_indexed_canonical_kmers), @@ -222,7 +253,9 @@ impl UnitigFileReader { let mmap = &*this.mmap; let seql = mmap[offset] as usize + k; let byte_len = (seql + 3) / 4; - let bytes = mmap[offset + 1..offset + 1 + byte_len].to_vec().into_boxed_slice(); + let bytes = mmap[offset + 1..offset + 1 + byte_len] + .to_vec() + .into_boxed_slice(); offset += 1 + byte_len; (chunk_id, Unitig::new((seql % 4) as u8, bytes)) }) @@ -238,8 +271,9 @@ fn read_idx(path: &Path) -> SKResult<(usize, usize, u8, Vec)> { let data = std::fs::read(path).map_err(SKError::Io)?; let mut pos = 0; - let magic_bytes = data.get(pos..pos + 4) - .ok_or(SKError::Truncated { context: "unitig index: magic" })?; + let magic_bytes = data.get(pos..pos + 4).ok_or(SKError::Truncated { + context: "unitig index: magic", + })?; if magic_bytes != &MAGIC { return Err(SKError::BadMagic { expected: "UIX3", @@ -248,8 +282,9 @@ fn read_idx(path: &Path) -> SKResult<(usize, usize, u8, Vec)> { } pos += 4; - let bb_bytes = data.get(pos..pos + 4) - .ok_or(SKError::Truncated { context: "unitig index: block_bits" })?; + let bb_bytes = data.get(pos..pos + 4).ok_or(SKError::Truncated { + context: "unitig index: block_bits", + })?; let block_bits_u32 = u32::from_le_bytes(bb_bytes.try_into().unwrap()); if block_bits_u32 > 31 { return Err(SKError::InvalidData { @@ -260,13 +295,15 @@ fn read_idx(path: &Path) -> SKResult<(usize, usize, u8, Vec)> { let block_bits = block_bits_u32 as u8; pos += 4; - let n_bytes = data.get(pos..pos + 4) - .ok_or(SKError::Truncated { context: "unitig index: n_unitigs" })?; + let n_bytes = data.get(pos..pos + 4).ok_or(SKError::Truncated { + context: "unitig index: n_unitigs", + })?; let n_unitigs = u32::from_le_bytes(n_bytes.try_into().unwrap()) as usize; pos += 4; - let nk_bytes = data.get(pos..pos + 8) - .ok_or(SKError::Truncated { context: "unitig index: n_kmers" })?; + let nk_bytes = data.get(pos..pos + 8).ok_or(SKError::Truncated { + context: "unitig index: n_kmers", + })?; let n_kmers = u64::from_le_bytes(nk_bytes.try_into().unwrap()) as usize; pos += 8; @@ -275,8 +312,9 @@ fn read_idx(path: &Path) -> SKResult<(usize, usize, u8, Vec)> { let n_offsets = n_blocks + 1; let mut block_offsets = Vec::with_capacity(n_offsets); for _ in 0..n_offsets { - let off_bytes = data.get(pos..pos + 4) - .ok_or(SKError::Truncated { context: "unitig index: block_offsets" })?; + let off_bytes = data.get(pos..pos + 4).ok_or(SKError::Truncated { + context: "unitig index: block_offsets", + })?; block_offsets.push(u32::from_le_bytes(off_bytes.try_into().unwrap())); pos += 4; } diff --git a/src/obisys/src/lock.rs b/src/obisys/src/lock.rs index 9b88bb31..b6a13aeb 100644 --- a/src/obisys/src/lock.rs +++ b/src/obisys/src/lock.rs @@ -23,18 +23,33 @@ impl DirLock { /// and the lock file within it if needed). Logs once if the wait is /// non-trivial, so a blocked command doesn't look silently hung. pub fn acquire(dir: &std::path::Path) -> std::io::Result { - std::fs::create_dir_all(dir)?; - let lock_path = dir.join(".obikmer.lock"); - let file = std::fs::OpenOptions::new() - .create(true) - .truncate(false) - .write(true) - .open(&lock_path)?; - + let file = Self::open_lock_file(dir)?; if file.try_lock().is_err() { info!(dir = %dir.display(), "waiting for another obikmer process to release this index"); file.lock()?; } Ok(Self { _file: file }) } + + /// Non-blocking probe: `Ok(Some(lock))` if `dir`'s lock was free and is + /// now held by the returned `DirLock`, `Ok(None)` if another process + /// currently holds it (not an error — contention is the expected, + /// common case this exists to distinguish from a genuine I/O failure). + pub fn try_acquire(dir: &std::path::Path) -> std::io::Result> { + let file = Self::open_lock_file(dir)?; + match file.try_lock() { + Ok(()) => Ok(Some(Self { _file: file })), + Err(_) => Ok(None), + } + } + + fn open_lock_file(dir: &std::path::Path) -> std::io::Result { + std::fs::create_dir_all(dir)?; + let lock_path = dir.join(".obikmer.lock"); + std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(&lock_path) + } }