// Copyright 2016 TiKV Project Authors. Licensed under Apache-2.0. // Modifications Copyright 2022 PingCAP, Inc. //! Storage configuration. use std::{ borrow::ToOwned, convert::TryFrom, error::Error, sync::{ Arc, atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}, }, time::Duration, }; use file_system::{IoPriority, IoRateLimitMode, IoRateLimiter, IoType}; use kvengine::config::{ DEFAULT_HARD_REGION_MEM_USAGE_LIMIT_MB, DEFAULT_MAX_REGION_SPEED_LIMIT_MB_PER_SEC, DEFAULT_MIN_REGION_SPEED_LIMIT_MB_PER_SEC, DEFAULT_SOFT_REGION_MEM_USAGE_LIMIT_MB, }; use kvproto::kvrpcpb::ApiVersion; use online_config::OnlineConfig; use tidb_query_common::storage::scanner::{ DEFAULT_RANGE_SCAN_MAX_TIME_SLICE_US, ScanRescheduleOptions, }; use tikv_util::{ config::{self, AbsoluteOrPercentSize, ReadableDuration, ReadableSize}, sys::SysQuota, }; use crate::config::BLOCK_CACHE_RATE; pub const DEFAULT_DATA_DIR: &str = "./"; const DEFAULT_GC_RATIO_THRESHOLD: f64 = 1.1; const DEFAULT_MAX_KEY_SIZE: usize = 8 * 1024; pub const DEFAULT_RANGE_SCAN_RESCHEDULE_CHECK_KEYS: usize = 32; pub const DEFAULT_BATCH_GET_FAIRNESS_CHECKPOINT_KEYS: usize = 32; pub const DEFAULT_CLOUD_STORE_FAIR_CPU_CHECKPOINT_ENTRIES: usize = 512; const MIN_RANGE_SCAN_MAX_TIME_SLICE_US: u64 = 100; const MAX_RANGE_SCAN_MAX_TIME_SLICE_US: u64 = 100_000; const DEFAULT_SCHED_CONCURRENCY: usize = 1024 * 512; const MAX_SCHED_CONCURRENCY: usize = 2 * 1024 * 1024; // the minimal thread count of scheduler-worker-pool pub const SCHED_MIN_WORKER_POOL_SIZE: usize = 4; // According to "Little's law", assuming you can write 100MB per // second, and it takes about 100ms to process the write requests // on average, in that situation the writing bytes estimated 10MB, // here we use 100MB as default value for tolerate 1s latency. // NOTE: Enlarge the default value to 1024MB to tolerate larger throughput. const DEFAULT_SCHED_PENDING_WRITE_MB: u64 = 1024; const DEFAULT_RESERVED_SPACE_GB: u64 = 5; const DEFAULT_RESERVED_RAFT_SPACE_GB: u64 = 1; /// Percentage of total system memory used to derive the default soft & hard /// limits for flow-control store memory usage when no explicit limit is /// provided. pub const SOFT_STORE_MEM_LIMIT_RATE: f64 = 0.15; pub const HARD_STORE_MEM_LIMIT_RATE: f64 = 0.2; // In tests, we've observed 1.2M entries in the TxnStatusCache. We // conservatively set the limit to 5M entries in total. // As TxnStatusCache have 128 slots by default. We round it to 5.12M. // This consumes at most around 300MB memory theoretically, but usually it's // much less as it's hard to see the capacity being used up. const DEFAULT_TXN_STATUS_CACHE_CAPACITY: usize = 40_000 * 128; const DEFAULT_MAX_TXN_FILE_LOCKS_PER_REGION: usize = 32; const DEFAULT_ACTION_ON_INVALID_MAX_TS_UPDATE: &str = "panic"; #[cfg(any(test, feature = "testexport"))] const DEFAULT_EXTRA_CF_SCAN_MODE: &str = "both"; #[cfg(not(any(test, feature = "testexport")))] const DEFAULT_EXTRA_CF_SCAN_MODE: &str = "pruned"; /// A request-local snapshot of the online read reschedule intervals. /// /// Callers should take one snapshot when a request starts so an online update /// does not change checkpoint intervals partway through that request. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct ReadRescheduleOptions { /// Number of range-scan keys between caller-owned reschedule checks. pub range_scan_reschedule_check_keys: usize, /// Wall time a range-scan coroutine may run between yields on the legacy /// path, used when no cooperative safe-point context is active. pub range_scan_max_time_slice_us: u64, /// Number of BatchGet keys between fair-CPU checkpoints. pub batch_get_fairness_checkpoint_keys: usize, /// Number of physical iterator entries between cooperative checkpoints in /// bottom-level CloudStore scans and the storage-backed KV ScanLock path. /// Zero disables checkpoints in both paths. pub cloud_store_fair_cpu_checkpoint_entries: usize, } impl Default for ReadRescheduleOptions { fn default() -> Self { Self { range_scan_reschedule_check_keys: DEFAULT_RANGE_SCAN_RESCHEDULE_CHECK_KEYS, range_scan_max_time_slice_us: DEFAULT_RANGE_SCAN_MAX_TIME_SLICE_US, batch_get_fairness_checkpoint_keys: DEFAULT_BATCH_GET_FAIRNESS_CHECKPOINT_KEYS, cloud_store_fair_cpu_checkpoint_entries: DEFAULT_CLOUD_STORE_FAIR_CPU_CHECKPOINT_ENTRIES, } } } impl ReadRescheduleOptions { /// Returns the rescheduling policy a range scan should use. pub fn range_scan(&self) -> ScanRescheduleOptions { ScanRescheduleOptions { check_keys: self.range_scan_reschedule_check_keys, max_time_slice: Duration::from_micros(self.range_scan_max_time_slice_us), } } } /// Shared runtime state for the online read reschedule intervals. /// /// The atomics are shared by Storage, the coprocessor endpoint, and the online /// config manager. Hot read loops use [`Self::snapshot`] instead of loading /// these atomics on every iteration. #[derive(Clone, Debug)] pub struct ReadRescheduleConfig { range_scan_reschedule_check_keys: Arc, range_scan_max_time_slice_us: Arc, batch_get_fairness_checkpoint_keys: Arc, cloud_store_fair_cpu_checkpoint_entries: Arc, } impl ReadRescheduleConfig { /// Creates runtime state initialized from the persisted storage config. pub fn new(config: &Config) -> Self { Self::from_options(ReadRescheduleOptions { range_scan_reschedule_check_keys: config.range_scan_reschedule_check_keys, range_scan_max_time_slice_us: config.range_scan_max_time_slice_us, batch_get_fairness_checkpoint_keys: config.batch_get_fairness_checkpoint_keys, cloud_store_fair_cpu_checkpoint_entries: config.cloud_store_fair_cpu_checkpoint_entries, }) } fn from_options(options: ReadRescheduleOptions) -> Self { Self { range_scan_reschedule_check_keys: Arc::new(AtomicUsize::new( options.range_scan_reschedule_check_keys, )), range_scan_max_time_slice_us: Arc::new(AtomicU64::new( options.range_scan_max_time_slice_us, )), batch_get_fairness_checkpoint_keys: Arc::new(AtomicUsize::new( options.batch_get_fairness_checkpoint_keys, )), cloud_store_fair_cpu_checkpoint_entries: Arc::new(AtomicUsize::new( options.cloud_store_fair_cpu_checkpoint_entries, )), } } /// Returns a stable copy of all intervals for one read request. pub fn snapshot(&self) -> ReadRescheduleOptions { ReadRescheduleOptions { range_scan_reschedule_check_keys: self .range_scan_reschedule_check_keys .load(Ordering::Relaxed), range_scan_max_time_slice_us: self.range_scan_max_time_slice_us.load(Ordering::Relaxed), batch_get_fairness_checkpoint_keys: self .batch_get_fairness_checkpoint_keys .load(Ordering::Relaxed), cloud_store_fair_cpu_checkpoint_entries: self .cloud_store_fair_cpu_checkpoint_entries .load(Ordering::Relaxed), } } pub(crate) fn set_range_scan_reschedule_check_keys(&self, value: usize) { self.range_scan_reschedule_check_keys .store(value, Ordering::Relaxed); } pub(crate) fn set_range_scan_max_time_slice_us(&self, value: u64) { self.range_scan_max_time_slice_us .store(value, Ordering::Relaxed); } pub(crate) fn set_batch_get_fairness_checkpoint_keys(&self, value: usize) { self.batch_get_fairness_checkpoint_keys .store(value, Ordering::Relaxed); } pub(crate) fn set_cloud_store_fair_cpu_checkpoint_entries(&self, value: usize) { self.cloud_store_fair_cpu_checkpoint_entries .store(value, Ordering::Relaxed); } } impl Default for ReadRescheduleConfig { fn default() -> Self { Self::from_options(ReadRescheduleOptions::default()) } } /// Shared state for storage settings that can change online. #[derive(Clone, Debug)] pub struct StorageRuntimeConfig { read_reschedule: ReadRescheduleConfig, enable_get_in_read_pool: Arc, } impl StorageRuntimeConfig { /// Creates runtime state initialized from the persisted storage config. pub fn new(config: &Config) -> Self { Self { read_reschedule: ReadRescheduleConfig::new(config), enable_get_in_read_pool: Arc::new(AtomicBool::new(config.enable_get_in_read_pool)), } } /// Returns the shared online read-reschedule configuration. pub fn read_reschedule(&self) -> &ReadRescheduleConfig { &self.read_reschedule } /// Whether a transactional point get should run entirely on the read pool /// when keyspace fairness is disabled. pub fn enable_get_in_read_pool(&self) -> bool { self.enable_get_in_read_pool.load(Ordering::Relaxed) } pub(crate) fn set_enable_get_in_read_pool(&self, value: bool) { self.enable_get_in_read_pool.store(value, Ordering::Relaxed); } } impl Default for StorageRuntimeConfig { fn default() -> Self { Self::new(&Config::default()) } } #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum ExtraCfScanMode { Legacy = 0, Pruned = 1, Both = 2, } impl Default for ExtraCfScanMode { fn default() -> Self { if cfg!(any(test, feature = "testexport")) { Self::Both } else { Self::Pruned } } } impl ExtraCfScanMode { pub fn as_str(self) -> &'static str { match self { Self::Legacy => "legacy", Self::Pruned => "pruned", Self::Both => "both", } } fn both_allowed_by_build() -> bool { cfg!(any(test, feature = "testexport")) } } impl TryFrom<&str> for ExtraCfScanMode { type Error = Box; fn try_from(value: &str) -> Result { match value { "legacy" => Ok(Self::Legacy), "pruned" => Ok(Self::Pruned), "both" if Self::both_allowed_by_build() => Ok(Self::Both), "both" => Err( "storage.extra-cf-scan-mode=both is only allowed in test/testexport builds".into(), ), _ => { let allowed = if Self::both_allowed_by_build() { "\"legacy\", \"pruned\", or \"both\"" } else { "\"legacy\" or \"pruned\"" }; Err( format!("storage.extra-cf-scan-mode must be one of {allowed}, got {value}") .into(), ) } } } } impl TryFrom for ExtraCfScanMode { type Error = Box; fn try_from(value: String) -> Result { Self::try_from(value.as_str()) } } fn default_scheduler_worker_pool_size(cpu_quota: f64) -> usize { // For example, a 16-core TiDB X machine reserves one core for sidecars, // leaving a 15-core quota for TiKV. const RESERVED_CPU_CORES: f64 = 1.0; // TiDB X reserves one machine core for sidecars. Add it back before applying // the original floor-based worker calculation. (((cpu_quota + RESERVED_CPU_CORES) / 2.0).floor() as usize).clamp(1, 8) } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct Config { #[online_config(skip)] pub data_dir: String, // Replaced by `GcConfig.ratio_threshold`. Keep it for backward compatibility. #[online_config(skip)] pub gc_ratio_threshold: f64, #[online_config(skip)] pub max_key_size: usize, /// Number of range-scan keys between reschedule checks. Zero disables /// checkpoints owned by the range-scan caller. pub range_scan_reschedule_check_keys: usize, /// Wall time a range-scan coroutine may run between yields on the legacy /// path. It applies only when keyspace fairness is not installing a /// poll-scoped slice for the request. pub range_scan_max_time_slice_us: u64, /// Number of BatchGet keys between fair-CPU checkpoints. Zero disables /// checkpoints owned by the BatchGet loop. pub batch_get_fairness_checkpoint_keys: usize, /// Number of physical iterator entries between cooperative checkpoints in /// bottom-level CloudStore scans and the storage-backed KV ScanLock path. /// Zero disables checkpoints in both paths. pub cloud_store_fair_cpu_checkpoint_entries: usize, #[online_config(skip)] pub scheduler_concurrency: usize, pub scheduler_worker_pool_size: usize, #[online_config(skip)] pub scheduler_low_priority_worker_pool_size: Option, #[online_config(skip)] pub scheduler_background_worker_pool_size: usize, #[online_config(skip)] pub scheduler_pending_write_threshold: ReadableSize, #[online_config(skip)] pub use_separated_scheduler_pool: bool, #[online_config(skip)] pub scheduler_use_tokio: bool, /// Runs the entire storage-side transactional KvGet task on the read pool. /// Keyspace fairness takes precedence and always uses the fair read pool. pub enable_get_in_read_pool: bool, #[online_config(skip)] // Reserve disk space to make tikv would have enough space to compact when disk is full. pub reserve_space: ReadableSize, #[online_config(skip)] pub reserve_raft_space: ReadableSize, #[online_config(skip)] pub low_space_threshold: AbsoluteOrPercentSize, #[online_config(skip)] pub enable_async_apply_prewrite: bool, /// Interval to batch async region-level resolve-lock requests. /// /// Setting this to 0 disables the async resolve-lock path. Requests marked /// with `is_async` then fall back to the ordinary region-level resolve /// path. pub scheduler_async_resolve_lock_batch_interval: ReadableDuration, pub max_txn_file_locks_per_region: usize, /// Tracks transactions against backup timestamps so backup barriers can /// wait for in-flight commit records. #[online_config(skip)] pub check_backup_ts: bool, #[online_config(skip)] pub api_version: u8, #[online_config(skip)] pub enable_ttl: bool, pub extra_cf_scan_mode: String, #[online_config(skip)] pub background_error_recovery_window: ReadableDuration, // Deprecated for nextgen. /// Interval to check TTL for all SSTs, pub ttl_check_poll_interval: ReadableDuration, #[online_config(skip)] pub txn_status_cache_capacity: usize, // TODO: next-gen's flow-control parameters (for kvengine) does not // support online config currently. #[online_config(submodule)] pub flow_control: FlowControlConfig, // TODO: Do not support online config currently. #[online_config(submodule)] pub block_cache: BlockCacheConfig, #[online_config(submodule)] pub io_rate_limit: IoRateLimitConfig, #[online_config(submodule)] pub max_ts: MaxTsConfig, } impl Default for Config { fn default() -> Config { let cpu_num = SysQuota::cpu_cores_quota(); Config { data_dir: DEFAULT_DATA_DIR.to_owned(), gc_ratio_threshold: DEFAULT_GC_RATIO_THRESHOLD, max_key_size: DEFAULT_MAX_KEY_SIZE, range_scan_reschedule_check_keys: DEFAULT_RANGE_SCAN_RESCHEDULE_CHECK_KEYS, range_scan_max_time_slice_us: DEFAULT_RANGE_SCAN_MAX_TIME_SLICE_US, batch_get_fairness_checkpoint_keys: DEFAULT_BATCH_GET_FAIRNESS_CHECKPOINT_KEYS, cloud_store_fair_cpu_checkpoint_entries: DEFAULT_CLOUD_STORE_FAIR_CPU_CHECKPOINT_ENTRIES, scheduler_concurrency: DEFAULT_SCHED_CONCURRENCY, scheduler_worker_pool_size: default_scheduler_worker_pool_size(cpu_num), scheduler_low_priority_worker_pool_size: None, scheduler_background_worker_pool_size: 1, scheduler_pending_write_threshold: ReadableSize::mb(DEFAULT_SCHED_PENDING_WRITE_MB), use_separated_scheduler_pool: true, scheduler_use_tokio: true, enable_get_in_read_pool: false, reserve_space: ReadableSize::gb(DEFAULT_RESERVED_SPACE_GB), reserve_raft_space: ReadableSize::gb(DEFAULT_RESERVED_RAFT_SPACE_GB), low_space_threshold: AbsoluteOrPercentSize::Percent(20.), enable_async_apply_prewrite: false, scheduler_async_resolve_lock_batch_interval: ReadableDuration::millis(500), max_txn_file_locks_per_region: DEFAULT_MAX_TXN_FILE_LOCKS_PER_REGION, check_backup_ts: true, api_version: 1, enable_ttl: false, extra_cf_scan_mode: DEFAULT_EXTRA_CF_SCAN_MODE.to_owned(), ttl_check_poll_interval: ReadableDuration::hours(12), txn_status_cache_capacity: DEFAULT_TXN_STATUS_CACHE_CAPACITY, flow_control: FlowControlConfig::default(), block_cache: BlockCacheConfig::default(), io_rate_limit: IoRateLimitConfig::default(), background_error_recovery_window: ReadableDuration::hours(1), max_ts: MaxTsConfig::default(), } } } impl Config { pub(crate) fn adjust_cpu_limited_worker_counts(&mut self, cpu_cores_quota: f64) { let limit = std::cmp::max(SCHED_MIN_WORKER_POOL_SIZE, cpu_cores_quota as usize); if self.scheduler_worker_pool_size > limit { warn!( "configured thread pool size exceeds its limit, adjusting to the limit"; "config" => "storage.scheduler-worker-pool-size", "configured" => self.scheduler_worker_pool_size, "limit" => limit, ); self.scheduler_worker_pool_size = limit; } if self.scheduler_background_worker_pool_size > limit { warn!( "configured thread pool size exceeds its limit, adjusting to the limit"; "config" => "storage.scheduler-background-worker-pool-size", "configured" => self.scheduler_background_worker_pool_size, "limit" => limit, ); self.scheduler_background_worker_pool_size = limit; } } pub fn validate(&mut self) -> Result<(), Box> { if self.data_dir != DEFAULT_DATA_DIR { self.data_dir = config::canonicalize_path(&self.data_dir)? } if self.scheduler_concurrency > MAX_SCHED_CONCURRENCY { warn!( "TiKV has optimized latch since v4.0, so it is not necessary to set large schedule \ concurrency. To save memory, change it from {:?} to {:?}", self.scheduler_concurrency, MAX_SCHED_CONCURRENCY ); self.scheduler_concurrency = MAX_SCHED_CONCURRENCY; } if !(MIN_RANGE_SCAN_MAX_TIME_SLICE_US..=MAX_RANGE_SCAN_MAX_TIME_SLICE_US) .contains(&self.range_scan_max_time_slice_us) { return Err(format!( "storage.range-scan-max-time-slice-us should be between {} and {}", MIN_RANGE_SCAN_MAX_TIME_SLICE_US, MAX_RANGE_SCAN_MAX_TIME_SLICE_US ) .into()); } if !matches!(self.api_version, 1 | 2) { return Err("storage.api_version can only be set to 1 or 2.".into()); } ExtraCfScanMode::try_from(self.extra_cf_scan_mode.as_str())?; if self.api_version == 2 && !self.enable_ttl { return Err( "storage.enable_ttl must be true in API V2 because API V2 forces to enable TTL." .into(), ); }; // max worker pool size should be at least 4. let max_pool_size = std::cmp::max( SCHED_MIN_WORKER_POOL_SIZE, SysQuota::cpu_cores_quota() as usize, ); if self.scheduler_worker_pool_size == 0 || self.scheduler_worker_pool_size > max_pool_size { return Err( format!( "storage.scheduler_worker_pool_size should be greater than 0 and less than or equal to {}", max_pool_size ).into() ); } if self.scheduler_low_priority_worker_pool_size.is_none() { let pool_size = std::cmp::max(1, self.scheduler_worker_pool_size / 4); self.scheduler_low_priority_worker_pool_size = Some(pool_size); } self.flow_control.validate()?; self.io_rate_limit.validate()?; if self.scheduler_background_worker_pool_size == 0 || self.scheduler_background_worker_pool_size > max_pool_size { return Err( format!( "storage.scheduler_background_worker_pool_size should be greater than 0 and less than or equal to {}", max_pool_size ).into() ); } self.max_ts.validate()?; Ok(()) } pub fn api_version(&self) -> ApiVersion { match self.api_version { 1 if self.enable_ttl => ApiVersion::V1ttl, 1 => ApiVersion::V1, 2 => ApiVersion::V2, _ => unreachable!(), } } pub fn set_api_version(&mut self, api_version: ApiVersion) { match api_version { ApiVersion::V1 => { self.api_version = 1; self.enable_ttl = false; } ApiVersion::V1ttl => { self.api_version = 1; self.enable_ttl = true; } ApiVersion::V2 => { self.api_version = 2; self.enable_ttl = true; } ApiVersion::V3 => { unreachable!("API V3 is not supported on the cloud-engine branch") } } } #[inline] pub fn extra_cf_scan_mode(&self) -> ExtraCfScanMode { ExtraCfScanMode::try_from(self.extra_cf_scan_mode.as_str()) .expect("storage.extra_cf_scan_mode should be validated before use") } } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct FlowControlConfig { // do not support online config currently for next-gen. #[online_config(skip)] pub enable: bool, #[online_config(skip)] pub soft_pending_compaction_bytes_limit: ReadableSize, #[online_config(skip)] pub hard_pending_compaction_bytes_limit: ReadableSize, #[online_config(skip)] pub memtables_threshold: u64, #[online_config(skip)] pub l0_files_threshold: u64, // Store level flow controls: #[online_config(skip)] pub soft_store_mem_limit: Option, #[online_config(skip)] pub hard_store_mem_limit: Option, // Region level flow controls: #[online_config(skip)] pub soft_region_mem_limit: ReadableSize, #[online_config(skip)] pub hard_region_mem_limit: ReadableSize, // Region level flow controls for L0 table size, defaults to the same as soft_region_mem_limit. #[online_config(skip)] pub soft_region_l0table_size_limit: Option, // Region level flow controls for L0 table size, defaults to the same as hard_region_mem_limit. #[online_config(skip)] pub hard_region_l0table_size_limit: Option, // Max speed limit of single region when size of memory usage reaches `soft_region_mem_limit`, // per second. #[online_config(skip)] pub max_region_speed_limit: ReadableSize, // Min speed limit of single region when size of memory usage reaches `hard_region_mem_limit`, // per second. #[online_config(skip)] pub min_region_speed_limit: ReadableSize, } impl Default for FlowControlConfig { fn default() -> FlowControlConfig { FlowControlConfig { enable: true, soft_pending_compaction_bytes_limit: ReadableSize::gb(192), hard_pending_compaction_bytes_limit: ReadableSize::gb(1024), memtables_threshold: 5, l0_files_threshold: 20, soft_store_mem_limit: None, hard_store_mem_limit: None, soft_region_mem_limit: ReadableSize::mb(DEFAULT_SOFT_REGION_MEM_USAGE_LIMIT_MB), hard_region_mem_limit: ReadableSize::mb(DEFAULT_HARD_REGION_MEM_USAGE_LIMIT_MB), soft_region_l0table_size_limit: None, hard_region_l0table_size_limit: None, max_region_speed_limit: ReadableSize::mb(DEFAULT_MAX_REGION_SPEED_LIMIT_MB_PER_SEC), min_region_speed_limit: ReadableSize::mb(DEFAULT_MIN_REGION_SPEED_LIMIT_MB_PER_SEC), } } } impl FlowControlConfig { pub fn validate(&mut self) -> Result<(), Box> { let total_mem = SysQuota::memory_limit_in_bytes() as f64; if self.soft_store_mem_limit.is_none() { let limit = (total_mem * SOFT_STORE_MEM_LIMIT_RATE) as u64; self.soft_store_mem_limit = Some(ReadableSize(limit)); } if self.hard_store_mem_limit.is_none() { let limit = (total_mem * HARD_STORE_MEM_LIMIT_RATE) as u64; self.hard_store_mem_limit = Some(ReadableSize(limit)); } if self.soft_store_mem_limit.unwrap() > self.hard_store_mem_limit.unwrap() { return Err( "storage.flow-control.soft-store-mem-limit should be less than or equal to storage.flow-control.hard-store-mem-limit." .into(), ); } if self.soft_region_mem_limit.0 > self.hard_region_mem_limit.0 { return Err( "storage.flow-control.soft-region-mem-limit should be less than or equal to storage.flow-control.hard-region-mem-limit." .into(), ); } if self.min_region_speed_limit.0 > self.max_region_speed_limit.0 { return Err( "storage.flow-control.min-region-speed-limit should be less than or equal to storage.flow-control.max-region-speed-limit." .into(), ); } Ok(()) } } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct BlockCacheConfig { #[online_config(skip)] pub shared: bool, pub capacity: Option, #[online_config(skip)] pub num_shard_bits: i32, #[online_config(skip)] pub strict_capacity_limit: bool, #[online_config(skip)] pub high_pri_pool_ratio: f64, #[online_config(skip)] pub memory_allocator: Option, } impl Default for BlockCacheConfig { fn default() -> BlockCacheConfig { BlockCacheConfig { shared: true, capacity: Some(Self::default_capacity()), num_shard_bits: 6, strict_capacity_limit: false, high_pri_pool_ratio: 0.8, memory_allocator: Some(String::from("nodump")), } } } impl BlockCacheConfig { pub fn default_capacity() -> ReadableSize { let total_mem = SysQuota::memory_limit_in_bytes(); ReadableSize(((total_mem as f64) * BLOCK_CACHE_RATE) as u64) } } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct IoRateLimitConfig { pub max_bytes_per_sec: ReadableSize, #[online_config(skip)] pub mode: IoRateLimitMode, /// When this flag is off, high-priority IOs are counted but not limited. /// Default set to false because the optimal throughput target provided /// by user might not be the maximum available bandwidth. For /// multi-tenancy use case, this flag should be turned on. #[online_config(skip)] pub strict: bool, pub foreground_read_priority: IoPriority, pub foreground_write_priority: IoPriority, pub flush_priority: IoPriority, pub level_zero_compaction_priority: IoPriority, pub compaction_priority: IoPriority, pub replication_priority: IoPriority, pub load_balance_priority: IoPriority, pub gc_priority: IoPriority, pub import_priority: IoPriority, pub export_priority: IoPriority, pub other_priority: IoPriority, } impl Default for IoRateLimitConfig { fn default() -> IoRateLimitConfig { IoRateLimitConfig { max_bytes_per_sec: ReadableSize::mb(0), mode: IoRateLimitMode::WriteOnly, strict: true, foreground_read_priority: IoPriority::High, foreground_write_priority: IoPriority::High, flush_priority: IoPriority::High, level_zero_compaction_priority: IoPriority::Medium, compaction_priority: IoPriority::Low, replication_priority: IoPriority::High, load_balance_priority: IoPriority::High, gc_priority: IoPriority::High, import_priority: IoPriority::Medium, export_priority: IoPriority::Medium, other_priority: IoPriority::High, } } } impl IoRateLimitConfig { pub fn build(&self, enable_statistics: bool) -> IoRateLimiter { let limiter = IoRateLimiter::new(self.mode, self.strict, enable_statistics); limiter.set_io_rate_limit(self.max_bytes_per_sec.0 as usize); limiter.set_io_priority(IoType::ForegroundRead, self.foreground_read_priority); limiter.set_io_priority(IoType::ForegroundWrite, self.foreground_write_priority); limiter.set_io_priority(IoType::Flush, self.flush_priority); limiter.set_io_priority( IoType::LevelZeroCompaction, self.level_zero_compaction_priority, ); limiter.set_io_priority(IoType::Compaction, self.compaction_priority); limiter.set_io_priority(IoType::Replication, self.replication_priority); limiter.set_io_priority(IoType::LoadBalance, self.load_balance_priority); limiter.set_io_priority(IoType::Gc, self.gc_priority); limiter.set_io_priority(IoType::Import, self.import_priority); limiter.set_io_priority(IoType::Export, self.export_priority); limiter.set_io_priority(IoType::Other, self.other_priority); limiter } fn validate(&mut self) -> Result<(), Box> { if self.other_priority != IoPriority::High { warn!( "Occasionally some critical IO operations are tagged as IOType::Other, \ e.g. IOs are fired from unmanaged threads, thread-local type storage exceeds \ capacity. To be on the safe side, change priority for IOType::Other from \ {:?} to {:?}", self.other_priority, IoPriority::High ); self.other_priority = IoPriority::High; } if self.gc_priority != self.foreground_write_priority { warn!( "GC writes are merged with foreground writes. To avoid priority inversion, change \ priority for IOType::Gc from {:?} to {:?}", self.gc_priority, self.foreground_write_priority, ); self.gc_priority = self.foreground_write_priority; } if self.mode != IoRateLimitMode::WriteOnly { return Err( "storage.io-rate-limit.mode other than write-only is not supported.".into(), ); } Ok(()) } } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct MaxTsConfig { /// Maximum max_ts deviation allowed from PD TSO pub max_drift: ReadableDuration, /// How often to refresh the max_ts limit from PD #[online_config(skip)] pub cache_sync_interval: ReadableDuration, pub action_on_invalid_update: String, } impl Default for MaxTsConfig { fn default() -> Self { Self { max_drift: ReadableDuration::secs(60), cache_sync_interval: ReadableDuration::secs(15), action_on_invalid_update: DEFAULT_ACTION_ON_INVALID_MAX_TS_UPDATE.to_owned(), } } } impl MaxTsConfig { fn validate(&mut self) -> Result<(), Box> { if self.max_drift <= self.cache_sync_interval { let msg = format!( "storage.max-ts.max-drift {:?} is smaller than or equal to storage.max-ts.cache-sync-interval {:?}", self.max_drift, self.cache_sync_interval, ); error!("{}", msg); return Err(msg.into()); } if let Err(e) = concurrency_manager::ActionOnInvalidMaxTs::try_from( self.action_on_invalid_update.as_str(), ) { error!( "storage.max-ts.action-on-invalid-update is set to an invalid value {}, \ change to action panic", self.action_on_invalid_update, ); return Err(e.into()); } Ok(()) } } #[cfg(test)] mod tests { use super::*; #[test] fn test_check_backup_ts_enabled_by_default() { assert!(Config::default().check_backup_ts); } #[test] fn test_validate_storage_config() { let mut cfg = Config::default(); assert_eq!(cfg.extra_cf_scan_mode, DEFAULT_EXTRA_CF_SCAN_MODE); assert_eq!(cfg.extra_cf_scan_mode(), ExtraCfScanMode::Both); cfg.validate().unwrap(); let cpu_cores_quota = SysQuota::cpu_cores_quota(); let max_pool_size = std::cmp::max(4, cpu_cores_quota as usize); cfg.scheduler_worker_pool_size = max_pool_size; cfg.validate().unwrap(); cfg.scheduler_worker_pool_size = 0; cfg.validate().unwrap_err(); let mut cfg = Config { scheduler_worker_pool_size: max_pool_size + 1, ..Default::default() }; cfg.validate().unwrap_err(); cfg.adjust_cpu_limited_worker_counts(cpu_cores_quota); cfg.validate().unwrap(); assert_eq!(cfg.scheduler_worker_pool_size, max_pool_size); let mut cfg = Config { scheduler_background_worker_pool_size: 0, ..Default::default() }; cfg.validate().unwrap_err(); let mut cfg = Config { scheduler_background_worker_pool_size: max_pool_size + 1, ..Default::default() }; cfg.validate().unwrap_err(); cfg.adjust_cpu_limited_worker_counts(cpu_cores_quota); cfg.validate().unwrap(); assert_eq!(cfg.scheduler_background_worker_pool_size, max_pool_size); cfg.scheduler_worker_pool_size = max_pool_size; cfg.extra_cf_scan_mode = "legacy".to_owned(); cfg.validate().unwrap(); cfg.scheduler_worker_pool_size = max_pool_size; cfg.extra_cf_scan_mode = "pruned".to_owned(); cfg.validate().unwrap(); cfg.scheduler_worker_pool_size = max_pool_size; cfg.extra_cf_scan_mode = "both".to_owned(); cfg.validate().unwrap(); cfg.extra_cf_scan_mode = "invalid".to_owned(); cfg.validate().unwrap_err(); } #[test] fn test_validate_range_scan_max_time_slice() { let mut cfg = Config::default(); assert_eq!( cfg.range_scan_max_time_slice_us, DEFAULT_RANGE_SCAN_MAX_TIME_SLICE_US ); for value in [ MIN_RANGE_SCAN_MAX_TIME_SLICE_US, MAX_RANGE_SCAN_MAX_TIME_SLICE_US, ] { cfg.range_scan_max_time_slice_us = value; cfg.validate().unwrap(); } for value in [ 0, MIN_RANGE_SCAN_MAX_TIME_SLICE_US - 1, MAX_RANGE_SCAN_MAX_TIME_SLICE_US + 1, ] { cfg.range_scan_max_time_slice_us = value; cfg.validate().unwrap_err(); } } #[test] fn test_read_reschedule_options_build_range_scan_policy() { let options = ReadRescheduleOptions { range_scan_reschedule_check_keys: 7, range_scan_max_time_slice_us: 1_500, ..Default::default() }; let range_scan = options.range_scan(); assert_eq!(range_scan.check_keys, 7); assert_eq!(range_scan.max_time_slice, Duration::from_micros(1_500)); } #[test] fn test_reserved_cpu_does_not_lower_scheduler_pool_size() { assert_eq!(default_scheduler_worker_pool_size(7.0), 4); assert_eq!(default_scheduler_worker_pool_size(15.0), 8); assert_eq!(default_scheduler_worker_pool_size(31.0), 8); assert_eq!(default_scheduler_worker_pool_size(8.1), 4); } }