diff --git a/chutoro-core/src/batch_metrics.rs b/chutoro-core/src/batch_metrics.rs new file mode 100644 index 00000000..42e20b32 --- /dev/null +++ b/chutoro-core/src/batch_metrics.rs @@ -0,0 +1,76 @@ +//! Bounded metrics emitted by the one-shot clustering execution path. +//! +//! This module owns the stable batch metric vocabulary. It accepts only +//! resource values and bounded labels, keeping source names and payload data +//! outside the metrics surface. + +use crate::Result; + +const RUNS_TOTAL: &str = "chutoro.batch.runs_total"; +#[cfg(feature = "cpu")] +const MAX_CONNECTIONS: &str = "chutoro.batch.max_connections"; +#[cfg(feature = "cpu")] +const EFFECTIVE_EF_CONSTRUCTION: &str = "chutoro.batch.effective_ef_construction"; +#[cfg(feature = "cpu")] +const ESTIMATED_BYTES: &str = "chutoro.batch.estimated_bytes"; +#[cfg(feature = "cpu")] +const MEMORY_LIMIT_BYTES: &str = "chutoro.batch.memory_limit_bytes"; + +/// Records the final outcome of a one-shot batch run. +pub(crate) fn record_outcome(backend: &'static str, result: &Result) { + let (outcome, error_code) = match result { + Ok(_) => ("success", "none"), + Err(error) => ("error", error.code().as_str()), + }; + + metrics::describe_counter!( + RUNS_TOTAL, + metrics::Unit::Count, + "Total one-shot batch runs by backend, outcome, and stable error code." + ); + metrics::counter!( + RUNS_TOTAL, + "backend" => backend, + "outcome" => outcome, + "error_code" => error_code + ) + .increment(1); +} + +/// Records CPU HNSW and memory observations for a one-shot batch run. +#[cfg(feature = "cpu")] +pub(crate) fn record_cpu_resources( + max_connections: usize, + effective_ef_construction: usize, + estimated_bytes: u64, + memory_limit_bytes: Option, +) { + metrics::describe_histogram!( + MAX_CONNECTIONS, + metrics::Unit::Count, + "Configured CPU HNSW maximum connections for one-shot batch runs." + ); + metrics::describe_histogram!( + EFFECTIVE_EF_CONSTRUCTION, + metrics::Unit::Count, + "Dataset-bounded CPU HNSW construction search width for one-shot batch runs." + ); + metrics::describe_histogram!( + ESTIMATED_BYTES, + metrics::Unit::Bytes, + "Estimated peak bytes for one-shot CPU batch runs." + ); + metrics::histogram!(MAX_CONNECTIONS, "backend" => "cpu").record(max_connections as f64); + metrics::histogram!(EFFECTIVE_EF_CONSTRUCTION, "backend" => "cpu") + .record(effective_ef_construction as f64); + metrics::histogram!(ESTIMATED_BYTES, "backend" => "cpu").record(estimated_bytes as f64); + + if let Some(limit) = memory_limit_bytes { + metrics::describe_histogram!( + MEMORY_LIMIT_BYTES, + metrics::Unit::Bytes, + "Configured memory-limit bytes for one-shot CPU batch runs." + ); + metrics::histogram!(MEMORY_LIMIT_BYTES, "backend" => "cpu").record(limit as f64); + } +} diff --git a/chutoro-core/src/builder.rs b/chutoro-core/src/builder.rs index 411b0949..100490f9 100644 --- a/chutoro-core/src/builder.rs +++ b/chutoro-core/src/builder.rs @@ -9,7 +9,7 @@ use std::sync::Arc; #[cfg(feature = "cpu")] use crate::{ClusteringSession, DataSource, HnswParams, SessionConfig, SessionRefreshPolicy}; -use crate::{Result, chutoro::Chutoro, error::ChutoroError}; +use crate::{Result, chutoro::Chutoro, error::ChutoroError, execution_config::ExecutionConfig}; #[cfg(feature = "cpu")] use tracing::debug; use tracing::warn; @@ -200,7 +200,7 @@ impl ChutoroBuilder { #[must_use] pub fn max_bytes(&self) -> Option { self.max_bytes } - /// Sets the HNSW parameters used when constructing clustering sessions. + /// Sets the HNSW parameters used by CPU execution and sessions. /// /// # Examples /// ``` @@ -217,7 +217,7 @@ impl ChutoroBuilder { self } - /// Returns the HNSW parameters used for session construction. + /// Returns the HNSW parameters used by CPU execution and sessions. #[cfg(feature = "cpu")] #[must_use] pub fn hnsw_params(&self) -> &HnswParams { @@ -266,8 +266,13 @@ impl ChutoroBuilder { (!cfg!(feature = "gpu")).then_some(GpuRejectionReason::BackendNotCompiled); self.validate_execution_strategy(gpu_rejection_reason)?; + #[cfg(feature = "cpu")] + let execution_config = ExecutionConfig::new(min_cluster_size, self.hnsw_params); + #[cfg(not(feature = "cpu"))] + let execution_config = ExecutionConfig::new(min_cluster_size); + Ok(Chutoro::new( - min_cluster_size, + execution_config, self.execution_strategy, self.max_bytes, )) @@ -306,11 +311,8 @@ impl ChutoroBuilder { ) -> Result> { let min_cluster_size = self.validate_min_cluster_size()?; self.validate_execution_strategy(Some(GpuRejectionReason::SessionsCpuOnly))?; - let config = SessionConfig::new( - min_cluster_size, - self.hnsw_params, - self.session_refresh_policy, - ); + let execution_config = ExecutionConfig::new(min_cluster_size, self.hnsw_params); + let config = SessionConfig::new(execution_config, self.session_refresh_policy); debug!( min_cluster_size = %config.min_cluster_size(), "build_session: constructing empty ClusteringSession" diff --git a/chutoro-core/src/chutoro.rs b/chutoro-core/src/chutoro.rs index 61c97a31..8298d9a9 100644 --- a/chutoro-core/src/chutoro.rs +++ b/chutoro-core/src/chutoro.rs @@ -7,8 +7,10 @@ use std::{num::NonZeroUsize, sync::Arc}; use crate::{ Result, builder::ExecutionStrategy, datasource::DataSource, error::ChutoroError, - result::ClusteringResult, + execution_config::ExecutionConfig, result::ClusteringResult, }; +#[cfg(feature = "cpu")] +use tracing::debug; use tracing::{instrument, warn}; const CPU_PATH_AVAILABLE: bool = cfg!(feature = "cpu"); @@ -53,19 +55,19 @@ enum BackendChoice { /// ``` #[derive(Debug, Clone)] pub struct Chutoro { - min_cluster_size: NonZeroUsize, + execution_config: ExecutionConfig, execution_strategy: ExecutionStrategy, max_bytes: Option, } impl Chutoro { pub(crate) fn new( - min_cluster_size: NonZeroUsize, + execution_config: ExecutionConfig, execution_strategy: ExecutionStrategy, max_bytes: Option, ) -> Self { Self { - min_cluster_size, + execution_config, execution_strategy, max_bytes, } @@ -85,7 +87,13 @@ impl Chutoro { /// ``` #[must_use] pub fn min_cluster_size(&self) -> NonZeroUsize { - self.min_cluster_size + self.execution_config.min_cluster_size() + } + + /// Returns the HNSW parameters configured for CPU execution. + #[cfg(feature = "cpu")] + pub(crate) fn hnsw_params(&self) -> &crate::HnswParams { + self.execution_config.hnsw_params() } /// Returns the execution strategy that will be used when running. @@ -164,13 +172,12 @@ impl Chutoro { #[instrument( name = "core.run", - err, skip(self, source), fields( - data_source = %source.name(), items = items, - min_cluster_size = %self.min_cluster_size, - strategy = ?self.execution_strategy + min_cluster_size = %self.min_cluster_size(), + strategy = ?self.execution_strategy, + backend = self.backend_label() ), )] fn run_with_len( @@ -178,32 +185,84 @@ impl Chutoro { source: &D, items: usize, ) -> Result { + let backend = self.backend_label(); if items == 0 { + let error = ChutoroError::EmptySource { + data_source: Arc::from(source.name()), + }; warn!( - data_source = source.name(), + backend, + error_code = error.code().as_str(), "data source is empty, returning error" ); - return Err(ChutoroError::EmptySource { - data_source: Arc::from(source.name()), - }); + return self.record_batch_result(backend, Err(error)); } - if items < self.min_cluster_size.get() { - return Err(ChutoroError::InsufficientItems { + if items < self.min_cluster_size().get() { + let error = ChutoroError::InsufficientItems { data_source: Arc::from(source.name()), items, - min_cluster_size: self.min_cluster_size, - }); + min_cluster_size: self.min_cluster_size(), + }; + warn!( + backend, + error_code = error.code().as_str(), + "data source has insufficient items for configured cluster size" + ); + return self.record_batch_result(backend, Err(error)); } if let Some(err) = self.backend_unavailable_error() { - return Err(err); + warn!( + backend, + error_code = err.code().as_str(), + "requested batch backend is unavailable" + ); + return self.record_batch_result(backend, Err(err)); } - self.check_memory_limit(source, items)?; + self.record_batch_resources(items); - match self.choose_backend() { + if let Err(error) = self.check_memory_limit(source, items) { + return self.record_batch_result(backend, Err(error)); + } + + let result = match self.choose_backend() { BackendChoice::Cpu => self.run_cpu(source, items), BackendChoice::Gpu => self.run_gpu(source, items), + }; + if let Err(error) = &result { + warn!( + backend, + error_code = error.code().as_str(), + "batch execution failed after precondition checks" + ); } + self.record_batch_result(backend, result) + } + + fn record_batch_result(&self, backend: &'static str, result: Result) -> Result { + #[cfg(feature = "metrics")] + crate::batch_metrics::record_outcome(backend, &result); + + #[cfg(not(feature = "metrics"))] + let _ = backend; + + result + } + + fn record_batch_resources(&self, items: usize) { + #[cfg(all(feature = "cpu", feature = "metrics"))] + { + let hnsw_params = self.hnsw_params(); + crate::batch_metrics::record_cpu_resources( + hnsw_params.max_connections(), + hnsw_params.effective_ef_construction(items), + crate::memory::estimate_peak_bytes_for_hnsw_params(items, hnsw_params), + self.max_bytes, + ); + } + + #[cfg(not(all(feature = "cpu", feature = "metrics")))] + let _ = items; } /// Returns an error if the estimated peak memory exceeds `max_bytes`. @@ -213,22 +272,42 @@ impl Chutoro { None => return Ok(()), }; - // Use the default HNSW max_connections for estimation. The pipeline - // always constructs params via `HnswParams::default()`, so this is - // consistent with actual usage. Validated by the - // `default_max_connections_matches_hnsw_params` test. - const DEFAULT_MAX_CONNECTIONS: usize = 16; - let estimated = crate::memory::estimate_peak_bytes(items, DEFAULT_MAX_CONNECTIONS); + #[cfg(feature = "cpu")] + let estimated = + crate::memory::estimate_peak_bytes_for_hnsw_params(items, self.hnsw_params()); + #[cfg(not(feature = "cpu"))] + let estimated = 0; + + #[cfg(feature = "cpu")] + debug!( + backend = "cpu", + max_connections = self.hnsw_params().max_connections(), + configured_ef_construction = self.hnsw_params().ef_construction(), + effective_ef_construction = self.hnsw_params().effective_ef_construction(items), + estimated_bytes = estimated, + max_bytes = limit, + "checked CPU memory limit" + ); if estimated > limit { - return Err(ChutoroError::MemoryLimitExceeded { + let error = ChutoroError::MemoryLimitExceeded { data_source: Arc::from(source.name()), point_count: items, estimated_bytes: estimated, max_bytes: limit, estimated_display: Arc::from(crate::memory::format_bytes(estimated)), limit_display: Arc::from(crate::memory::format_bytes(limit)), - }); + }; + #[cfg(feature = "cpu")] + warn!( + backend = "cpu", + max_connections = self.hnsw_params().max_connections(), + estimated_bytes = estimated, + max_bytes = limit, + error_code = error.code().as_str(), + "CPU memory estimate exceeds configured limit" + ); + return Err(error); } Ok(()) } @@ -247,17 +326,32 @@ impl Chutoro { } } + fn backend_label(&self) -> &'static str { + if self.is_backend_unavailable() { + return "unavailable"; + } + + match self.choose_backend() { + BackendChoice::Cpu => "cpu", + BackendChoice::Gpu => "gpu", + } + } + /// Execute the CPU FISHDBC pipeline; available with the `cpu` feature. #[instrument( name = "core.run_cpu", - err, skip(self, source), - fields(items = items, min_cluster_size = %self.min_cluster_size), + fields(items = items, min_cluster_size = %self.min_cluster_size()), )] fn run_cpu(&self, source: &D, items: usize) -> Result { #[cfg(feature = "cpu")] { - crate::cpu_pipeline::run_cpu_pipeline_with_len(source, items, self.min_cluster_size) + crate::cpu_pipeline::run_cpu_pipeline_with_len( + source, + items, + self.min_cluster_size(), + self.hnsw_params(), + ) } #[cfg(not(feature = "cpu"))] { @@ -296,79 +390,9 @@ impl Chutoro { } #[cfg(test)] -mod tests { - //! Unit tests for the Chutoro builder facade. - - use super::*; - use crate::ChutoroBuilder; - - #[test] - fn gpu_preferred_requires_gpu_feature() { - let chutoro = Chutoro::new( - NonZeroUsize::new(1).expect("literal 1 is non-zero"), - ExecutionStrategy::GpuPreferred, - None, - ); - let err = chutoro.backend_unavailable_error(); - assert!(matches!( - err, - Some(ChutoroError::BackendUnavailable { - requested: ExecutionStrategy::GpuPreferred - }) - )); - } - - #[test] - fn backend_available_when_features_enabled() { - if cfg!(feature = "cpu") { - for strategy in [ExecutionStrategy::Auto, ExecutionStrategy::CpuOnly] { - let chutoro = Chutoro::new( - NonZeroUsize::new(1).expect("literal 1 is non-zero"), - strategy, - None, - ); - assert!(chutoro.backend_unavailable_error().is_none()); - } - } - - let chutoro = Chutoro::new( - NonZeroUsize::new(1).expect("literal 1 is non-zero"), - ExecutionStrategy::GpuPreferred, - None, - ); - assert!(matches!( - chutoro.backend_unavailable_error(), - Some(ChutoroError::BackendUnavailable { - requested: ExecutionStrategy::GpuPreferred - }) - )); - } - - #[test] - fn max_bytes_none_imposes_no_limit() { - let chutoro = ChutoroBuilder::new().build().expect("build must succeed"); - assert_eq!(chutoro.max_bytes(), None); - } - - #[test] - fn max_bytes_propagates_through_builder() { - let chutoro = ChutoroBuilder::new() - .with_max_bytes(1_000_000) - .build() - .expect("build must succeed"); - assert_eq!(chutoro.max_bytes(), Some(1_000_000)); - } +#[path = "chutoro_tests.rs"] +mod tests; - /// Guards against silent drift if `HnswParams::default().max_connections` - /// ever changes. The constant in `check_memory_limit` must stay in sync. - #[cfg(feature = "cpu")] - #[test] - fn default_max_connections_matches_hnsw_params() { - let params = crate::HnswParams::default(); - assert_eq!( - params.max_connections(), - 16, - "DEFAULT_MAX_CONNECTIONS in check_memory_limit must be updated to match" - ); - } -} +#[cfg(all(test, feature = "cpu"))] +#[path = "chutoro/properties.rs"] +mod properties; diff --git a/chutoro-core/src/chutoro/properties.rs b/chutoro-core/src/chutoro/properties.rs new file mode 100644 index 00000000..81a28477 --- /dev/null +++ b/chutoro-core/src/chutoro/properties.rs @@ -0,0 +1,100 @@ +//! Property tests for shared one-shot and session execution configuration. + +use std::sync::{Arc, atomic::AtomicUsize}; + +use proptest::prelude::*; +use tracing_subscriber::layer::SubscriberExt; + +use super::*; +use crate::{ + ChutoroBuilder, ExecutionStrategy, HnswParams, estimate_peak_bytes_for_hnsw_params, + test_utils::{CountingSource, suite_proptest_config}, +}; +use chutoro_test_support::tracing::RecordingLayer; + +fn shared_execution_config_strategy() -> impl Strategy { + (4_usize..=8, 1_usize..=4, 0_usize..=4).prop_flat_map( + |(point_count, max_connections, extra_construction_width)| { + ( + Just(point_count), + 1_usize..=point_count, + Just(max_connections), + Just(max_connections + extra_construction_width), + ) + }, + ) +} + +proptest! { + #![proptest_config(suite_proptest_config(4))] + + #[test] + fn shared_execution_config_reaches_batch_and_session_paths( + (point_count, min_cluster_size, max_connections, ef_construction) + in shared_execution_config_strategy(), + ) { + let params = HnswParams::new(max_connections, ef_construction) + .map_err(|error| TestCaseError::fail(error.to_string()))?; + let source = CountingSource::new( + (0..point_count).map(|point| point as f32).collect(), + Arc::new(AtomicUsize::new(0)), + ); + let estimate = estimate_peak_bytes_for_hnsw_params(point_count, ¶ms); + + let memory_limited = ChutoroBuilder::new() + .with_min_cluster_size(min_cluster_size) + .with_execution_strategy(ExecutionStrategy::CpuOnly) + .with_hnsw_params(params.clone()) + .with_max_bytes(estimate - 1) + .build() + .map_err(|error| TestCaseError::fail(error.to_string()))?; + let memory_error = memory_limited + .run(&source) + .expect_err("one byte below the generated estimate must reject the batch"); + let ChutoroError::MemoryLimitExceeded { estimated_bytes, .. } = memory_error else { + return Err(TestCaseError::fail("memory guard must return MemoryLimitExceeded")); + }; + prop_assert_eq!(estimated_bytes, estimate); + + let session = ChutoroBuilder::new() + .with_min_cluster_size(min_cluster_size) + .with_execution_strategy(ExecutionStrategy::CpuOnly) + .with_hnsw_params(params.clone()) + .build_session(Arc::new(source.clone())) + .map_err(|error| TestCaseError::fail(error.to_string()))?; + prop_assert_eq!(session.config().min_cluster_size().get(), min_cluster_size); + prop_assert_eq!(session.config().hnsw_params(), ¶ms); + + let layer = RecordingLayer::default(); + let subscriber = tracing_subscriber::registry().with(layer.clone()); + let runnable = ChutoroBuilder::new() + .with_min_cluster_size(min_cluster_size) + .with_execution_strategy(ExecutionStrategy::CpuOnly) + .with_hnsw_params(params.clone()) + .with_max_bytes(estimate) + .build() + .map_err(|error| TestCaseError::fail(error.to_string()))?; + let result = tracing::subscriber::with_default(subscriber, || runnable.run(&source)); + prop_assert!(result.is_ok()); + + let event = layer + .events() + .into_iter() + .find(|event| { + event + .fields + .get("message") + .is_some_and(|message| message == "building CPU HNSW index") + }) + .ok_or_else(|| TestCaseError::fail("CPU HNSW construction event must be recorded"))?; + prop_assert_eq!(event.fields.get("max_connections"), Some(&max_connections.to_string())); + prop_assert_eq!( + event.fields.get("configured_ef_construction"), + Some(&ef_construction.to_string()), + ); + prop_assert_eq!( + event.fields.get("effective_ef_construction"), + Some(¶ms.effective_ef_construction(point_count).to_string()), + ); + } +} diff --git a/chutoro-core/src/chutoro_tests.rs b/chutoro-core/src/chutoro_tests.rs new file mode 100644 index 00000000..73237707 --- /dev/null +++ b/chutoro-core/src/chutoro_tests.rs @@ -0,0 +1,86 @@ +//! Unit tests for the Chutoro runtime facade. + +use std::num::NonZeroUsize; + +use super::*; +use crate::ChutoroBuilder; + +fn execution_config(min_cluster_size: NonZeroUsize) -> ExecutionConfig { + #[cfg(feature = "cpu")] + { + ExecutionConfig::new(min_cluster_size, crate::HnswParams::default()) + } + #[cfg(not(feature = "cpu"))] + { + ExecutionConfig::new(min_cluster_size) + } +} + +#[test] +fn gpu_preferred_requires_gpu_feature() { + let chutoro = Chutoro::new( + execution_config(NonZeroUsize::new(1).expect("literal 1 is non-zero")), + ExecutionStrategy::GpuPreferred, + None, + ); + let err = chutoro.backend_unavailable_error(); + assert!(matches!( + err, + Some(ChutoroError::BackendUnavailable { + requested: ExecutionStrategy::GpuPreferred + }) + )); +} + +#[test] +fn backend_available_when_features_enabled() { + if cfg!(feature = "cpu") { + for strategy in [ExecutionStrategy::Auto, ExecutionStrategy::CpuOnly] { + let chutoro = Chutoro::new( + execution_config(NonZeroUsize::new(1).expect("literal 1 is non-zero")), + strategy, + None, + ); + assert!(chutoro.backend_unavailable_error().is_none()); + } + } + + let chutoro = Chutoro::new( + execution_config(NonZeroUsize::new(1).expect("literal 1 is non-zero")), + ExecutionStrategy::GpuPreferred, + None, + ); + assert!(matches!( + chutoro.backend_unavailable_error(), + Some(ChutoroError::BackendUnavailable { + requested: ExecutionStrategy::GpuPreferred + }) + )); +} + +#[test] +fn max_bytes_none_imposes_no_limit() { + let chutoro = ChutoroBuilder::new().build().expect("build must succeed"); + assert_eq!(chutoro.max_bytes(), None); +} + +#[test] +fn max_bytes_propagates_through_builder() { + let chutoro = ChutoroBuilder::new() + .with_max_bytes(1_000_000) + .build() + .expect("build must succeed"); + assert_eq!(chutoro.max_bytes(), Some(1_000_000)); +} + +#[cfg(feature = "cpu")] +#[test] +fn builder_hnsw_params_reach_chutoro_execution_config() { + let params = crate::HnswParams::new(4, 16).expect("parameters must be valid"); + let chutoro = ChutoroBuilder::new() + .with_hnsw_params(params.clone()) + .build() + .expect("build must succeed"); + + assert_eq!(chutoro.hnsw_params(), ¶ms); +} diff --git a/chutoro-core/src/cpu_pipeline.rs b/chutoro-core/src/cpu_pipeline.rs index 82cc75c4..55cf23a7 100644 --- a/chutoro-core/src/cpu_pipeline.rs +++ b/chutoro-core/src/cpu_pipeline.rs @@ -15,48 +15,30 @@ use crate::{ CandidateEdge, ClusterId, CpuHnsw, DataSource, EdgeHarvest, HierarchyConfig, HnswError, HnswParams, MstError, Result, error::ChutoroError, parallel_kruskal, result::ClusteringResult, }; - -/// Runs the CPU pipeline end-to-end for the provided [`DataSource`]. -/// -/// # Errors -/// Returns the same errors as [`crate::Chutoro::run`], including empty or -/// undersized sources, data source failures, and CPU pipeline failures. -#[cfg(feature = "cpu")] -pub fn run_cpu_pipeline( - source: &D, - min_cluster_size: NonZeroUsize, -) -> Result { - let items = source.len(); - if items == 0 { - return Err(ChutoroError::EmptySource { - data_source: Arc::from(source.name()), - }); - } - if items < min_cluster_size.get() { - return Err(ChutoroError::InsufficientItems { - data_source: Arc::from(source.name()), - items, - min_cluster_size, - }); - } - - run_cpu_pipeline_with_len(source, items, min_cluster_size) -} +use tracing::debug; #[cfg(feature = "cpu")] pub(crate) fn run_cpu_pipeline_with_len( source: &D, items: usize, min_cluster_size: NonZeroUsize, + hnsw_params: &HnswParams, ) -> Result { - let params = HnswParams::default(); - let (index, harvested) = CpuHnsw::build_with_edges(source, params.clone()) + let configured_ef_construction = hnsw_params.ef_construction(); + let hnsw_params = hnsw_params.clone().bounded_for_point_count(items); + debug!( + max_connections = hnsw_params.max_connections(), + configured_ef_construction, + effective_ef_construction = hnsw_params.ef_construction(), + "building CPU HNSW index" + ); + let (index, harvested) = CpuHnsw::build_with_edges(source, hnsw_params.clone()) .map_err(|error| map_cpu_hnsw_error(source, error))?; let desired = min_cluster_size .get() .saturating_add(1) - .max(params.ef_construction()) + .max(hnsw_params.ef_construction()) .min(items); let Some(ef) = NonZeroUsize::new(desired) else { unreachable!("ef_construction is non-zero so the computed ef is non-zero"); diff --git a/chutoro-core/src/execution_config.rs b/chutoro-core/src/execution_config.rs new file mode 100644 index 00000000..e15f6bc6 --- /dev/null +++ b/chutoro-core/src/execution_config.rs @@ -0,0 +1,46 @@ +//! Validated execution policy shared by batch and session construction. +//! +//! `ExecutionConfig` is an internal composition value. The builder creates it +//! once after validation, then passes it to either `Chutoro` or `SessionConfig` +//! so both execution paths preserve the same clustering policy. + +use std::num::NonZeroUsize; + +#[cfg(feature = "cpu")] +use crate::HnswParams; + +/// Validated clustering settings shared by all execution paths. +#[derive(Clone, Debug, PartialEq)] +pub(crate) struct ExecutionConfig { + min_cluster_size: NonZeroUsize, + #[cfg(feature = "cpu")] + hnsw_params: HnswParams, +} + +impl ExecutionConfig { + /// Creates a configuration from values already validated by the builder. + #[cfg(feature = "cpu")] + pub(crate) fn new(min_cluster_size: NonZeroUsize, hnsw_params: HnswParams) -> Self { + Self { + min_cluster_size, + hnsw_params, + } + } + + /// Creates a configuration from values already validated by the builder. + #[cfg(not(feature = "cpu"))] + pub(crate) fn new(min_cluster_size: NonZeroUsize) -> Self { + Self { min_cluster_size } + } + + /// Returns the validated minimum cluster size. + pub(crate) fn min_cluster_size(&self) -> NonZeroUsize { + self.min_cluster_size + } + + /// Returns the HNSW parameters for CPU execution. + #[cfg(feature = "cpu")] + pub(crate) fn hnsw_params(&self) -> &HnswParams { + &self.hnsw_params + } +} diff --git a/chutoro-core/src/hnsw/params.rs b/chutoro-core/src/hnsw/params.rs index de28f9f5..f0dbbad4 100644 --- a/chutoro-core/src/hnsw/params.rs +++ b/chutoro-core/src/hnsw/params.rs @@ -106,6 +106,22 @@ impl HnswParams { self.ef_construction } + /// Returns the construction search width useful for `point_count` items. + /// + /// The returned width preserves the `ef_construction >= max_connections` + /// invariant while avoiding allocations for neighbours that cannot exist + /// in the supplied batch. + pub(crate) fn effective_ef_construction(&self, point_count: usize) -> usize { + self.ef_construction + .min(point_count.max(self.max_connections)) + } + + /// Caps construction search width to the useful width for a batch. + pub(crate) fn bounded_for_point_count(mut self, point_count: usize) -> Self { + self.ef_construction = self.effective_ef_construction(point_count); + self + } + pub(crate) fn max_level(&self) -> usize { self.max_level } diff --git a/chutoro-core/src/hnsw/tests/params.rs b/chutoro-core/src/hnsw/tests/params.rs index 48f5ac6a..2c0d2132 100644 --- a/chutoro-core/src/hnsw/tests/params.rs +++ b/chutoro-core/src/hnsw/tests/params.rs @@ -11,6 +11,19 @@ fn accepts_equal_search_and_connection_width() { assert_eq!(params.ef_construction(), 8); } +#[test] +fn effective_ef_construction_preserves_minimum_search_width() { + let params = HnswParams::new(4, 64).expect("parameters must be valid"); + + assert_eq!(params.effective_ef_construction(3), 4); + assert_eq!(params.effective_ef_construction(12), 12); + assert_eq!(params.effective_ef_construction(128), 64); + assert_eq!( + params.clone().bounded_for_point_count(12).ef_construction(), + 12 + ); +} + #[test] fn preserves_distance_cache_ttl_when_overriding_capacity() { let ttl = Some(Duration::from_secs(5)); diff --git a/chutoro-core/src/lib.rs b/chutoro-core/src/lib.rs index effd26fa..33eab190 100644 --- a/chutoro-core/src/lib.rs +++ b/chutoro-core/src/lib.rs @@ -1,5 +1,7 @@ //! Chutoro core library. +#[cfg(feature = "metrics")] +mod batch_metrics; mod builder; mod chutoro; mod clustering_quality; @@ -8,6 +10,7 @@ mod cpu_pipeline; mod datasource; mod distance; mod error; +mod execution_config; #[cfg(feature = "cpu")] mod hierarchy; #[cfg(feature = "cpu")] @@ -37,7 +40,7 @@ pub use crate::{ }; #[cfg(feature = "cpu")] -pub use crate::cpu_pipeline::run_cpu_pipeline; +pub use crate::memory::estimate_peak_bytes_for_hnsw_params; #[cfg(feature = "cpu")] /// CPU-accelerated HNSW index components; requires the `cpu` feature. diff --git a/chutoro-core/src/memory.rs b/chutoro-core/src/memory.rs index 00de66d4..55c114fd 100644 --- a/chutoro-core/src/memory.rs +++ b/chutoro-core/src/memory.rs @@ -16,10 +16,12 @@ const SAFETY_MULTIPLIER_NUMERATOR: u64 = 3; const SAFETY_MULTIPLIER_DENOMINATOR: u64 = 2; -/// Default maximum distance cache entries. Mirrors the value in -/// `DistanceCacheConfig::DEFAULT_MAX_ENTRIES` but is duplicated here so the -/// estimation module compiles without the `cpu` feature gate. -const DEFAULT_CACHE_MAX_ENTRIES: u64 = 1_048_576; +/// Default maximum distance-cache entries used by [`estimate_peak_bytes`]. +/// +/// This mirrors `DistanceCacheConfig::DEFAULT_MAX_ENTRIES` while allowing the +/// legacy estimator to compile without the `cpu` feature gate. Parameter-aware +/// estimates use the capacity configured on [`crate::HnswParams`] instead. +const DEFAULT_CACHE_MAX_ENTRIES: usize = 1_048_576; /// Estimated overhead per node in the HNSW graph: `Option`, `Vec` /// headers for the per-level neighbour lists, sequence counter, and alignment @@ -40,6 +42,10 @@ const CACHE_ENTRY_BYTES: u64 = 80; /// Size of an `f32` — used for the core-distances vector. const F32_BYTES: u64 = 4; +/// Conservative allocation budget for one `SearchState` width unit. It covers +/// two binary heaps and two hash sets, including their expected spare capacity. +const SEARCH_STATE_BYTES_PER_WIDTH: u64 = 256; + /// Size of a `usize` — derived at compile time so the estimate adapts to the /// target platform (8 bytes on 64-bit, 4 bytes on 32-bit). const USIZE_BYTES: u64 = std::mem::size_of::() as u64; @@ -56,7 +62,7 @@ const USIZE_BYTES: u64 = std::mem::size_of::() as u64; /// /// - HNSW level-0 adjacency lists (`2 × M` neighbours per node). /// - Per-node struct overhead (Vec headers, sequence counter, alignment). -/// - Distance cache (full configured capacity of 1,048,576 entries). +/// - Distance cache at `DistanceCacheConfig::DEFAULT_MAX_ENTRIES` capacity. /// - Candidate edges harvested during HNSW build (`≈ n × M`). /// - Core-distance vector (`n × sizeof(f32)`). /// - Mutual-reachability edge rewrite (same count as candidate edges). @@ -78,12 +84,55 @@ const USIZE_BYTES: u64 = std::mem::size_of::() as u64; /// ``` #[must_use] pub fn estimate_peak_bytes(point_count: usize, max_connections: usize) -> u64 { + estimate_peak_bytes_with_search_width( + point_count, + max_connections, + 0, + DEFAULT_CACHE_MAX_ENTRIES, + ) +} + +/// Returns the guarded peak estimate for concrete CPU HNSW parameters. +/// +/// This extends [`estimate_peak_bytes`] with the temporary search-state +/// allocation and distance-cache capacity configured for the CPU HNSW index. +/// +/// # Examples +/// +/// ``` +/// use chutoro_core::{HnswParams, estimate_peak_bytes_for_hnsw_params}; +/// +/// let params = HnswParams::new(16, 64).expect("parameters must be valid"); +/// let bytes = estimate_peak_bytes_for_hnsw_params(1_000, ¶ms); +/// assert!(bytes > 0, "a non-empty CPU run requires memory"); +/// ``` +#[cfg(feature = "cpu")] +#[must_use] +pub fn estimate_peak_bytes_for_hnsw_params( + point_count: usize, + hnsw_params: &crate::HnswParams, +) -> u64 { + estimate_peak_bytes_with_search_width( + point_count, + hnsw_params.max_connections(), + hnsw_params.effective_ef_construction(point_count), + hnsw_params.distance_cache_config().max_entries().get(), + ) +} + +fn estimate_peak_bytes_with_search_width( + point_count: usize, + max_connections: usize, + search_width: usize, + distance_cache_capacity: usize, +) -> u64 { if point_count == 0 { return 0; } let n = point_count as u64; let m = max_connections as u64; + let search_width = search_width as u64; // HNSW level-0 adjacency: each node keeps up to 2*M neighbour IDs. let hnsw_adjacency = n.saturating_mul(2_u64.saturating_mul(m).saturating_mul(USIZE_BYTES)); @@ -91,10 +140,9 @@ pub fn estimate_peak_bytes(point_count: usize, max_connections: usize) -> u64 { // Per-node struct overhead (Option, Vec headers, sequence, etc.). let hnsw_nodes = n.saturating_mul(NODE_OVERHEAD_BYTES); - // Distance cache — always allocates up to DEFAULT_CACHE_MAX_ENTRIES - // entries regardless of point count, because pairwise lookups during - // HNSW construction can fill the cache to capacity even for small n. - let distance_cache = DEFAULT_CACHE_MAX_ENTRIES.saturating_mul(CACHE_ENTRY_BYTES); + // Pairwise lookups can fill the configured cache capacity even for small + // batches, so account for every configured entry rather than point count. + let distance_cache = (distance_cache_capacity as u64).saturating_mul(CACHE_ENTRY_BYTES); // Candidate edges: approximately n * M edges from the HNSW build. let candidate_edges = n.saturating_mul(m).saturating_mul(CANDIDATE_EDGE_BYTES); @@ -108,13 +156,17 @@ pub fn estimate_peak_bytes(point_count: usize, max_connections: usize) -> u64 { // MST forest: up to n edges (n − 1 for a connected graph, rounded up). let mst_forest = n.saturating_mul(MST_EDGE_BYTES); + // CPU construction creates search queues sized by the effective `ef`. + let search_state = search_width.saturating_mul(SEARCH_STATE_BYTES_PER_WIDTH); + let subtotal = hnsw_adjacency .saturating_add(hnsw_nodes) .saturating_add(distance_cache) .saturating_add(candidate_edges) .saturating_add(core_distances) .saturating_add(mutual_edges) - .saturating_add(mst_forest); + .saturating_add(mst_forest) + .saturating_add(search_state); // Apply safety multiplier (3/2 = 1.5×) using integer arithmetic. subtotal @@ -213,6 +265,42 @@ mod tests { ); } + #[cfg(feature = "cpu")] + #[rstest] + fn parameter_estimate_grows_with_effective_search_width() { + let narrow = crate::HnswParams::new(4, 4).expect("parameters must be valid"); + let wide = crate::HnswParams::new(4, 64).expect("parameters must be valid"); + + let narrow_bytes = estimate_peak_bytes_for_hnsw_params(100, &narrow); + let wide_bytes = estimate_peak_bytes_for_hnsw_params(100, &wide); + + assert!( + wide_bytes > narrow_bytes, + "a wider effective search state must increase the memory estimate" + ); + } + + #[cfg(feature = "cpu")] + #[rstest] + fn parameter_estimate_grows_with_distance_cache_capacity() { + let default_params = crate::HnswParams::new(4, 16).expect("parameters must be valid"); + let cache_capacity = std::num::NonZeroUsize::new( + crate::DistanceCacheConfig::DEFAULT_MAX_ENTRIES.saturating_mul(2), + ) + .expect("doubled default cache capacity must be non-zero"); + let custom_cache_params = default_params + .clone() + .with_distance_cache_max_entries(cache_capacity); + + let default_bytes = estimate_peak_bytes_for_hnsw_params(100, &default_params); + let custom_cache_bytes = estimate_peak_bytes_for_hnsw_params(100, &custom_cache_params); + + assert!( + custom_cache_bytes > default_bytes, + "a larger configured distance cache must increase the memory estimate" + ); + } + #[rstest] #[case::hundred_vs_thousand(100, 1_000, 16)] #[case::thousand_vs_million(1_000, 1_000_000, 16)] diff --git a/chutoro-core/src/session/config.rs b/chutoro-core/src/session/config.rs index 1bdf5902..d61a02e6 100644 --- a/chutoro-core/src/session/config.rs +++ b/chutoro-core/src/session/config.rs @@ -8,7 +8,7 @@ use std::num::NonZeroUsize; -use crate::HnswParams; +use crate::{HnswParams, execution_config::ExecutionConfig}; /// Refresh behaviour for a [`super::ClusteringSession`]. /// @@ -97,20 +97,17 @@ impl SessionRefreshPolicy { /// ``` #[derive(Clone, Debug, PartialEq)] pub struct SessionConfig { - min_cluster_size: NonZeroUsize, - hnsw_params: HnswParams, + execution_config: ExecutionConfig, refresh_policy: SessionRefreshPolicy, } impl SessionConfig { pub(crate) fn new( - min_cluster_size: NonZeroUsize, - hnsw_params: HnswParams, + execution_config: ExecutionConfig, refresh_policy: SessionRefreshPolicy, ) -> Self { Self { - min_cluster_size, - hnsw_params, + execution_config, refresh_policy, } } @@ -118,13 +115,13 @@ impl SessionConfig { /// Returns the minimum cluster size carried into the session. #[must_use] pub fn min_cluster_size(&self) -> NonZeroUsize { - self.min_cluster_size + self.execution_config.min_cluster_size() } /// Returns the HNSW parameters used for the session index. #[must_use] pub fn hnsw_params(&self) -> &HnswParams { - &self.hnsw_params + self.execution_config.hnsw_params() } /// Returns the session refresh policy. diff --git a/chutoro-core/src/session/tests/builder.rs b/chutoro-core/src/session/tests/builder.rs index 647c9c1f..f23ce9a0 100644 --- a/chutoro-core/src/session/tests/builder.rs +++ b/chutoro-core/src/session/tests/builder.rs @@ -11,6 +11,7 @@ use std::{num::NonZeroUsize, sync::Arc}; use rstest::rstest; use super::common::{SessionTestSource, session_builder}; +use crate::execution_config::ExecutionConfig; use crate::{ ChutoroBuilder, ChutoroError, ClusteringSession, ExecutionStrategy, HnswParams, SessionConfig, SessionRefreshPolicy, @@ -115,8 +116,10 @@ fn build_session_rejects_gpu_preferred_execution_strategy(session_builder: Chuto fn build_session_maps_hnsw_construction_failure_to_cpu_hnsw_failure() { let source = Arc::new(SessionTestSource::with_len(0)); let config = SessionConfig::new( - NonZeroUsize::new(5).expect("minimum cluster size must be non-zero"), - HnswParams::default(), + ExecutionConfig::new( + NonZeroUsize::new(5).expect("minimum cluster size must be non-zero"), + HnswParams::default(), + ), SessionRefreshPolicy::manual(), ); diff --git a/chutoro-core/tests/batch_metrics.rs b/chutoro-core/tests/batch_metrics.rs new file mode 100644 index 00000000..702f7d2a --- /dev/null +++ b/chutoro-core/tests/batch_metrics.rs @@ -0,0 +1,243 @@ +//! Metrics tests for one-shot batch execution. + +#![cfg(feature = "metrics")] + +mod common; + +use metrics_util::debugging::DebugValue; +#[cfg(feature = "cpu")] +use tracing_subscriber::layer::SubscriberExt; + +#[cfg(feature = "cpu")] +use chutoro_core::DataSource; +use chutoro_core::{ChutoroBuilder, ChutoroError, ExecutionStrategy}; +#[cfg(feature = "cpu")] +use chutoro_test_support::tracing::RecordingLayer; + +use common::Dummy; + +const RUNS_TOTAL: &str = "chutoro.batch.runs_total"; +#[cfg(feature = "cpu")] +const MAX_CONNECTIONS: &str = "chutoro.batch.max_connections"; +#[cfg(feature = "cpu")] +const EFFECTIVE_EF_CONSTRUCTION: &str = "chutoro.batch.effective_ef_construction"; +#[cfg(feature = "cpu")] +const ESTIMATED_BYTES: &str = "chutoro.batch.estimated_bytes"; +#[cfg(feature = "cpu")] +const MEMORY_LIMIT_BYTES: &str = "chutoro.batch.memory_limit_bytes"; + +macro_rules! metric_value { + ($snapshot:expr, $name:expr, $labels:expr $(,)?) => { + $snapshot + .iter() + .find(|(key, _, _, _)| { + key.key().name() == $name + && key.key().labels().count() == $labels.len() + && $labels.iter().all(|(expected_key, expected_value)| { + key.key().labels().any(|label| { + label.key() == *expected_key && label.value() == *expected_value + }) + }) + }) + .map(|(_, _, _, value)| value) + .expect("metric with the expected bounded labels must be recorded") + }; +} + +#[cfg(feature = "cpu")] +fn assert_histogram_sample(value: &DebugValue, expected: f64) { + let DebugValue::Histogram(samples) = value else { + panic!("expected a Histogram metric value, got {value:?}"); + }; + assert!( + samples.iter().any(|sample| sample.into_inner() == expected), + "expected histogram sample {expected}, got {samples:?}" + ); +} + +macro_rules! assert_outcome { + ($snapshot:expr, $backend:expr, $outcome:expr, $error_code:expr $(,)?) => { + assert_eq!( + metric_value!( + $snapshot, + RUNS_TOTAL, + &[ + ("backend", $backend), + ("outcome", $outcome), + ("error_code", $error_code), + ], + ), + &DebugValue::Counter(1), + ); + }; +} + +#[cfg(feature = "cpu")] +#[test] +fn successful_cpu_run_records_bounded_resources_and_tracing() { + use chutoro_core::{HnswParams, estimate_peak_bytes_for_hnsw_params}; + + let params = HnswParams::new(2, 4).expect("parameters must be valid"); + let source = Dummy::new(vec![1.0, 3.0, 6.0, 10.0]); + let estimate = estimate_peak_bytes_for_hnsw_params(source.len(), ¶ms); + let chutoro = ChutoroBuilder::new() + .with_min_cluster_size(2) + .with_execution_strategy(ExecutionStrategy::CpuOnly) + .with_hnsw_params(params.clone()) + .build() + .expect("configuration must be valid"); + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let layer = RecordingLayer::default(); + let subscriber = tracing_subscriber::registry().with(layer.clone()); + + metrics::with_local_recorder(&recorder, || { + tracing::subscriber::with_default(subscriber, || { + chutoro.run(&source).expect("CPU batch run must succeed"); + }); + }); + + let snapshot = snapshotter.snapshot().into_vec(); + assert_outcome!(&snapshot, "cpu", "success", "none"); + assert_histogram_sample( + metric_value!(&snapshot, MAX_CONNECTIONS, &[("backend", "cpu")]), + 2.0, + ); + assert_histogram_sample( + metric_value!(&snapshot, EFFECTIVE_EF_CONSTRUCTION, &[("backend", "cpu")],), + 4.0, + ); + assert_histogram_sample( + metric_value!(&snapshot, ESTIMATED_BYTES, &[("backend", "cpu")]), + estimate as f64, + ); + + let run_span = layer + .spans() + .into_iter() + .find(|span| span.name == "core.run") + .expect("batch run span must be recorded"); + assert_eq!(run_span.fields.get("backend"), Some(&"cpu".to_owned())); + assert!(!run_span.fields.contains_key("data_source")); +} + +#[cfg(feature = "cpu")] +#[test] +fn memory_limit_rejection_records_bounded_metrics_and_tracing() { + use chutoro_core::{HnswParams, estimate_peak_bytes_for_hnsw_params}; + + let params = HnswParams::new(2, 4).expect("parameters must be valid"); + let source = Dummy::new(vec![1.0, 3.0, 6.0, 10.0]); + let estimate = estimate_peak_bytes_for_hnsw_params(source.len(), ¶ms); + let limit = estimate - 1; + let chutoro = ChutoroBuilder::new() + .with_min_cluster_size(2) + .with_execution_strategy(ExecutionStrategy::CpuOnly) + .with_hnsw_params(params) + .with_max_bytes(limit) + .build() + .expect("configuration must be valid"); + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let layer = RecordingLayer::default(); + let subscriber = tracing_subscriber::registry().with(layer.clone()); + + metrics::with_local_recorder(&recorder, || { + let error = tracing::subscriber::with_default(subscriber, || chutoro.run(&source)) + .expect_err("one byte below the estimate must reject the run"); + assert!(matches!(error, ChutoroError::MemoryLimitExceeded { .. })); + }); + + let snapshot = snapshotter.snapshot().into_vec(); + assert_outcome!(&snapshot, "cpu", "error", "CHUTORO_MEMORY_LIMIT_EXCEEDED",); + assert_histogram_sample( + metric_value!(&snapshot, MAX_CONNECTIONS, &[("backend", "cpu")]), + 2.0, + ); + assert_histogram_sample( + metric_value!(&snapshot, EFFECTIVE_EF_CONSTRUCTION, &[("backend", "cpu")],), + 4.0, + ); + assert_histogram_sample( + metric_value!(&snapshot, ESTIMATED_BYTES, &[("backend", "cpu")]), + estimate as f64, + ); + assert_histogram_sample( + metric_value!(&snapshot, MEMORY_LIMIT_BYTES, &[("backend", "cpu")],), + limit as f64, + ); + let event = layer + .events() + .into_iter() + .find(|event| { + event + .fields + .get("message") + .is_some_and(|message| message == "CPU memory estimate exceeds configured limit") + }) + .expect("memory limit rejection event must be recorded"); + assert_eq!( + event.fields.get("error_code"), + Some(&"CHUTORO_MEMORY_LIMIT_EXCEEDED".to_owned()) + ); + assert!(!event.fields.contains_key("data_source")); +} + +#[test] +fn empty_and_insufficient_rejections_record_stable_outcomes() { + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + let empty_error = ChutoroBuilder::new() + .build() + .expect("configuration must be valid") + .run(&Dummy::new(Vec::new())) + .expect_err("empty source must be rejected"); + assert!(matches!(empty_error, ChutoroError::EmptySource { .. })); + + let insufficient_error = ChutoroBuilder::new() + .with_min_cluster_size(3) + .build() + .expect("configuration must be valid") + .run(&Dummy::new(vec![1.0, 2.0])) + .expect_err("undersized source must be rejected"); + assert!(matches!( + insufficient_error, + ChutoroError::InsufficientItems { .. } + )); + }); + + let backend = if cfg!(feature = "cpu") { + "cpu" + } else { + "unavailable" + }; + let snapshot = snapshotter.snapshot().into_vec(); + assert_outcome!(&snapshot, backend, "error", "CHUTORO_EMPTY_SOURCE",); + assert_outcome!(&snapshot, backend, "error", "CHUTORO_INSUFFICIENT_ITEMS",); +} + +#[cfg(not(feature = "cpu"))] +#[test] +fn unavailable_cpu_backend_records_stable_outcome() { + let recorder = metrics_util::debugging::DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + metrics::with_local_recorder(&recorder, || { + let error = ChutoroBuilder::new() + .with_min_cluster_size(2) + .with_execution_strategy(ExecutionStrategy::CpuOnly) + .build() + .expect("configuration must be valid") + .run(&Dummy::new(vec![1.0, 2.0])) + .expect_err("unavailable CPU backend must be rejected"); + assert!(matches!(error, ChutoroError::BackendUnavailable { .. })); + }); + + let snapshot = snapshotter.snapshot().into_vec(); + assert_outcome!( + &snapshot, + "unavailable", + "error", + "CHUTORO_BACKEND_UNAVAILABLE", + ); +} diff --git a/chutoro-core/tests/chutoro.rs b/chutoro-core/tests/chutoro.rs index 3208d508..ab0ecc0f 100644 --- a/chutoro-core/tests/chutoro.rs +++ b/chutoro-core/tests/chutoro.rs @@ -6,9 +6,13 @@ use chutoro_core::{ ChutoroBuilder, ChutoroError, ClusterId, ClusteringResult, DataSource, DataSourceError, ExecutionStrategy, NonContiguousClusterIds, }; +#[cfg(feature = "cpu")] +use chutoro_core::{ + DistanceCacheConfig, HnswParams, estimate_peak_bytes, estimate_peak_bytes_for_hnsw_params, +}; use common::Dummy; use rstest::{fixture, rstest}; -use std::sync::Arc; +use std::{num::NonZeroUsize, sync::Arc}; use tracing::Level; use tracing_subscriber::layer::SubscriberExt; @@ -147,6 +151,110 @@ fn run_insufficient_items_errors(small_dummy: Dummy) { )); } +#[cfg(feature = "cpu")] +#[rstest] +fn run_memory_limit_uses_builder_hnsw_params(dummy: Dummy) { + let hnsw_params = HnswParams::new(4, 16).expect("parameters must be valid"); + let estimated_bytes = estimate_peak_bytes_for_hnsw_params(dummy.len(), &hnsw_params); + assert_ne!(estimated_bytes, estimate_peak_bytes(dummy.len(), 16)); + + let chutoro = ChutoroBuilder::new() + .with_min_cluster_size(2) + .with_execution_strategy(ExecutionStrategy::CpuOnly) + .with_hnsw_params(hnsw_params) + .with_max_bytes(estimated_bytes - 1) + .build() + .expect("configuration must be valid"); + + let err = chutoro + .run(&dummy) + .expect_err("configured memory limit must be enforced"); + assert!(matches!( + err, + ChutoroError::MemoryLimitExceeded { + point_count, + estimated_bytes: actual_estimated_bytes, + .. + } if point_count == dummy.len() && actual_estimated_bytes == estimated_bytes + )); +} + +#[cfg(feature = "cpu")] +#[rstest] +fn run_memory_limit_uses_configured_distance_cache_capacity(dummy: Dummy) { + let default_params = HnswParams::new(16, 64).expect("parameters must be valid"); + let cache_capacity = + NonZeroUsize::new(DistanceCacheConfig::DEFAULT_MAX_ENTRIES.saturating_mul(2)) + .expect("doubled default cache capacity must be non-zero"); + let hnsw_params = default_params + .clone() + .with_distance_cache_max_entries(cache_capacity); + let default_estimate = estimate_peak_bytes_for_hnsw_params(dummy.len(), &default_params); + let parameter_aware_estimate = estimate_peak_bytes_for_hnsw_params(dummy.len(), &hnsw_params); + let max_bytes = parameter_aware_estimate - 1; + assert!( + max_bytes > default_estimate, + "the configured cache must increase the estimate beyond the default cache" + ); + + let chutoro = ChutoroBuilder::new() + .with_min_cluster_size(2) + .with_execution_strategy(ExecutionStrategy::CpuOnly) + .with_hnsw_params(hnsw_params) + .with_max_bytes(max_bytes) + .build() + .expect("configuration must be valid"); + + let error = chutoro + .run(&dummy) + .expect_err("parameter-aware memory limit must be enforced"); + assert!(matches!( + error, + ChutoroError::MemoryLimitExceeded { + estimated_bytes, + .. + } if estimated_bytes == parameter_aware_estimate + )); +} + +#[cfg(feature = "cpu")] +#[rstest] +fn run_passes_builder_hnsw_params_to_cpu_pipeline(dummy: Dummy) { + let hnsw_params = HnswParams::new(1, 2).expect("parameters must be valid"); + let layer = RecordingLayer::default(); + let subscriber = tracing_subscriber::registry().with(layer.clone()); + let chutoro = ChutoroBuilder::new() + .with_min_cluster_size(2) + .with_execution_strategy(ExecutionStrategy::CpuOnly) + .with_hnsw_params(hnsw_params) + .build() + .expect("configuration must be valid"); + + let result = tracing::subscriber::with_default(subscriber, || chutoro.run(&dummy)) + .expect("custom HNSW configuration must complete the CPU pipeline"); + assert_eq!(result.assignments().len(), dummy.len()); + + let event = layer + .events() + .into_iter() + .find(|event| { + event + .fields + .get("message") + .is_some_and(|message| message == "building CPU HNSW index") + }) + .expect("CPU pipeline must record its HNSW construction parameters"); + assert_eq!(event.fields.get("max_connections"), Some(&"1".to_owned())); + assert_eq!( + event.fields.get("configured_ef_construction"), + Some(&"2".to_owned()) + ); + assert_eq!( + event.fields.get("effective_ef_construction"), + Some(&"2".to_owned()) + ); +} + #[cfg(not(feature = "gpu"))] #[rstest] fn builder_rejects_gpu_preferred_without_feature(dummy: Dummy) { diff --git a/chutoro-core/tests/session_api_surface.rs b/chutoro-core/tests/session_api_surface.rs index d3b88396..a53acda0 100644 --- a/chutoro-core/tests/session_api_surface.rs +++ b/chutoro-core/tests/session_api_surface.rs @@ -8,6 +8,7 @@ fn session_api_compiles_when_cpu_feature_is_enabled() { let cases = trybuild::TestCases::new(); cases.pass("tests/trybuild/session_api_cpu_enabled.rs"); cases.compile_fail("tests/trybuild/session_api_non_send_sync_source.rs"); + cases.compile_fail("tests/trybuild/run_cpu_pipeline_is_private.rs"); } #[test] diff --git a/chutoro-core/tests/trybuild/run_cpu_pipeline_is_private.rs b/chutoro-core/tests/trybuild/run_cpu_pipeline_is_private.rs new file mode 100644 index 00000000..1bb90e41 --- /dev/null +++ b/chutoro-core/tests/trybuild/run_cpu_pipeline_is_private.rs @@ -0,0 +1,5 @@ +use chutoro_core::run_cpu_pipeline; + +fn main() { + let _ = run_cpu_pipeline; +} diff --git a/chutoro-core/tests/trybuild/run_cpu_pipeline_is_private.stderr b/chutoro-core/tests/trybuild/run_cpu_pipeline_is_private.stderr new file mode 100644 index 00000000..10a587e2 --- /dev/null +++ b/chutoro-core/tests/trybuild/run_cpu_pipeline_is_private.stderr @@ -0,0 +1,8 @@ +error[E0432]: unresolved import `chutoro_core::run_cpu_pipeline` + --> tests/trybuild/run_cpu_pipeline_is_private.rs:1:5 + | +1 | use chutoro_core::run_cpu_pipeline; + | ^^^^^^^^^^^^^^---------------- + | | | + | | help: a similar name exists in the module: `cpu_pipeline` + | no `run_cpu_pipeline` in the root diff --git a/docs/chutoro-design.md b/docs/chutoro-design.md index bed6ff05..d175dc66 100644 --- a/docs/chutoro-design.md +++ b/docs/chutoro-design.md @@ -1788,14 +1788,22 @@ use std::sync::Arc; /// Builder for the chutoro implementation of the FISHDBC algorithm. pub struct ChutoroBuilder { min_cluster_size: usize, - // HNSW parameters (e.g., ef_construction, M) - //... other configuration... + execution_strategy: ExecutionStrategy, + max_bytes: Option, + #[cfg(feature = "cpu")] + hnsw_params: HnswParams, } impl ChutoroBuilder { pub fn new() -> Self { // Default parameters - Self { min_cluster_size: 5, /*... */ } + Self { + min_cluster_size: 5, + execution_strategy: ExecutionStrategy::Auto, + max_bytes: None, + #[cfg(feature = "cpu")] + hnsw_params: HnswParams::default(), + } } pub fn min_cluster_size(mut self, size: usize) -> Self { @@ -1803,30 +1811,39 @@ impl ChutoroBuilder { self } - //... other builder methods for HNSW/HDBSCAN parameters... - pub fn build(self) -> Result { let min_cluster_size = NonZeroUsize::new(self.min_cluster_size) .ok_or(ChutoroError::InvalidMinClusterSize { got: self.min_cluster_size })?; + // `ExecutionConfig` is created only after builder validation and is + // shared by batch and session construction. + #[cfg(feature = "cpu")] + let execution_config = + ExecutionConfig::new(min_cluster_size, self.hnsw_params); + #[cfg(not(feature = "cpu"))] + let execution_config = ExecutionConfig::new(min_cluster_size); + Ok(Chutoro { - min_cluster_size, + execution_config, execution_strategy: self.execution_strategy, + max_bytes: self.max_bytes, }) } } /// The main chutoro clustering algorithm struct. pub struct Chutoro { - min_cluster_size: NonZeroUsize, + execution_config: ExecutionConfig, execution_strategy: ExecutionStrategy, + max_bytes: Option, } impl Chutoro { /// Runs the clustering algorithm on the given data source. /// - /// The CPU implementation validates the dataset before dispatch and then - /// executes the FISHDBC pipeline (HNSW → MST → hierarchy extraction). + /// The CPU implementation validates the dataset and memory estimate before + /// dispatch, then executes the FISHDBC pipeline (HNSW → MST → hierarchy + /// extraction). pub fn run(&self, source: &D) -> Result { let len = source.len(); if len == 0 { @@ -1834,14 +1851,18 @@ impl Chutoro { data_source: Arc::from(source.name()), }); } - if len < self.min_cluster_size.get() { + if len < self.execution_config.min_cluster_size().get() { return Err(ChutoroError::InsufficientItems { data_source: Arc::from(source.name()), items: len, - min_cluster_size: self.min_cluster_size, + min_cluster_size: self.execution_config.min_cluster_size(), }); } + // The memory guard uses the configured HNSW connectivity whenever a + // limit is set; it runs before backend dispatch. + self.check_memory_limit(source, len)?; + // If neither the `cpu` nor `gpu` feature is enabled, the orchestrator // cannot select a backend and returns `BackendUnavailable`. match self.execution_strategy { @@ -1868,7 +1889,12 @@ impl Chutoro { // The detailed CPU pipeline lives in `chutoro-core/src/cpu_pipeline.rs`. // // Precondition: `items > 0` due to earlier source validation in `run`. - cpu_pipeline::run_cpu_pipeline_with_len(source, items, self.min_cluster_size) + cpu_pipeline::run_cpu_pipeline_with_len( + source, + items, + self.execution_config.min_cluster_size(), + self.execution_config.hnsw_params(), + ) } #[cfg(feature = "gpu")] @@ -1885,10 +1911,18 @@ impl Chutoro { `ChutoroBuilder::build` rejects zero-sized clusters while deferring backend availability to runtime so GPU-preferred configurations can be constructed -ahead of accelerated support. The struct stores the validated -`min_cluster_size` and `execution_strategy`, and [`Chutoro::run`] fails fast on -empty or undersized sources while sharing `Arc` handles for the -data-source name so repeated errors avoid cloning. +ahead of accelerated support. It creates one validated internal +`ExecutionConfig` carrying `min_cluster_size` and, with the `cpu` feature, +`HnswParams`. `Chutoro` stores that config alongside `execution_strategy` and +the optional `max_bytes` limit. [`Chutoro::run`] checks for an empty source, +insufficient items, backend availability, and a memory-limit violation in that +order before dispatching. The memory estimate uses the configured HNSW +`max_connections`, so `with_hnsw_params` applies to both batch CPU execution +and its memory guard. For one-shot CPU execution, the configured +`ef_construction` is bounded to the effective width +`min(ef_construction, max(point_count, max_connections))`. The memory guard +uses `estimate_peak_bytes_for_hnsw_params` so that same effective width is +included in its estimate. _Implementation update (2025-12-18)._ The CPU pipeline is available when the `cpu` feature is enabled (it is part of the default feature set). The `Auto` @@ -2299,9 +2333,12 @@ deterministic queries with `ef_search = 64`. Results are written to controls insertion thoroughness. They are complementary: increasing `M` without sufficient `ef_construction` wastes connectivity, while high `ef_construction` with low `M` is limited by the graph's fan-out capacity. -- Memory footprint is primarily a function of `M` (not `ef_construction`) - since `ef_construction` only affects the construction search beam, not the - stored graph structure. Memory scaling is tracked in §11.2. +- One-shot CPU execution bounds the effective construction width to + `min(ef_construction, max(point_count, max_connections))` before building + HNSW. The batch memory guard calls + `estimate_peak_bytes_for_hnsw_params(point_count, &HnswParams)`, including + that effective width as well as the stored graph's `M`-dependent footprint. + Memory scaling is tracked in §11.2. ### 11.4. Memory guards and estimation (roadmap 2.1.5) @@ -2311,13 +2348,14 @@ rejects datasets whose estimated peak memory exceeds the configured limit. The guard fires before any pipeline allocation, avoiding wasted work and out-of-memory crashes. -**Estimation formula.** The function `estimate_peak_bytes(n, M)` computes a -conservative upper bound on the peak memory that the CPU pipeline will require: +**Estimation formula.** The function +`estimate_peak_bytes(n, M)` computes a conservative upper bound on the peak +memory that the CPU pipeline will require: ```text hnsw_adjacency = n × (2 × M) × 8 (level-0 neighbour IDs) hnsw_node_overhead = n × 80 (Node structs, Vec headers) -distance_cache = 1 048 576 × 80 (full cache capacity) +distance_cache = cache_entries × 80 (configured cache capacity) candidate_edges = n × M × 32 (CandidateEdge structs) core_distances = n × 4 (f32 per point) mutual_edges = n × M × 32 (recomputed edges) @@ -2326,6 +2364,18 @@ mst_forest = n × 32 (MstEdge structs) estimated_bytes = (sum of above) × 1.5 (safety multiplier) ``` +For one-shot runs, the memory guard calls +`estimate_peak_bytes_for_hnsw_params(point_count, &HnswParams)`. That helper +derives `effective_ef_construction` as +`min(ef_construction, max(point_count, max_connections))` and includes the +construction-search state required by that bounded width. Retaining +`max_connections` as the lower bound keeps the estimate aligned with the +allocation made by the batch pipeline. It also sets `cache_entries` from the +configured `DistanceCacheConfig::max_entries` value, so enlarged or reduced +distance caches are reflected in the guard. The legacy public +`estimate_peak_bytes(n, M)` helper continues to use +`DistanceCacheConfig::DEFAULT_MAX_ENTRIES` for compatibility. + The 1.5× safety multiplier covers heap fragmentation, Rayon thread-local buffers, and transient allocations that are difficult to predict statically. The estimate is intentionally pessimistic: it is better to reject a dataset @@ -2350,11 +2400,13 @@ work, ensure at least 4 GiB of headroom above the estimate. The `--max-bytes` flag accepts human-readable suffixes: `--max-bytes 2G`, `--max-bytes 512M`, or plain byte counts. -**Limitations.** The estimate assumes the default HNSW `M = 16` and -`DistanceCacheConfig::DEFAULT_MAX_ENTRIES = 1 048 576`. Custom HNSW parameters -or enlarged caches will shift actual memory usage. The formula does not account -for the data source's own memory footprint (e.g., the in-memory Parquet column -or text corpus), which must be added separately for a complete picture. +**Limitations.** The table above assumes the default HNSW `M = 16` and +`DistanceCacheConfig::DEFAULT_MAX_ENTRIES = 1 048 576`; the parameter-aware +one-shot estimate instead uses the configured cache capacity. Custom HNSW +parameters or enlarged caches will shift actual memory usage. The formula does +not account for the data source's own memory footprint (e.g., the in-memory +Parquet column or text corpus), which must be added separately for a complete +picture. ### 11.5. Optional Gaussian clustering-quality tracking (roadmap 2.1.6) @@ -2522,11 +2574,16 @@ clustering state, in contrast to the stateless `Chutoro::run()` path. - The returned session does **not** seed existing source items into the index. This keeps `11.1.2` limited to configuration and lifecycle scaffolding while leaving bootstrap behaviour to `11.3.1` and `11.3.2`. - - `SessionConfig` stores validated types directly: - `NonZeroUsize` for `min_cluster_size`, `HnswParams` for HNSW tuning, and the + - `SessionConfig` composes the same validated internal `ExecutionConfig` + used by `Chutoro`, rather than storing `min_cluster_size` and `HnswParams` + separately. Its accessors continue to expose those values alongside the v1 `SessionRefreshPolicy { refresh_every_n: Option }`. Later roadmap items extend the refresh policy with drift-trigger and baseline-management fields once those semantics are implemented. + - Batch CPU execution consumes the configured HNSW parameters through the + internal `run_cpu_pipeline_with_len(source, items, min_cluster_size, + hnsw_params)` boundary. Session construction and one-shot execution + therefore use one validated source for HNSW tuning. - Session construction is CPU-only in v1. `build_session(...)` therefore rejects `ExecutionStrategy::GpuPreferred` even if batch execution later gains a separate GPU backend. diff --git a/docs/developers-guide.md b/docs/developers-guide.md index 722bc505..9435181a 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -96,6 +96,14 @@ pub fn snapshot_version(&self) -> u64; returns an inert session whose initial observable state is `point_count() == 0` and `snapshot_version() == 0`. +The internal `ExecutionConfig` is the single validated source for +`min_cluster_size` and CPU `HnswParams`. `ChutoroBuilder` creates it after +validation and passes it unchanged to batch `Chutoro` or `SessionConfig`. +One-shot runs use these same HNSW settings for both CPU pipeline construction +and memory estimates. It is a composition boundary only: execution and session +code may read it, but must not create or revalidate alternate copies of those +settings. + `append` inserts source indices into the live HNSW index by calling `CpuHnsw::insert_harvesting` for each index. It must not duplicate HNSW insertion logic or inspect private HNSW adapter internals. The session stores diff --git a/docs/users-guide.md b/docs/users-guide.md index b44b6bfd..9bfec78d 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -53,6 +53,58 @@ assert_eq!(result.cluster_count(), 1); # Ok::<(), chutoro_core::ChutoroError>(()) ``` +On CPU builds, `with_hnsw_params` configures both one-shot batch runs and +incremental sessions. For a batch run, `max_connections` and `ef_construction` +are passed to CPU HNSW construction. The configured `max_connections` and the +effective, dataset-bounded construction-search width derived from +`ef_construction` are included when estimating the run's peak memory for a +limit set with `with_max_bytes`: + +```rust +use chutoro_core::{ChutoroBuilder, HnswParams}; + +let chutoro = ChutoroBuilder::new() + .with_min_cluster_size(2) + .with_hnsw_params(HnswParams::new(32, 128)?) + .with_max_bytes(1_073_741_824) + .build()?; +let result = chutoro.run(&Dummy(vec![1.0, 2.0, 4.0, 8.0]))?; +# Ok::<(), chutoro_core::ChutoroError>(()) +``` + +When the estimate exceeds the configured limit, `run` returns +`ChutoroError::MemoryLimitExceeded` before allocating the pipeline. Omitting +`with_max_bytes` leaves this guard disabled. + +When the `metrics` feature is enabled, one-shot runs emit bounded batch +metrics. The `chutoro.batch.runs_total` counter has the labels `backend`, +`outcome`, and `error_code`. `backend` is one of `cpu`, `gpu`, or `unavailable`; +`outcome` is `success` or `error`; and `error_code` is `none` for successful +runs or the stable `ChutoroErrorCode` string for failures. The resource +histograms use only the bounded `backend` label: + +- `chutoro.batch.max_connections` (Count) records the configured HNSW + connection width. +- `chutoro.batch.effective_ef_construction` (Count) records the dataset- + bounded HNSW construction-search width. +- `chutoro.batch.estimated_bytes` (Bytes) records the estimated peak memory. +- `chutoro.batch.memory_limit_bytes` (Bytes) records the configured memory + limit when one is present. + +The stable `ChutoroErrorCode` vocabulary includes +`CHUTORO_INVALID_MIN_CLUSTER_SIZE`, `CHUTORO_EMPTY_SOURCE`, +`CHUTORO_INSUFFICIENT_ITEMS`, `CHUTORO_BACKEND_UNAVAILABLE`, +`CHUTORO_DATA_SOURCE_FAILURE`, `CHUTORO_CPU_HNSW_FAILURE`, +`CHUTORO_CPU_MST_FAILURE`, `CHUTORO_CPU_HIERARCHY_FAILURE`, and +`CHUTORO_MEMORY_LIMIT_EXCEEDED`. The `CHUTORO_INVALID_MIN_CLUSTER_SIZE` code is +reported during builder validation; the remaining applicable codes can be +reported as `run` outcomes. Neither these metrics nor the batch decision-point +tracing includes source names, source paths, or source payload data. + +The public `run_cpu_pipeline` entry point has been removed. Migrate callers to +`ChutoroBuilder::build()?.run(&source)`, which applies the same validated +configuration and run preconditions as the supported batch API. + `ExecutionStrategy::Auto` runs the CPU backend. The `gpu` feature prepares the orchestration surface for a future accelerator backend; requesting `ExecutionStrategy::GpuPreferred` currently yields `BackendUnavailable`.