From 4803ed3af649591e62250bca917406a4dbf605f1 Mon Sep 17 00:00:00 2001 From: alejandrogzi Date: Thu, 10 Sep 2026 10:13:52 +0200 Subject: [PATCH 1/4] feat(run): unit-level static partition, per-unit ledger, receiver guards (experimental rounds 88/90/90b) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sync of experimental HEAD (alejandrogzi/hspZ @ 04d5985) into production. Sources are experimental HEAD with the production packaging re-applied (file headers, clap attribute, hspZ naming). No dependency changes. - round 90: HSPZ_UNIT_PARTITION splits the frozen plan's work units by count quotas (whole bins first, then contiguous query slices of as few bins as possible, one extra reference build per split bin); --time gains the per-unit ledger, and every run prints schedule lines; the receiver rejects duplicate/out-of-plan units and verifies completeness after joining the workers, reporting a worker's own error first. - round 90b: unset = auto — partition only with W>=2, one worker per device, and every used device the same class (equal SM count and L2, nominal clocks within 10%); the reason is printed in the schedule line. 1/0 force. - round 88 also lands here: v0.0.4 was synced from ce1de86 and cherry-picked b44cc04's window budget only, so -B 0 automatic layout (W>=2; W=1 is exactly the 500 Mbp plan), the ref-buckets inconclusive-block bound, and the --time gap/phase attribution were never in production. Round 90 builds on that code. - the round-89 find_hits overlap prototype (043bd30, reverted upstream) is absent. Verified: cargo oxide build --release; 138 tests; chr20 digest 70152625c1a75cb3 (3,638 HSPs); schedule: policy=whole-bin W=1 ... (auto: W=1). --- src/gpu/mod.rs | 365 +++++++++++++++++--- src/plan.rs | 877 ++++++++++++++++++++++++++++++++++++++++++++++--- src/run.rs | 698 ++++++++++++++++++++++++++++++++++++--- src/timing.rs | 9 + 4 files changed, 1822 insertions(+), 127 deletions(-) diff --git a/src/gpu/mod.rs b/src/gpu/mod.rs index 2e1a1af..8d8707a 100644 --- a/src/gpu/mod.rs +++ b/src/gpu/mod.rs @@ -961,6 +961,11 @@ pub struct Engine { /// which is structural. The single largest gap cannot tell those apart. Linear scan /// over a handful of distinct stage pairs. gap_pairs: Vec<((&'static str, &'static str), f32, u64)>, + /// Intervals the gap accounting discards because they exceed 1,000 ms + /// (unit-transition work: query pack, ref prep). Counted separately so + /// the residual is visible instead of silently dropped. + gap_long_n: u64, + gap_long_ms: f32, /// Generate the query seed stream on the device instead of walking it on the /// host (round 68). Both paths are compiled; the executor sets this from the /// worker count because the trade reverses with GPU count — on one GPU the @@ -1049,6 +1054,45 @@ unsafe fn reserve( Ok(()) } +/// `cuDeviceGetAttribute` for `ordinal`, no context required (round 90b: the +/// unit-partition guard runs before any worker owns a device). +fn device_attribute(ordinal: i32, attr: cuda_core::sys::CUdevice_attribute) -> Result { + let mut value: i32 = 0; + // SAFETY: driver query; `value` is a local and the status is checked. + unsafe { + if cuda_core::sys::cuDeviceGetAttribute(&mut value, attr, ordinal) + != cuda_core::sys::cudaError_enum_CUDA_SUCCESS + { + return Err(format!("cuDeviceGetAttribute failed for device {ordinal}")); + } + } + Ok(value) +} + +/// The three attributes the round-90b same-device guard compares: resident +/// multiprocessors, the driver-reported clock in kHz, and L2 cache bytes. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct DeviceProfile { + pub(crate) sms: i32, + pub(crate) clock_khz: i32, + pub(crate) l2_bytes: u64, +} + +/// [`DeviceProfile`] of `ordinal`, queried once per used device before any +/// worker spawns (the auto unit-partition guard). +pub(crate) fn device_profile(ordinal: i32) -> Result { + use cuda_core::sys::{ + CUdevice_attribute_enum_CU_DEVICE_ATTRIBUTE_CLOCK_RATE as CLOCK_RATE, + CUdevice_attribute_enum_CU_DEVICE_ATTRIBUTE_L2_CACHE_SIZE as L2_CACHE_SIZE, + CUdevice_attribute_enum_CU_DEVICE_ATTRIBUTE_MULTIPROCESSOR_COUNT as MULTIPROCESSOR_COUNT, + }; + Ok(DeviceProfile { + sms: device_attribute(ordinal, MULTIPROCESSOR_COUNT)?, + clock_khz: device_attribute(ordinal, CLOCK_RATE)?, + l2_bytes: device_attribute(ordinal, L2_CACHE_SIZE)?.max(0) as u64, + }) +} + /// Device L2 cache bytes, or 0 if the driver will not say. /// /// `CU_DEVICE_ATTRIBUTE_L2_CACHE_SIZE` on the context's device. cuda-oxide keeps @@ -1057,23 +1101,18 @@ unsafe fn reserve( #[cfg(feature = "ref-loc-buckets")] fn l2_cache_bytes() -> u64 { let mut dev: i32 = 0; - let mut bytes: i32 = 0; - // SAFETY: driver queries against the current context; both outputs are - // locals, and a failed call leaves the zero initialiser in place. + // SAFETY: driver query against the current context; `dev` is a local. unsafe { if cuda_core::sys::cuCtxGetDevice(&mut dev) != cuda_core::sys::cudaError_enum_CUDA_SUCCESS { return 0; } - if cuda_core::sys::cuDeviceGetAttribute( - &mut bytes, - cuda_core::sys::CUdevice_attribute_enum_CU_DEVICE_ATTRIBUTE_L2_CACHE_SIZE, - dev, - ) != cuda_core::sys::cudaError_enum_CUDA_SUCCESS - { - return 0; - } } - bytes.max(0) as u64 + device_attribute( + dev, + cuda_core::sys::CUdevice_attribute_enum_CU_DEVICE_ATTRIBUTE_L2_CACHE_SIZE, + ) + .map(|bytes| bytes.max(0) as u64) + .unwrap_or(0) } /// Human-readable window size for a bucket shift. Shifts below 20 (a small @@ -1303,10 +1342,24 @@ struct BlockDecision { on_ns: Vec, } +/// A block closed without a usable settled comparison: `span_ms >= +/// 2 * T_BLOCK` with fewer than 4 settled chunks (e.g. a tail of small +/// chunks). Exactness-neutral: the block ran its production path, only +/// the accounting is discarded. +#[cfg(feature = "ref-loc-buckets")] +#[derive(Clone, Debug, PartialEq)] +struct InconclusiveBlock { + k: usize, + on: bool, + span_ms: f64, + settled_chunks: u32, +} + /// `Auto` phase: untimed warm-up, then alternating OFF, ON, OFF, ON, OFF, ON /// blocks (3 per path). `k` is the block index and always equals the number /// of blocks closed so far. #[cfg(feature = "ref-loc-buckets")] +#[derive(Debug, PartialEq)] enum AutoPhase { Warm { n: u32, @@ -1326,6 +1379,7 @@ enum AutoPhase { /// auto-off forcing OFF before any block), the block-alternation phase, or /// the committed decision. #[cfg(feature = "ref-loc-buckets")] +#[derive(Debug, PartialEq)] enum BucketMode { Forced(bool), Auto { @@ -1474,6 +1528,13 @@ impl BucketMode { /// `ns_per_hit = settled_span_ms * 1e6 / settled_hits`. Returns the /// closed block (index + sample) plus, after the 6th block, the decision /// (`Decided(true)` iff the median adjacent-pair ratio is < 0.97). + /// + /// Inconclusive close (exactness-neutral): a block that reaches + /// `span_ms >= 2 * t_block` with fewer than 4 settled chunks (a tail of + /// small chunks) can never decide. An inconclusive OFF block is dropped + /// and a fresh OFF block starts at the same index; an inconclusive ON + /// block ends the trial as `Decided(false)`. Reported as the third + /// element; conclusive closes return it as `None`. fn commit_block_span( &mut self, span_ms: f64, @@ -1481,9 +1542,13 @@ impl BucketMode { max_hits: u32, t_block: f64, t_settle: f64, - ) -> (Option<(usize, BlockSample)>, Option) { + ) -> ( + Option<(usize, BlockSample)>, + Option, + Option, + ) { let BucketMode::Auto { phase, blocks } = self else { - return (None, None); + return (None, None, None); }; let AutoPhase::Block { on, @@ -1495,7 +1560,7 @@ impl BucketMode { settled_chunks, } = phase else { - return (None, None); + return (None, None, None); }; debug_assert_eq!(*k, blocks.len()); let settled_before = *acc >= t_settle; @@ -1507,7 +1572,29 @@ impl BucketMode { *settled_chunks += 1; } if *acc < t_block || *settled_chunks < 4 { - return (None, None); + if *acc >= 2.0 * t_block && *settled_chunks < 4 { + let inc = InconclusiveBlock { + k: *k, + on: *on, + span_ms: *acc, + settled_chunks: *settled_chunks, + }; + if *on { + *self = BucketMode::Decided(false); + } else { + *phase = AutoPhase::Block { + on: false, + k: *k, + span_ms: 0.0, + settled_span_ms: 0.0, + settled_hits: 0, + chunks: 0, + settled_chunks: 0, + }; + } + return (None, None, Some(inc)); + } + return (None, None, None); } debug_assert!(*settled_hits > 0); let closed_k = *k; @@ -1530,7 +1617,7 @@ impl BucketMode { chunks: 0, settled_chunks: 0, }; - return (Some((closed_k, closed)), None); + return (Some((closed_k, closed)), None, None); } let ratios = block_pair_ratios(blocks); let use_on = decide_bucket_autotune(&ratios); @@ -1547,7 +1634,17 @@ impl BucketMode { .collect(), }; *self = BucketMode::Decided(use_on); - (Some((closed_k, closed)), Some(decision)) + (Some((closed_k, closed)), Some(decision), None) + } + + /// Closed blocks so far, if the engine is still in `Auto` (`None` once + /// forced or decided). The engine-end ledger uses this to say why there + /// is no decision. + fn undecided_blocks(&self) -> Option { + match self { + BucketMode::Auto { blocks, .. } => Some(blocks.len()), + BucketMode::Forced(_) | BucketMode::Decided(_) => None, + } } } @@ -1781,6 +1878,8 @@ impl Engine { gap_max: 0.0, gap_max_pair: ("", ""), gap_pairs: Vec::new(), + gap_long_n: 0, + gap_long_ms: 0.0, device_seeds: false, reference_uploads: 1, query_swaps: 0, @@ -1834,7 +1933,7 @@ impl Engine { /// How many times this engine uploaded a reference index (§9.7 / AM-A1). /// - /// The executor asserts this equals the *reference-bin* count, not the + /// The executor asserts this equals the *visit* count, not the /// work-unit count. Counting host `SeedTable::build` calls alone would miss a /// refactor that rebuilt the `Engine` per pair, which is the regression the /// amendment is guarding against — so this counts device uploads. @@ -3135,7 +3234,7 @@ impl Engine { .expect("block span events ensured above"), )? as f64; let max_hits = self.max_hits; - let (closed, decided) = self.bucket_mode.commit_block_span( + let (closed, decided, inconclusive) = self.bucket_mode.commit_block_span( span_ms, iter_num_hits, max_hits, @@ -3157,6 +3256,19 @@ impl Engine { sample.ns_per_hit, ); } + if let Some(inc) = &inconclusive { + if inc.on { + eprintln!( + " ref buckets: auto -> off (inconclusive: block k={} mode=on span {:.1} ms settled {} chunks)", + inc.k, inc.span_ms, inc.settled_chunks, + ); + } else { + eprintln!( + " ref buckets: dropped block eng={} k={} mode=off (inconclusive; span_ms={:.1} settled={})", + self.bucket_eng_id, inc.k, inc.span_ms, inc.settled_chunks, + ); + } + } if let Some(d) = &decided { let r3 = |v: &[f64]| -> Vec { v.iter().map(|x| (x * 1000.0).round() / 1000.0).collect() @@ -3608,6 +3720,8 @@ impl Engine { // on one stream, so the delta is GPU-timeline idle — the bubble the host put // there. A negative or absurd reading would mean the events are unordered, // so anything outside [0, 1000] ms is dropped rather than trusted. + // Intervals above 1,000 ms are unit-transition work (query pack, ref + // prep) counted separately so they stay visible instead of silent. if let (Some(prev), Some((start, _))) = (self.last_end.as_ref(), p.events.as_ref()) { if let Ok(gap) = prev.elapsed_ms(start) { if (0.0f32..=1000.0f32).contains(&gap) { @@ -3625,6 +3739,9 @@ impl Engine { } None => self.gap_pairs.push((key, gap, 1)), } + } else if gap > 1000.0 { + self.gap_long_n += 1; + self.gap_long_ms += gap; } } } @@ -3651,6 +3768,8 @@ impl Engine { self.gap_max = 0.0; self.gap_max_pair = ("", ""); self.gap_pairs.clear(); + self.gap_long_n = 0; + self.gap_long_ms = 0.0; self.last_end = None; self.last_name = ""; } @@ -3662,11 +3781,38 @@ impl Engine { v } + /// Intervals discarded by the gap accounting because they exceed 1,000 ms + /// (unit-transition work), as `(count, summed ms)`. + pub fn discarded_gaps(&self) -> (u64, f32) { + (self.gap_long_n, self.gap_long_ms) + } + /// Round 71: `(summed gap ms, pairs measured, largest gap, its stage pair)`. pub fn stage_gaps(&self) -> (f32, u64, f32, (&'static str, &'static str)) { (self.gap_ms, self.gap_n, self.gap_max, self.gap_max_pair) } + /// Engine-end autotune ledger: if still in `Auto` (the chunks ran out + /// before 6 blocks), the count of closed blocks so the log says why + /// there is no decision. `None` once forced or decided. + #[cfg(feature = "ref-loc-buckets")] + pub fn bucket_undecided_blocks(&self) -> Option { + self.bucket_mode.undecided_blocks() + } + + /// Prints the engine-end ledger line when still undecided. Call once per + /// engine after its last chunk, under `--time` only via `self.timing`. + pub fn finish_bucket_autotune(&self) { + #[cfg(feature = "ref-loc-buckets")] + { + if let Some(k) = self.bucket_undecided_blocks() { + if self.timing { + eprintln!("ref buckets: auto undecided (engine ended after {k} blocks)"); + } + } + } + } + /// A genuine host dependency (Phase 1 §2): the host is about to read a device /// result, free a buffer a queued kernel touches, or return to the caller. fn sync_pipeline(&mut self) -> Result<(), DriverError> { @@ -3706,8 +3852,10 @@ impl Engine { /// /// Not statistics: the executor asserts these, because the failure they guard /// against — rebuilding or re-uploading the reference index per work unit instead -/// of per reference bin — is invisible in output and only shows up as ~1 GB of -/// extra H->D traffic per avoided reuse. +/// of per reference visit — is invisible in output and only shows up as ~1 GB of +/// extra H->D traffic per avoided reuse. Under the round-90 unit partition one +/// visit is one reference build/upload, so `visits` is the bin count on the +/// whole-bin path and the visit count on the unit path. #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] pub struct Lifecycle { pub seed_table_builds: u32, @@ -3722,11 +3870,11 @@ impl Lifecycle { /// /// Returns the first violation as a message rather than panicking, so a /// caller can report it alongside the rest of a profile. - pub fn check(&self, reference_bins: u32, work_units: u32) -> Result<(), String> { + pub fn check(&self, visits: u32, work_units: u32) -> Result<(), String> { let want = [ - ("seed_table_builds", self.seed_table_builds, reference_bins), - ("engine_creations", self.engine_creations, reference_bins), - ("reference_uploads", self.reference_uploads, reference_bins), + ("seed_table_builds", self.seed_table_builds, visits), + ("engine_creations", self.engine_creations, visits), + ("reference_uploads", self.reference_uploads, visits), ("query_swaps", self.query_swaps, work_units), ("work_units_executed", self.work_units_executed, work_units), ]; @@ -3734,7 +3882,7 @@ impl Lifecycle { if got != expect { return Err(format!( "lifecycle: {name} = {got}, expected {expect} \ - (reference_bins {reference_bins}, work_units {work_units})" + (visits {visits}, work_units {work_units})" )); } } @@ -4459,32 +4607,34 @@ mod tests { // a tail joins the span but is never settled. assert_eq!( mode.commit_block_span(10.0, MAX, MAX, T_BLOCK, T_SETTLE), - (None, None) + (None, None, None) ); assert_eq!( mode.commit_block_span(10.0, MAX, MAX, T_BLOCK, T_SETTLE), - (None, None) + (None, None, None) ); assert_eq!( mode.commit_block_span(10.0, 1, MAX, T_BLOCK, T_SETTLE), - (None, None) + (None, None, None) ); assert_eq!( mode.commit_block_span(10.0, MAX, MAX, T_BLOCK, T_SETTLE), - (None, None) + (None, None, None) ); assert_eq!( mode.commit_block_span(10.0, MAX, MAX, T_BLOCK, T_SETTLE), - (None, None) + (None, None, None) ); assert_eq!( mode.commit_block_span(10.0, MAX, MAX, T_BLOCK, T_SETTLE), - (None, None) + (None, None, None) ); // Span 60 ms but only 3 settled: both conditions are required, so // the next chunk closes block 0. - let (closed, decided) = mode.commit_block_span(10.0, MAX, MAX, T_BLOCK, T_SETTLE); + let (closed, decided, inconclusive) = + mode.commit_block_span(10.0, MAX, MAX, T_BLOCK, T_SETTLE); assert_eq!(decided, None); + assert_eq!(inconclusive, None); let Some((k, sample)) = closed else { panic!("70 ms span with 4 settled chunks must close block 0"); }; @@ -4505,10 +4655,15 @@ mod tests { for expect_k in 1..BUCKET_BLOCKS { let expect_on = expect_k % 2 == 1; let span = if expect_on { 9.0 } else { 10.0 }; - let (closed, decided) = loop { - let (closed, decided) = mode.commit_block_span(span, MAX, MAX, T_BLOCK, T_SETTLE); + let (closed, decided, inconclusive) = loop { + let (closed, decided, inconclusive) = + mode.commit_block_span(span, MAX, MAX, T_BLOCK, T_SETTLE); + assert!( + inconclusive.is_none(), + "settled path must not go inconclusive" + ); if closed.is_some() || decided.is_some() { - break (closed, decided); + break (closed, decided, inconclusive); } }; let Some((k, sample)) = closed else { @@ -4539,6 +4694,142 @@ mod tests { assert!(!BucketMode::Decided(false).production_path()); } + /// Inconclusive OFF blocks (span >= 2*T_BLOCK, <4 settled) are dropped + /// and a fresh OFF block starts at the same index; the trial stays in + /// Auto with no decided path. + #[cfg(feature = "ref-loc-buckets")] + #[test] + fn bucket_inconclusive_off_drops_and_retries_same_block() { + use super::{AutoPhase, BucketMode, InconclusiveBlock}; + const T_BLOCK: f64 = 60.0; + const T_SETTLE: f64 = 15.0; + const MAX: u32 = 16_711_680; + let mut mode = BucketMode::Auto { + phase: AutoPhase::Block { + on: false, + k: 0, + span_ms: 0.0, + settled_span_ms: 0.0, + settled_hits: 0, + chunks: 0, + settled_chunks: 0, + }, + blocks: Vec::new(), + }; + // 11 tail chunks (ineligible): span 110 < 120, no decision, no drop. + for _ in 0..11 { + assert_eq!( + mode.commit_block_span(10.0, 1, MAX, T_BLOCK, T_SETTLE), + (None, None, None) + ); + } + // 12th tail chunk: span 120 >= 2*T_BLOCK with 0 settled -> drop. + let (closed, decided, inconclusive) = + mode.commit_block_span(10.0, 1, MAX, T_BLOCK, T_SETTLE); + assert_eq!(closed, None); + assert_eq!(decided, None); + assert_eq!( + inconclusive, + Some(InconclusiveBlock { + k: 0, + on: false, + span_ms: 120.0, + settled_chunks: 0, + }) + ); + // Dropped: no block recorded, fresh OFF block at the same index. + assert!(matches!(mode, BucketMode::Auto { .. })); + assert_eq!(mode.undecided_blocks(), Some(0)); + assert!(!mode.production_path()); + assert!(mode.auto_chunk(MAX, MAX)); + // The retry can still close conclusively. + for _ in 0..5 { + let (_, _, inc) = mode.commit_block_span(10.0, MAX, MAX, T_BLOCK, T_SETTLE); + assert_eq!(inc, None); + } + let (closed, decided, inc) = mode.commit_block_span(10.0, MAX, MAX, T_BLOCK, T_SETTLE); + assert_eq!(inc, None); + assert_eq!(decided, None); + let Some((k, sample)) = closed else { + panic!("retried OFF block must close"); + }; + assert_eq!(k, 0); + assert!(!sample.on); + assert_eq!(mode.undecided_blocks(), Some(1)); + } + + /// An inconclusive ON block ends the trial as `Decided(false)`: the + /// production default path, with the ledger fields for the + /// `auto -> off (inconclusive: ...)` line. + #[cfg(feature = "ref-loc-buckets")] + #[test] + fn bucket_inconclusive_on_falls_back_to_off() { + use super::{AutoPhase, BlockSample, BucketMode, InconclusiveBlock}; + const T_BLOCK: f64 = 60.0; + const T_SETTLE: f64 = 15.0; + const MAX: u32 = 16_711_680; + let mut mode = BucketMode::Auto { + phase: AutoPhase::Block { + on: true, + k: 1, + span_ms: 0.0, + settled_span_ms: 0.0, + settled_hits: 0, + chunks: 0, + settled_chunks: 0, + }, + blocks: vec![BlockSample { + on: false, + ns_per_hit: 1.0, + chunks: 7, + settled_chunks: 4, + hits: 4 * u64::from(MAX), + span_ms: 70.0, + }], + }; + for _ in 0..11 { + assert_eq!( + mode.commit_block_span(10.0, 1, MAX, T_BLOCK, T_SETTLE), + (None, None, None) + ); + } + let (closed, decided, inconclusive) = + mode.commit_block_span(10.0, 1, MAX, T_BLOCK, T_SETTLE); + assert_eq!(closed, None); + assert_eq!(decided, None); + assert_eq!( + inconclusive, + Some(InconclusiveBlock { + k: 1, + on: true, + span_ms: 120.0, + settled_chunks: 0, + }) + ); + assert_eq!(mode, BucketMode::Decided(false)); + assert!(!mode.production_path()); + assert!(!mode.auto_chunk(MAX, MAX)); + assert_eq!(mode.undecided_blocks(), None); + } + + /// An engine that finishes in `Auto` reports its closed-block count so + /// the ledger can say why there is no decision; forced/decided engines + /// report nothing. + #[cfg(feature = "ref-loc-buckets")] + #[test] + fn bucket_engine_end_reports_undecided_blocks() { + use super::{AutoPhase, BucketMode}; + let warm = BucketMode::Auto { + phase: AutoPhase::Warm { n: 0 }, + blocks: Vec::new(), + }; + assert_eq!(warm.undecided_blocks(), Some(0)); + assert_eq!(BucketMode::Forced(true).undecided_blocks(), None); + assert_eq!(BucketMode::Forced(false).undecided_blocks(), None); + assert_eq!(BucketMode::Decided(true).undecided_blocks(), None); + assert_eq!(BucketMode::Decided(false).undecided_blocks(), None); + } + /// Two distributions with the *same* mean hits/seed, the same seed count /// and the same total work must come out with very different numbers, or /// these statistics cannot do the job the mean was failing at. This checks diff --git a/src/plan.rs b/src/plan.rs index fdc5665..d391aba 100644 --- a/src/plan.rs +++ b/src/plan.rs @@ -95,6 +95,192 @@ pub fn assign_bins(bins: &[Bin], workers: usize) -> Vec> { out } +/// One contiguous query slice of a single reference bin (round 90). +/// +/// A worker executes its visits in ascending bin id, and the units inside a +/// visit in ascending query id. Whole-bin ownership is the special case +/// `queries == 0..Q`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Visit { + pub bin: usize, + pub queries: std::ops::Range, +} + +/// Static unit-level partition of a frozen plan over `workers` workers. +/// +/// Pure and deterministic. With `N = R*Q` units, quotas are `N div W` and one +/// more (exactly `N mod W` large). A quota `c` gives `c div Q` whole-bin slots +/// plus a residual of `c mod Q` units. Whole slots take the largest bins first +/// (LPT on reference bp exactly as [`assign_bins`], ties by bin id, only workers +/// with a free slot). The leftover bins form one residual stream of units in bin +/// order then query order, cut into consecutive pieces; the large/small quota +/// order along the stream is chosen by an `O(W^2)` DP over how many of each have +/// been placed, minimising total bin visits (a piece at stream offset `S` with +/// length `d` costs `1 + (((S mod Q) + d - 1) / Q)` visits, `0` when `d == 0`). +/// Ties break towards fewer visits on large-quota workers, then earliest large +/// quotas. `W == 1` is one whole-bin visit per bin, identical to today. +pub fn unit_partition(plan: &Plan, workers: usize) -> Vec> { + let w = workers.max(1); + let r = plan.reference_bins.len(); + let q = plan.query_bins.len(); + let n = plan.units.len(); + if r == 0 || q == 0 || n == 0 { + return vec![Vec::new(); w]; + } + if w == 1 { + let mut v: Vec = plan + .reference_bins + .iter() + .map(|b| Visit { + bin: b.id as usize, + queries: 0..q, + }) + .collect(); + v.sort_by_key(|x| x.bin); + return vec![v]; + } + let a = n / w; + let rem = n % w; + let (nl, large_c, small_c) = if rem == 0 { (0, a, a) } else { (rem, a + 1, a) }; + let ns = w - nl; + let (d_large, d_small) = (large_c % q, small_c % q); + // DP over (#small placed, #large placed); stream offset is fixed by the + // counts, so the cost of the next piece is known. Best is (total visits, + // visits on large-quota pieces); exact ties keep the small arm, which + // places large quotas earliest (fixed order). + let visits_at = |s: usize, d: usize| -> u32 { + if d == 0 { + 0 + } else { + (1 + ((s % q) + d - 1) / q) as u32 + } + }; + let mut best: Vec>> = vec![vec![None; nl + 1]; ns + 1]; + let mut large_last: Vec> = vec![vec![false; nl + 1]; ns + 1]; + best[0][0] = Some((0, 0)); + for i in 0..=ns { + for j in 0..=nl { + let Some((tot, lg)) = best[i][j] else { + continue; + }; + let s = i * d_small + j * d_large; + if i < ns { + let cand = (tot + visits_at(s, d_small), lg); + let replace = match best[i + 1][j] { + None => true, + Some(cur) => cand < cur, + }; + if replace { + best[i + 1][j] = Some(cand); + large_last[i + 1][j] = false; + } + } + if j < nl { + let v = visits_at(s, d_large); + let cand = (tot + v, lg + v); + let replace = match best[i][j + 1] { + None => true, + Some(cur) => cand < cur, + }; + if replace { + best[i][j + 1] = Some(cand); + large_last[i][j + 1] = true; + } + } + } + } + let mut is_large = vec![false; w]; + { + let (mut i, mut j) = (ns, nl); + while i + j > 0 { + if large_last[i][j] { + is_large[i + j - 1] = true; + j -= 1; + } else { + i -= 1; + } + } + } + let quota = |x: usize| { + if is_large[x] { large_c } else { small_c } + }; + // Whole-bin slots, largest bins first into the lightest worker with a free + // slot (ties by worker index), exactly like `assign_bins`. + let mut order: Vec<&Bin> = plan.reference_bins.iter().collect(); + order.sort_by(|x, y| y.total_bp.cmp(&x.total_bp).then(x.id.cmp(&y.id))); + let mut load = vec![0u64; w]; + let mut used = vec![0usize; w]; + let mut taken = vec![false; r]; + let mut visits: Vec> = vec![Vec::new(); w]; + for b in order { + let mut pick: Option = None; + for x in 0..w { + if used[x] >= quota(x) / q { + continue; + } + pick = Some(match pick { + None => x, + Some(p) => { + if (load[x], x) < (load[p], p) { + x + } else { + p + } + } + }); + } + let Some(p) = pick else { break }; + load[p] += b.total_bp; + used[p] += 1; + taken[b.id as usize] = true; + visits[p].push(Visit { + bin: b.id as usize, + queries: 0..q, + }); + } + // Residual stream: leftover bins (descending bp, ties by id), query order. + let mut rest: Vec<&Bin> = plan + .reference_bins + .iter() + .filter(|b| !taken[b.id as usize]) + .collect(); + rest.sort_by(|x, y| y.total_bp.cmp(&x.total_bp).then(x.id.cmp(&y.id))); + let mut stream: Vec<(usize, usize)> = Vec::with_capacity(rest.len() * q); + for b in &rest { + for qq in 0..q { + stream.push((b.id as usize, qq)); + } + } + let mut off = 0usize; + for x in 0..w { + let d = quota(x) % q; + let mut k = off; + let end = off + d; + while k < end { + let b = stream[k].0; + let mut e = k + 1; + while e < end && stream[e].0 == b { + e += 1; + } + visits[x].push(Visit { + bin: b, + queries: stream[k].1..stream[e - 1].1 + 1, + }); + k = e; + } + off = end; + } + debug_assert_eq!(off, stream.len()); + for v in &mut visits { + v.sort_by_key(|t| t.bin); + debug_assert!( + v.windows(2).all(|w| w[0].bin != w[1].bin), + "two visits of one bin on a worker" + ); + } + visits +} + /// KegAlign's block rule, reproduced exactly, for matched-granularity runs /// (benchmark plan §5 Mode A). /// @@ -288,6 +474,232 @@ pub fn candidate_plans(reference: &[RecordMeta], base: &Plan, workers: usize) -> out } +/// Default block target both sides fall back to: the KegAlign-matched digest +/// (`-B`'s CLI default). `-B 0` resolves to this at `workers <= 1`, and the +/// automatic layout derives its reference count from it. +pub const DEFAULT_BLOCK_TARGET: u64 = 500_000_000; + +/// Rejected combination: the automatic layout bins by LPT record balance, +/// which cannot reproduce KegAlign's sequential fill. +pub const AUTO_KEGALIGN_ERROR: &str = + "automatic layout uses LPT bins; drop --kegalign-bins or give -B explicitly"; + +/// Automatic reference-bin count for `-B 0` at `workers >= 2` (rank-2 rule): +/// the smallest multiple of `workers` that does not coarsen the default +/// layout (`R_def`), capped at one bin per record so no worker idles by +/// construction. Divisibility is the whole point: `R % W == 0` keeps every +/// worker holding the same number of bins. +pub fn auto_ref_bin_count(n_records: usize, r_def: usize, workers: usize) -> usize { + if n_records == 0 { + return 0; + } + let w = workers.max(1); + n_records.min(r_def.max(1).div_ceil(w).saturating_mul(w)) +} + +/// Resolved automatic layout: the plan plus the targets that reproduce it +/// through the normal planner, for the manifest and the layout line. +pub struct AutoLayout { + pub plan: Plan, + /// `Some((R, Q))` when the automatic candidate did not fit the device budget + /// and the plan fell back to the ordinary default policy (Grok r88 review: + /// `-B 0` must never be harder to admit than the default). + pub fell_back: Option<(usize, usize)>, + /// `ceil(total_ref_bp / R)`: any explicit `-B` in + /// `(total/R, total/(R-1)]` round-trips through [`layout_n_bins`] to the + /// same `R`, and [`bin_records_n`] is a pure function of the count, so an + /// explicit rerun with this target rebuilds these bins exactly. + pub seq_target: u64, + /// Whole-query total when Q=1 collapses the query (same round-trip), else + /// the target actually used (explicit override or default). + pub query_target: u64, +} + +/// Everything [`auto_layout`]'s fit checks need beyond metadata and budgets. +/// `threads`/`max_seeds` feed [`host_estimate`]; the caller derives them the +/// same way `run` does so the layout-time check agrees with the real preflight. +pub struct AutoCtx { + pub kmer_size: usize, + pub step: u32, + pub max_hits: u32, + pub wga_chunk_size: u32, + pub transitions: bool, + pub threads: usize, + pub max_seeds: usize, +} + +/// Automatic layout for `-B 0` (rank-2 policy). +/// +/// `workers <= 1` resolves trivially to today's default plan — no Q=1 attempt, +/// no new bins — so a one-worker run is byte-identical to `-B 500000000`. +/// At `workers >= 2` the reference takes [`auto_ref_bin_count`] bins via +/// [`bin_records_n`], and the query collapses to one bin iff *both* the device +/// fit ([`worst_unit_bytes`] against `budget_bytes`, the same budget +/// [`plan_within_budget`] uses) and the host fit ([`host_preflight`] against +/// `host_available`, already discounted; `None` means unknown and admits) +/// allow it — otherwise the default query blocks are kept. An explicit +/// `query_override` is honoured as given and skips the fit check. The query +/// target is never derived from the reference target. +/// +/// Fit-checks and falls back, never shrinks: a final plan that still exceeds +/// the device budget errors with an explicit-flags hint, exactly like +/// matched-granularity mode, instead of coarsening the layout behind the +/// caller's back. Kernels, chunking and emitter ordinals are untouched — +/// [`from_bins`] keeps dense ref-outer ordinals. +#[allow(clippy::too_many_arguments)] +pub fn auto_layout( + reference: &[RecordMeta], + query: &[RecordMeta], + workers: usize, + exec_workers: usize, + budget_bytes: u64, + host_available: Option, + query_override: Option, + ctx: &AutoCtx, +) -> Result { + let ref_total: u64 = reference.iter().map(|r| r.len).sum(); + let qry_total: u64 = query.iter().map(|r| r.len).sum(); + if workers.max(1) <= 1 { + let q = query_override.unwrap_or(DEFAULT_BLOCK_TARGET); + let plan = plan_with(reference, query, DEFAULT_BLOCK_TARGET, q.max(1), false); + return Ok(AutoLayout { + plan, + fell_back: None, + seq_target: DEFAULT_BLOCK_TARGET, + query_target: q, + }); + } + let r_def = layout_n_bins(ref_total, DEFAULT_BLOCK_TARGET, reference.len()); + let r = auto_ref_bin_count(reference.len(), r_def, workers); + let reference_bins = bin_records_n(reference, r); + let seq_target = match r { + 0 => DEFAULT_BLOCK_TARGET, + _ => ref_total.div_ceil(r as u64).max(1), + }; + let (query_bins, query_target) = match query_override { + Some(q) => (bin_records(query, q.max(1)), q), + None => { + let single = bin_records_n(query, 1); + let cand = from_bins(reference_bins.clone(), single.clone()); + let device_ok = worst_unit_bytes( + &cand, + ctx.kmer_size, + ctx.step, + ctx.max_hits, + ctx.wga_chunk_size, + ctx.transitions, + ) + .map(|worst| worst <= budget_bytes) + .unwrap_or(false); + let host_ok = match host_available { + None => true, + Some(avail) => { + let est = host_estimate( + &cand, + ref_total, + qry_total, + ctx.kmer_size, + ctx.step, + ctx.threads, + ctx.max_seeds, + ); + host_preflight( + &est, + &assign_bins(&reference_bins, exec_workers.max(1)), + avail, + ) + .is_ok() + } + }; + if device_ok && host_ok { + (single, qry_total.max(1)) + } else { + ( + bin_records(query, DEFAULT_BLOCK_TARGET), + DEFAULT_BLOCK_TARGET, + ) + } + } + }; + let plan = from_bins(reference_bins, query_bins); + let worst = worst_unit_bytes( + &plan, + ctx.kmer_size, + ctx.step, + ctx.max_hits, + ctx.wga_chunk_size, + ctx.transitions, + )?; + if worst > budget_bytes { + if query_override.is_none() { + // The candidate does not fit: run exactly what `-B 500000000` would + // (including plan_within_budget's own halving), never an error. + let (candidate_r, candidate_q) = (plan.reference_bins.len(), plan.query_bins.len()); + let (plan, _worst) = plan_within_budget( + reference, + query, + DEFAULT_BLOCK_TARGET, + DEFAULT_BLOCK_TARGET, + budget_bytes, + ctx.kmer_size, + ctx.step, + ctx.max_hits, + false, + ctx.wga_chunk_size, + ctx.transitions, + )?; + return Ok(AutoLayout { + plan, + fell_back: Some((candidate_r, candidate_q)), + seq_target: DEFAULT_BLOCK_TARGET, + query_target: DEFAULT_BLOCK_TARGET, + }); + } + return Err(format!( + "automatic layout (R={} Q={}) with the explicit --query-block-size needs {:.1} GB per work unit, only {:.1} GB budgeted: \ + drop --query-block-size or give -B explicitly", + plan.reference_bins.len(), + plan.query_bins.len(), + worst as f64 / 1e9, + budget_bytes as f64 / 1e9 + )); + } + Ok(AutoLayout { + plan, + fell_back: None, + seq_target, + query_target, + }) +} + +/// The one stderr line an automatic layout prints: the explicit flags that +/// reproduce it plus the per-worker bin counts (`owners`) under the layout +/// worker count, so the run is reproducible without `-B 0`. +pub fn auto_layout_line(workers: usize, seq_target: u64, query_target: u64, plan: &Plan) -> String { + let owners = assign_bins(&plan.reference_bins, workers.max(1)) + .iter() + .map(|bins| bins.len().to_string()) + .collect::>() + .join("+"); + let ref_total: u64 = plan.reference_bins.iter().map(|b| b.total_bp).sum(); + let n_records: usize = plan.reference_bins.iter().map(|b| b.record_ids.len()).sum(); + let r = plan.reference_bins.len(); + // The printed -B reproduces these bins only if the count-first planner + // derives the same R from it (true for genome-scale inputs; tiny toy sets + // can break it). Say so rather than print flags that would not replay. + let hint = if layout_n_bins(ref_total, seq_target.max(1), n_records) == r { + String::new() + } else { + " [flags do not reproduce this R on this input; use --dump-manifest/--from-manifest]".into() + }; + format!( + "layout: auto W={workers} -> --seq-block-size {seq_target} --query-block-size \ + {query_target} (R={r} Q={}, {} units, owners {owners}){hint}", + plan.query_bins.len(), + plan.units.len() + ) +} + fn from_bins(reference_bins: Vec, query_bins: Vec) -> Plan { let mut units = Vec::with_capacity(reference_bins.len() * query_bins.len()); let mut ordinal = 0u32; @@ -518,14 +930,7 @@ pub fn max_hit_capacity( if max_hits == 0 { return Err("max_hits must be positive".into()); } - let base = worst_unit_bytes( - plan, - kmer_size, - step, - max_hits, - wga_chunk_size, - transitions, - )?; + let base = worst_unit_bytes(plan, kmer_size, step, max_hits, wga_chunk_size, transitions)?; if base > budget_bytes { return Err(format!( "max_hits {max_hits} needs {base} bytes per work unit, only {budget_bytes} bytes budgeted" @@ -1495,7 +1900,7 @@ mod tests { /// *over* target — so blocks overshoot and the last one may be short. #[test] fn sequential_bins_match_kegalign_block_fill() { - use super::{bin_records_sequential, RecordMeta}; + use super::{RecordMeta, bin_records_sequential}; let rec = |id: u32, len: u64| RecordMeta { id, name: format!("chr{id}"), @@ -1594,7 +1999,7 @@ mod tests { #[test] fn assign_bins_is_a_balanced_deterministic_partition() { - use super::{assign_bins, Bin}; + use super::{Bin, assign_bins}; let bins: Vec = [100u64, 90, 80, 70, 10] .iter() .enumerate() @@ -1651,7 +2056,7 @@ mod tests { /// count. #[test] fn host_preflight_keeps_prefetch_then_falls_back_then_errors() { - use super::{host_peak, host_preflight, HostEstimate}; + use super::{HostEstimate, host_peak, host_preflight}; // shared 100, per-worker prefetch 300, no-prefetch 150. let est = HostEstimate { shared: 100, @@ -1659,7 +2064,7 @@ mod tests { per_worker_no_prefetch: 150, }; let multi = |w: usize| vec![vec![0u32, 1]; w]; // every worker owns 2 bins - // 1 worker: 100 + 300 = 400 fits → prefetch kept. + // 1 worker: 100 + 300 = 400 fits → prefetch kept. assert_eq!(host_preflight(&est, &multi(1), 400), Ok(true)); // 4 workers: 100 + 4*300 = 1300 fails, 100 + 4*150 = 700 fits → disabled. assert_eq!(host_preflight(&est, &multi(4), 1000), Ok(false)); @@ -1931,6 +2336,152 @@ mod tests { assert_eq!(layout_n_bins(450, 150, 6), 3); } + fn auto_test_ctx() -> AutoCtx { + AutoCtx { + kmer_size: 12, + step: 1, + max_hits: 1000, + wga_chunk_size: 250_000, + transitions: true, + threads: 4, + max_seeds: 10_000, + } + } + + /// `-B 0` rank-2 rule: `R = min(n_records, ceil(R_def / W) * W)` — the + /// smallest multiple of W that does not coarsen the default layout. + #[test] + fn auto_ref_bin_count_follows_the_rank2_rule() { + // 7 x 499 Mbp + 5 x 1 bp: total 3,493,000,005 bp, so R_def = 7 over 12 + // records against the 500 Mbp default. + let mut lens = vec![499_000_000u64; 7]; + lens.extend_from_slice(&[1; 5]); + let r = recs(&lens); + assert_eq!(r.len(), 12); + assert_eq!(layout_n_bins(3_493_000_005, 500_000_000, 12), 7); + assert_eq!(auto_ref_bin_count(12, 7, 2), 8); + assert_eq!(auto_ref_bin_count(12, 7, 4), 8); + assert_eq!(auto_ref_bin_count(12, 7, 3), 9); + assert_eq!(auto_ref_bin_count(12, 7, 8), 8); + assert_eq!(auto_ref_bin_count(5, 7, 4), 5, "one bin per record caps R"); + assert_eq!(auto_ref_bin_count(0, 7, 4), 0); + // The full path agrees, and its printed targets rebuild the same plan. + let q = recs(&[10, 10]); + let auto = auto_layout(&r, &q, 2, 2, u64::MAX, None, None, &auto_test_ctx()).unwrap(); + assert_eq!(auto.plan.reference_bins.len(), 8); + assert_eq!( + plan_with(&r, &q, auto.seq_target, auto.query_target, false), + auto.plan + ); + } + + /// Resolving 0 at W=1 is the default plan exactly — no Q=1 attempt, no new + /// bins — which is what makes one-worker `-B 0` byte-identical to today. + #[test] + fn auto_layout_at_w1_is_the_default_plan() { + let r = recs(&[250, 240, 200, 190, 180, 170, 160]); + let q = recs(&[195, 180, 160, 155, 150, 145, 60]); + let ctx = auto_test_ctx(); + let auto = auto_layout(&r, &q, 1, 1, u64::MAX, None, None, &ctx).unwrap(); + assert_eq!( + auto.plan, + plan_with(&r, &q, 500_000_000, 500_000_000, false) + ); + assert_eq!( + (auto.seq_target, auto.query_target), + (500_000_000, 500_000_000) + ); + // An explicit query target is honoured as given, still with no Q=1. + let auto = auto_layout(&r, &q, 1, 1, u64::MAX, None, Some(10_000), &ctx).unwrap(); + assert_eq!(auto.plan, plan_with(&r, &q, 500_000_000, 10_000, false)); + assert_eq!(auto.query_target, 10_000); + } + + /// Q=1 needs *both* budgets; either failure falls back to the default + /// query blocks, and an explicit query target is honoured as given. + #[test] + fn auto_layout_q1_needs_both_budgets() { + // 2 x 100 bp reference over 2 workers: R_def = 1, so R = 2. + let r = recs(&[100, 100]); + // 2 x 600 Mbp query: Q=1 holds 1.2 Gbp where the default holds 600 + // Mbp, so a budget between the two estimates separates the paths. + let q = recs(&[600_000_000, 600_000_000]); + let ctx = auto_test_ctx(); + let wb = |p: &Plan| { + worst_unit_bytes( + p, + ctx.kmer_size, + ctx.step, + ctx.max_hits, + ctx.wga_chunk_size, + ctx.transitions, + ) + .unwrap() + }; + let q1 = from_bins(bin_records_n(&r, 2), bin_records_n(&q, 1)); + let defq = from_bins(bin_records_n(&r, 2), bin_records(&q, 500_000_000)); + assert_eq!(defq.query_bins.len(), 2); + let (w1, w0) = (wb(&q1), wb(&defq)); + assert!( + w1 > w0, + "fixture must separate Q=1 ({w1}) from default-Q ({w0})" + ); + + // Both budgets admit: Q=1, and the printed targets rebuild the plan + // through the normal planner. + let one = auto_layout(&r, &q, 2, 2, w1, None, None, &ctx).unwrap(); + assert_eq!(one.plan, q1); + assert_eq!(one.query_target, 1_200_000_000); + assert_eq!( + plan_with(&r, &q, one.seq_target, one.query_target, false), + one.plan + ); + assert_eq!( + auto_layout_line(2, one.seq_target, one.query_target, &one.plan), + "layout: auto W=2 -> --seq-block-size 100 --query-block-size 1200000000 \ + (R=2 Q=1, 2 units, owners 1+1)" + ); + + // Device budget admits the fallback but not Q=1: default query blocks. + let poor = auto_layout(&r, &q, 2, 2, w0, None, None, &ctx).unwrap(); + assert_eq!(poor.plan, defq); + assert_eq!(poor.query_target, 500_000_000); + assert_eq!( + plan_with(&r, &q, poor.seq_target, poor.query_target, false), + poor.plan + ); + + // Host budget rejects Q=1 under ample device memory: same fallback. + let cramped = auto_layout(&r, &q, 2, 2, w1, Some(1), None, &ctx).unwrap(); + assert_eq!(cramped.plan, defq); + + // Explicit query target is honoured as given, fit permitting. + let expl = auto_layout(&r, &q, 2, 2, w0, None, Some(500_000_000), &ctx).unwrap(); + assert_eq!(expl.plan, defq); + assert_eq!(expl.query_target, 500_000_000); + } + + /// The manifest stores the resolved targets (never 0) and round-trips. + #[test] + fn auto_manifest_round_trips_resolved_targets() { + let r = recs(&[100, 100]); + let q = recs(&[50, 50]); + let auto = auto_layout(&r, &q, 2, 2, u64::MAX, None, None, &auto_test_ctx()).unwrap(); + assert_eq!(auto.plan.reference_bins.len(), 2); + assert_eq!(auto.plan.query_bins.len(), 1); + let mut m = manifest_fixture(auto.plan.clone()); + m.seq_block_size = auto.seq_target; + m.query_block_size = auto.query_target; + assert_ne!(m.seq_block_size, 0); + assert_ne!(m.query_block_size, 0); + let back = PlanManifest::read(&manifest_text(&m)).unwrap(); + assert_eq!(back, m); + assert_eq!( + (back.seq_block_size, back.query_block_size), + (auto.seq_target, auto.query_target) + ); + } + #[test] fn candidate_bin_counts_dedup_and_clamp() { assert_eq!(candidate_bin_counts(24, 4, 7), vec![4, 7, 8]); @@ -2020,15 +2571,18 @@ mod tests { worst, worst_unit_bytes(&p, 12, 1, 16_711_680, m.wga_chunk_size, m.transitions).unwrap() ); - assert!(m - .check_fit(1, 12) - .unwrap_err() - .contains("refusing to replan")); + assert!( + m.check_fit(1, 12) + .unwrap_err() + .contains("refusing to replan") + ); let zero = text.replace("max_hits 16711680", "max_hits 0"); - assert!(PlanManifest::read(&zero) - .unwrap_err() - .contains("resolved cap")); + assert!( + PlanManifest::read(&zero) + .unwrap_err() + .contains("resolved cap") + ); } #[test] @@ -2080,10 +2634,12 @@ mod tests { manifest_fixture(plan(&empty, &q, 1)) .validate_records(&empty, &q) .unwrap(); - assert!(manifest_fixture(plan(&empty, &q, 1)) - .validate_records(&r, &q) - .unwrap_err() - .contains("reference")); + assert!( + manifest_fixture(plan(&empty, &q, 1)) + .validate_records(&r, &q) + .unwrap_err() + .contains("reference") + ); let mut alt = m0.clone(); alt.plan.units.swap(0, 1); @@ -2124,10 +2680,11 @@ mod tests { let mut miss = manifest_fixture(plan(&r1, &q1, 1000)); miss.plan.reference_bins[0].record_ids = vec![0]; miss.plan.reference_bins[0].total_bp = 10; - assert!(miss - .validate_records(&r1, &q1) - .unwrap_err() - .contains("missing")); + assert!( + miss.validate_records(&r1, &q1) + .unwrap_err() + .contains("missing") + ); let big = vec![RecordMeta { id: 0, @@ -2229,10 +2786,11 @@ mod tests { worst, worst_unit_bytes(&p, 12, 1, m.max_hits, m.wga_chunk_size, m.transitions).unwrap() ); - assert!(m - .check_fit(1, 12) - .unwrap_err() - .contains("refusing to replan")); + assert!( + m.check_fit(1, 12) + .unwrap_err() + .contains("refusing to replan") + ); let mut bad = m.clone(); bad.plan.units[0].reference_bin = 99; @@ -2473,10 +3031,11 @@ mod tests { let b = m.check_fit(u64::MAX, 12).unwrap(); assert_ne!(a, b, "frozen fit must use manifest wga_chunk_size"); assert_eq!(m.max_hits, cap_before); - assert!(m - .check_fit(1, 12) - .unwrap_err() - .contains("refusing to replan")); + assert!( + m.check_fit(1, 12) + .unwrap_err() + .contains("refusing to replan") + ); assert_eq!(m.max_hits, cap_before); } @@ -2671,9 +3230,8 @@ mod tests { #[test] fn max_hit_capacity_exact_boundary_and_brute_force_agreement() { - let cost = |p: &super::Plan, h: u32| { - super::worst_unit_bytes(p, 12, 1, h, TEST_C, true).unwrap() - }; + let cost = + |p: &super::Plan, h: u32| super::worst_unit_bytes(p, 12, 1, h, TEST_C, true).unwrap(); // Multi-query plan charges the swap overlap (3 Qmax vs 2 Qmax). let p = super::plan_with(&recs(&[100]), &recs(&[50, 60]), 10_000, 1, false); assert_eq!(p.query_bins.len(), 2); @@ -2735,3 +3293,246 @@ mod tests { ); } } + +/// Round 90: unit-level static partition (task.md Part 2). +#[cfg(test)] +mod unit_partition_tests { + use super::{Bin, Plan, Visit, WorkUnit, unit_partition}; + + fn toy_plan(ref_bps: &[u64], q: usize) -> Plan { + let reference_bins = ref_bps + .iter() + .enumerate() + .map(|(i, &bp)| Bin { + id: i as u32, + record_ids: Vec::new(), + total_bp: bp, + }) + .collect::>(); + let query_bins = (0..q) + .map(|i| Bin { + id: i as u32, + record_ids: Vec::new(), + total_bp: 1, + }) + .collect::>(); + let mut units = Vec::new(); + let mut ordinal = 0u32; + for r in &reference_bins { + for qq in &query_bins { + units.push(WorkUnit { + ordinal, + reference_bin: r.id, + query_bin: qq.id, + }); + ordinal += 1; + } + } + Plan { + reference_bins, + query_bins, + units, + } + } + + /// Canonical whole-genome shape: descending bp order 3,4,2,1,0,6,5. + fn canonical() -> Plan { + let mut bps = [0u64; 7]; + bps[3] = 700; + bps[4] = 600; + bps[2] = 500; + bps[1] = 400; + bps[0] = 300; + bps[6] = 200; + bps[5] = 100; + toy_plan(&bps, 6) + } + + fn v(bin: usize, lo: usize, hi: usize) -> Visit { + Visit { + bin, + queries: lo..hi, + } + } + + fn total_visits(part: &[Vec]) -> usize { + part.iter().map(Vec::len).sum() + } + + /// Exact cover, quotas, contiguity and in-bin order for any partition. + fn check_invariants(plan: &Plan, workers: usize, part: &[Vec]) { + let r = plan.reference_bins.len(); + let q = plan.query_bins.len(); + let n = plan.units.len(); + let w = workers.max(1); + assert_eq!(part.len(), w); + let (a, rem) = (n / w, n % w); + let mut seen = vec![vec![false; q]; r.max(1)]; + let mut large = 0; + for (x, visits) in part.iter().enumerate() { + let mut units = 0; + let mut bins: Vec = Vec::new(); + for t in visits { + assert!(!t.queries.is_empty(), "worker {x}: empty visit"); + assert!(t.bin < r, "worker {x}: bin out of range"); + assert!(t.queries.end <= q, "worker {x}: range past Q"); + bins.push(t.bin); + for qq in t.queries.clone() { + assert!(!seen[t.bin][qq], "worker {x}: unit covered twice"); + seen[t.bin][qq] = true; + units += 1; + } + } + let mut sorted = bins.clone(); + sorted.sort_unstable(); + assert_eq!(bins, sorted, "worker {x}: visits not in bin order"); + assert!(units == a || units == a + 1, "worker {x}: quota {units}"); + if rem > 0 && units == a + 1 { + large += 1; + } + } + for b in 0..r { + assert!(seen[b].iter().all(|&s| s), "bin {b}: unit missing"); + } + if rem > 0 { + assert_eq!(large, rem, "exactly N mod W large quotas"); + } + } + + #[test] + fn canonical_w2() { + let p = canonical(); + let got = unit_partition(&p, 2); + assert_eq!( + got, + vec![ + vec![v(0, 0, 6), v(1, 0, 6), v(3, 0, 6), v(5, 0, 3)], + vec![v(2, 0, 6), v(4, 0, 6), v(5, 3, 6), v(6, 0, 6)], + ] + ); + assert_eq!(total_visits(&got), 8); + assert_eq!(total_visits(&got) - 7, 1, "1 extra replica"); + check_invariants(&p, 2, &got); + } + + #[test] + fn canonical_w4() { + let p = canonical(); + let got = unit_partition(&p, 4); + assert_eq!( + got, + vec![ + vec![v(0, 0, 5), v(3, 0, 6)], + vec![v(0, 5, 6), v(4, 0, 6), v(6, 0, 3)], + vec![v(2, 0, 6), v(5, 0, 1), v(6, 3, 6)], + vec![v(1, 0, 6), v(5, 1, 6)], + ] + ); + assert_eq!(total_visits(&got), 10); + assert_eq!(total_visits(&got) - 7, 3, "3 extra replicas"); + check_invariants(&p, 4, &got); + } + + #[test] + fn w1_is_one_visit_per_bin() { + let p = canonical(); + let got = unit_partition(&p, 1); + let want: Vec = (0..7).map(|b| v(b, 0, 6)).collect(); + assert_eq!(got, vec![want]); + check_invariants(&p, 1, &got); + } + + #[test] + fn q1_needs_no_split() { + let p = toy_plan(&[50, 40, 30, 20], 1); + for w in 1..=5 { + let got = unit_partition(&p, w); + assert_eq!(total_visits(&got), 4, "W={w}"); + for visits in &got { + for t in visits { + assert_eq!(t.queries, 0..1); + } + } + check_invariants(&p, w, &got); + } + } + + #[test] + fn fewer_units_than_workers_leaves_empty_quotas() { + let p = toy_plan(&[30, 20], 1); + let got = unit_partition(&p, 5); + let counts: Vec = got + .iter() + .map(|visits| visits.iter().map(|t| t.queries.len()).sum()) + .collect(); + assert_eq!(counts, vec![1, 1, 0, 0, 0]); + assert_eq!(total_visits(&got), 2); + check_invariants(&p, 5, &got); + } + + #[test] + fn deterministic_and_quota_shaped_elsewhere() { + for (bps, q, w) in [ + (vec![9, 8, 7, 6, 5], 3, 2), + (vec![100, 1, 1, 1], 4, 3), + (vec![30, 20, 10], 2, 4), + (vec![5], 5, 3), + ] { + let p = toy_plan(&bps, q); + let (a, b) = (unit_partition(&p, w), unit_partition(&p, w)); + assert_eq!(a, b, "deterministic for {bps:?} Q={q} W={w}"); + check_invariants(&p, w, &a); + } + } + + #[test] + fn minimal_visits_against_brute_force_oracle() { + // Every R,Q,W <= 3: enumerate all quota-respecting assignments of the + // N units to W workers and take the minimum (bin,worker) touch count. + // Visit count is contiguity-blind, so the oracle needs no contiguity + // filter; the construction must still attain the minimum. + for r in 1..=3 { + for q in 1..=3 { + for w in 1..=3 { + let bps: Vec = (0..r).map(|i| (r - i) as u64 * 10 + 1).collect(); + let p = toy_plan(&bps, q); + let n = r * q; + let (a, rem) = (n / w, n % w); + let mut best = usize::MAX; + let mut assign = vec![0usize; n]; + loop { + let mut counts = vec![0usize; w]; + for &x in &assign { + counts[x] += 1; + } + if counts.iter().all(|&c| c == a || (rem > 0 && c == a + 1)) { + let mut pair = vec![vec![false; w]; r]; + for (u, &x) in assign.iter().enumerate() { + pair[u / q][x] = true; + } + let mut touches = 0; + for b in 0..r { + touches += pair[b].iter().filter(|&&t| t).count(); + } + best = best.min(touches); + } + let mut k = 0; + while k < n { + assign[k] += 1; + if assign[k] < w { + break; + } + assign[k] = 0; + k += 1; + } + if k == n { + break; + } + } + let got = total_visits(&unit_partition(&p, w)); + assert_eq!(got, best, "R={r} Q={q} W={w}"); + } + } + } + } +} diff --git a/src/run.rs b/src/run.rs index 3800fa0..678f969 100644 --- a/src/run.rs +++ b/src/run.rs @@ -19,7 +19,7 @@ use crate::Fallible; use crate::cli::RunArgs; -use crate::gpu::{Engine, EngineConfig, HitStats, Lifecycle}; +use crate::gpu::{DeviceProfile, Engine, EngineConfig, HitStats, Lifecycle}; use crate::hsp::{self, SegmentPair}; use crate::partition::{Partitioner, Plan}; use crate::plan::{self, PackedBin, RecordMeta}; @@ -94,10 +94,23 @@ pub(crate) fn prepare(args: &RunArgs, phases: &mut Phases) -> Fallible let t = Instant::now(); // PLAN.md §1: reference and query input are timed separately and kept out // of `core`, so a format change can never be confused with a core change. - let query = Genome::load(&args.query, &args.query_prefix, args.seq_block_size)?; + // `-B 0` (automatic layout) is a planner decision this single-block path + // never makes: resolve to the default target so the guard below keeps + // today's behavior. `Genome::load` packs the whole input as one block and + // uses the target only to reject multi-block input, so no per-bin target + // applies here (the GPU `run` path below does not call `Genome::load`). + if args.seq_block_size == 0 && args.kegalign_bins { + return Err(plan::AUTO_KEGALIGN_ERROR.into()); + } + let seq_block_size = if args.seq_block_size == 0 { + plan::DEFAULT_BLOCK_TARGET as u32 + } else { + args.seq_block_size + }; + let query = Genome::load(&args.query, &args.query_prefix, seq_block_size)?; phases.add("input.query", t.elapsed()); let t = Instant::now(); - let reference = Genome::load(&args.reference, &args.target_prefix, args.seq_block_size)?; + let reference = Genome::load(&args.reference, &args.target_prefix, seq_block_size)?; phases.add("input.reference", t.elapsed()); if reference.block_len <= shape.size || query.block_len <= shape.size { return Err("reference and query blocks must be longer than the seed".into()); @@ -817,6 +830,29 @@ struct UnitOutput { pass: Pass, } +/// One executed work unit for the `--time` unit ledger (round 90). GPU busy +/// is the `engine.phases.gpu_ms()` delta across the unit's own +/// `seed_and_filter_all` — no new device syncs, events resolve at the +/// existing pipeline boundaries. Shared reference setup is not charged here. +pub(crate) struct UnitLedgerRow { + ordinal: u32, + worker: usize, + device: usize, + reference_bin: u32, + query_bin: u32, + reference_bp: u64, + query_bp: u64, + gpu_ms: f64, + host_start_ms: f64, + host_end_ms: f64, + pack_ms: f64, + swap_ms: f64, + seeds: u64, + hits: u64, + raw_hsps: u64, + hsps: u64, +} + /// What one worker reports at join. Everything the serial executor used to /// accumulate inline, now per worker and summed by the caller. #[derive(Default)] @@ -824,6 +860,10 @@ struct WorkerReport { stats: Stats, phases: Phases, lifecycle: Lifecycle, + ledger: Vec, + /// Wall ms since run start when this worker returned (after its last send and + /// engine teardown) — the per-worker critical path, unlike the last unit's end. + finished_ms: f64, launches: u64, stage_syncs: u64, pipeline_syncs: u64, @@ -835,6 +875,12 @@ struct WorkerReport { /// any batch-level pipelining. gap_ms: f32, gap_n: u64, + /// Per-pair gap totals merged across this worker's engines (round 72), + /// plus the discarded >1,000 ms intervals (unit-transition work) as a + /// separate count and sum so they stay visible instead of silent. + gap_pairs: Vec<((&'static str, &'static str), f32, u64)>, + gap_long_n: u64, + gap_long_ms: f32, prefetched_ms: Duration, hit_stats: HitStats, audit: Option, @@ -927,6 +973,112 @@ fn check_manifest_params( } } +/// Layout worker count for `-B 0`: requested GPUs clamped to visible devices, +/// so one device always resolves to the default layout. `forced` (the env hook +/// below) overrides the clamp for testing the W>=2 path on a one-GPU box. +pub(crate) fn layout_workers(gpus: usize, devices: usize, forced: Option) -> usize { + if let Some(w) = forced.filter(|&w| w >= 1) { + return w; + } + gpus.max(1).min(devices.max(1)) +} + +/// Debug/test hook: forces the `-B 0` layout worker count without touching +/// execution (worker threads, device ids, budgets are all derived from +/// `--gpus` as before). Deliberately ungated (no `cfg`): the release binary +/// must exercise the W>=2 path on a one-GPU box, and the layout line always +/// prints when it fires, so a stray setting is visible. Invalid values are +/// ignored. +fn forced_layout_workers() -> Option { + std::env::var("HSPZ_LAYOUT_FORCE_WORKERS") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|&w| w >= 1) +} + +/// Round 90b `HSPZ_UNIT_PARTITION` policy: unset is auto, `0` forces whole-bin, +/// `1` forces the unit partition, anything else is a startup error. +/// +/// Auto partitions iff `workers >= 2` and every device the run will use +/// reports the same [`DeviceProfile`] (SM count, clock, L2). The guard exists +/// because a heterogeneous pair pays -10.8% or +12.7% depending on which +/// device got which units, so the count-balanced partition must not silently +/// turn on there. Pure over its inputs so the decision is unit-testable with +/// no env access and no device; `profiles` holds one entry per used device +/// (worker `w` runs on `w % devices`), in ordinal order. +pub(crate) fn partition_policy( + env: Option<&str>, + workers: usize, + profiles: &[DeviceProfile], +) -> Result<(bool, String), String> { + let forced = match env { + Some("1") => Some((true, "forced by HSPZ_UNIT_PARTITION=1")), + Some("0") => Some((false, "forced by HSPZ_UNIT_PARTITION=0")), + Some(v) => { + return Err(format!( + "HSPZ_UNIT_PARTITION must be unset, 0 or 1, got {v:?}" + )); + } + None => None, + }; + if let Some((on, why)) = forced { + return Ok((on, why.to_string())); + } + if workers <= 1 { + return Ok((false, "auto: W=1".to_string())); + } + let first = match profiles.first() { + Some(p) => p, + None => return Ok((false, "auto: no device profiles".to_string())), + }; + // Workers sharing one device gain nothing from balancing and pay the replicas. + if workers > profiles.len() { + return Ok(( + false, + format!( + "auto: {workers} workers time-slice {} device{}", + profiles.len(), + if profiles.len() == 1 { "" } else { "s" } + ), + )); + } + // Same class = same SM count and L2 (the architecture/model); clocks within 10%, + // because otherwise-identical cards from different vendors report different + // nominal boost clocks (RTX 4090: 2520-2610 MHz) and the partition is worth + // -12% on such a pair. A same-model card that is slower under load is not + // detectable here (HSPZ_UNIT_PARTITION=0 restores whole-bin ownership). + let same_class = |p: &DeviceProfile| { + p.sms == first.sms + && p.l2_bytes == first.l2_bytes + && (p.clock_khz - first.clock_khz).abs() * 10 <= first.clock_khz.abs() + }; + if let Some((i, p)) = profiles.iter().enumerate().find(|(_, p)| !same_class(p)) { + return Ok(( + false, + format!( + "auto: device {i} differs from device 0 ({} vs {} SMs, {} vs {} MHz, {} vs {} MiB L2)", + p.sms, + first.sms, + p.clock_khz / 1000, + first.clock_khz / 1000, + p.l2_bytes >> 20, + first.l2_bytes >> 20, + ), + )); + } + Ok(( + true, + format!( + "auto: {} matching device{} ({} SMs, {} MHz, {} MiB L2; static attributes only, clocks within 10%)", + profiles.len(), + if profiles.len() == 1 { "" } else { "s" }, + first.sms, + first.clock_khz / 1000, + first.l2_bytes >> 20, + ), + )) +} + /// Whether to generate the query seed stream on the device (round 69). /// /// The trade reverses with GPU count. On one GPU the device seeder adds ~165 s of @@ -942,17 +1094,19 @@ pub fn device_seeds_for(workers: usize) -> bool { } } -/// Runs one worker's reference bins on `device`, streaming finished units to the -/// emitter (§Phase 5: build/upload each reference once, reuse it across its -/// queries; no GPU is shared for performance). +/// Runs one worker's visits on `device`, streaming finished units to the +/// emitter (§Phase 5: build/upload each visited reference once, reuse it +/// across its slice; no GPU is shared for performance). /// -/// This is the serial executor, parameterised by which bins it owns: with one -/// worker it is exactly the old path, which is what makes `serial == multi-GPU` -/// (§20) a property of the assignment rather than of two code paths. +/// This is the serial executor, parameterised by which visits it owns: with one +/// worker (or the whole-bin policy) every visit is a whole bin, which is what +/// makes `serial == multi-GPU` (§20) a property of the assignment rather than +/// of two code paths. One engine per visit, reused across the slice. #[allow(clippy::too_many_arguments)] fn run_bins( device: usize, - bins: &[u32], + worker: usize, + visits: &[plan::Visit], plan: &plan::Plan, ref_records: &[(String, Vec)], qry_records: &[(String, Vec)], @@ -967,9 +1121,10 @@ fn run_bins( device_seeds: bool, contract: crate::gpu::ExecutionContract, tx: &std::sync::mpsc::SyncSender, + run_started: Instant, ) -> Fallible { let mut rep = WorkerReport::default(); - if bins.is_empty() { + if visits.is_empty() { return Ok(rep); } let t = Instant::now(); @@ -997,8 +1152,8 @@ fn run_bins( }; let mut pending: Option<(PackedBin, SeedTable)> = None; - for (bin_index, bin_id) in bins.iter().enumerate() { - let rbin = &plan.reference_bins[*bin_id as usize]; + for (visit_index, visit) in visits.iter().enumerate() { + let rbin = &plan.reference_bins[visit.bin]; let t = Instant::now(); let (mut packed_ref, table) = match pending.take() { Some(built) => built, @@ -1040,10 +1195,10 @@ fn run_bins( drop(packed_ref); drop(table); - // The next bin *this worker owns* rides along with this bin's GPU work. - let next_bin = bins - .get(bin_index + 1) - .map(|id| &plan.reference_bins[*id as usize]) + // The next visit *this worker owns* rides along with this visit's GPU work. + let next_bin = visits + .get(visit_index + 1) + .map(|v| &plan.reference_bins[v.bin]) .filter(|_| prefetch); std::thread::scope(|scope| -> Fallible<()> { let prefetch = next_bin.map(|nb| { @@ -1053,8 +1208,11 @@ fn run_bins( }) }); - for unit in plan.units.iter().filter(|u| u.reference_bin == rbin.id) { + for unit in plan.units.iter().filter(|u| { + u.reference_bin == rbin.id && visit.queries.contains(&(u.query_bin as usize)) + }) { let qbin = &plan.query_bins[unit.query_bin as usize]; + let host_start_ms = run_started.elapsed().as_secs_f64() * 1000.0; let t = Instant::now(); let packed_q = PackedBin::build( qbin.record_ids.iter().map(|&id| { @@ -1064,7 +1222,8 @@ fn run_bins( &args.query_prefix, true, ); - rep.phases.add("query pack", t.elapsed()); + let pack_dur = t.elapsed(); + rep.phases.add("query pack", pack_dur); // Per-bin intervals + q_block_len (AM-B2). `intervals` is empty for a // block <= seed, and `q_block_len` is then never read; saturating // avoids the underflow the single-block path guards with an error. @@ -1074,13 +1233,15 @@ fn run_bins( let t = Instant::now(); engine.swap_query(&packed_q.enc, &packed_q.enc_rc)?; - rep.phases.add("swap_query", t.elapsed()); + let swap_dur = t.elapsed(); + rep.phases.add("swap_query", swap_dur); let qpass = QueryPass { fwd: &packed_q.buf[..packed_q.block_len], rc: &packed_q.rc, intervals: &intervals, q_block_len, }; + let gpu_before = engine.phases.gpu_ms(); let pass = seed_and_filter_all(&mut engine, &qpass, shape, transitions, args, threads) .map_err(|e| { @@ -1089,12 +1250,34 @@ fn run_bins( unit.ordinal, rbin.id, qbin.id ) })?; + let busy_ms = engine.phases.gpu_ms() - gpu_before; + let host_end_ms = run_started.elapsed().as_secs_f64() * 1000.0; rep.stats.seeds += pass.stats.seeds; rep.stats.seed_hits += pass.stats.seed_hits; rep.stats.raw_hsps += pass.stats.raw_hsps; rep.stats.hsps += pass.stats.hsps; rep.lifecycle.work_units_executed += 1; + if args.time { + rep.ledger.push(UnitLedgerRow { + ordinal: unit.ordinal, + worker, + device, + reference_bin: rbin.id, + query_bin: qbin.id, + reference_bp: rbin.total_bp, + query_bp: qbin.total_bp, + gpu_ms: busy_ms, + host_start_ms, + host_end_ms, + pack_ms: pack_dur.as_secs_f64() * 1000.0, + swap_ms: swap_dur.as_secs_f64() * 1000.0, + seeds: pass.stats.seeds, + hits: pass.stats.seed_hits, + raw_hsps: pass.stats.raw_hsps, + hsps: pass.stats.hsps, + }); + } // A dead emitter means the run is already failing; propagate rather // than block forever on a channel nobody drains. @@ -1128,7 +1311,22 @@ fn run_bins( let (g, n, _, _) = engine.stage_gaps(); rep.gap_ms += g; rep.gap_n += n; + for ((a, b), ms, c) in engine.gap_pairs() { + match rep.gap_pairs.iter_mut().find(|e| e.0 == (a, b)) { + Some(e) => { + e.1 += ms; + e.2 += c; + } + None => rep.gap_pairs.push(((a, b), ms, c)), + } + } + let (ln, lms) = engine.discarded_gaps(); + rep.gap_long_n += ln; + rep.gap_long_ms += lms; } + // Engine-end autotune ledger: still in Auto means the chunks ran + // out before a decision, so say so under --time. + engine.finish_bucket_autotune(); rep.launches += engine.launches; rep.stage_syncs += engine.stage_syncs(); rep.pipeline_syncs += engine.pipeline_syncs(); @@ -1168,6 +1366,7 @@ fn run_bins( ); } } + rep.finished_ms = run_started.elapsed().as_secs_f64() * 1000.0; Ok(rep) } @@ -1175,6 +1374,12 @@ fn run_bins( // run pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallible { + // Round 90b gate, read once here and validated before the `--cpu-only` + // branch returns (a bad value is a startup error there too). The auto arm + // is resolved with the other scheduling decisions below, where W and the + // per-device profiles are known; until then the env value is just carried. + let unit_env = std::env::var("HSPZ_UNIT_PARTITION").ok(); + partition_policy(unit_env.as_deref(), 1, &[])?; let mut phases = Phases::new(); phases.add_ms("process startup", pre_main_ms); @@ -1232,27 +1437,101 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl // ponytail: min(gpus, n_records) overcharges when bins < records; upgrade on measured false rejections let workers_upper = args.gpus.max(1).min(ref_meta.len().max(1)); let budget = plan::worker_device_budget(free, workers_upper, devices); - let q_target = args.query_block_size.unwrap_or(args.seq_block_size) as u64; - let (plan, _worst, mut contract) = if let Some(m) = &loaded_manifest { + // `-B 0` is the automatic layout, resolved here before planning. It needs + // LPT bins, so `--kegalign-bins` (sequential fill) is rejected in every + // mode, including replay, where `-B` itself is otherwise ignored. + if args.seq_block_size == 0 && args.kegalign_bins { + return Err(plan::AUTO_KEGALIGN_ERROR.into()); + } + let (plan, _worst, mut contract, res_seq, res_qry) = if let Some(m) = &loaded_manifest { let worst = m.check_fit(budget, shape.kmer_size)?; let contract = crate::gpu::ExecutionContract::from_resolved(m.max_hits, m.hsp_blocks); - (m.plan.clone(), worst, contract) + ( + m.plan.clone(), + worst, + contract, + m.seq_block_size, + m.query_block_size, + ) } else { let contract = crate::gpu::ExecutionContract::resolve(&ctx, args.max_hits, args.hsp_blocks); - let (plan, worst) = plan::plan_within_budget( - &ref_meta, - &qry_meta, - args.seq_block_size as u64, - q_target, - budget, - shape.kmer_size, - args.step, - contract.max_hits, - args.kegalign_bins, - args.wga_chunk_size, - transitions, - )?; - (plan, worst, contract) + if args.seq_block_size == 0 { + let w = layout_workers(args.gpus, devices, forced_layout_workers()); + if w <= 1 { + // Trivial resolution: exactly today's default call, so a + // one-worker `-B 0` run is byte-identical to `-B 500000000`. + let q = args + .query_block_size + .map(u64::from) + .unwrap_or(plan::DEFAULT_BLOCK_TARGET); + let (plan, worst) = plan::plan_within_budget( + &ref_meta, + &qry_meta, + plan::DEFAULT_BLOCK_TARGET, + q, + budget, + shape.kmer_size, + args.step, + contract.max_hits, + args.kegalign_bins, + args.wga_chunk_size, + transitions, + )?; + (plan, worst, contract, plan::DEFAULT_BLOCK_TARGET, q) + } else { + let auto = plan::auto_layout( + &ref_meta, + &qry_meta, + w, + args.gpus.max(1).min(ref_meta.len().max(1)), + budget, + timing::available_host_bytes().map(|b| b * 9 / 10), + args.query_block_size.map(u64::from), + &plan::AutoCtx { + kmer_size: shape.kmer_size, + step: args.step, + max_hits: contract.max_hits, + wga_chunk_size: args.wga_chunk_size, + transitions, + threads: resolve_threads(args.threads), + max_seeds: seed::max_seeds(args.wga_chunk_size, &shape, transitions), + }, + )?; + match auto.fell_back { + Some((r, q)) => eprintln!( + "layout: auto W={w} -> default --seq-block-size {} --query-block-size {} \ + (candidate R={r} Q={q} did not fit the device budget)", + plan::DEFAULT_BLOCK_TARGET, + plan::DEFAULT_BLOCK_TARGET + ), + None => eprintln!( + "{}", + plan::auto_layout_line(w, auto.seq_target, auto.query_target, &auto.plan) + ), + } + let (seq, qry) = (auto.seq_target, auto.query_target); + (auto.plan, 0, contract, seq, qry) + } + } else { + let q_target = args + .query_block_size + .map(u64::from) + .unwrap_or(args.seq_block_size as u64); + let (plan, worst) = plan::plan_within_budget( + &ref_meta, + &qry_meta, + args.seq_block_size as u64, + q_target, + budget, + shape.kmer_size, + args.step, + contract.max_hits, + args.kegalign_bins, + args.wga_chunk_size, + transitions, + )?; + (plan, worst, contract, args.seq_block_size as u64, q_target) + } }; // Physical capacity from the frozen/chosen plan at H under the same // per-worker budget; never replans. Clamped to the kernel-safe ceiling. @@ -1310,8 +1589,8 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl wga_chunk_size: args.wga_chunk_size, lastz_interval_size: args.lastz_interval_size, kegalign_bins: args.kegalign_bins, - seq_block_size: args.seq_block_size as u64, - query_block_size: q_target, + seq_block_size: res_seq, + query_block_size: res_qry, ref_hash: plan::records_hash(&ref_records), qry_hash: plan::records_hash(&qry_records), executable_hash: plan::executable_hash()?, @@ -1329,13 +1608,82 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl f.flush()?; } - // §18: reference bins to workers, deterministic LPT. Every query bin runs - // against every reference bin, so `cost(R) = reference_bp x total_query_bp` is + // §18: reference bins to workers, deterministic LPT (whole-bin), or the + // round-90 unit partition. `cost(R) = reference_bp x total_query_bp` is // monotone in the bin's own bp — LPT on `total_bp` is the same schedule with - // less arithmetic (plan::assign_bins). + // less arithmetic (plan::assign_bins). W = 1 always takes today's path. let devices = crate::gpu::device_count().max(1); let workers = args.gpus.max(1).min(plan.reference_bins.len().max(1)); - let assignment = plan::assign_bins(&plan.reference_bins, workers); + // Round 90b: query each device the run will use (worker `w` maps to + // `w % devices`, so the used ordinals are `0..min(workers, devices)`) once, + // before any worker spawns. Only auto needs the profiles; a forced value + // short-circuits in `partition_policy` without them. + let profiles: Vec = if unit_env.is_none() && workers > 1 { + (0..workers.min(devices)) + .map(|d| crate::gpu::device_profile(d as i32)) + .collect::>()? + } else { + Vec::new() + }; + let (unit_enabled, reason) = partition_policy(unit_env.as_deref(), workers, &profiles)?; + let use_units = unit_enabled && workers > 1; + let part: Vec> = if use_units { + plan::unit_partition(&plan, workers) + } else { + let q = plan.query_bins.len(); + plan::assign_bins(&plan.reference_bins, workers) + .into_iter() + .map(|bins| { + bins.into_iter() + .map(|id| plan::Visit { + bin: id as usize, + queries: 0..q, + }) + .collect() + }) + .collect() + }; + // Distinct reference bins per worker: the host peak counts one bin build at + // a time, never multiplied by the replica count. + let visit_bins: Vec> = part + .iter() + .map(|visits| { + let mut bins: Vec = visits.iter().map(|t| t.bin as u32).collect(); + bins.sort_unstable(); + bins.dedup(); + bins + }) + .collect(); + let total_visits: usize = part.iter().map(Vec::len).sum(); + eprintln!( + "schedule: policy={} W={workers} visits={total_visits} extra_replicas={} ({reason})", + if use_units { + "unit-partition" + } else { + "whole-bin" + }, + total_visits.saturating_sub(plan.reference_bins.len()) + ); + for (x, visits) in part.iter().enumerate() { + let units: usize = visits.iter().map(|t| t.queries.len()).sum(); + let desc = visits + .iter() + .map(|t| format!("R{}[{}..{})", t.bin, t.queries.start, t.queries.end)) + .collect::>() + .join(" "); + eprintln!( + "schedule: worker {x} device {}: {desc} units={units} visits={}", + x % devices, + visits.len() + ); + } + eprintln!( + "schedule: device mapping: {}", + (0..workers) + .map(|x| format!("worker{x}->device{}", x % devices)) + .collect::>() + .join(" ") + ); if workers > devices { eprintln!( "note: {workers} workers over {devices} device(s) — they time-slice one GPU. \ @@ -1367,7 +1715,7 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl let mut host_status = "unknown (no cgroup/meminfo reading)"; if let Some(available) = timing::available_host_bytes().map(|b| b * 9 / 10) { host_budget = Some(available); - let fits = plan::host_preflight(&est, &assignment, available)?; + let fits = plan::host_preflight(&est, &visit_bins, available)?; host_status = if fits { "fits with prefetch" } else { @@ -1383,7 +1731,7 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl } // §9: the estimate that the decision was made on, so a run can be checked // against its own measured RSS (§11) without re-deriving the model. - let host_peak_est = plan::host_peak(&est, &assignment, prefetch); + let host_peak_est = plan::host_peak(&est, &visit_bins, prefetch); let mut emitter = Emitter::new(args)?; let mut raw_all: Vec<(char, Vec)> = Vec::new(); @@ -1413,7 +1761,7 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl let (tx, rx) = std::sync::mpsc::sync_channel::(2 * workers); let reports = std::thread::scope(|scope| -> Fallible> { let mut handles = Vec::new(); - for (w, bins) in assignment.iter().enumerate() { + for (w, visits) in part.iter().enumerate() { let tx = tx.clone(); let (plan, ref_records, qry_records, shape, sub_mat) = (&plan, &ref_records, &qry_records, &shape, &sub_mat); @@ -1422,7 +1770,8 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl handles.push(scope.spawn(move || { run_bins( w % devices, - bins, + w, + visits, plan, ref_records, qry_records, @@ -1435,6 +1784,7 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl device_seeds, contract, &tx, + started, ) .map_err(|e| e.to_string()) })); @@ -1445,7 +1795,24 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl std::collections::BTreeMap::new(); let mut next = 0u32; for unit in rx { - buffered.insert(unit.ordinal, unit); + let ord = unit.ordinal; + // The ordinal alone is not identity: the unit must be the plan's unit. + let pu = plan.units.get(ord as usize).ok_or_else(|| { + format!( + "work unit ordinal {ord} is outside the plan ({} units)", + plan.units.len() + ) + })?; + if (pu.reference_bin, pu.query_bin) != (unit.reference_bin, unit.query_bin) { + return Err(format!( + "work unit ordinal {ord} arrived as R{} Q{} but the plan has R{} Q{}", + unit.reference_bin, unit.query_bin, pu.reference_bin, pu.query_bin + ) + .into()); + } + if ord < next || buffered.insert(ord, unit).is_some() { + return Err(format!("duplicate work unit ordinal {ord} from a worker").into()); + } while let Some(u) = buffered.remove(&next) { emitter.emit_unit( u.reference_bin, @@ -1461,6 +1828,12 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl next += 1; } } + let mut out = Vec::new(); + for h in handles { + out.push(h.join().expect("gpu worker panicked")?); + } + // Join first so a worker's own error (OOM, CUDA failure) is reported + // instead of the completeness check it also trips (codex review, round 90). if !buffered.is_empty() { return Err(format!( "emitter has {} unit(s) it can never reach: expected ordinal {next}, \ @@ -1470,9 +1843,12 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl ) .into()); } - let mut out = Vec::new(); - for h in handles { - out.push(h.join().expect("gpu worker panicked")?); + if next != plan.units.len() as u32 { + return Err(format!( + "emitter reached ordinal {next} of {} work units — a unit ran twice or never", + plan.units.len() + ) + .into()); } Ok(out) })?; @@ -1505,6 +1881,111 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl } ); } + // Residual attribution: per-worker notes that keep every existing line + // above unchanged (cycle-4/5 report scripts still parse). Printed under + // --time for any worker count, so a single-worker run still shows them. + if args.time { + let pairs: Vec = reports + .iter() + .map(|r| { + let mut v = r.gap_pairs.clone(); + v.sort_by(|a, b| b.1.total_cmp(&a.1)); + let top: Vec = v + .iter() + .take(5) + .map(|((a, b), ms, n)| format!("({a}->{b}, {ms:.1}, {n})")) + .collect(); + format!("[{}]", top.join(", ")) + }) + .collect(); + eprintln!("note: per-worker gap pairs: {pairs:?}"); + let long_n: Vec = reports.iter().map(|r| r.gap_long_n).collect(); + let long_ms: Vec = reports.iter().map(|r| r.gap_long_ms).collect(); + eprintln!("note: per-worker long transitions: n={long_n:?} sum={long_ms:?} ms"); + let qp: Vec = reports.iter().map(|r| r.phases.ms("query pack")).collect(); + let sq: Vec = reports.iter().map(|r| r.phases.ms("swap_query")).collect(); + let cp: Vec = reports + .iter() + .map(|r| r.phases.ms("chunk prep (lower_bound)")) + .collect(); + let st: Vec = reports + .iter() + .map(|r| r.seed_table_ms.as_secs_f64() * 1000.0) + .collect(); + // Standalone (non-overlapped) reference prep lives in the worker report, + // not in the worker Phases (that name is added globally after this note). + let rp: Vec = reports + .iter() + .map(|r| r.prefetched_ms.as_secs_f64() * 1000.0) + .collect(); + eprintln!( + "note: per-worker host phases ms: [query pack={qp:?}, swap_query={sq:?}, \ + chunk prep (lower_bound)={cp:?}, seed table build={st:?}, \ + reference bin prep (standalone)={rp:?}]" + ); + // Round 90 unit ledger: one tab-separated line per executed unit in + // ordinal order, then per-worker totals. The summed unit busy must agree + // with the per-worker gpu-busy note within 1% (warning only). + let mut rows: Vec<&UnitLedgerRow> = reports.iter().flat_map(|r| r.ledger.iter()).collect(); + rows.sort_by_key(|r| r.ordinal); + eprintln!( + "unit ledger:\tordinal\tworker\tdevice\tref_bin\tquery_bin\tref_bp\tquery_bp\t\ + gpu_ms\thost_start_ms\thost_end_ms\tpack_ms\tswap_ms\tseeds\thits\traw_hsps\thsps" + ); + for r in &rows { + eprintln!( + "unit ledger:\t{}\t{}\t{}\t{}\t{}\t{}\t{}\t{:.3}\t{:.2}\t{:.2}\t{:.2}\t{:.2}\t{}\t{}\t{}\t{}", + r.ordinal, + r.worker, + r.device, + r.reference_bin, + r.query_bin, + r.reference_bp, + r.query_bp, + r.gpu_ms, + r.host_start_ms, + r.host_end_ms, + r.pack_ms, + r.swap_ms, + r.seeds, + r.hits, + r.raw_hsps, + r.hsps + ); + } + for (x, rep) in reports.iter().enumerate() { + let busy: f64 = rep.ledger.iter().map(|r| r.gpu_ms).sum(); + let host: f64 = rep + .ledger + .iter() + .map(|r| r.host_end_ms - r.host_start_ms) + .sum(); + let end: f64 = rep.ledger.iter().map(|r| r.host_end_ms).fold(0.0, f64::max); + let gpu = rep.phases.gpu_ms(); + eprintln!( + "unit ledger: worker {x} finished at {:.2} ms (last unit end {end:.2}), busy {busy:.2} ms, unit-host {host:.2} ms", + rep.finished_ms + ); + eprintln!( + "unit ledger: worker {x} self-check unit_busy_sum={busy:.2} ms gpu_busy={gpu:.2} ms" + ); + let tol = 0.01 * gpu.max(1e-9); + if (busy - gpu).abs() > tol { + eprintln!( + "unit ledger: worker {x} WARNING unit busy and gpu-busy differ by over 1%" + ); + } + debug_assert!((busy - gpu).abs() <= tol, "unit ledger self-check"); + } + } + + for (x, r) in reports.iter().enumerate() { + let vw = part[x].len() as u32; + let uw: u32 = part[x].iter().map(|t| t.queries.len() as u32).sum(); + r.lifecycle + .check(vw, uw) + .map_err(|e| format!("worker {x}: {e}"))?; + } let mut stats = Stats::default(); let mut lifecycle = Lifecycle::default(); @@ -1550,7 +2031,7 @@ pub(crate) fn run(args: &RunArgs, pre_main_ms: f64, started: Instant) -> Fallibl if prefetched_ms > Duration::ZERO { phases.add_overlapped("reference bin prep (standalone)", prefetched_ms); } - lifecycle.check(plan.reference_bins.len() as u32, plan.units.len() as u32)?; + lifecycle.check(total_visits as u32, plan.units.len() as u32)?; if let Some(path) = &args.dump_raw { let mut f = std::io::BufWriter::new(std::fs::File::create(path)?); @@ -2072,6 +2553,32 @@ mod tests { assert_eq!(bare, spelled_out); } + /// `-B 0` needs LPT bins, so `--kegalign-bins` (sequential fill) is + /// rejected before any input is read; the GPU `run` path repeats the + /// check before planning. + #[test] + fn prepare_rejects_kegalign_with_auto_layout() { + let mut phases = Phases::new(); + let mut auto = base_args(); + auto.reference = PathBuf::from("/nonexistent/ref.fa"); + auto.query = PathBuf::from("/nonexistent/qry.fa"); + auto.seq_block_size = 0; + auto.kegalign_bins = true; + let err = prepare(&auto, &mut phases).err().unwrap(); + assert!(err.to_string().contains("automatic layout"), "{err}"); + } + + #[test] + fn layout_workers_clamps_to_devices_with_env_override() { + assert_eq!(layout_workers(0, 0, None), 1); + assert_eq!(layout_workers(1, 1, None), 1); + assert_eq!(layout_workers(2, 1, None), 1); + assert_eq!(layout_workers(2, 2, None), 2); + assert_eq!(layout_workers(8, 2, None), 2); + assert_eq!(layout_workers(1, 4, Some(2)), 2); + assert_eq!(layout_workers(2, 2, Some(0)), 2, "zero override is ignored"); + } + /// `prepare` is the GPU-free path `benchmark` and `--cpu-only` share; neither /// implements frozen-plan replay or manifest dumps, so both flags must fail /// fast here rather than being silently dropped. Both paths point at inputs @@ -2442,3 +2949,90 @@ mod tests { assert!(line.contains("eligible hits 28 of 904 (3.0973%)")); } } + +/// Round 90b: the `HSPZ_UNIT_PARTITION` auto arm and its forced overrides are +/// pure, so every branch (W=1, identical, one attribute differing, forced +/// either way, invalid) is covered without CUDA or env access. +#[cfg(test)] +mod partition_policy_tests { + use super::partition_policy; + use crate::gpu::DeviceProfile; + + fn p(sms: i32, mhz: i32, mib: u64) -> DeviceProfile { + DeviceProfile { + sms, + clock_khz: mhz * 1000, + l2_bytes: mib << 20, + } + } + + #[test] + fn w1_stays_whole_bin() { + assert_eq!( + partition_policy(None, 1, &[p(128, 2520, 72)]), + Ok((false, "auto: W=1".to_string())) + ); + } + + #[test] + fn identical_devices_partition() { + assert_eq!( + partition_policy(None, 2, &[p(128, 2520, 72), p(128, 2520, 72)]), + Ok(( + true, + "auto: 2 matching devices (128 SMs, 2520 MHz, 72 MiB L2; static attributes only, clocks within 10%)".to_string() + )) + ); + // Vendor OC variants of one model report different nominal clocks; within 10% is the same class. + assert!(partition_policy(None, 2, &[p(128, 2520, 72), p(128, 2610, 72)]).unwrap().0); + } + + #[test] + fn time_sliced_workers_stay_whole_bin() { + let (on, why) = partition_policy(None, 2, &[p(128, 2520, 72)]).unwrap(); + assert!(!on); + assert_eq!(why, "auto: 2 workers time-slice 1 device"); + } + + #[test] + fn any_differing_attribute_forces_whole_bin() { + let base = p(40, 1590, 4); + for other in [p(48, 1590, 4), p(40, 1200, 4), p(40, 1590, 8)] { + let (on, why) = partition_policy(None, 2, &[base, other]).unwrap(); + assert!(!on); + assert!( + why.starts_with("auto: device 1 differs from device 0 ("), + "{why}" + ); + } + // The tuple is (differing device) vs (device 0), differing first. + let (_, why) = partition_policy(None, 2, &[base, p(48, 1500, 8)]).unwrap(); + assert_eq!( + why, + "auto: device 1 differs from device 0 \ + (48 vs 40 SMs, 1500 vs 1590 MHz, 8 vs 4 MiB L2)" + ); + } + + #[test] + fn forced_values_override_the_profiles() { + let same = [p(128, 2520, 72), p(128, 2520, 72)]; + let diff = [p(40, 1590, 4), p(48, 1500, 8)]; + assert_eq!( + partition_policy(Some("1"), 2, &diff), + Ok((true, "forced by HSPZ_UNIT_PARTITION=1".to_string())) + ); + assert_eq!( + partition_policy(Some("0"), 2, &same), + Ok((false, "forced by HSPZ_UNIT_PARTITION=0".to_string())) + ); + } + + #[test] + fn invalid_value_is_an_error() { + assert_eq!( + partition_policy(Some("2"), 2, &[]).unwrap_err(), + "HSPZ_UNIT_PARTITION must be unset, 0 or 1, got \"2\"" + ); + } +} diff --git a/src/timing.rs b/src/timing.rs index 40e6599..a7ac414 100644 --- a/src/timing.rs +++ b/src/timing.rs @@ -141,6 +141,15 @@ impl Phases { .map_or(0, |r| r.calls) } + /// Summed host ms for a named phase, or 0 when it never ran. The + /// per-worker attribution line reads host residuals through this. + pub fn ms(&self, name: &str) -> f64 { + self.rows + .iter() + .find(|r| r.name == name) + .map_or(0.0, |r| r.host_ms) + } + pub fn total_ms(&self) -> f64 { self.rows .iter() From c4b8079a36bfffd2544cae90563f042eef96de55 Mon Sep 17 00:00:00 2001 From: alejandrogzi Date: Thu, 10 Sep 2026 10:13:56 +0200 Subject: [PATCH 2/4] =?UTF-8?q?release:=20v0.0.5=20=E2=80=94=20unit-level?= =?UTF-8?q?=20static=20partition,=20per-unit=20ledger,=20receiver=20checks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - version 0.0.5 in Cargo.toml and the hspZ entry of Cargo.lock only; features unchanged (default = simd-prelude, find-hits-warp, ref-loc-buckets) and no dependency changes (experimental's [dependencies] unchanged since b44cc04). - changelog 0.0.5: unit-level static partition with the measured seed-and-filter numbers (-11.9% on 2x RTX 4090, 1,400 -> 1,233 s; -5.1% on 4x RTX 4090, 718 -> 681 s at MAX_HITS=16,711,680; identical output), the auto rule (class match, 10% clocks, one worker per device), the per-unit ledger, and the receiver checks; the entry also records the -B 0 automatic layout, the inconclusive-block bound and the --time attribution that land with round 88. --- Cargo.lock | 2 +- Cargo.toml | 2 +- assets/changelog/changelog.md | 42 +++++++++++++++++++++++++++++++++++ 3 files changed, 44 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1b60063..e8e2b20 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -385,7 +385,7 @@ checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" [[package]] name = "hspz" -version = "0.0.4" +version = "0.0.5" dependencies = [ "clap", "cuda-core", diff --git a/Cargo.toml b/Cargo.toml index 1cd3c23..54ace64 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,7 +1,7 @@ [package] name = "hspz" authors = ["Alejandro Gonzales-Irribarren "] -version = "0.0.4" +version = "0.0.5" edition = "2024" license = "GPL-3.0" description = "GPU-accelerated high-scoring ungapped alignment pair backend" diff --git a/assets/changelog/changelog.md b/assets/changelog/changelog.md index cdd1012..db4662f 100644 --- a/assets/changelog/changelog.md +++ b/assets/changelog/changelog.md @@ -36,6 +36,48 @@ All notable changes to `hspZ` are documented here, newest first. +## [0.0.5] — 2026-09-10 + +Unit-level static partitioning behind a device-class guard, with a per-unit ledger +and emitter-side receiver checks. Output bytes are unchanged. + +- **Unit-level static partition (default on for matching GPUs).** With two or more + GPUs the frozen plan's work units are now split across workers by count quotas + (whole bins first, then contiguous query slices of as few bins as possible, one + extra reference build per split bin) instead of whole-bin ownership, when every + device the run uses is the same class (equal SM count and L2, nominal clocks + within 10%) and each worker has its own device. Output bytes are unchanged. + Measured on canonical hg38 × mm39 at `MAX_HITS=16,711,680`: seed-and-filter + wall **−11.9%** on 2× RTX 4090 (1,400 → 1,233 s) and **−5.1%** on 4× RTX 4090 + (718 → 681 s), identical output. `HSPZ_UNIT_PARTITION=0` restores whole-bin + ownership, `1` forces the partition; W=1, time-sliced workers and mixed device + classes keep whole-bin ownership. The static-attribute guard cannot detect + same-model cards with different sustained clocks (on such a pair the balance + depends on device order). +- **Per-unit ledger.** `--time` prints one `unit ledger:` row per work unit + (worker, device, bins, bp, GPU ms, host start/end, pack/swap ms, seeds, hits, + raw/final HSPs) plus per-worker finish/busy lines, and every run prints + `schedule:` lines with the policy, its reason, and each worker's visits. +- **Receiver checks.** The emitter now rejects duplicate or out-of-plan work units + and verifies completeness after joining the workers, reporting a worker's own + error first. +- **Automatic layout (`-B 0`).** `-B 0` resolves the layout for the current + machine instead of the fixed 500 Mbp default. At `--gpus W ≥ 2` the reference + takes `R = min(records, ceil(R_default/W)·W)` LPT-balanced bins, and the query + collapses to one bin when the whole query fits both the per-worker device budget + and the host preflight (an explicit `--query-block-size` is honoured); `W=1` + resolves to exactly the 500 Mbp plan, and a candidate that fails the final fit + falls back to the default policy rather than erroring. Resolved sizes are stored + in the manifest/`--dump-plan`, and one `layout: auto W=…` line prints the flags + that reproduce them. `-B 0` with `--kegalign-bins` is an error. + `HSPZ_LAYOUT_FORCE_WORKERS` exercises the W≥2 path on one device. +- **Diagnostics.** The ref-buckets autotune drops and restarts an OFF block that + reaches twice the block span with fewer than four settled chunks (an + inconclusive ON block decides OFF, with a ledger line). `--time` attributes + gap pairs per worker, counts the >1 s transitions the gap accounting discards, + and totals the host phases (query pack, swap, chunk prep, seed table, + standalone reference prep). + ## [0.0.4] — 2026-09-09 Two exact speed-ups, both byte-identical to 0.0.3 on every device tested (NVIDIA L4, From a93c4a6db8731bcd53b0e1d71e6956c1b9fd247d Mon Sep 17 00:00:00 2001 From: alejandrogzi Date: Thu, 10 Sep 2026 10:14:00 +0200 Subject: [PATCH 3/4] docs: HSPZ_UNIT_PARTITION knob row, (0.0.5) section title - gpus.mdx: the runtime tuning knobs section is retitled (0.0.5) and gains the HSPZ_UNIT_PARTITION=auto|0|1 row (device-class guard, whole-bin at W=1, time-sliced workers and mixed classes; -11.9% on 2x RTX 4090 at MAX_HITS 16,711,680, -5.1% on 4x RTX 4090). "Both knobs" -> "Every knob". - docs/src/generated/cli-reference.txt regenerated from the 0.0.5 binary and byte-identical: no CLI text changed. - docs validate (9 journeys, 43 slides) and astro build (16 pages) pass. --- docs/src/pages/docs/gpus.mdx | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/docs/src/pages/docs/gpus.mdx b/docs/src/pages/docs/gpus.mdx index 21703dc..0b4d57c 100644 --- a/docs/src/pages/docs/gpus.mdx +++ b/docs/src/pages/docs/gpus.mdx @@ -27,15 +27,16 @@ HSPZ_DEVICE_SEEDS=0 hspZ run ... # force CPU HSPZ_DEVICE_SEEDS=1 hspZ run ... # force device ``` -## Runtime tuning knobs (0.0.4) +## Runtime tuning knobs (0.0.5) -Both knobs are exact: every mode produces the same HSP set and the same bytes. +Every knob is exact: every mode produces the same HSP set and the same bytes. | variable | default | what it does | | --- | --- | --- | | `HSPZ_REF_BUCKETS=auto\|1\|0` | `auto` | Reference-locality bucketing before the score gate. It wins where lower DRAM traffic lets a power-capped GPU raise its clock (L4: −6.5% whole-genome wall) and loses where the card already runs near its boost clock (RTX 4090: +3.7%). `auto` measures both paths per engine in blocks of ≥3 s, discards each block's first second, and keeps the faster one; `--time` prints the blocks and the decision. Cards whose L2 cannot hold the smallest window (T4) stay off. | | `HSPZ_REF_BUCKET_SHIFT=<16..31>` | derived from L2 | Pins the bucket window size (implies on). | | `HSPZ_CHUNK_WALK=full` | sparse | Restores the pre-0.0.4 chunk walk that copied the whole cumulative hit array per seed batch (rollback only). | +| `HSPZ_UNIT_PARTITION=auto\|0\|1` | `auto` | Splits the frozen plan's work units across workers by count quotas when every device the run uses is the same class and each worker has its own device (W=1, time-sliced workers and mixed classes keep whole-bin ownership); output is unchanged. −11.9% seed-and-filter wall on 2× RTX 4090 (1,400 → 1,233 s) and −5.1% on 4× RTX 4090 (718 → 681 s) at `MAX_HITS=16,711,680`. | ## Choosing `-B` From 7afedca76e1f5004b23f2016fa1b0960f5a3b204 Mon Sep 17 00:00:00 2001 From: alejandrogzi Date: Thu, 10 Sep 2026 10:15:42 +0200 Subject: [PATCH 4/4] changelog: -B 0 is an opt-in layout tool, not a default --- assets/changelog/changelog.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/assets/changelog/changelog.md b/assets/changelog/changelog.md index db4662f..d11be47 100644 --- a/assets/changelog/changelog.md +++ b/assets/changelog/changelog.md @@ -61,8 +61,10 @@ and emitter-side receiver checks. Output bytes are unchanged. - **Receiver checks.** The emitter now rejects duplicate or out-of-plan work units and verifies completeness after joining the workers, reporting a worker's own error first. -- **Automatic layout (`-B 0`).** `-B 0` resolves the layout for the current - machine instead of the fixed 500 Mbp default. At `--gpus W ≥ 2` the reference +- **Automatic layout (`-B 0`, opt-in measurement tool).** `-B 0` resolves a + count-balanced layout for the current machine instead of the fixed 500 Mbp + default; it changes the HSP set (a different frozen plan), so measure per machine + before adopting it (a 2× T4 screen was 17–22% slower than the default). At `--gpus W ≥ 2` the reference takes `R = min(records, ceil(R_default/W)·W)` LPT-balanced bins, and the query collapses to one bin when the whole query fits both the per-worker device budget and the host preflight (an explicit `--query-block-size` is honoured); `W=1`