From 454489fcc99d2e39fdf6861165cdb02b31661cca Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed, 3 Apr 2024 02:50:45 +0000 Subject: [PATCH] refactor(storage): refactor compact iter recreate stream (#15919) (#16110) Co-authored-by: Li0k --- src/common/src/config.rs | 9 ++++-- src/config/docs.md | 1 + src/config/example.toml | 1 + .../src/hummock/compactor/compaction_utils.rs | 7 ++--- .../src/hummock/compactor/compactor_runner.rs | 16 +++++------ src/storage/src/hummock/compactor/iterator.rs | 28 ++++++++++--------- src/storage/src/opts.rs | 3 ++ 7 files changed, 38 insertions(+), 27 deletions(-) diff --git a/src/common/src/config.rs b/src/common/src/config.rs index c10796c3fb3f3..29c5a488c52df 100644 --- a/src/common/src/config.rs +++ b/src/common/src/config.rs @@ -684,6 +684,7 @@ pub struct StorageConfig { #[serde(default = "default::storage::compactor_max_sst_key_count")] pub compactor_max_sst_key_count: u64, + // DEPRECATED: This config will be deprecated in the future version, use `storage.compactor_iter_max_io_retry_times` instead. #[serde(default = "default::storage::compact_iter_recreate_timeout_ms")] pub compact_iter_recreate_timeout_ms: u64, #[serde(default = "default::storage::compactor_max_sst_size")] @@ -694,12 +695,12 @@ pub struct StorageConfig { pub check_compaction_result: bool, #[serde(default = "default::storage::max_preload_io_retry_times")] pub max_preload_io_retry_times: usize, - #[serde(default = "default::storage::compactor_fast_max_compact_delete_ratio")] pub compactor_fast_max_compact_delete_ratio: u32, - #[serde(default = "default::storage::compactor_fast_max_compact_task_size")] pub compactor_fast_max_compact_task_size: u64, + #[serde(default = "default::storage::compactor_iter_max_io_retry_times")] + pub compactor_iter_max_io_retry_times: usize, #[serde(default, flatten)] #[config_doc(omitted)] @@ -1319,6 +1320,10 @@ pub mod default { 10 * 60 * 1000 } + pub fn compactor_iter_max_io_retry_times() -> usize { + 8 + } + pub fn compactor_max_sst_size() -> u64 { 512 * 1024 * 1024 // 512m } diff --git a/src/config/docs.md b/src/config/docs.md index f673ea0c186b4..d1295bf6f1989 100644 --- a/src/config/docs.md +++ b/src/config/docs.md @@ -100,6 +100,7 @@ This page is automatically generated by `./risedev generate-example-config` | compact_iter_recreate_timeout_ms | | 600000 | | compactor_fast_max_compact_delete_ratio | | 40 | | compactor_fast_max_compact_task_size | | 2147483648 | +| compactor_iter_max_io_retry_times | | 8 | | compactor_max_sst_key_count | | 2097152 | | compactor_max_sst_size | | 536870912 | | compactor_max_task_multiplier | Compactor calculates the maximum number of tasks that can be executed on the node based on worker_num and compactor_max_task_multiplier. max_pull_task_count = worker_num * compactor_max_task_multiplier | 2.5 | diff --git a/src/config/example.toml b/src/config/example.toml index da9ed4ea5c263..8b96eeef8e6b5 100644 --- a/src/config/example.toml +++ b/src/config/example.toml @@ -131,6 +131,7 @@ check_compaction_result = false max_preload_io_retry_times = 3 compactor_fast_max_compact_delete_ratio = 40 compactor_fast_max_compact_task_size = 2147483648 +compactor_iter_max_io_retry_times = 8 mem_table_spill_threshold = 4194304 [storage.cache.block_cache_eviction] diff --git a/src/storage/src/hummock/compactor/compaction_utils.rs b/src/storage/src/hummock/compactor/compaction_utils.rs index 9282b41c7726d..a0f0c1991ae2f 100644 --- a/src/storage/src/hummock/compactor/compaction_utils.rs +++ b/src/storage/src/hummock/compactor/compaction_utils.rs @@ -348,7 +348,6 @@ pub async fn check_compaction_result( let mut table_iters = Vec::new(); let mut del_iter = ForwardMergeRangeIterator::default(); - let compact_io_retry_time = context.storage_opts.compact_iter_recreate_timeout_ms; for level in &compact_task.input_ssts { if level.table_infos.is_empty() { continue; @@ -365,7 +364,7 @@ pub async fn check_compaction_result( KeyRange::inf(), context.sstable_store.clone(), Arc::new(TaskProgress::default()), - compact_io_retry_time, + context.storage_opts.compactor_iter_max_io_retry_times, )); } else { let mut stats = StoreLocalStatistic::default(); @@ -381,7 +380,7 @@ pub async fn check_compaction_result( KeyRange::inf(), context.sstable_store.clone(), Arc::new(TaskProgress::default()), - compact_io_retry_time, + context.storage_opts.compactor_iter_max_io_retry_times, )); } } @@ -401,7 +400,7 @@ pub async fn check_compaction_result( KeyRange::inf(), context.sstable_store.clone(), Arc::new(TaskProgress::default()), - compact_io_retry_time, + context.storage_opts.compactor_iter_max_io_retry_times, ); let right_iter = UserIterator::new( SkipWatermarkIterator::from_safe_epoch_watermarks(iter, &compact_task.table_watermarks), diff --git a/src/storage/src/hummock/compactor/compactor_runner.rs b/src/storage/src/hummock/compactor/compactor_runner.rs index 0bbe41e6649f8..c22d8eb5a30a9 100644 --- a/src/storage/src/hummock/compactor/compactor_runner.rs +++ b/src/storage/src/hummock/compactor/compactor_runner.rs @@ -154,12 +154,6 @@ impl CompactorRunner { task_progress: Arc, ) -> HummockResult> { let mut table_iters = Vec::new(); - let compact_io_retry_time = self - .compactor - .context - .storage_opts - .compact_iter_recreate_timeout_ms; - for level in &self.compact_task.input_ssts { if level.table_infos.is_empty() { continue; @@ -189,7 +183,10 @@ impl CompactorRunner { self.compactor.task_config.key_range.clone(), self.sstable_store.clone(), task_progress.clone(), - compact_io_retry_time, + self.compactor + .context + .storage_opts + .compactor_iter_max_io_retry_times, )); } else { for table_info in &level.table_infos { @@ -209,7 +206,10 @@ impl CompactorRunner { self.compactor.task_config.key_range.clone(), self.sstable_store.clone(), task_progress.clone(), - compact_io_retry_time, + self.compactor + .context + .storage_opts + .compactor_iter_max_io_retry_times, )); } } diff --git a/src/storage/src/hummock/compactor/iterator.rs b/src/storage/src/hummock/compactor/iterator.rs index 13fafe720be4c..f89d70a756486 100644 --- a/src/storage/src/hummock/compactor/iterator.rs +++ b/src/storage/src/hummock/compactor/iterator.rs @@ -55,8 +55,8 @@ pub struct SstableStreamIterator { sstable_info: SstableInfo, existing_table_ids: HashSet, task_progress: Arc, - io_retry_timeout_ms: u64, - create_time: Instant, + io_retry_times: usize, + max_io_retry_times: usize, } impl SstableStreamIterator { @@ -82,7 +82,7 @@ impl SstableStreamIterator { stats: &StoreLocalStatistic, task_progress: Arc, sstable_store: SstableStoreRef, - io_retry_timeout_ms: u64, + max_io_retry_times: usize, ) -> Self { Self { block_stream: None, @@ -92,10 +92,10 @@ impl SstableStreamIterator { stats_ptr: stats.remote_io_time.clone(), existing_table_ids, sstable_info, - create_time: Instant::now(), sstable_store, task_progress, - io_retry_timeout_ms, + io_retry_times: 0, + max_io_retry_times, } } @@ -178,13 +178,11 @@ impl SstableStreamIterator { } Ok(None) => break, Err(e) => { - if !e.is_object_error() - || self.create_time.elapsed().as_millis() as u64 - > self.io_retry_timeout_ms - { + if !e.is_object_error() || !self.need_recreate_io_stream() { return Err(e); } self.block_stream.take(); + self.io_retry_times += 1; fail_point!("create_stream_err"); } } @@ -250,6 +248,10 @@ impl SstableStreamIterator { self.sstable_info.table_ids ) } + + fn need_recreate_io_stream(&self) -> bool { + self.io_retry_times < self.max_io_retry_times + } } impl Drop for SstableStreamIterator { @@ -280,7 +282,7 @@ pub struct ConcatSstableIterator { stats: StoreLocalStatistic, task_progress: Arc, - io_retry_timeout_ms: u64, + max_io_retry_times: usize, } impl ConcatSstableIterator { @@ -293,7 +295,7 @@ impl ConcatSstableIterator { key_range: KeyRange, sstable_store: SstableStoreRef, task_progress: Arc, - io_retry_timeout_ms: u64, + max_io_retry_times: usize, ) -> Self { Self { key_range, @@ -304,7 +306,7 @@ impl ConcatSstableIterator { sstable_store, task_progress, stats: StoreLocalStatistic::default(), - io_retry_timeout_ms, + max_io_retry_times, } } @@ -405,7 +407,7 @@ impl ConcatSstableIterator { &self.stats, self.task_progress.clone(), self.sstable_store.clone(), - self.io_retry_timeout_ms, + self.max_io_retry_times, ); sstable_iter.seek(seek_key).await?; diff --git a/src/storage/src/opts.rs b/src/storage/src/opts.rs index 6a39cad111268..8e4f342bac9ed 100644 --- a/src/storage/src/opts.rs +++ b/src/storage/src/opts.rs @@ -68,6 +68,7 @@ pub struct StorageOpts { /// Capacity of sstable meta cache. pub compactor_memory_limit_mb: usize, /// compactor streaming iterator recreate timeout. + /// deprecated pub compact_iter_recreate_timeout_ms: u64, /// Number of SST ids fetched from meta per RPC pub sstable_id_remote_fetch_number: u32, @@ -77,6 +78,7 @@ pub struct StorageOpts { pub max_sub_compaction: u32, pub max_concurrent_compaction_task_number: u64, pub max_version_pinning_duration_sec: u64, + pub compactor_iter_max_io_retry_times: usize, pub data_file_cache_dir: String, pub data_file_cache_capacity_mb: usize, @@ -269,6 +271,7 @@ impl From<(&RwConfig, &SystemParamsReader, &StorageMemoryConfig)> for StorageOpt .storage .compactor_fast_max_compact_delete_ratio, compactor_fast_max_compact_task_size: c.storage.compactor_fast_max_compact_task_size, + compactor_iter_max_io_retry_times: c.storage.compactor_iter_max_io_retry_times, } } }