From aefc04dc9ca652dc4f59d2f572e8ac162a0a0ee0 Mon Sep 17 00:00:00 2001 From: loutPhilipps Date: Mon, 3 Aug 2026 20:01:20 +0200 Subject: [PATCH 1/4] Spread indexing pipelines --- docs/configuration/node-config.md | 1 + .../quickwit-config/src/node_config/mod.rs | 8 ++ .../src/node_config/serialize.rs | 1 + .../src/actors/cooperative_indexing.rs | 76 ++++++++++--------- .../quickwit-indexing/src/actors/indexer.rs | 59 +++++++------- .../src/actors/indexing_pipeline.rs | 8 ++ .../src/actors/indexing_service.rs | 4 + 7 files changed, 94 insertions(+), 63 deletions(-) diff --git a/docs/configuration/node-config.md b/docs/configuration/node-config.md index cbe3f8b0022..eba5411fe14 100644 --- a/docs/configuration/node-config.md +++ b/docs/configuration/node-config.md @@ -244,6 +244,7 @@ This section contains the configuration options for an indexer. The split store | `enable_otlp_endpoint` | If true, enables the OpenTelemetry exporter endpoint to ingest logs and traces via the OpenTelemetry Protocol (OTLP). | `false` | | `cpu_capacity` | Advisory parameter used by the control plane. The value can expressed be in threads (e.g. `2`) or in term of millicpus (`2000m`). The control plane will attempt to schedule indexing pipelines on the different nodes proportionally to the cpu capacity advertised by the indexer. It is NOT used as a limit. All pipelines will be scheduled regardless of whether the cluster has sufficient capacity or not. The control plane does not attempt to spread the work equally when the load is well below the `cpu_capacity`. Users who need a balanced load on all of their indexer nodes can set the `cpu_capacity` to an arbitrarily low value as long as they keep it proportional to the number of threads available. | `num threads available` | | `enable_cooperative_indexing` | Enable sharing resources more efficiently when the number of indexes actively written to is significantly higher than the number of cores but might decrease the overall indexing throughput. | `false` | +| `enable_spread_indexing_pipelines` | Enable spreading indexing pipelines in time to make sure they are not in sync. | `false` | Example: diff --git a/quickwit/quickwit-config/src/node_config/mod.rs b/quickwit/quickwit-config/src/node_config/mod.rs index c9daa723982..377e2610fa9 100644 --- a/quickwit/quickwit-config/src/node_config/mod.rs +++ b/quickwit/quickwit-config/src/node_config/mod.rs @@ -210,6 +210,8 @@ pub struct IndexerConfig { pub enable_otlp_endpoint: bool, #[serde(default = "IndexerConfig::default_enable_cooperative_indexing")] pub enable_cooperative_indexing: bool, + #[serde(default = "IndexerConfig::default_enable_spread_indexing_pipelines")] + pub enable_spread_indexing_pipelines: bool, #[serde(default = "IndexerConfig::default_cpu_capacity")] pub cpu_capacity: CpuCapacity, /// If true, run Parquet merges through the streaming column-major engine @@ -229,6 +231,10 @@ impl IndexerConfig { false } + fn default_enable_spread_indexing_pipelines() -> bool { + false + } + fn default_enable_otlp_endpoint() -> bool { #[cfg(any(test, feature = "testsuite"))] { @@ -269,6 +275,7 @@ impl IndexerConfig { use quickwit_proto::indexing::PIPELINE_FULL_CAPACITY; let indexer_config = IndexerConfig { enable_cooperative_indexing: false, + enable_spread_indexing_pipelines: false, enable_otlp_endpoint: true, split_store_max_num_bytes: ByteSize::mb(1), split_store_max_num_splits: 3, @@ -286,6 +293,7 @@ impl Default for IndexerConfig { fn default() -> Self { Self { enable_cooperative_indexing: Self::default_enable_cooperative_indexing(), + enable_spread_indexing_pipelines: Self::default_enable_spread_indexing_pipelines(), enable_otlp_endpoint: Self::default_enable_otlp_endpoint(), split_store_max_num_bytes: Self::default_split_store_max_num_bytes(), split_store_max_num_splits: Self::default_split_store_max_num_splits(), diff --git a/quickwit/quickwit-config/src/node_config/serialize.rs b/quickwit/quickwit-config/src/node_config/serialize.rs index f92e28446c0..599ab0b4fd2 100644 --- a/quickwit/quickwit-config/src/node_config/serialize.rs +++ b/quickwit/quickwit-config/src/node_config/serialize.rs @@ -850,6 +850,7 @@ mod tests { merge_concurrency: NonZeroUsize::new(2).unwrap(), cpu_capacity: IndexerConfig::default_cpu_capacity(), enable_cooperative_indexing: false, + enable_spread_indexing_pipelines: false, max_merge_write_throughput: Some(ByteSize::mb(100)), parquet_merge_use_streaming_engine: true, } diff --git a/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs b/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs index f39f97b39d5..e1cb4dcb8c1 100644 --- a/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs +++ b/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs @@ -70,20 +70,20 @@ static ORIGIN_OF_TIME: LazyLock = LazyLock::new(Instant::now); /// /// We then allow ourselves to tweak the sleep time one way or another by at /// most two seconds to eventually nudge the system toward the desired phase. -pub(crate) struct CooperativeIndexingCycle { +pub(crate) struct IndexingCycle { target_phase: Duration, commit_timeout: Duration, - indexing_permits: Arc, + indexing_permits_opt: Option>, } -impl CooperativeIndexingCycle { +impl IndexingCycle { /// Creates a new cooperative indexing cycle object. /// `phase_id` is hashed to compute the target phase. pub fn new( phase_id: &(impl Hash + ?Sized), commit_timeout: Duration, - indexing_permits: Arc, - ) -> CooperativeIndexingCycle { + indexing_permits_opt: Option>, + ) -> IndexingCycle { assert!(commit_timeout.as_millis() > 0); let mut hasher = DefaultHasher::new(); phase_id.hash(&mut hasher); @@ -91,21 +91,21 @@ impl CooperativeIndexingCycle { Self::new_with_phase( Duration::from_millis(target_phase_millis), commit_timeout, - indexing_permits, + indexing_permits_opt, ) } fn new_with_phase( target_phase: Duration, commit_timeout: Duration, - indexing_permits: Arc, - ) -> CooperativeIndexingCycle { + indexing_permits_opt: Option>, + ) -> IndexingCycle { // Force the initial of the origin of time. let _t0 = *ORIGIN_OF_TIME; - CooperativeIndexingCycle { + IndexingCycle { target_phase, commit_timeout, - indexing_permits, + indexing_permits_opt, } } @@ -125,33 +125,36 @@ impl CooperativeIndexingCycle { Duration::from_millis(initial_sleep_millis) } - pub async fn cooperative_indexing_period(&self) -> CooperativeIndexingPeriod { + pub async fn indexing_period(&self) -> IndexingPeriod { let t_wake = Instant::now(); - let permit = Semaphore::acquire_owned(self.indexing_permits.clone()) - .await - .unwrap(); + let permit_opt = if let Some(permits) = self.indexing_permits_opt.clone() { + Some(permits.acquire_owned().await.unwrap()) + } else { + None + }; + let t_work_start = Instant::now(); - CooperativeIndexingPeriod { + IndexingPeriod { t_wake, t_work_start, commit_timeout: self.commit_timeout, target_phase: self.target_phase, - _permit: permit, + _permit_opt: permit_opt, } } } -pub(crate) struct CooperativeIndexingPeriod { +pub(crate) struct IndexingPeriod { // measured right before the acquisition of the indexing semaphore t_wake: Instant, // measured after the acquisition of the semaphore. t_work_start: Instant, commit_timeout: Duration, target_phase: Duration, - _permit: OwnedSemaphorePermit, + _permit_opt: Option, } -impl CooperativeIndexingPeriod { +impl IndexingPeriod { fn compute_pipeline_metrics( &self, end: Instant, @@ -248,16 +251,15 @@ mod tests { let target_phase = Duration::from_secs(target_phase_secs); let semaphore = Arc::new(Semaphore::new(1)); tokio::time::sleep(Duration::from_secs(start_time_secs)).await; - let cooperative_indexing = CooperativeIndexingCycle::new_with_phase( + let indexing_cycle = IndexingCycle::new_with_phase( target_phase, Duration::from_secs(30), - semaphore.clone(), + Some(semaphore.clone()), ); - let initial_sleep_duration: Duration = - cooperative_indexing.initial_sleep_duration(); + let initial_sleep_duration: Duration = indexing_cycle.initial_sleep_duration(); tokio::time::sleep(initial_sleep_duration).await; - let target_phase_millis = cooperative_indexing.target_phase.as_millis() as i64; - let commit_timeout_ms = cooperative_indexing.commit_timeout.as_millis() as i64; + let target_phase_millis = indexing_cycle.target_phase.as_millis() as i64; + let commit_timeout_ms = indexing_cycle.commit_timeout.as_millis() as i64; let phase_millis = (t0.elapsed().as_millis() as i64 - target_phase_millis) % commit_timeout_ms; assert!(phase_millis >= -100, "{phase_millis}"); @@ -270,9 +272,9 @@ mod tests { async fn test_cooperative_indexing_simple() { tokio::time::pause(); let semaphore = Arc::new(Semaphore::new(1)); - let cooperative_indexing = - CooperativeIndexingCycle::new("id", Duration::from_secs(30), semaphore.clone()); - let guard = cooperative_indexing.cooperative_indexing_period().await; + let indexing_cycle = + IndexingCycle::new("id", Duration::from_secs(30), Some(semaphore.clone())); + let guard = indexing_cycle.indexing_period().await; tokio::time::advance(Duration::from_secs(10)).await; let (sleep_time, metrics) = guard.end_of_work(100_000_000); assert_approx_equal_sleep_time(sleep_time, Duration::from_secs(20)); @@ -294,11 +296,11 @@ mod tests { async fn test_cooperative_indexing_maximum_throughput() { tokio::time::pause(); let semaphore = Arc::new(Semaphore::new(1)); - let cooperative_indexing = - CooperativeIndexingCycle::new("id", Duration::from_secs(30), semaphore.clone()); + let indexing_cycle = + IndexingCycle::new("id", Duration::from_secs(30), Some(semaphore.clone())); let semaphore_guard = Semaphore::acquire_owned(semaphore).await; drop_after(semaphore_guard, Duration::from_secs(30)); - let cycle_guard = cooperative_indexing.cooperative_indexing_period().await; + let cycle_guard = indexing_cycle.indexing_period().await; tokio::time::advance(Duration::from_secs(15)).await; let (sleep_time, metrics) = cycle_guard.end_of_work(30_000_000); let expected_metrics = PipelineMetrics { @@ -313,11 +315,11 @@ mod tests { async fn test_cooperative_indexing_simple_contention() { tokio::time::pause(); let semaphore = Arc::new(Semaphore::new(1)); - let cooperative_indexing = - CooperativeIndexingCycle::new("id", Duration::from_secs(30), semaphore.clone()); + let indexing_cycle = + IndexingCycle::new("id", Duration::from_secs(30), Some(semaphore.clone())); let semaphore_guard = Semaphore::acquire_owned(semaphore).await; drop_after(semaphore_guard, Duration::from_secs(10)); - let cycle_guard = cooperative_indexing.cooperative_indexing_period().await; + let cycle_guard = indexing_cycle.indexing_period().await; tokio::time::advance(Duration::from_secs(10)).await; let (sleep_time, metrics) = cycle_guard.end_of_work(100_000_000); assert_approx_equal_sleep_time(sleep_time, Duration::from_secs(10)); @@ -341,15 +343,15 @@ mod tests { for i in 0..num_pipelines { let target_phase = Duration::from_millis(commit_timeout.as_millis() as u64 * i / num_pipelines); - let cooperative_indexing = CooperativeIndexingCycle::new_with_phase( + let indexing_cycle = IndexingCycle::new_with_phase( target_phase, commit_timeout, - semaphore.clone(), + Some(semaphore.clone()), ); let join_handle = tokio::task::spawn(async move { let mut last_phase = 0; for _ in 0..num_steps { - let cycle_guard = cooperative_indexing.cooperative_indexing_period().await; + let cycle_guard = indexing_cycle.indexing_period().await; let work_time = Duration::from_millis(10); tokio::time::sleep(work_time).await; last_phase = diff --git a/quickwit/quickwit-indexing/src/actors/indexer.rs b/quickwit/quickwit-indexing/src/actors/indexer.rs index c8412ce9bdf..42194fe84ba 100644 --- a/quickwit/quickwit-indexing/src/actors/indexer.rs +++ b/quickwit/quickwit-indexing/src/actors/indexer.rs @@ -51,7 +51,7 @@ use tracing::{Span, debug, info_span, warn}; use ulid::Ulid; use super::IndexSerializer; -use super::cooperative_indexing::{CooperativeIndexingCycle, CooperativeIndexingPeriod}; +use super::cooperative_indexing::{IndexingCycle, IndexingPeriod}; use crate::docs_clustering::{DocIdClusterer, Fingerprinter}; use crate::metrics::SPLIT_BUILDERS; use crate::models::{ @@ -84,7 +84,8 @@ pub struct IndexerCounters { pub num_doc_batches_in_workbench: u64, /// Metrics describing the load and indexing performance of the - /// pipeline. This is only updated for cooperative indexers. + /// pipeline. This is only updated for cooperative indexers or + /// when indexing pipeline spreading is enabled. pub pipeline_metrics_opt: Option, } @@ -100,7 +101,7 @@ struct IndexerState { tokenizer_manager: TokenizerManager, max_num_partitions: NonZeroU32, index_settings: IndexSettings, - cooperative_indexing_opt: Option, + indexing_cycle_opt: Option, } impl IndexerState { @@ -199,15 +200,11 @@ impl IndexerState { workbench_id=%workbench_id, ); let indexing_span = info_span!(parent: batch_parent_span.id(), "indexer"); - let cooperative_indexing_period = - if let Some(cooperative_indexing) = &self.cooperative_indexing_opt { - Some( - ctx.protect_future(cooperative_indexing.cooperative_indexing_period()) - .await, - ) - } else { - None - }; + let indexing_period_opt = if let Some(indexing_cycle) = &self.indexing_cycle_opt { + Some(ctx.protect_future(indexing_cycle.indexing_period()).await) + } else { + None + }; let last_delete_opstamp_request = LastDeleteOpstampRequest { index_uid: Some(self.pipeline_id.index_uid.clone()), @@ -239,7 +236,7 @@ impl IndexerState { publish_lock, last_delete_opstamp, memory_usage: GaugeGuard::new(&IN_FLIGHT_INDEX_WRITER, 0.0), - cooperative_indexing_period, + indexing_period_opt, split_builders_guard, }; Ok(workbench) @@ -366,7 +363,7 @@ struct IndexingWorkbench { // Number of bytes declared as used by tantivy. memory_usage: GaugeGuard, split_builders_guard: GaugeGuard, - cooperative_indexing_period: Option, + indexing_period_opt: Option, } pub struct Indexer { @@ -402,8 +399,8 @@ impl Actor for Indexer { } async fn initialize(&mut self, ctx: &ActorContext) -> Result<(), ActorExitStatus> { - if let Some(cooperative_indexing_cycle) = &self.indexer_state.cooperative_indexing_opt { - let initial_sleep_duration = cooperative_indexing_cycle.initial_sleep_duration(); + if let Some(indexing_cycle) = &self.indexer_state.indexing_cycle_opt { + let initial_sleep_duration = indexing_cycle.initial_sleep_duration(); ctx.pause(); ctx.schedule_self_msg(initial_sleep_duration, Command::Resume); } @@ -418,9 +415,7 @@ impl Actor for Indexer { return Ok(()); }; - let Some(cooperative_indexing_period) = - indexing_workbench.cooperative_indexing_period.take() - else { + let Some(indexing_period) = indexing_workbench.indexing_period_opt.take() else { return Ok(()); }; @@ -431,8 +426,7 @@ impl Actor for Indexer { .sum::(); // This also drops the indexing permit. - let (sleep_duration, pipeline_metrics) = - cooperative_indexing_period.end_of_work(uncompressed_num_bytes); + let (sleep_duration, pipeline_metrics) = indexing_period.end_of_work(uncompressed_num_bytes); self.counters.pipeline_metrics_opt = Some(pipeline_metrics); @@ -532,6 +526,7 @@ impl Indexer { indexing_directory: TempDirectory, indexing_settings: IndexingSettings, cooperative_indexing_permits_opt: Option>, + spread_indexing_pipelines: bool, index_serializer_mailbox: Mailbox, fingerprinter_opt: Option, ) -> Self { @@ -548,12 +543,12 @@ impl Indexer { // A configured fingerprinter supplies the mapping when the split is finalized. manual_doc_id_mapping: fingerprinter_opt.is_some(), }; - let cooperative_indexing_opt: Option = - cooperative_indexing_permits_opt.map(|cooperative_indexing_permits| { - CooperativeIndexingCycle::new( + let indexing_cycle_opt: Option = + (spread_indexing_pipelines || cooperative_indexing_permits_opt.is_some()).then(|| { + IndexingCycle::new( &pipeline_id, indexing_settings.commit_timeout(), - cooperative_indexing_permits, + cooperative_indexing_permits_opt, ) }); Self { @@ -569,7 +564,7 @@ impl Indexer { tokenizer_manager: tokenizer_manager.tantivy_manager().clone(), index_settings, max_num_partitions: doc_mapper.max_num_partitions(), - cooperative_indexing_opt, + indexing_cycle_opt, }, index_serializer_mailbox, indexing_workbench_opt: None, @@ -778,6 +773,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -919,6 +915,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -997,6 +994,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1081,6 +1079,7 @@ mod tests { indexing_directory, indexing_settings, Some(Arc::new(Semaphore::new(1))), + false, index_serializer_mailbox, None, ); @@ -1170,6 +1169,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1238,6 +1238,7 @@ mod tests { TempDirectory::for_test(), IndexingSettings::for_test(), None, + false, index_serializer_mailbox, Some(Fingerprinter::new( &serde_json::from_value::(serde_json::json!([ @@ -1358,6 +1359,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1457,6 +1459,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1530,6 +1533,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1604,6 +1608,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1670,6 +1675,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); @@ -1740,6 +1746,7 @@ mod tests { indexing_directory, indexing_settings, None, + false, index_serializer_mailbox, None, ); diff --git a/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs b/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs index b3d197bf000..5332bcd15c8 100644 --- a/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs +++ b/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs @@ -368,6 +368,7 @@ impl IndexingPipeline { self.params.indexing_directory.clone(), self.params.indexing_settings.clone(), self.params.cooperative_indexing_permits.clone(), + self.params.spread_indexing_pipelines, index_serializer_mailbox, self.params.fingerprinter_opt.clone(), ); @@ -553,6 +554,7 @@ pub struct IndexingPipelineParams { pub split_store: IndexingSplitStore, pub max_concurrent_split_uploads_index: usize, pub cooperative_indexing_permits: Option>, + pub spread_indexing_pipelines: bool, // Merge-related parameters pub merge_policy: Arc, @@ -697,6 +699,7 @@ mod tests { max_concurrent_split_uploads_index: 4, max_concurrent_split_uploads_merge: 5, cooperative_indexing_permits: None, + spread_indexing_pipelines: false, merge_planner_mailbox_opt: Some(merge_planner_mailbox), event_broker: EventBroker::default(), params_fingerprint: 42u64, @@ -804,6 +807,7 @@ mod tests { max_concurrent_split_uploads_index: 4, max_concurrent_split_uploads_merge: 5, cooperative_indexing_permits: None, + spread_indexing_pipelines: false, merge_planner_mailbox_opt: Some(merge_planner_mailbox), params_fingerprint: 42u64, event_broker: EventBroker::default(), @@ -934,6 +938,7 @@ mod tests { max_concurrent_split_uploads_index: 4, max_concurrent_split_uploads_merge: 5, cooperative_indexing_permits: None, + spread_indexing_pipelines: false, merge_planner_mailbox_opt: Some(merge_planner_mailbox), event_broker: Default::default(), params_fingerprint: 42u64, @@ -1035,6 +1040,7 @@ mod tests { max_concurrent_split_uploads_index: 4, max_concurrent_split_uploads_merge: 5, cooperative_indexing_permits: None, + spread_indexing_pipelines: false, merge_planner_mailbox_opt: Some(merge_planner_mailbox.clone()), event_broker: Default::default(), params_fingerprint: 42u64, @@ -1119,6 +1125,7 @@ mod tests { max_concurrent_split_uploads_index: 4, max_concurrent_split_uploads_merge: 5, cooperative_indexing_permits: None, + spread_indexing_pipelines: false, merge_planner_mailbox_opt: None, event_broker: Default::default(), params_fingerprint: 42u64, @@ -1273,6 +1280,7 @@ mod tests { max_concurrent_split_uploads_index: 4, max_concurrent_split_uploads_merge: 5, cooperative_indexing_permits: None, + spread_indexing_pipelines: false, merge_planner_mailbox_opt: Some(merge_planner_mailbox), params_fingerprint: 42u64, event_broker: Default::default(), diff --git a/quickwit/quickwit-indexing/src/actors/indexing_service.rs b/quickwit/quickwit-indexing/src/actors/indexing_service.rs index 2bcb5131946..871253e06b5 100644 --- a/quickwit/quickwit-indexing/src/actors/indexing_service.rs +++ b/quickwit/quickwit-indexing/src/actors/indexing_service.rs @@ -123,6 +123,7 @@ pub struct IndexingService { #[cfg(feature = "metrics")] parquet_merge_pipeline_handles: HashMap, cooperative_indexing_permits: Option>, + spread_indexing_pipelines: bool, fingerprinter_opt: Option, merge_io_throughput_limiter_opt: Option, pub(crate) event_broker: EventBroker, @@ -166,6 +167,7 @@ impl IndexingService { } else { None }; + let spread_indexing_pipelines = indexer_config.enable_spread_indexing_pipelines; Ok(IndexingService { node_id, indexing_root_directory, @@ -189,6 +191,7 @@ impl IndexingService { fingerprinter_opt, merge_io_throughput_limiter_opt, cooperative_indexing_permits, + spread_indexing_pipelines, event_broker, }) } @@ -420,6 +423,7 @@ impl IndexingService { split_store, max_concurrent_split_uploads_index, cooperative_indexing_permits: self.cooperative_indexing_permits.clone(), + spread_indexing_pipelines: self.spread_indexing_pipelines, merge_policy, retention_policy, max_concurrent_split_uploads_merge, From 0dfc0f429379adcab7bf5f49e870f443b0710ffb Mon Sep 17 00:00:00 2001 From: loutPhilipps Date: Mon, 3 Aug 2026 20:31:02 +0200 Subject: [PATCH 2/4] Update comments --- .../quickwit-config/src/node_config/mod.rs | 2 + .../src/actors/cooperative_indexing.rs | 47 ++++++++++--------- .../quickwit-indexing/src/actors/indexer.rs | 4 +- .../src/actors/indexing_pipeline.rs | 2 + 4 files changed, 31 insertions(+), 24 deletions(-) diff --git a/quickwit/quickwit-config/src/node_config/mod.rs b/quickwit/quickwit-config/src/node_config/mod.rs index 377e2610fa9..b52c4d22375 100644 --- a/quickwit/quickwit-config/src/node_config/mod.rs +++ b/quickwit/quickwit-config/src/node_config/mod.rs @@ -210,6 +210,8 @@ pub struct IndexerConfig { pub enable_otlp_endpoint: bool, #[serde(default = "IndexerConfig::default_enable_cooperative_indexing")] pub enable_cooperative_indexing: bool, + /// Spreads the indexing pipelines of a node uniformly in time, so that they don't all use + /// the same resources at the same time. Implied by `enable_cooperative_indexing`. #[serde(default = "IndexerConfig::default_enable_spread_indexing_pipelines")] pub enable_spread_indexing_pipelines: bool, #[serde(default = "IndexerConfig::default_cpu_capacity")] diff --git a/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs b/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs index e1cb4dcb8c1..c424443b564 100644 --- a/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs +++ b/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs @@ -27,20 +27,22 @@ const NUDGE_TOLERANCE: Duration = Duration::from_secs(5); // Origin of time. It is used to compute the phase of the pipeline. static ORIGIN_OF_TIME: LazyLock = LazyLock::new(Instant::now); -/// Cooperative indexing is a mechanism to deal with a large amount of pipelines. +/// The indexing cycle is a mechanism to deal with a large amount of pipelines. /// -/// Instead of having all pipelines index concurrently, cooperative indexing: -/// - have them take turn, making sure that at most only N pipelines are indexing at the same time. -/// This has the benefit is reducing RAM using (by having a limited number of `IndexWriter` at the -/// same time), reducing context switching. -/// - keeps the different pipelines work uniformously spread in time. If the system is not at -/// capacity, we prefer to have the indexing pipeline as desynchronized as possible to make sure -/// they don't all use the same resources (disk/cpu/network) at the same time. +/// Instead of having all pipelines index concurrently, the cycle keeps the different +/// pipelines' work uniformly spread in time. If the system is not +/// at capacity, we prefer to have the indexing pipelines as desynchronized as possible +/// to make sure they don't all use the same resources (disk/cpu/network) at the +/// same time. +/// +/// Optionally, it is possible to have them take turns, making sure that at most N pipelines are +/// indexing at the same time. This has the benefit of reducing RAM usage (by having a limited +/// number of `IndexWriter` at the same time), reducing context switching. /// /// It works by: -/// - a semaphore is used to restrict the number of pipelines indexing at the same time. -/// - in the indexer when `on_drain` is called, the indexer will cut a split and "go to sleep" for a -/// given amount of time. +/// - an optional semaphore is used to restrict the number of pipelines indexing at the same time. +/// - in the indexer when `on_drained_messages` is called, the indexer will cut a split and "go to +/// sleep" for a given amount of time. /// /// The key logic is in the computation of that sleep time. /// @@ -56,20 +58,20 @@ static ORIGIN_OF_TIME: LazyLock = LazyLock::new(Instant::now); /// /// Each period of this cycle is divided into three phases. /// - waking [t_wake..t_work_start) acquisition of the period guard (this is instantaneous) -/// acquisition of the semaphore +/// acquisition of the semaphore if any /// - working [t_work_start..t_work_end) /// - sleeping [t=t_work_end..t_sleep_end) /// -/// The idea is to first pick the sleep time to to create a cycle of period +/// The idea is to first pick the sleep time to create a cycle of period /// `commit_timeout`. -/// sleep_time := max(0, commit_timeout - (t_workend - t_wake)) +/// sleep_time := max(0, commit_timeout - (t_work_end - t_wake)) /// /// If the work phase is too long, the regular commit timeout mechanism -/// kicks in an the pipeline will create a split without waiting for the +/// kicks in and the pipeline will create a split without waiting for the /// mailbox to be drained. /// /// We then allow ourselves to tweak the sleep time one way or another by at -/// most two seconds to eventually nudge the system toward the desired phase. +/// most `NUDGE_TOLERANCE` to eventually nudge the system toward the desired phase. pub(crate) struct IndexingCycle { target_phase: Duration, commit_timeout: Duration, @@ -77,7 +79,7 @@ pub(crate) struct IndexingCycle { } impl IndexingCycle { - /// Creates a new cooperative indexing cycle object. + /// Creates a new indexing cycle. /// `phase_id` is hashed to compute the target phase. pub fn new( phase_id: &(impl Hash + ?Sized), @@ -100,7 +102,7 @@ impl IndexingCycle { commit_timeout: Duration, indexing_permits_opt: Option>, ) -> IndexingCycle { - // Force the initial of the origin of time. + // Force the initialization of the origin of time. let _t0 = *ORIGIN_OF_TIME; IndexingCycle { target_phase, @@ -145,9 +147,10 @@ impl IndexingCycle { } pub(crate) struct IndexingPeriod { - // measured right before the acquisition of the indexing semaphore + // measured right before the acquisition of the indexing semaphore, if any t_wake: Instant, - // measured after the acquisition of the semaphore. + // measured after the acquisition of the semaphore, if any. Equal to `t_wake` when the cycle + // has no semaphore. t_work_start: Instant, commit_timeout: Duration, target_phase: Duration, @@ -200,8 +203,8 @@ impl IndexingPeriod { } } - /// This drops the indexing permit, allowing another indexer to start indexing. - /// This function also returns the amount of time to sleep until the next period. + /// If an indexing permit is being held, it is released here, allowing another indexer to start + /// indexing. This function also returns the amount of time to sleep until the next period. pub fn end_of_work(self, uncompressed_num_bytes: u64) -> (Duration, PipelineMetrics) { let end = Instant::now(); let sleep_duration = self.compute_sleep_duration(end); diff --git a/quickwit/quickwit-indexing/src/actors/indexer.rs b/quickwit/quickwit-indexing/src/actors/indexer.rs index 42194fe84ba..d162478d09f 100644 --- a/quickwit/quickwit-indexing/src/actors/indexer.rs +++ b/quickwit/quickwit-indexing/src/actors/indexer.rs @@ -287,7 +287,7 @@ impl IndexerState { .get_or_create_workbench(indexing_workbench_opt, ctx) .await?; if publish_lock.is_dead() { - // Release indexing permit early. + // Release indexing permit early if there is one. indexing_workbench_opt.take(); return Ok(()); } @@ -425,7 +425,7 @@ impl Actor for Indexer { .map(|split| split.split_attrs.uncompressed_docs_size_in_bytes) .sum::(); - // This also drops the indexing permit. + // This also drops the indexing permit, if any. let (sleep_duration, pipeline_metrics) = indexing_period.end_of_work(uncompressed_num_bytes); self.counters.pipeline_metrics_opt = Some(pipeline_metrics); diff --git a/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs b/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs index 5332bcd15c8..9b06a738bb0 100644 --- a/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs +++ b/quickwit/quickwit-indexing/src/actors/indexing_pipeline.rs @@ -554,6 +554,8 @@ pub struct IndexingPipelineParams { pub split_store: IndexingSplitStore, pub max_concurrent_split_uploads_index: usize, pub cooperative_indexing_permits: Option>, + /// Spreads the indexing pipelines of a node uniformly in time. Implied by cooperative + /// indexing. pub spread_indexing_pipelines: bool, // Merge-related parameters From 9402a36f9aee60c762567bb00f96ee0807113e3e Mon Sep 17 00:00:00 2001 From: loutPhilipps Date: Mon, 3 Aug 2026 20:31:45 +0200 Subject: [PATCH 3/4] Rename file --- quickwit/quickwit-indexing/src/actors/indexer.rs | 5 +++-- .../actors/{cooperative_indexing.rs => indexing_cycle.rs} | 0 quickwit/quickwit-indexing/src/actors/mod.rs | 2 +- 3 files changed, 4 insertions(+), 3 deletions(-) rename quickwit/quickwit-indexing/src/actors/{cooperative_indexing.rs => indexing_cycle.rs} (100%) diff --git a/quickwit/quickwit-indexing/src/actors/indexer.rs b/quickwit/quickwit-indexing/src/actors/indexer.rs index d162478d09f..87970566a8c 100644 --- a/quickwit/quickwit-indexing/src/actors/indexer.rs +++ b/quickwit/quickwit-indexing/src/actors/indexer.rs @@ -51,7 +51,7 @@ use tracing::{Span, debug, info_span, warn}; use ulid::Ulid; use super::IndexSerializer; -use super::cooperative_indexing::{IndexingCycle, IndexingPeriod}; +use super::indexing_cycle::{IndexingCycle, IndexingPeriod}; use crate::docs_clustering::{DocIdClusterer, Fingerprinter}; use crate::metrics::SPLIT_BUILDERS; use crate::models::{ @@ -426,7 +426,8 @@ impl Actor for Indexer { .sum::(); // This also drops the indexing permit, if any. - let (sleep_duration, pipeline_metrics) = indexing_period.end_of_work(uncompressed_num_bytes); + let (sleep_duration, pipeline_metrics) = + indexing_period.end_of_work(uncompressed_num_bytes); self.counters.pipeline_metrics_opt = Some(pipeline_metrics); diff --git a/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs b/quickwit/quickwit-indexing/src/actors/indexing_cycle.rs similarity index 100% rename from quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs rename to quickwit/quickwit-indexing/src/actors/indexing_cycle.rs diff --git a/quickwit/quickwit-indexing/src/actors/mod.rs b/quickwit/quickwit-indexing/src/actors/mod.rs index fdab5df4bcc..4cc267438ad 100644 --- a/quickwit/quickwit-indexing/src/actors/mod.rs +++ b/quickwit/quickwit-indexing/src/actors/mod.rs @@ -12,10 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -mod cooperative_indexing; mod doc_processor; mod index_serializer; mod indexer; +mod indexing_cycle; mod indexing_pipeline; mod indexing_service; mod log_publisher_impl; From e84a52557e178584845c55c32aa85ceb9fca18e1 Mon Sep 17 00:00:00 2001 From: loutPhilipps Date: Mon, 3 Aug 2026 20:50:42 +0200 Subject: [PATCH 4/4] Update tests --- .../src/actors/indexing_cycle.rs | 24 +++++-------------- 1 file changed, 6 insertions(+), 18 deletions(-) diff --git a/quickwit/quickwit-indexing/src/actors/indexing_cycle.rs b/quickwit/quickwit-indexing/src/actors/indexing_cycle.rs index c424443b564..fb7aef1d3bc 100644 --- a/quickwit/quickwit-indexing/src/actors/indexing_cycle.rs +++ b/quickwit/quickwit-indexing/src/actors/indexing_cycle.rs @@ -252,13 +252,9 @@ mod tests { for target_phase_secs in [0, 1, 2, 5, 10, 15, 20, 25, 29, 30, 1_000] { for start_time_secs in [0, 1, 2, 5, 10, 15, 20, 25, 29, 30] { let target_phase = Duration::from_secs(target_phase_secs); - let semaphore = Arc::new(Semaphore::new(1)); tokio::time::sleep(Duration::from_secs(start_time_secs)).await; - let indexing_cycle = IndexingCycle::new_with_phase( - target_phase, - Duration::from_secs(30), - Some(semaphore.clone()), - ); + let indexing_cycle = + IndexingCycle::new_with_phase(target_phase, Duration::from_secs(30), None); let initial_sleep_duration: Duration = indexing_cycle.initial_sleep_duration(); tokio::time::sleep(initial_sleep_duration).await; let target_phase_millis = indexing_cycle.target_phase.as_millis() as i64; @@ -272,11 +268,9 @@ mod tests { } #[tokio::test] - async fn test_cooperative_indexing_simple() { + async fn test_indexing_cycle_simple() { tokio::time::pause(); - let semaphore = Arc::new(Semaphore::new(1)); - let indexing_cycle = - IndexingCycle::new("id", Duration::from_secs(30), Some(semaphore.clone())); + let indexing_cycle = IndexingCycle::new("id", Duration::from_secs(30), None); let guard = indexing_cycle.indexing_period().await; tokio::time::advance(Duration::from_secs(10)).await; let (sleep_time, metrics) = guard.end_of_work(100_000_000); @@ -334,23 +328,17 @@ mod tests { } #[tokio::test] - async fn test_cooperative_indexing_nudge_to_phase() { + async fn test_indexing_cycle_nudge_to_phase() { tokio::time::pause(); - let num_threads = 10; let num_pipelines = 100; let num_steps = 15; - let semaphore = Arc::new(Semaphore::new(num_threads)); let commit_timeout = Duration::from_secs(30); let t0 = Instant::now(); let mut handles = Vec::new(); for i in 0..num_pipelines { let target_phase = Duration::from_millis(commit_timeout.as_millis() as u64 * i / num_pipelines); - let indexing_cycle = IndexingCycle::new_with_phase( - target_phase, - commit_timeout, - Some(semaphore.clone()), - ); + let indexing_cycle = IndexingCycle::new_with_phase(target_phase, commit_timeout, None); let join_handle = tokio::task::spawn(async move { let mut last_phase = 0; for _ in 0..num_steps {