// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0. // Modifications Copyright 2022 PingCAP, Inc. use std::{convert::TryFrom, path::Path, time::Duration}; use batch_system::Config as BatchSystemConfig; use fail::fail_point; use online_config::OnlineConfig; use raftstore::{ Result, coprocessor::config::{SPLIT_KEYS_PER_MB, SPLIT_SIZE_MB}, store::config::{MAX_RAFT_ENTRY_SIZE, SPLIT_REGION_MAX_KEYS_DEF}, }; use serde::{Deserialize, Deserializer, Serialize}; use serde_with::with_prefix; use tikv_util::{ box_err, config::{ReadableDuration, ReadableSize}, sys::{SysQuota, path_in_diff_mount_point}, warn, }; use time::Duration as TimeDuration; /// Maximum number of peer messages processed per scheduler tick. /// /// Background: /// - Derived from OSS TiKV's effective cap: `store-max-batch-size × /// messages-per-tick`. /// - In Next-Gen, `store-max-batch-size` no longer means "number of FSMs per /// tick"; we keep the same effective ceiling for compatibility. /// - Example (OSS defaults): `256 × 4096`. const MAX_BATCH_SIZE_PER_TICK: usize = 256 * 4096; const RFENGINE_WAL_BATCH_FRAMING_HEADROOM: ReadableSize = ReadableSize::mb(16); // An idle worker can overshoot its target by one entry or append message. The // main Raft worker and IO worker can each combine it with almost one more // target-sized batch. const MAX_RFENGINE_WAL_BATCH_DATA_SIZE: ReadableSize = ReadableSize(rfengine::WAL_BATCH_SIZE_LIMIT.0 - RFENGINE_WAL_BATCH_FRAMING_HEADROOM.0); fn max_raft_wal_batch_data_size( raft_entry_max_size: ReadableSize, raft_max_size_per_msg: ReadableSize, raft_worker_max_batch_size: ReadableSize, ) -> u64 { raft_worker_max_batch_size .0 .saturating_mul(3) .saturating_add(raft_entry_max_size.0.max(raft_max_size_per_msg.0)) } pub(crate) fn validate_raft_batch_sizes( raft_entry_max_size: ReadableSize, raft_max_size_per_msg: ReadableSize, raft_worker_max_batch_size: ReadableSize, ) -> std::result::Result<(), String> { if raft_max_size_per_msg.0 == 0 || raft_max_size_per_msg.0 > ReadableSize::gb(3).0 { return Err( "raft max size per message should be greater than 0 and less than or equal to 3GiB" .to_owned(), ); } if raft_entry_max_size.0 == 0 || raft_entry_max_size.0 > MAX_RAFT_ENTRY_SIZE.0 { return Err(format!( "raft entry max size should be greater than 0 and less than or equal to {}MiB", MAX_RAFT_ENTRY_SIZE.as_mb() )); } if max_raft_wal_batch_data_size( raft_entry_max_size, raft_max_size_per_msg, raft_worker_max_batch_size, ) > MAX_RFENGINE_WAL_BATCH_DATA_SIZE.0 { return Err(format!( "the larger of raft entry max size and raft max size per message plus three times raft worker max batch size should be less than or equal to {}MiB", MAX_RFENGINE_WAL_BATCH_DATA_SIZE.as_mb() )); } Ok(()) } const DISK_HANG_TIMEOUT_MIN: ReadableDuration = ReadableDuration::secs(30); /// For an average Raft message size of 1KB, 262,144 messages will use 256MB of /// memory. const NOTIFY_CAPACITY_DEF: usize = 262144; const RAFT_ENTRY_MAX_SIZE_DEFAULT: ReadableSize = ReadableSize::mb(16); // 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; fn default_aux_worker_count(cpu_quota: f64) -> usize { ((cpu_quota + RESERVED_CPU_CORES) / 8.0).floor() as usize } fn default_schema_worker_count(cpu_quota: f64) -> usize { (((cpu_quota + RESERVED_CPU_CORES) / 16.0).floor() as usize).max(1) } fn default_apply_batch_system(cpu_quota: f64) -> BatchSystemConfig { BatchSystemConfig { pool_size: (((cpu_quota + RESERVED_CPU_CORES) / 2.0).floor() as usize).max(2), ..Default::default() } } fn default_store_batch_system() -> BatchSystemConfig { BatchSystemConfig { max_batch_size: Some(1024), ..Default::default() } } #[derive(Default, Deserialize)] #[serde(default)] #[serde(rename_all = "kebab-case")] // Flattened fields are deserialized even when no matching prefixed key exists. // Keep overrides optional so omitted keys preserve rfstore's outer defaults. struct BatchSystemConfigOverrides { max_batch_size: Option, pool_size: Option, reschedule_duration: Option, } impl BatchSystemConfigOverrides { fn apply_to(self, mut config: BatchSystemConfig) -> BatchSystemConfig { // Keep this pattern exhaustive so adding a batch-system field requires // handling its override here as well. let BatchSystemConfig { max_batch_size, pool_size, reschedule_duration, } = &mut config; if let Some(value) = self.max_batch_size { *max_batch_size = Some(value); } if let Some(value) = self.pool_size { *pool_size = value; } if let Some(value) = self.reschedule_duration { *reschedule_duration = value; } config } } with_prefix!(prefix_apply "apply-"); with_prefix!(prefix_store "store-"); fn deserialize_apply_batch_system<'de, D>( deserializer: D, ) -> std::result::Result where D: Deserializer<'de>, { let overrides: BatchSystemConfigOverrides = prefix_apply::deserialize(deserializer)?; Ok(overrides.apply_to(default_apply_batch_system(SysQuota::cpu_cores_quota()))) } fn deserialize_store_batch_system<'de, D>( deserializer: D, ) -> std::result::Result where D: Deserializer<'de>, { let overrides: BatchSystemConfigOverrides = prefix_store::deserialize(deserializer)?; Ok(overrides.apply_to(default_store_batch_system())) } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct TxnFileLockActiveResolveConfig { /// Minimum lock age required to trigger active resolve. /// `0` disables active resolve. pub threshold: ReadableDuration, /// Maximum concurrent active-resolve tasks on this store. /// Must be `>= 1`. pub concurrency: usize, } impl Default for TxnFileLockActiveResolveConfig { fn default() -> Self { Self { threshold: ReadableDuration::secs(30), concurrency: 4, } } } #[derive(Clone, Copy, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "kebab-case")] pub enum ColdRegionPolicy { /// Do not allow peers to enter Cold. #[default] Disabled, /// Allow only followers to stop their regular Raft ticks. EnableFollower, /// Allow both leaders and followers to stop their regular Raft ticks. Enabled, } impl ColdRegionPolicy { /// Returns whether both leaders and followers may enter Cold. pub fn is_enabled(self) -> bool { self == Self::Enabled } pub(crate) fn allows_cold(self, is_leader: bool) -> bool { self.is_enabled() || (self == Self::EnableFollower && !is_leader) } } impl std::fmt::Display for ColdRegionPolicy { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::Disabled => f.write_str("disabled"), Self::EnableFollower => f.write_str("enable-follower"), Self::Enabled => f.write_str("enabled"), } } } impl From for online_config::ConfigValue { fn from(policy: ColdRegionPolicy) -> Self { Self::String(policy.to_string()) } } impl TryFrom for ColdRegionPolicy { type Error = String; fn try_from(value: online_config::ConfigValue) -> std::result::Result { let online_config::ConfigValue::String(value) = value else { return Err(format!("expected string cold-region policy, got {value:?}")); }; match value.as_str() { "disabled" => Ok(Self::Disabled), "enable-follower" => Ok(Self::EnableFollower), "enabled" => Ok(Self::Enabled), _ => Err(format!("invalid cold-region policy: {value}")), } } } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct ColdRegionConfig { /// Controls which Raft roles may enter Cold. pub policy: ColdRegionPolicy, /// The minimum age of shard max_ts before a region can enter Cold. #[online_config(skip)] pub max_ts_threshold: ReadableDuration, /// Required PD Store heartbeat age before cold recovery is allowed. #[online_config(skip)] pub store_down_grace_period: ReadableDuration, /// Nominal interval for a full cold-region sweep. #[online_config(skip)] pub region_heartbeat_interval: ReadableDuration, /// Nominal Raft heartbeat interval for cold leaders. #[online_config(skip)] pub leader_heartbeat_interval: ReadableDuration, } impl Default for ColdRegionConfig { fn default() -> Self { Self { policy: ColdRegionPolicy::Disabled, max_ts_threshold: ReadableDuration::hours(24), store_down_grace_period: ReadableDuration::minutes(10), region_heartbeat_interval: ReadableDuration::minutes(5), leader_heartbeat_interval: ReadableDuration::minutes(30), } } } #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, OnlineConfig)] #[serde(default)] #[serde(rename_all = "kebab-case")] pub struct Config { // ===================================================================== // Deprecated configs (for compatibility with Raftstore) // --------------------------------------------------------------------- // These fields are retained for legacy compatibility and migration. // They may be ignored or have no effect in next-gen implementations. // ===================================================================== // Interval to compact unnecessary raft log. #[online_config(skip)] pub raft_log_compact_sync_interval: ReadableDuration, // A threshold to gc stale raft log, must >= 1. #[online_config(skip)] pub raft_log_gc_threshold: u64, // Deprecated in next-gen. // When entry count exceed this value, gc will be forced trigger. #[online_config(skip)] pub raft_log_gc_count_limit: Option, // Old Raft logs could be reserved if `raft_log_gc_threshold` is not reached. // GC them after ticks `raft_log_reserve_max_ticks` times. #[doc(hidden)] #[online_config(hidden)] pub raft_log_reserve_max_ticks: usize, // Old logs in Raft engine needs to be purged periodically. #[online_config(skip)] pub raft_engine_purge_interval: ReadableDuration, /// If set, rfstore exits when a raft-disk probe cannot complete for this /// long. /// /// The fail-fast monitor always probes the sync WAL directory and also /// probes the rfengine data directory when it is on a different mount /// point. Set this to `0s` to disable disk hang fail-fast explicitly. /// Any non-zero value must be at least 30 seconds. /// /// This config is online-configurable. The monitor reads the latest value /// on each check/probe loop, so updates take effect without restart. pub disk_hang_timeout: Option, // When a peer is not responding for this time, leader will not keep entry cache for it. #[online_config(skip)] pub raft_entry_cache_life_time: ReadableDuration, // Deprecated! The configuration has no effect. // They are preserved for compatibility check. // When a peer is newly added, reject transferring leader to the peer for a while. #[doc(hidden)] #[serde(skip_serializing)] #[online_config(skip)] pub raft_reject_transfer_leader_duration: ReadableDuration, /// When size change of region exceed the diff since last check, it /// will be checked again whether it should be split. #[online_config(skip)] pub region_split_check_diff: Option, /// Interval (ms) to check whether start compaction for a region. #[online_config(skip)] pub region_compact_check_interval: ReadableDuration, /// Number of regions for each time checking. #[online_config(skip)] pub region_compact_check_step: u64, /// Minimum number of tombstones to trigger manual compaction. #[online_config(skip)] pub region_compact_min_tombstones: u64, /// Minimum percentage of tombstones to trigger manual compaction. #[online_config(skip)] /// Should between 1 and 100. #[online_config(hidden)] pub region_compact_tombstones_percent: u64, // used to periodically check whether schedule pending applies in region runner #[doc(hidden)] #[online_config(skip)] pub region_worker_tick_interval: ReadableDuration, #[online_config(hidden)] pub report_region_flow_interval: ReadableDuration, // Right region derive origin region id when split. #[online_config(hidden)] pub right_derive_when_split: bool, #[online_config(skip)] pub snap_apply_batch_size: ReadableSize, #[online_config(skip)] pub snap_gc_timeout: ReadableDuration, #[online_config(skip)] pub snap_generator_pool_size: usize, #[online_config(skip)] pub snap_mgr_gc_tick_interval: ReadableDuration, // used to periodically check whether we should delete a stale peer's range in // region runner #[doc(hidden)] #[online_config(skip)] pub clean_stale_ranges_tick: usize, // Interval (ms) to check region whether the data is consistent. #[online_config(skip)] pub consistency_check_interval: ReadableDuration, #[online_config(skip)] pub lock_cf_compact_interval: ReadableDuration, #[online_config(skip)] pub lock_cf_compact_bytes_threshold: ReadableSize, #[online_config(skip)] pub peer_stale_state_check_interval: ReadableDuration, // Interval of scheduling a tick to check the leader lease. // It will be set to raft_store_max_leader_lease/4 by default. #[online_config(skip)] pub check_leader_lease_interval: ReadableDuration, #[online_config(skip)] pub notify_capacity: usize, #[online_config(skip)] pub messages_per_tick: usize, /// Max log gap allowed to propose merge. #[online_config(hidden)] pub merge_max_log_gap: u64, /// Interval to re-propose merge. #[online_config(skip)] pub merge_check_tick_interval: ReadableDuration, #[online_config(hidden)] pub use_delete_range: bool, pub cleanup_import_sst_interval: ReadableDuration, /// If it is 0, it means io tasks are handled in store threads. #[online_config(skip)] pub store_io_pool_size: usize, #[online_config(skip)] pub future_poll_size: usize, #[online_config(hidden)] pub apply_yield_duration: ReadableDuration, /// yield the fsm when apply flushed data size exceeds this threshold. /// the yield is check after commit, so the actual handled messages can be /// bigger than the configed value. // NOTE: the default value is much smaller than the default max raft batch msg size(0.2 // * raft_entry_max_size), this is intentional because in the common case, a raft entry // is unlikely to exceed this threshold, but in case when raftstore is the bottleneck, // we still allow big raft batch for better throughput. #[online_config(skip)] pub apply_yield_write_size: ReadableSize, /// Whether to enable `hibernate` feature. /// /// Not implemented in Next-Gen. Use `true` by default for compatibility. #[online_config(skip)] pub hibernate_regions: bool, #[doc(hidden)] #[online_config(hidden)] pub dev_assert: bool, #[doc(hidden)] #[online_config(skip)] /// Disable this feature by set to 0, logic will be removed in other pr. /// When TiKV memory usage reaches `memory_usage_high_water` it will try to /// limit memory increasing. For raftstore layer entries will be evicted /// from entry cache, if they utilize memory more than /// `evict_cache_on_memory_ratio` * total. /// /// Set it to 0 can disable cache evict. // By default it's 0.2. So for different system memory capacity, cache evict happens: // * system=8G, memory_usage_limit=6G, evict=1.2G // * system=16G, memory_usage_limit=12G, evict=2.4G // * system=32G, memory_usage_limit=24G, evict=4.8G pub evict_cache_on_memory_ratio: f64, #[online_config(skip)] /// When the count of concurrent ready exceeds this value, command will not /// be proposed until the previous ready has been persisted. /// If `cmd_batch` is 0, this config will have no effect. /// If it is 0, it means no limit. pub cmd_batch_concurrent_ready_max_count: usize, #[online_config(skip)] /// When the size of raft db writebatch exceeds this value, write will be /// triggered. pub raft_write_size_limit: ReadableSize, #[online_config(skip)] pub waterfall_metrics: bool, #[online_config(skip)] pub io_reschedule_concurrent_max_count: usize, #[online_config(skip)] pub io_reschedule_hotpot_duration: ReadableDuration, // Deprecated. Batch is done in raft client. #[doc(hidden)] #[serde(skip_serializing)] #[online_config(skip)] pub raft_msg_flush_interval: ReadableDuration, #[doc(hidden)] #[serde(skip_serializing)] #[online_config(skip)] pub clean_stale_peer_delay: ReadableDuration, // TODO: resolved_ts is not support yet. // Interval to report min resolved ts, if it is zero, it means disabled. pub report_min_resolved_ts_interval: ReadableDuration, /// Interval to check whether to reactivate in-memory pessimistic lock after /// being disabled before transferring leader. pub reactive_memory_lock_tick_interval: ReadableDuration, /// Max tick count before reactivating in-memory pessimistic lock. pub reactive_memory_lock_timeout_tick: usize, // Interval of scheduling a tick to report region buckets. pub report_region_buckets_tick_interval: ReadableDuration, /// Interval to check long uncommitted proposals. #[doc(hidden)] #[online_config(skip)] pub check_long_uncommitted_interval: ReadableDuration, /// Base threshold of long uncommitted proposal. #[doc(hidden)] #[online_config(skip)] pub long_uncommitted_base_threshold: ReadableDuration, #[online_config(skip)] /// Max duration for the entry cache to be warmed up. /// Set it to 0 to disable warmup. pub max_entry_cache_warmup_duration: ReadableDuration, #[doc(hidden)] #[online_config(skip)] pub max_snapshot_file_raw_size: ReadableSize, #[online_config(skip)] pub unreachable_backoff: ReadableDuration, #[doc(hidden)] #[serde(skip_serializing)] #[online_config(hidden)] // Interval to check peers availability info. pub check_peers_availability_interval: ReadableDuration, // Deprecated. /// Maximum size of every local read task batch. #[online_config(hidden)] pub local_read_batch_size: u64, // Deprecated. #[online_config(hidden)] pub cmd_batch: bool, // Deprecated. /// Capacity of the internal channel for raftstore batch processing. /// Larger values allow more concurrent requests but increase memory usage. #[online_config(hidden)] pub channel_capacity: usize, // ===================================================================== // Valid configurations // --------------------------------------------------------------------- // These fields are actively used and maintained in the current engine. // They represent the main operational parameters for Raftstore. // ===================================================================== // minimizes disruption when a partitioned node rejoins the cluster by using a two phase // election. #[online_config(skip)] pub prevote: bool, #[online_config(skip)] pub raftdb_path: String, // store capacity. 0 means no limit. #[online_config(skip)] pub capacity: ReadableSize, // raft_base_tick_interval is a base tick interval (ms). #[online_config(hidden)] pub raft_base_tick_interval: ReadableDuration, #[online_config(hidden)] pub raft_heartbeat_ticks: usize, #[online_config(hidden)] pub raft_election_timeout_ticks: usize, #[online_config(hidden)] pub raft_min_election_timeout_ticks: usize, #[online_config(hidden)] pub raft_max_election_timeout_ticks: usize, pub raft_max_size_per_msg: ReadableSize, pub raft_max_inflight_msgs: usize, // When the entry exceed the max size, reject to propose it. pub raft_entry_max_size: ReadableSize, // When the approximate size of raft log entries exceed this value, // gc will be forced trigger. pub raft_log_gc_size_limit: Option, // Interval to gc unnecessary raft log. pub raft_log_gc_tick_interval: ReadableDuration, /// The maximum raft log numbers that applied_index can be ahead of /// persisted_index. pub max_apply_unpersisted_log_limit: u64, // Interval (ms) to check region whether need to be split or not. pub split_region_check_tick_interval: ReadableDuration, pub pd_heartbeat_tick_interval: ReadableDuration, pub pd_store_heartbeat_tick_interval: ReadableDuration, /// When a peer is not active for max_peer_down_duration, /// the peer is considered to be down and is reported to PD. pub max_peer_down_duration: ReadableDuration, /// If the leader of a peer is missing for longer than /// max_leader_missing_duration, the peer would ask pd to confirm /// whether it is valid in any region. If the peer is stale and is not /// valid in any region, it will destroy itself. pub max_leader_missing_duration: ReadableDuration, /// Similar to the max_leader_missing_duration, instead it will log warnings /// and try to alert monitoring systems, if there is any. pub abnormal_leader_missing_duration: ReadableDuration, /// Interval to check peer states with ver low frequency. /// Currently used for checking stale peer and retrigering failed /// compaction. pub peer_long_check_interval: ReadableDuration, #[online_config(hidden)] pub leader_transfer_max_log_lag: u64, // The lease provided by a successfully proposed and applied entry. pub raft_store_max_leader_lease: ReadableDuration, // Check if leader lease will expire at `current_time + renew_leader_lease_advance_duration`. // It will be set to raft_store_max_leader_lease/4 by default. // NOTE: `renew_leader_lease_advance_duration` is only used to determine the `max_lease` of // `Lease`, so it's actually not support OnlineConfig. #[online_config(skip)] pub renew_leader_lease_advance_duration: ReadableDuration, /// This setting can only ensure conf remove will not be proposed by the /// peer being removed. But it can't guarantee the remove is applied /// when the target is not leader. That means we always need to check if /// it's working as expected when a leader applies a self-remove conf /// change. Keep the configuration only for convenient test. pub allow_remove_leader: bool, #[online_config(submodule)] #[serde( flatten, serialize_with = "prefix_apply::serialize", deserialize_with = "deserialize_apply_batch_system" )] pub apply_batch_system: BatchSystemConfig, #[online_config(submodule)] #[serde( flatten, serialize_with = "prefix_store::serialize", deserialize_with = "deserialize_store_batch_system" )] pub store_batch_system: BatchSystemConfig, #[online_config(skip)] pub store_io_notify_capacity: usize, // Deprecated! These configuration has been moved to Coprocessor. // They are preserved for compatibility check. #[doc(hidden)] #[serde(skip_serializing)] #[online_config(skip)] pub region_max_size: ReadableSize, #[doc(hidden)] #[serde(skip_serializing)] #[online_config(skip)] pub region_split_size: ReadableSize, /// The minimal count of region pending on applying raft logs. /// Only when the count of regions which not pending on applying logs is /// less than the threshold, can the raftstore supply service. #[online_config(hidden)] pub min_pending_apply_region_count: u64, /// Time window after startup to check if the store is still busy /// applying raft logs before being marked as ready. #[online_config(hidden)] pub store_busy_apply_check_window: ReadableDuration, /// Number of peers per deterministic startup batch for stores with many /// local peers. /// /// The number of startup batches is derived from: /// `local_peer_count.div_ceil(startup_peer_per_batch)`. /// /// If the derived batch count is less than or equal to 1, batched startup /// is disabled. #[online_config(skip)] pub startup_peer_per_batch: usize, /// Minimum local peer count required to enable deterministic batched peer /// startup. #[online_config(skip)] pub startup_batch_region_threshold: u64, /// Minimum connected peer percentage required to advance a startup batch /// before all peers in the batch are connected. /// /// Should be between 1 and 100. Set to 100 to keep the previous "all /// connected or timeout" behavior. #[online_config(skip)] pub startup_batch_connected_threshold_percent: u64, #[online_config(hidden)] // Interval to inspect the latency of rfstore for slow store detection. pub inspect_interval: ReadableDuration, #[online_config(hidden)] // Interval to inspect the latency of flushes on kvdb for slow store detection. // If the kvdb uses the same mount path with raftdb, the default value will be // optimized to `0` to avoid duplicated inspection. pub inspect_kvdb_interval: ReadableDuration, #[online_config(hidden)] // Interval to inspect the network latency between PD and tikv for slow store detection. pub inspect_network_interval: ReadableDuration, #[doc(hidden)] #[online_config(hidden)] /// Threshold of CPU utilization to inspect for slow store detection. pub inspect_cpu_util_thd: f64, // ===================================================================== // Extra configs for Next-gen // --------------------------------------------------------------------- // These fields are exclusive to the next-generation engine features. // They enable advanced optimizations and new capabilities. // ===================================================================== // cloud engine configs pub local_file_gc_tick_interval: ReadableDuration, pub local_file_gc_timeout: ReadableDuration, /// Number of auxiliary worker threads for background tasks. /// Used to offload non-critical work and improve throughput. pub aux_worker_count: usize, /// Number of schema worker threads for schema-related background tasks. /// Used for DDL, schema changes, and metadata management in next-gen /// engine. #[online_config(skip)] pub schema_worker_count: usize, /// Maximum CPU utilization threshold for auxiliary workers. /// Lower than main worker to ensure synchronization and avoid contention. pub aux_worker_max_util: usize, /// Maximum CPU utilization threshold for main raft worker before offloading /// to aux workers. When exceeded, auxiliary workers are activated to /// help process tasks. pub main_worker_max_util: usize, /// The maximum batch size for raft worker in bytes. /// Controls how much data is processed in each raft worker batch for /// efficiency. pub raft_worker_max_batch_size: ReadableSize, /// The minimum duration for IO worker to write data (ms). /// Used to avoid too frequent writes and reduce write amplification in /// next-gen engine. pub io_worker_min_write_duration: ReadableDuration, /// Enable inner key offset optimization for next-gen engine. /// Used for advanced key encoding and fast lookups. #[doc(hidden)] #[online_config(skip)] pub enable_inner_key_offset: bool, /// Enable region bucket feature for fine-grained region management. /// When true, regions are split into buckets for parallelism and isolation. pub enable_region_bucket: bool, /// Size of each region bucket in bytes. /// Used only when enable_region_bucket is true. pub region_bucket_size: ReadableSize, /// Number of split keys to use when splitting regions in next-gen engine. /// Controls granularity of region splits for balancing and compaction. pub region_split_keys: u64, // Maximum number of split_keys in a split region request. pub split_region_max_keys: usize, // Interval (ms) to do maintenance task like // - switch mem-table // - shrink lock_cache // - compaction to gc tombstone pub maintenance_tick_interval: ReadableDuration, /// Interval of the leader-side tick that scans for file-level GC candidate /// SSTs (tombstone-heavy or old-version-heavy) and enqueues a file-level GC /// compaction. `0s` disables the tick. pub file_gc_compaction_check_tick_interval: ReadableDuration, /// Interval (ms) to update GC safe point (aka update_safe_ts_interval). /// Controls how frequently the system advances the global GC timestamp for /// MVCC. #[serde(alias = "update_safe_ts_interval")] pub update_gc_safe_point_interval: ReadableDuration, /// When mem-table is empty and applied to last index, /// if no kv raft log entries exceeds this value, gc will be triggered. pub raft_log_gc_no_kv_count: u64, pub peer_idle_duration: ReadableDuration, // Deprecated! This configuration no longer has any effect. // It is retained so old configuration files continue to be accepted. #[doc(hidden)] #[serde(skip_serializing)] #[online_config(skip)] pub ia_kv_size_discount: f64, pub idle_worker_tick_slow: bool, // Enable kv engine meta persist by diff. Default is false. // NOTE: Not support change to disable once enabled. pub enable_kv_engine_meta_diff: bool, // The percent of kv engine meta diff to rewrite. Default is 50. pub kv_engine_meta_diff_rewrite_percent: usize, // When the storage size of a whole keyspace is smaller than this value, skip reporting. #[serde(alias = "skip-store-size-metrics-threshold")] pub skip_storage_size_metrics_threshold: ReadableSize, #[online_config(submodule)] pub cold_region: ColdRegionConfig, #[online_config(submodule)] pub txn_file_lock_active_resolve: TxnFileLockActiveResolveConfig, } impl Default for Config { fn default() -> Config { let cpu_quota = SysQuota::cpu_cores_quota(); Config { // Deprecated configs [for compatibility to Raftstore] raft_log_compact_sync_interval: ReadableDuration::secs(2), raft_log_gc_threshold: 50, raft_log_gc_count_limit: None, raft_log_reserve_max_ticks: 6, raft_engine_purge_interval: ReadableDuration::secs(10), disk_hang_timeout: Some(ReadableDuration::minutes(1)), raft_entry_cache_life_time: ReadableDuration::secs(30), raft_reject_transfer_leader_duration: ReadableDuration::secs(3), region_split_check_diff: None, region_compact_check_interval: ReadableDuration::minutes(5), region_compact_check_step: 100, region_compact_min_tombstones: 10000, region_compact_tombstones_percent: 30, region_worker_tick_interval: if cfg!(any(test, feature = "testexport")) { ReadableDuration::millis(200) } else { ReadableDuration::millis(1000) }, report_region_flow_interval: ReadableDuration::minutes(1), right_derive_when_split: true, snap_apply_batch_size: ReadableSize::mb(10), snap_gc_timeout: ReadableDuration::hours(4), snap_generator_pool_size: 2, snap_mgr_gc_tick_interval: ReadableDuration::minutes(1), clean_stale_ranges_tick: if cfg!(any(test, feature = "testexport")) { 1 } else { 10 }, // Disable consistency check by default as it will hurt performance. // We should turn on this only in our tests. consistency_check_interval: ReadableDuration::secs(0), lock_cf_compact_interval: ReadableDuration::minutes(10), lock_cf_compact_bytes_threshold: ReadableSize::mb(256), peer_stale_state_check_interval: ReadableDuration::minutes(5), check_leader_lease_interval: ReadableDuration::secs(0), notify_capacity: NOTIFY_CAPACITY_DEF, messages_per_tick: 4096, merge_max_log_gap: 10, merge_check_tick_interval: ReadableDuration::secs(2), use_delete_range: false, cleanup_import_sst_interval: ReadableDuration::minutes(10), store_io_pool_size: 0, future_poll_size: 1, apply_yield_duration: ReadableDuration::millis(500), apply_yield_write_size: ReadableSize::kb(32), hibernate_regions: true, dev_assert: false, evict_cache_on_memory_ratio: 0.0, cmd_batch_concurrent_ready_max_count: 1, raft_write_size_limit: ReadableSize::mb(1), waterfall_metrics: true, io_reschedule_concurrent_max_count: 4, io_reschedule_hotpot_duration: ReadableDuration::secs(5), raft_msg_flush_interval: ReadableDuration::micros(250), clean_stale_peer_delay: ReadableDuration::minutes(0), report_min_resolved_ts_interval: ReadableDuration::secs(1), report_region_buckets_tick_interval: ReadableDuration::secs(10), max_snapshot_file_raw_size: ReadableSize::mb(100), unreachable_backoff: ReadableDuration::secs(10), check_peers_availability_interval: ReadableDuration::secs(30), reactive_memory_lock_tick_interval: ReadableDuration::secs(2), reactive_memory_lock_timeout_tick: 5, check_long_uncommitted_interval: ReadableDuration::secs(10), // In some cases, such as rolling upgrade, some regions' commit log // duration can be 12 seconds. Before #13078 is merged, // the commit log duration can be 2.8 minutes. So maybe // 20s is a relatively reasonable base threshold. Generally, // the log commit duration is less than 1s. Feel free to adjust // this config :) long_uncommitted_base_threshold: ReadableDuration::secs(20), max_entry_cache_warmup_duration: ReadableDuration::secs(1), // Valid configs prevote: true, raftdb_path: String::new(), capacity: ReadableSize(0), raft_base_tick_interval: ReadableDuration::secs(1), raft_heartbeat_ticks: 2, raft_election_timeout_ticks: 10, raft_min_election_timeout_ticks: 0, raft_max_election_timeout_ticks: 0, raft_max_size_per_msg: ReadableSize::mb(1), raft_max_inflight_msgs: 256, raft_entry_max_size: RAFT_ENTRY_MAX_SIZE_DEFAULT, raft_log_gc_size_limit: Some(ReadableSize::mb(32)), raft_log_gc_tick_interval: ReadableDuration::secs(3), // Enlarged to 16k. NextGen currently lacks entry cache eviction, so memory // usage can be higher than OP; pick a more conservative default. max_apply_unpersisted_log_limit: 16384, split_region_check_tick_interval: ReadableDuration::secs(3), txn_file_lock_active_resolve: TxnFileLockActiveResolveConfig::default(), pd_heartbeat_tick_interval: ReadableDuration::minutes(1), pd_store_heartbeat_tick_interval: ReadableDuration::secs(10), max_peer_down_duration: ReadableDuration::minutes(10), max_leader_missing_duration: ReadableDuration::hours(2), abnormal_leader_missing_duration: ReadableDuration::minutes(10), peer_long_check_interval: ReadableDuration::minutes(5), leader_transfer_max_log_lag: 512, raft_store_max_leader_lease: ReadableDuration::secs(9), renew_leader_lease_advance_duration: ReadableDuration::secs(0), allow_remove_leader: false, local_read_batch_size: 1024, apply_batch_system: default_apply_batch_system(cpu_quota), store_batch_system: default_store_batch_system(), store_io_notify_capacity: 40960, cmd_batch: true, region_max_size: ReadableSize(0), region_split_size: ReadableSize::mb(SPLIT_SIZE_MB), inspect_interval: ReadableDuration::millis(100), inspect_kvdb_interval: ReadableDuration::millis(100), inspect_network_interval: ReadableDuration::millis(0), // disabled by default // Default `inspect_cpu_util_thd` is 0.6: when CPU utilization falls below // `inspect_cpu_util_thd`, delayed inspected messages are treated as a slow-store // signal. This keeps the false-positive rate low for typical workloads. inspect_cpu_util_thd: 0.6, // Extra configs for Next-gen local_file_gc_tick_interval: ReadableDuration::minutes(10), local_file_gc_timeout: ReadableDuration::minutes(30), aux_worker_count: default_aux_worker_count(cpu_quota), schema_worker_count: default_schema_worker_count(cpu_quota), // The CPU utilization determines how many aux worker we use. // When the main raft worker utilization exceeds this value, we use aux worker // to help. main_worker_max_util: 80, // The aux worker's max CPU utilization is lower because it must sync with main // worker. aux_worker_max_util: 60, raft_worker_max_batch_size: ReadableSize::mb(1), io_worker_min_write_duration: ReadableDuration::millis(1), channel_capacity: 40960, enable_inner_key_offset: false, enable_region_bucket: false, region_bucket_size: ReadableSize::mb(96), region_split_keys: SPLIT_SIZE_MB * SPLIT_KEYS_PER_MB, split_region_max_keys: SPLIT_REGION_MAX_KEYS_DEF, maintenance_tick_interval: ReadableDuration::minutes(1), file_gc_compaction_check_tick_interval: ReadableDuration::secs(10), update_gc_safe_point_interval: ReadableDuration::secs(60), raft_log_gc_no_kv_count: 4, cold_region: ColdRegionConfig::default(), min_pending_apply_region_count: 10, store_busy_apply_check_window: ReadableDuration::hours(3), startup_peer_per_batch: 10_000, startup_batch_region_threshold: 30_000, startup_batch_connected_threshold_percent: 90, peer_idle_duration: ReadableDuration::secs(180), ia_kv_size_discount: 0.0, idle_worker_tick_slow: true, enable_kv_engine_meta_diff: true, kv_engine_meta_diff_rewrite_percent: 20, skip_storage_size_metrics_threshold: ReadableSize::mb(50), } } } impl Config { pub(crate) fn validate_disk_hang_timeout(timeout: Option) -> Result<()> { if let Some(timeout) = timeout { if !timeout.is_zero() && timeout < DISK_HANG_TIMEOUT_MIN { return Err(box_err!( "raftstore.disk-hang-timeout must be at least {} ms", DISK_HANG_TIMEOUT_MIN.as_millis() )); } } Ok(()) } pub fn new() -> Config { Config::default() } pub fn raft_store_max_leader_lease(&self) -> TimeDuration { TimeDuration::try_from(self.raft_store_max_leader_lease.0).unwrap() } pub fn renew_leader_lease_advance_duration(&self) -> TimeDuration { TimeDuration::try_from(self.renew_leader_lease_advance_duration.0).unwrap() } pub fn raft_heartbeat_interval(&self) -> Duration { self.raft_base_tick_interval.0 * self.raft_heartbeat_ticks as u32 } pub fn raft_election_timeout(&self) -> Duration { self.raft_base_tick_interval.0 * self.raft_election_timeout_ticks as u32 } pub fn startup_batch_wait_timeout(&self) -> Duration { self.raft_election_timeout() * 2 } pub fn raft_log_gc_count_limit(&self) -> u64 { self.raft_log_gc_count_limit.unwrap() } pub fn raft_log_gc_size_limit(&self) -> ReadableSize { self.raft_log_gc_size_limit.unwrap() } pub fn region_split_check_diff(&self) -> ReadableSize { self.region_split_check_diff.unwrap() } #[cfg(any(test, feature = "testexport"))] pub fn allow_remove_leader(&self) -> bool { self.allow_remove_leader } #[cfg(not(any(test, feature = "testexport")))] pub fn allow_remove_leader(&self) -> bool { false } pub(crate) fn validate_cold_region(&self) -> Result<()> { if self.cold_region.policy == ColdRegionPolicy::Disabled { return Ok(()); } if self.split_region_check_tick_interval.as_millis() == 0 { return Err(box_err!( "raftstore.split-region-check-tick-interval should be greater than 0 when \ cold-region is enabled" )); } if self.cold_region.max_ts_threshold.as_millis() == 0 { return Err(box_err!( "raftstore.cold-region.max-ts-threshold should be greater than 0" )); } if self.cold_region.store_down_grace_period.as_millis() == 0 { return Err(box_err!( "raftstore.cold-region.store-down-grace-period should be greater than 0" )); } let region_heartbeat_interval = self.cold_region.region_heartbeat_interval.as_millis(); if region_heartbeat_interval == 0 || region_heartbeat_interval < self.raft_base_tick_interval.as_millis() { return Err(box_err!( "raftstore.cold-region.region-heartbeat-interval should be greater than \ or equal to raft-base-tick-interval" )); } let configured_leader_heartbeat_interval = self.cold_region.leader_heartbeat_interval.as_millis(); if configured_leader_heartbeat_interval == 0 { return Err(box_err!( "raftstore.cold-region.leader-heartbeat-interval should be greater than 0" )); } let leader_heartbeat_interval = configured_leader_heartbeat_interval .div_ceil(region_heartbeat_interval) * region_heartbeat_interval; if self.cold_region.policy.is_enabled() && leader_heartbeat_interval >= self.max_leader_missing_duration.as_millis() { return Err(box_err!( "cold leader heartbeat interval should be less than \ raftstore.max-leader-missing-duration" )); } Ok(()) } /// Adjusts worker counts that are bounded by the current CPU quota. pub fn adjust_cpu_limited_worker_counts(&mut self, cpu_cores_quota: f64) { let pool_size_limit = max_apply_pool_size(cpu_cores_quota); if self.apply_batch_system.pool_size > pool_size_limit { warn!( "configured thread pool size exceeds its limit, adjusting to the limit"; "config" => "raftstore.apply-pool-size", "configured" => self.apply_batch_system.pool_size, "limit" => pool_size_limit, ); self.apply_batch_system.pool_size = pool_size_limit; } if self.store_batch_system.pool_size > pool_size_limit { warn!( "configured thread pool size exceeds its limit, adjusting to the limit"; "config" => "raftstore.store-pool-size", "configured" => self.store_batch_system.pool_size, "limit" => pool_size_limit, ); self.store_batch_system.pool_size = pool_size_limit; } let aux_worker_limit = cpu_cores_quota.ceil() as usize; if self.aux_worker_count > aux_worker_limit { warn!( "configured worker count exceeds its limit, adjusting to the limit"; "config" => "raftstore.aux-worker-count", "configured" => self.aux_worker_count, "limit" => aux_worker_limit, ); self.aux_worker_count = aux_worker_limit; } } pub fn validate( &mut self, region_split_size: ReadableSize, region_split_keys: Option, enable_region_bucket: bool, region_bucket_size: ReadableSize, ) -> Result<()> { if self.raft_base_tick_interval.as_millis() == 0 { return Err(box_err!("raft base tick interval must greater than 0")); } if self.raft_heartbeat_ticks == 0 { return Err(box_err!("heartbeat tick must greater than 0")); } if self.raft_election_timeout_ticks != 10 { warn!( "Election timeout ticks needs to be same across all the cluster, \ otherwise it may lead to inconsistency." ); } if self.raft_election_timeout_ticks <= self.raft_heartbeat_ticks { return Err(box_err!( "election tick must be greater than heartbeat tick" )); } if self.raft_min_election_timeout_ticks == 0 { self.raft_min_election_timeout_ticks = self.raft_election_timeout_ticks; } if self.raft_max_election_timeout_ticks == 0 { self.raft_max_election_timeout_ticks = self.raft_election_timeout_ticks * 2; } if self.raft_min_election_timeout_ticks < self.raft_election_timeout_ticks || self.raft_min_election_timeout_ticks >= self.raft_max_election_timeout_ticks { return Err(box_err!( "invalid timeout range [{}, {}) for timeout {}", self.raft_min_election_timeout_ticks, self.raft_max_election_timeout_ticks, self.raft_election_timeout_ticks )); } // The adjustment of this value is related to the number of regions, usually // 16384 is already a large enough value if self.raft_max_inflight_msgs == 0 || self.raft_max_inflight_msgs > 16384 { return Err(box_err!( "raft max inflight msgs should be greater than 0 and less than or equal to 16384" )); } if let Err(error) = validate_raft_batch_sizes( self.raft_entry_max_size, self.raft_max_size_per_msg, self.raft_worker_max_batch_size, ) { return Err(box_err!(error)); } if self.txn_file_lock_active_resolve.concurrency == 0 { return Err(box_err!( "raftstore.txn-file-lock-active-resolve.concurrency should be >= 1" )); } self.validate_cold_region()?; Self::validate_disk_hang_timeout(self.disk_hang_timeout)?; let election_timeout = self.raft_base_tick_interval.as_millis() * self.raft_election_timeout_ticks as u64; let lease = self.raft_store_max_leader_lease.as_millis(); if election_timeout < lease { return Err(box_err!( "election timeout {} ms is less than lease {} ms", election_timeout, lease )); } let tick = self.raft_base_tick_interval.as_millis(); if lease > election_timeout - tick { return Err(box_err!( "lease {} ms should not be greater than election timeout {} ms - 1 tick({} ms)", lease, election_timeout, tick )); } let stale_state_check = self.peer_stale_state_check_interval.as_millis(); if stale_state_check < election_timeout * 2 { return Err(box_err!( "peer stale state check interval {} ms is less than election timeout x 2 {} ms", stale_state_check, election_timeout * 2 )); } if self.leader_transfer_max_log_lag < 10 { return Err(box_err!( "raftstore.leader-transfer-max-log-lag should be >= 10." )); } let abnormal_leader_missing = self.abnormal_leader_missing_duration.as_millis(); if abnormal_leader_missing < stale_state_check { return Err(box_err!( "abnormal leader missing {} ms is less than peer stale state check interval {} ms", abnormal_leader_missing, stale_state_check )); } let max_leader_missing = self.max_leader_missing_duration.as_millis(); if max_leader_missing < abnormal_leader_missing { return Err(box_err!( "max leader missing {} ms is less than abnormal leader missing {} ms", max_leader_missing, abnormal_leader_missing )); } if self.io_worker_min_write_duration.as_millis() > 1000 { return Err(box_err!( "io-worker-min-write-duration should be less than 1000ms, current value is {}ms", self.io_worker_min_write_duration.as_millis() )); } // As the nextgen apply thread does not involve IO anymore, we set the pool size // limit to MAX(2, CPU_CORES/2) to avoid creating useless threads. let limit = get_max_apply_pool_size(); if self.apply_batch_system.pool_size == 0 || self.apply_batch_system.pool_size > limit { return Err(box_err!( "total apply pool size (apply-pool-size) should be greater than \ 0 and less than or equal to: {}, current values: apply-pool-size={}", limit, self.apply_batch_system.pool_size )); } if let Some(size) = self.apply_batch_system.max_batch_size { if size == 0 || size > MAX_BATCH_SIZE_PER_TICK { return Err(box_err!( "apply-max-batch-size should be greater than 0 and less than or equal to {}", MAX_BATCH_SIZE_PER_TICK )); } } else { self.apply_batch_system.max_batch_size = Some(256); } if self.store_batch_system.pool_size == 0 || self.store_batch_system.pool_size > limit { return Err(box_err!( "store-pool-size should be greater than 0 and less than or equal to: {}", limit )); } if let Some(size) = self.store_batch_system.max_batch_size { if size == 0 || size > MAX_BATCH_SIZE_PER_TICK { return Err(box_err!( "store-max-batch-size should be greater than 0 and less than or equal to {}", MAX_BATCH_SIZE_PER_TICK )); } } else if self.hibernate_regions { self.store_batch_system.max_batch_size = Some(256); } else { self.store_batch_system.max_batch_size = Some(1024); } // Avoid hibernated peer being reported as down peer. if self.hibernate_regions { self.max_peer_down_duration = std::cmp::max( self.max_peer_down_duration, self.peer_stale_state_check_interval * 2, ); } if self.renew_leader_lease_advance_duration.as_millis() == 0 && self.hibernate_regions { self.renew_leader_lease_advance_duration = self.raft_store_max_leader_lease / 4; } #[cfg(not(any(test, feature = "testexport")))] if self.max_snapshot_file_raw_size.0 != 0 && self.max_snapshot_file_raw_size.as_mb() < 100 { return Err(box_err!( "max_snapshot_file_raw_size should be no less than 100MB." )); } if self.local_file_gc_timeout.as_millis() < self.local_file_gc_tick_interval.as_millis() { return Err(box_err!( "local file gc timeout should be no less than local_file_gc_tick_interval" )); } if self.local_file_gc_tick_interval.as_millis() < self.raft_base_tick_interval.as_millis() { return Err(box_err!( "local_file_gc_tick_interval should be at least than raft_base_tick_interval" )); } // `0` is the ticker's "disabled" sentinel; otherwise enforce the // `raft_base_tick_interval` floor so the tick can run. let file_gc_compaction_check_tick_interval_ms = self.file_gc_compaction_check_tick_interval.as_millis(); if file_gc_compaction_check_tick_interval_ms > 0 && file_gc_compaction_check_tick_interval_ms < self.raft_base_tick_interval.as_millis() { return Err(box_err!( "file_gc_compaction_check_tick_interval should be 0 (disabled) or at least \ raft_base_tick_interval" )); } if self.raft_log_gc_threshold < 1 { return Err(box_err!( "raft log gc threshold must >= 1, not {}", self.raft_log_gc_threshold )); } self.region_split_size = region_split_size; if let Some(split_keys) = region_split_keys { self.region_split_keys = split_keys; } self.enable_region_bucket = enable_region_bucket; self.region_bucket_size = region_bucket_size; match self.raft_log_gc_size_limit { Some(size_limit) => { if size_limit.0 == 0 { return Err(box_err!("raft log gc size limit should large than 0.")); } } None => self.raft_log_gc_size_limit = Some(region_split_size * 3 / 4), } match self.raft_log_gc_count_limit { Some(count_limit) => { if self.merge_max_log_gap >= count_limit { return Err(box_err!( "merge log gap {} should be less than log gc limit {}.", self.merge_max_log_gap, count_limit )); } } None => { // Assume the average size of entries is 1k. self.raft_log_gc_count_limit = Some(region_split_size * 3 / 4 / ReadableSize::kb(1)); } } match self.region_split_check_diff { Some(split_check_diff) => { if split_check_diff.0 == 0 { return Err(box_err!("region split check diff should large than 0.")); } } None => { self.region_split_check_diff = if !enable_region_bucket { Some(region_split_size / 16) } else { Some(ReadableSize(std::cmp::min( region_split_size.0 / 16, region_bucket_size.0, ))) } } } if self.min_pending_apply_region_count == 0 { return Err(box_err!( "min_pending_apply_region_count must be greater than 0" )); } if self.startup_peer_per_batch == 0 { return Err(box_err!("startup-peer-per-batch must be greater than 0")); } if !(1..=100).contains(&self.startup_batch_connected_threshold_percent) { return Err(box_err!( "startup_batch_connected_threshold_percent must be between 1 and 100" )); } // For tests only. if cfg!(debug_assertions) && self.raft_base_tick_interval.as_millis() < 100 { // It is a test config, adjust the fields not included in the old. if self.update_gc_safe_point_interval.0 > self.raft_base_tick_interval.0 * 60 { self.update_gc_safe_point_interval.0 = self.raft_base_tick_interval.0 * 60; } if self.maintenance_tick_interval.0 > self.raft_base_tick_interval.0 * 60 { self.maintenance_tick_interval.0 = self.raft_base_tick_interval.0 * 60; } if self.local_file_gc_timeout.0 > self.raft_base_tick_interval.0 * 20 * 30 { self.local_file_gc_timeout.0 = self.raft_base_tick_interval.0 * 20 * 30; // 30s, see `new_test_config`. self.local_file_gc_tick_interval.0 = self.raft_base_tick_interval.0 * 20 * 10; // 10s, see `new_test_config`. } // cover aux worker in test if self.aux_worker_count == 0 { self.aux_worker_count = 1; } self.main_worker_max_util = 16; self.aux_worker_max_util = 12; self.schema_worker_count = self.schema_worker_count.max(2); self.peer_idle_duration = ReadableDuration::secs(5); } let cpu_cores_quota = SysQuota::cpu_cores_quota().ceil() as usize; if self.aux_worker_count > cpu_cores_quota { return Err(box_err!( "aux_worker_count must be smaller or equal to {}", cpu_cores_quota )); } (|| { fail_point!( "rfstore_config_from_old_force_short_update_gc_safe_point_interval", |_| { self.update_gc_safe_point_interval = self.raft_base_tick_interval; } ) })(); Ok(()) } pub fn optimize_for( &mut self, rf_wal_dir: &str, rfdb_dir: &str, kvdb_dirs: &[impl AsRef], ) { // Optimize the inspector configurations. let kvengine_independent_with_rfengine = kvdb_dirs.iter().any(|dir| { dir.as_ref().to_str().is_none_or(|kvdb_dir| { path_in_diff_mount_point(rf_wal_dir, kvdb_dir) && path_in_diff_mount_point(rfdb_dir, kvdb_dir) }) }); self.tune_inspector_configs(kvengine_independent_with_rfengine); } /// Optimize the interval of different inspectors according to the /// configuration. fn tune_inspector_configs(&mut self, separated_raft_mount_path: bool) { // If the kvdb uses the same mount path with raftdb, the health status // of kvdb will be inspected by raftstore automatically. So it's not necessary // to inspect kvdb. if !separated_raft_mount_path { self.inspect_kvdb_interval = ReadableDuration::ZERO; } } // TODO pub fn write_into_metrics(&self) {} } fn max_apply_pool_size(cpu_cores_quota: f64) -> usize { let cpu_cores_quota = cpu_cores_quota.ceil() as usize; std::cmp::max(cpu_cores_quota.div_ceil(2), 2) } // the maximum thread count of apply pool size is MAX(2, ceil(CPU_CORES/2)). pub(crate) fn get_max_apply_pool_size() -> usize { max_apply_pool_size(SysQuota::cpu_cores_quota()) } #[cfg(test)] mod tests { use super::*; #[test] fn test_cold_region_policy_semantics() { assert!(!ColdRegionPolicy::Disabled.is_enabled()); assert!(!ColdRegionPolicy::Disabled.allows_cold(false)); assert!(!ColdRegionPolicy::Disabled.allows_cold(true)); assert!(!ColdRegionPolicy::EnableFollower.is_enabled()); assert!(ColdRegionPolicy::EnableFollower.allows_cold(false)); assert!(!ColdRegionPolicy::EnableFollower.allows_cold(true)); assert!(ColdRegionPolicy::Enabled.is_enabled()); assert!(ColdRegionPolicy::Enabled.allows_cold(false)); assert!(ColdRegionPolicy::Enabled.allows_cold(true)); } #[test] fn test_partial_config_preserves_batch_system_defaults() { let default_cfg = Config::default(); let cfg: Config = toml::from_str("aux-worker-count = 1").unwrap(); assert_eq!(cfg.apply_batch_system, default_cfg.apply_batch_system); assert_eq!(cfg.store_batch_system, default_cfg.store_batch_system); } #[test] fn test_raft_batch_size_limits() { let mut cfg = Config::new(); assert_eq!(cfg.raft_entry_max_size, ReadableSize::mb(16)); cfg.raft_entry_max_size = MAX_RAFT_ENTRY_SIZE; cfg.validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap(); cfg.raft_entry_max_size = ReadableSize(MAX_RAFT_ENTRY_SIZE.0 + 1); let err = cfg .validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap_err(); assert!(err.to_string().contains("less than or equal to 128MiB")); cfg.raft_entry_max_size = MAX_RAFT_ENTRY_SIZE; cfg.raft_worker_max_batch_size = ReadableSize::mb(37); cfg.validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap(); cfg.raft_max_size_per_msg = ReadableSize::mb(129); cfg.validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap(); cfg.raft_max_size_per_msg = ReadableSize::mb(130); assert!( cfg.validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap_err() .to_string() .contains("plus three times") ); cfg.raft_max_size_per_msg = ReadableSize::mb(1); cfg.raft_worker_max_batch_size = ReadableSize::mb(38); assert!( cfg.validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap_err() .to_string() .contains("plus three times") ); } #[test] fn test_reserved_cpu_does_not_lower_worker_defaults() { assert_eq!(default_apply_batch_system(7.0).pool_size, 4); assert_eq!(default_apply_batch_system(15.0).pool_size, 8); assert_eq!(default_apply_batch_system(31.0).pool_size, 16); assert_eq!(default_aux_worker_count(7.0), 1); assert_eq!(default_aux_worker_count(15.0), 2); assert_eq!(default_aux_worker_count(31.0), 4); assert_eq!(default_schema_worker_count(7.0), 1); assert_eq!(default_schema_worker_count(15.0), 1); assert_eq!(default_schema_worker_count(31.0), 2); } #[test] fn test_fractional_cpu_does_not_advance_worker_tiers() { assert_eq!(default_apply_batch_system(8.1).pool_size, 4); assert_eq!(default_aux_worker_count(8.1), 1); assert_eq!(default_schema_worker_count(16.1), 1); } #[test] fn test_serde_with_custom_config() { let content = r#" ### deprecated report-region-flow-interval = "55s" raft-log-gc-threshold = 128 ### derived from RaftStore raftdb-path = "tmp1_path" store-pool-size = 4 apply-pool-size = 3 store-max-batch-size = 128 apply-max-batch-size = 64 store-reschedule-duration = "9s" apply-reschedule-duration = "7s" ### extra config raft-log-gc-no-kv-count = 55 startup-peer-per-batch = 16 startup-batch-region-threshold = 20480 startup-batch-connected-threshold-percent = 95 [txn-file-lock-active-resolve] threshold = "45s" concurrency = 6 [cold-region] policy = "enable-follower" max-ts-threshold = "12h" store-down-grace-period = "2m" region-heartbeat-interval = "6m" leader-heartbeat-interval = "36m" "#; let cfg: Config = toml::from_str(content).unwrap(); assert_eq!(cfg.report_region_flow_interval, ReadableDuration::secs(55)); assert_eq!(cfg.raft_log_gc_threshold, 128); assert_eq!(cfg.raftdb_path, "tmp1_path"); assert_eq!(cfg.store_batch_system.pool_size, 4); assert_eq!(cfg.store_batch_system.max_batch_size.unwrap(), 128); assert_eq!( cfg.store_batch_system.reschedule_duration, ReadableDuration::secs(9) ); assert_eq!(cfg.apply_batch_system.pool_size, 3); assert_eq!(cfg.apply_batch_system.max_batch_size.unwrap(), 64); assert_eq!( cfg.apply_batch_system.reschedule_duration, ReadableDuration::secs(7) ); assert_eq!(cfg.raft_log_gc_no_kv_count, 55); assert_eq!(cfg.startup_peer_per_batch, 16); assert_eq!(cfg.startup_batch_region_threshold, 20_480); assert_eq!(cfg.startup_batch_connected_threshold_percent, 95); assert_eq!( cfg.txn_file_lock_active_resolve.threshold, ReadableDuration::secs(45) ); assert_eq!(cfg.txn_file_lock_active_resolve.concurrency, 6); assert_eq!(cfg.cold_region.policy, ColdRegionPolicy::EnableFollower); assert_eq!( cfg.cold_region.max_ts_threshold, ReadableDuration::hours(12) ); assert_eq!( cfg.cold_region.store_down_grace_period, ReadableDuration::minutes(2) ); assert_eq!( cfg.cold_region.region_heartbeat_interval, ReadableDuration::minutes(6) ); assert_eq!( cfg.cold_region.leader_heartbeat_interval, ReadableDuration::minutes(36) ); } #[test] fn test_validate_apply_pool_size() { let content = r#" apply-pool-size = 1 # Legacy config, kept for backward compatibility. It should be ignored. apply-low-priority-pool-size = 1 apply-max-batch-size = 2048 "#; let mut cfg: Config = toml::from_str(content).unwrap(); cfg.validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap(); assert_eq!(cfg.apply_batch_system.pool_size, 1); assert_eq!(cfg.apply_batch_system.max_batch_size.unwrap(), 2048); } #[test] fn test_adjust_cpu_limited_worker_counts() { let cpu_cores_quota = SysQuota::cpu_cores_quota(); let pool_size_limit = get_max_apply_pool_size(); let aux_worker_limit = cpu_cores_quota.ceil() as usize; let validate = |cfg: &mut Config| { cfg.validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) }; let mut cfg = Config::new(); cfg.apply_batch_system.pool_size = pool_size_limit + 1; cfg.store_batch_system.pool_size = pool_size_limit + 1; cfg.aux_worker_count = aux_worker_limit + 1; validate(&mut cfg).unwrap_err(); cfg.adjust_cpu_limited_worker_counts(cpu_cores_quota); validate(&mut cfg).unwrap(); assert_eq!(cfg.apply_batch_system.pool_size, pool_size_limit); assert_eq!(cfg.store_batch_system.pool_size, pool_size_limit); assert_eq!(cfg.aux_worker_count, aux_worker_limit); let adjusted = cfg.clone(); cfg.adjust_cpu_limited_worker_counts(cpu_cores_quota); assert_eq!(cfg, adjusted); let mut cfg = Config::new(); cfg.apply_batch_system.pool_size = 0; validate(&mut cfg).unwrap_err(); let mut cfg = Config::new(); cfg.store_batch_system.pool_size = 0; validate(&mut cfg).unwrap_err(); let mut cfg = Config::new(); cfg.aux_worker_count = 0; validate(&mut cfg).unwrap(); assert_eq!(cfg.aux_worker_count, 0); } #[test] fn test_serde_with_invalid_custom_config() { let content = r#" ### deprecated report_region_flow_interval = "55s" ### derived from RaftStore raftdb_path = "tmp1_path" store_pool_size = 4 ### extra config raft-log-gc-no-count = 55 "#; let default_cfg = Config::new(); let cfg: Config = toml::from_str(content).unwrap(); assert_eq!( cfg.report_region_flow_interval, default_cfg.report_region_flow_interval ); assert_eq!(cfg.raftdb_path, default_cfg.raftdb_path); assert_eq!( cfg.store_batch_system.pool_size, default_cfg.store_batch_system.pool_size ); assert_eq!( cfg.raft_log_gc_no_kv_count, default_cfg.raft_log_gc_no_kv_count ); } #[test] fn test_config_optimization() { let mut cfg = Config::new(); cfg.optimize_for("./", "./", &["./"]); assert_eq!(cfg.inspect_kvdb_interval, ReadableDuration::ZERO); cfg.inspect_kvdb_interval = ReadableDuration::secs(10); cfg.tune_inspector_configs(false); assert_eq!(cfg.inspect_kvdb_interval, ReadableDuration::ZERO); cfg = Config::new(); cfg.inspect_kvdb_interval = ReadableDuration::secs(1); cfg.tune_inspector_configs(false); assert_eq!(cfg.inspect_kvdb_interval, ReadableDuration::ZERO); cfg.tune_inspector_configs(true); assert_eq!(cfg.inspect_kvdb_interval, ReadableDuration::ZERO); cfg.inspect_kvdb_interval = ReadableDuration::secs(1); cfg.tune_inspector_configs(true); assert_eq!(cfg.inspect_kvdb_interval, ReadableDuration::secs(1)); cfg = Config::new(); cfg.inspect_kvdb_interval = ReadableDuration::millis(1); cfg.tune_inspector_configs(true); assert_eq!(cfg.inspect_kvdb_interval, ReadableDuration::millis(1)); } #[test] fn test_validate_rejects_zero_raft_base_tick_interval() { let mut cfg = Config::new(); cfg.raft_base_tick_interval = ReadableDuration::ZERO; let err = cfg .validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap_err(); assert!( err.to_string() .contains("raft base tick interval must greater than 0") ); } #[test] fn test_validate_rejects_zero_txn_file_lock_active_resolve_concurrency() { let mut cfg = Config::new(); cfg.txn_file_lock_active_resolve.concurrency = 0; let err = cfg .validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap_err(); assert!( err.to_string() .contains("raftstore.txn-file-lock-active-resolve.concurrency should be >= 1") ); } #[test] fn test_validate_rejects_invalid_startup_batch_connected_threshold_percent() { let mut cfg = Config::new(); cfg.startup_batch_connected_threshold_percent = 101; let err = cfg .validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap_err(); assert!( err.to_string() .contains("startup_batch_connected_threshold_percent must be between 1 and 100") ); } #[test] fn test_validate_disk_hang_timeout() { let mut cfg = Config::new(); cfg.disk_hang_timeout = Some(ReadableDuration::millis(999)); let err = cfg .validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap_err(); assert!( err.to_string() .contains("raftstore.disk-hang-timeout must be at least") ); let mut cfg = Config::new(); cfg.disk_hang_timeout = Some(ReadableDuration::secs(0)); cfg.validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap(); let mut cfg = Config::new(); cfg.disk_hang_timeout = Some(ReadableDuration::secs(29)); assert!( cfg.validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .is_err() ); let mut cfg = Config::new(); cfg.disk_hang_timeout = Some(ReadableDuration::secs(30)); cfg.validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap(); } #[test] fn test_validate_rejects_file_gc_compaction_check_tick_interval_below_raft_base() { let mut cfg = Config::new(); cfg.raft_base_tick_interval = ReadableDuration::secs(5); cfg.file_gc_compaction_check_tick_interval = ReadableDuration::secs(1); let err = cfg .validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap_err(); assert!( err.to_string().contains( "file_gc_compaction_check_tick_interval should be 0 (disabled) or at least raft_base_tick_interval" ), "unexpected error: {}", err ); } #[test] fn test_validate_rejects_invalid_cold_region_config() { let mut heartbeat_too_frequent = Config::new(); heartbeat_too_frequent.cold_region.policy = ColdRegionPolicy::Enabled; heartbeat_too_frequent.raft_base_tick_interval = ReadableDuration::secs(5); heartbeat_too_frequent.cold_region.region_heartbeat_interval = ReadableDuration::secs(1); let mut zero_down_grace = Config::new(); zero_down_grace.cold_region.policy = ColdRegionPolicy::Enabled; zero_down_grace.cold_region.store_down_grace_period = ReadableDuration::secs(0); let mut zero_leader_heartbeat = Config::new(); zero_leader_heartbeat.cold_region.policy = ColdRegionPolicy::Enabled; zero_leader_heartbeat.cold_region.leader_heartbeat_interval = ReadableDuration::secs(0); let mut split_check_disabled = Config::new(); split_check_disabled.cold_region.policy = ColdRegionPolicy::Enabled; split_check_disabled.split_region_check_tick_interval = ReadableDuration::secs(0); for (mut cfg, expected_error) in [ ( heartbeat_too_frequent, "cold-region.region-heartbeat-interval should be greater than", ), ( zero_down_grace, "cold-region.store-down-grace-period should be greater than 0", ), ( zero_leader_heartbeat, "cold-region.leader-heartbeat-interval should be greater than 0", ), ( split_check_disabled, "split-region-check-tick-interval should be greater than 0 when cold-region is \ enabled", ), ] { let err = cfg .validate( ReadableSize::mb(96), Some(960_000), false, ReadableSize::mb(96), ) .unwrap_err(); assert!( err.to_string().contains(expected_error), "unexpected error: {}", err ); } } } // TODO: online configurations for RfStoreConfig