diff --git a/contrib/cloud-storage-engine b/contrib/cloud-storage-engine index eb32cf11d43..b49d94ae594 160000 --- a/contrib/cloud-storage-engine +++ b/contrib/cloud-storage-engine @@ -1 +1 @@ -Subproject commit eb32cf11d43a6bff9e51ecccb7f1f88862c70d70 +Subproject commit b49d94ae594914225390e3914f92025f1486d95b diff --git a/contrib/tiflash-columnar-hub/Cargo.lock b/contrib/tiflash-columnar-hub/Cargo.lock index 3cec2980c2d..94e8c7f1a3b 100644 --- a/contrib/tiflash-columnar-hub/Cargo.lock +++ b/contrib/tiflash-columnar-hub/Cargo.lock @@ -3803,7 +3803,7 @@ dependencies = [ [[package]] name = "kvproto" version = "0.0.2" -source = "git+https://github.com/pingcap/kvproto.git#1e6ff1c80df24a9ead2b4408133af23c09273718" +source = "git+https://github.com/pingcap/kvproto.git?rev=683dad8fa3689deb243f2ff8ab5847c97af53e38#683dad8fa3689deb243f2ff8ab5847c97af53e38" dependencies = [ "bytes", "futures 0.3.32", diff --git a/contrib/tiflash-columnar-hub/Cargo.toml b/contrib/tiflash-columnar-hub/Cargo.toml index 85f31932e02..6db56db4e6f 100644 --- a/contrib/tiflash-columnar-hub/Cargo.toml +++ b/contrib/tiflash-columnar-hub/Cargo.toml @@ -10,6 +10,7 @@ cloud_encryption = { git = "https://github.com/tidbcloud/cloud-storage-engine.gi keys = { git = "https://github.com/tidbcloud/cloud-storage-engine.git", branch = "cloud-engine" } kvengine = { git = "https://github.com/tidbcloud/cloud-storage-engine.git", branch = "cloud-engine" } kvenginepb = { git = "https://github.com/tidbcloud/cloud-storage-engine.git", branch = "cloud-engine" } +kvproto = { git = "https://github.com/pingcap/kvproto.git", rev = "683dad8fa3689deb243f2ff8ab5847c97af53e38" } pd_client = { git = "https://github.com/tidbcloud/cloud-storage-engine.git", branch = "cloud-engine", default-features = false } prometheus = { version = "=0.13.0", features = ["nightly", "push"], default-features = true } security = { git = "https://github.com/tidbcloud/cloud-storage-engine.git", branch = "cloud-engine", default-features = false } @@ -28,7 +29,6 @@ lindera = { git = "https://github.com/breezewish/lindera", branch = "v0.43.1-tok tantivy = { git = "https://github.com/breezewish/tikv-tantivy.git", branch = "patch-0.22.1" } pprof = { git = "https://github.com/tikv/pprof-rs.git", rev = "01cff82dbe6fe110a707bf2b38d8ebb1d14a18f8" } prometheus = { git = "https://github.com/solotzg/rust-prometheus.git", rev = "b4fe98a06a58d29f9b9987a0d7186f6ed5230193", features = ["nightly", "push"], default-features = true } -kvproto = { git = "https://github.com/pingcap/kvproto.git" } tipb = { git = "https://github.com/pingcap/tipb.git" } # cloud-storage-engine now is private. We add it as a submodule to workaround diff --git a/contrib/tiflash-columnar-hub/hub-runtime/Cargo.toml b/contrib/tiflash-columnar-hub/hub-runtime/Cargo.toml index ee5f3335820..6fa4e9b1046 100644 --- a/contrib/tiflash-columnar-hub/hub-runtime/Cargo.toml +++ b/contrib/tiflash-columnar-hub/hub-runtime/Cargo.toml @@ -29,7 +29,7 @@ hyper = { version = "0.14.26", features = ["full", "runtime"] } keys = { workspace = true } kvengine = { workspace = true } kvenginepb = { workspace = true } -kvproto = { git = "https://github.com/pingcap/kvproto.git" } +kvproto = { workspace = true } lazy_static = "1.5.0" num_cpus = "1" pd_client = { workspace = true } diff --git a/contrib/tiflash-columnar-hub/hub-runtime/src/cloud_helper.rs b/contrib/tiflash-columnar-hub/hub-runtime/src/cloud_helper.rs index a59875bdba7..99e44699388 100644 --- a/contrib/tiflash-columnar-hub/hub-runtime/src/cloud_helper.rs +++ b/contrib/tiflash-columnar-hub/hub-runtime/src/cloud_helper.rs @@ -13,6 +13,7 @@ // limitations under the License. use std::{ + collections::HashMap, fs, ops::Deref, path::{Path, PathBuf}, @@ -23,7 +24,7 @@ use std::{ time::{Duration, UNIX_EPOCH}, }; -use api_version::api_v2::KEYSPACE_PREFIX_LEN; +use api_version::{api_v2::KEYSPACE_PREFIX_LEN, ApiV2}; use bytes::Bytes; use cloud_encryption::MasterKey; use dashmap::DashMap; @@ -76,6 +77,7 @@ const BACKOFF_RETRY_COUNT: usize = 5; const SNAPSHOT_CACHE_SIZE: usize = 10240; // 10k shards const SNAPSHOT_CACHE_CAP: u64 = 1024 * 1024 * 1024; // 1GB snapshot size with memtable const SNAPSHOT_CACHE_CAPABILITY_HEADER: &str = "x-cse-snapshot-cache-version"; +pub(crate) const DEFAULT_SCHEMA_FILE_CACHE_MAX_KEYSPACES: usize = 1024; #[derive(Clone)] struct RefreshingHttpClient { @@ -207,6 +209,24 @@ pub struct CloudEngineBackends { pub txn_chunk_mgr: TxnChunkManager, } +#[derive(Clone, Copy, Debug)] +struct SchemaFileCacheEntry { + latest_version: i64, + last_access_seq: u64, +} + +#[derive(Debug, Default)] +struct SchemaFileCacheState { + access_seq: u64, + keyspaces: HashMap, +} + +#[derive(Clone, Copy, Debug)] +struct LatestSchemaFile { + file_id: u64, + version: i64, +} + #[derive(Clone)] struct RegionBucketCacheEntry { region_ver: u64, @@ -330,6 +350,8 @@ pub struct CloudHelper { shared_snap_access_cache: SharedSnapAccessCache, meta_file_cache: Arc, MetaFileCacheWeighter>>, schema_files: Arc>, + schema_file_cache_state: Arc>, + schema_file_cache_max_keyspaces: usize, runtime: Arc, read_concurrency: usize, ia_ctx: IaCtx, @@ -346,6 +368,7 @@ impl CloudHelper { fts_cache_config: FtsCacheConfig, fts_delta_cache_config: FtsDeltaCacheConfig, block_cache_size: AbsoluteOrPercentSize, + schema_file_cache_max_keyspaces: usize, ) -> Self { let thread_count = std::cmp::max(4, (SysQuota::cpu_cores_quota() * 7.0 / 8.0) as usize); let runtime = Arc::new( @@ -395,6 +418,8 @@ impl CloudHelper { shared_snap_access_cache, meta_file_cache, schema_files: Arc::new(DashMap::new()), + schema_file_cache_state: Arc::new(Mutex::new(SchemaFileCacheState::default())), + schema_file_cache_max_keyspaces, runtime, read_concurrency: thread_count, ia_ctx, @@ -477,6 +502,14 @@ impl CloudHelper { } } + pub fn gc_schema_file_cache(&self) { + gc_schema_file_cache_global( + &self.schema_files, + &self.schema_file_cache_state, + self.schema_file_cache_max_keyspaces, + ); + } + pub fn make_columnar_reader( &self, shard_id: u64, @@ -492,6 +525,7 @@ impl CloudHelper { let dfs = self.dfs.clone(); let pd_client = self.pd_client.clone(); let schema_files = self.schema_files.clone(); + let schema_file_cache_state = self.schema_file_cache_state.clone(); let txn_mgr = self.txn_chunk_mgr.clone(); let master_key = self.master_key.clone(); let ia_ctx = self.ia_ctx.clone(); @@ -529,6 +563,7 @@ impl CloudHelper { snap_cache_capable_stores, meta_file_cache, schema_files, + schema_file_cache_state, txn_mgr, block_cache, fts_cache, @@ -682,6 +717,7 @@ async fn request_snapshot_from_leader( snap_cache_capable_stores: Arc>, meta_file_cache: Arc, MetaFileCacheWeighter>>, schema_files: Arc>, + schema_file_cache_state: Arc>, txn_chunk_manager: TxnChunkManager, block_cache: BlockCache, fts_cache: FtsCache, @@ -847,7 +883,7 @@ async fn request_snapshot_from_leader( fts_cache, fts_delta_cache, meta_file_cache, - schema_files: Some(schema_files), + schema_files: Some(schema_files.clone()), txn_chunk_manager, ia_ctx, prepare_type, @@ -866,6 +902,9 @@ async fn request_snapshot_from_leader( memory_limiter, ) .await?; + if let Some(keyspace_id) = get_snapshot_keyspace_id(tables) { + touch_schema_file_cache_keyspace(&schema_file_cache_state, keyspace_id); + } // Fill cache if not hit. if !cache_valid && server_supports_snap_cache { snap_cache.insert( @@ -1230,6 +1269,7 @@ async fn get_or_request_shared_snapshot( snap_cache_capable_stores: Arc>, meta_file_cache: Arc, MetaFileCacheWeighter>>, schema_files: Arc>, + schema_file_cache_state: Arc>, txn_chunk_manager: TxnChunkManager, block_cache: BlockCache, fts_cache: FtsCache, @@ -1289,6 +1329,7 @@ async fn get_or_request_shared_snapshot( snap_cache_capable_stores, meta_file_cache, schema_files, + schema_file_cache_state, txn_chunk_manager, block_cache, fts_cache, @@ -1309,11 +1350,177 @@ async fn get_or_request_shared_snapshot( snap } +fn get_snapshot_keyspace_id(tables: &[TableCtx]) -> Option { + tables + .first() + .and_then(|table| table.ranges.first()) + .and_then(|range| ApiV2::get_u32_keyspace_id_by_key(range.get_low())) +} + +fn touch_schema_file_cache_keyspace( + schema_file_cache_state: &Mutex, + active_keyspace_id: u32, +) { + // The read path only updates recency. Stale schema files are pruned by the + // background GC task so a keyspace may temporarily keep multiple versions. + let mut cache_state = schema_file_cache_state.lock().unwrap(); + let access_seq = cache_state.access_seq.wrapping_add(1); + cache_state.access_seq = access_seq; + cache_state + .keyspaces + .entry(active_keyspace_id) + .and_modify(|entry| entry.last_access_seq = access_seq) + .or_insert(SchemaFileCacheEntry { + latest_version: 0, + last_access_seq: access_seq, + }); +} + +fn gc_schema_file_cache_global( + schema_files: &DashMap, + schema_file_cache_state: &Mutex, + max_keyspaces: usize, +) { + let latest_by_keyspace = prune_stale_schema_file_versions(schema_files); + let mut cache_state = schema_file_cache_state.lock().unwrap(); + cache_state + .keyspaces + .retain(|keyspace_id, _| latest_by_keyspace.contains_key(keyspace_id)); + + for (&keyspace_id, latest_schema_file) in &latest_by_keyspace { + if let Some(entry) = cache_state.keyspaces.get_mut(&keyspace_id) { + entry.latest_version = latest_schema_file.version; + } else { + // Treat first discovery in the cache as a recent access so GC does + // not immediately evict a schema file that was just loaded. + let access_seq = cache_state.access_seq.wrapping_add(1); + cache_state.access_seq = access_seq; + cache_state.keyspaces.insert(keyspace_id, SchemaFileCacheEntry { + latest_version: latest_schema_file.version, + last_access_seq: access_seq, + }); + } + } + + if max_keyspaces == 0 { + return; + } + + while cache_state.keyspaces.len() > max_keyspaces { + let Some((victim_keyspace_id, victim_entry)) = cache_state + .keyspaces + .iter() + .map(|(keyspace_id, entry)| (*keyspace_id, *entry)) + .min_by_key(|(_, entry)| entry.last_access_seq) + else { + break; + }; + cache_state.keyspaces.remove(&victim_keyspace_id); + evict_schema_file_cache_keyspace( + schema_files, + victim_keyspace_id, + victim_entry.latest_version, + max_keyspaces, + ); + } +} + +fn prune_stale_schema_file_versions( + schema_files: &DashMap, +) -> HashMap { + let mut latest_by_keyspace = HashMap::new(); + let mut stale_files = Vec::new(); + + for entry in schema_files.iter() { + let schema_file = entry.value(); + let keyspace_id = schema_file.get_keyspace_id(); + let candidate = LatestSchemaFile { + file_id: *entry.key(), + version: schema_file.get_version(), + }; + + match latest_by_keyspace.get_mut(&keyspace_id) { + None => { + latest_by_keyspace.insert(keyspace_id, candidate); + } + Some(current) if is_newer_schema_file(candidate, *current) => { + stale_files.push((keyspace_id, current.file_id, current.version)); + *current = candidate; + } + Some(_) => stale_files.push((keyspace_id, candidate.file_id, candidate.version)), + } + } + + for (_, file_id, _) in &stale_files { + schema_files.remove(file_id); + } + + if !stale_files.is_empty() { + info!( + "removed stale schema file versions during full cache gc"; + "removed_file_count" => stale_files.len(), + "removed_files" => ?stale_files, + ); + } + + latest_by_keyspace +} + +fn evict_schema_file_cache_keyspace( + schema_files: &DashMap, + keyspace_id: u32, + latest_version: i64, + max_keyspaces: usize, +) { + let removed_files = remove_schema_files_for_keyspace(schema_files, keyspace_id); + if !removed_files.is_empty() { + info!( + "evicted schema file cache for inactive keyspace"; + "victim_keyspace_id" => keyspace_id, + "victim_latest_version" => latest_version, + "removed_file_count" => removed_files.len(), + "removed_files" => ?removed_files, + "max_keyspaces" => max_keyspaces, + ); + } +} + +fn remove_schema_files_for_keyspace( + schema_files: &DashMap, + keyspace_id: u32, +) -> Vec<(u64, i64)> { + let removed_files: Vec<_> = schema_files + .iter() + .filter_map(|entry| { + let schema_file = entry.value(); + (schema_file.get_keyspace_id() == keyspace_id) + .then_some((*entry.key(), schema_file.get_version())) + }) + .collect(); + + for (file_id, _) in &removed_files { + schema_files.remove(file_id); + } + removed_files +} + +fn is_newer_schema_file(lhs: LatestSchemaFile, rhs: LatestSchemaFile) -> bool { + (lhs.version, lhs.file_id) > (rhs.version, rhs.file_id) +} + #[cfg(test)] mod tests { use super::*; + use bytes::Bytes; + use kvengine::table::{file::InMemFile, schema_file::build_schema_file}; use security::SecurityConfig; + fn new_test_schema_file(file_id: u64, keyspace_id: u32, version: i64) -> SchemaFile { + let schema_file_data = build_schema_file(keyspace_id, version, Vec::new(), 0); + let file = Arc::new(InMemFile::new(file_id, Bytes::from(schema_file_data))); + SchemaFile::open(file).unwrap() + } + #[test] fn shared_snap_access_eviction_is_sticky_for_in_flight_loader() { let cache = SharedSnapAccessCache::new(); @@ -1358,4 +1565,106 @@ mod tests { assert_eq!(http_client.refresh_count(), 1); } + + #[test] + fn schema_file_cache_keeps_latest_version_per_keyspace() { + let schema_files = DashMap::new(); + schema_files.insert(11, new_test_schema_file(11, 1, 1)); + schema_files.insert(13, new_test_schema_file(13, 1, 3)); + schema_files.insert(12, new_test_schema_file(12, 1, 2)); + schema_files.insert(21, new_test_schema_file(21, 2, 8)); + + let latest = prune_stale_schema_file_versions(&schema_files); + + assert!(!schema_files.contains_key(&11)); + assert!(!schema_files.contains_key(&12)); + assert!(schema_files.contains_key(&13)); + assert!(schema_files.contains_key(&21)); + assert_eq!(latest.len(), 2); + assert_eq!(latest.get(&1).unwrap().version, 3); + assert_eq!(latest.get(&1).unwrap().file_id, 13); + } + + #[test] + fn schema_file_cache_touch_updates_lru_state() { + let cache_state = Mutex::new(SchemaFileCacheState::default()); + + touch_schema_file_cache_keyspace(&cache_state, 1); + touch_schema_file_cache_keyspace(&cache_state, 2); + touch_schema_file_cache_keyspace(&cache_state, 1); + + let cache_state = cache_state.lock().unwrap(); + assert_eq!(cache_state.keyspaces.len(), 2); + assert!(cache_state.keyspaces.contains_key(&1)); + assert!(cache_state.keyspaces.contains_key(&2)); + assert_eq!(cache_state.keyspaces.get(&1).unwrap().latest_version, 0); + assert!( + cache_state.keyspaces.get(&1).unwrap().last_access_seq + > cache_state.keyspaces.get(&2).unwrap().last_access_seq + ); + } + + #[test] + fn schema_file_cache_global_gc_removes_stale_versions_and_enforces_limit() { + let schema_files = DashMap::new(); + let cache_state = Mutex::new(SchemaFileCacheState { + access_seq: 3, + keyspaces: HashMap::from([ + ( + 1, + SchemaFileCacheEntry { + latest_version: 1, + last_access_seq: 3, + }, + ), + ( + 2, + SchemaFileCacheEntry { + latest_version: 2, + last_access_seq: 1, + }, + ), + ]), + }); + + schema_files.insert(11, new_test_schema_file(11, 1, 1)); + schema_files.insert(12, new_test_schema_file(12, 1, 2)); + schema_files.insert(21, new_test_schema_file(21, 2, 5)); + schema_files.insert(31, new_test_schema_file(31, 3, 7)); + + gc_schema_file_cache_global(&schema_files, &cache_state, 2); + + assert!(!schema_files.contains_key(&11)); + assert!(!schema_files.contains_key(&21)); + assert!(schema_files.contains_key(&12)); + assert!(schema_files.contains_key(&31)); + + let cache_state = cache_state.lock().unwrap(); + assert_eq!(cache_state.access_seq, 4); + assert_eq!(cache_state.keyspaces.get(&1).unwrap().latest_version, 2); + assert!(!cache_state.keyspaces.contains_key(&2)); + assert_eq!(cache_state.keyspaces.get(&3).unwrap().last_access_seq, 4); + assert!(cache_state.keyspaces.contains_key(&3)); + } + + #[test] + fn schema_file_cache_global_gc_keeps_all_keyspaces_when_limit_is_zero() { + let schema_files = DashMap::new(); + let cache_state = Mutex::new(SchemaFileCacheState::default()); + + schema_files.insert(11, new_test_schema_file(11, 1, 1)); + schema_files.insert(21, new_test_schema_file(21, 2, 2)); + schema_files.insert(31, new_test_schema_file(31, 3, 3)); + + gc_schema_file_cache_global(&schema_files, &cache_state, 0); + + assert!(schema_files.contains_key(&11)); + assert!(schema_files.contains_key(&21)); + assert!(schema_files.contains_key(&31)); + + let cache_state = cache_state.lock().unwrap(); + assert!(cache_state.keyspaces.contains_key(&1)); + assert!(cache_state.keyspaces.contains_key(&2)); + assert!(cache_state.keyspaces.contains_key(&3)); + } } diff --git a/contrib/tiflash-columnar-hub/hub-runtime/src/run.rs b/contrib/tiflash-columnar-hub/hub-runtime/src/run.rs index 944b18906ce..28b2e9c12bf 100644 --- a/contrib/tiflash-columnar-hub/hub-runtime/src/run.rs +++ b/contrib/tiflash-columnar-hub/hub-runtime/src/run.rs @@ -47,13 +47,13 @@ use pd_client::{Config as PdConfig, Error as PdError, PdClient, RpcClient}; use security::{RestfulClient, SecurityConfig, SecurityManager}; use serde::{Deserialize, Serialize}; use tikv_util::{ - config::{AbsoluteOrPercentSize, LogFormat, ReadableSize}, + config::{AbsoluteOrPercentSize, LogFormat, ReadableDuration, ReadableSize}, logger, sys::disk::get_disks_stats, }; use crate::{ - cloud_helper::{CloudEngineBackends, CloudHelper}, + cloud_helper::{CloudEngineBackends, CloudHelper, DEFAULT_SCHEMA_FILE_CACHE_MAX_KEYSPACES}, columnar_impls::{ ffi_clear_shared_snap_access_by_start_ts, ffi_columnar_scan_stats, ffi_get_region_bucket_keys, ffi_make_columnar_reader, ffi_physical_table_id, @@ -163,6 +163,9 @@ struct ConfigFile { fts_cache: FtsCacheConfig, fts_delta_cache: FtsDeltaCacheConfig, block_cache_size: AbsoluteOrPercentSize, + gc_interval: ReadableDuration, + ia_meta_cap: AbsoluteOrPercentSize, + schema_file_cache_max_keyspaces: usize, log: HubLogConfig, log_level: String, log_file: String, @@ -180,7 +183,7 @@ fn update_default_ia_config(ia: &mut IaConfig) { #[cfg(target_os = "linux")] fn update_default_ia_config(ia: &mut IaConfig) { - ia.mem_cap = AbsoluteOrPercentSize::Percent(20.0); + ia.mem_cap = AbsoluteOrPercentSize::Percent(10.0); ia.disk_cap = AbsoluteOrPercentSize::Percent(70.0); } @@ -200,6 +203,9 @@ impl Default for ConfigFile { fts_cache: FtsCacheConfig::default(), fts_delta_cache: FtsDeltaCacheConfig::default(), block_cache_size: AbsoluteOrPercentSize::default(), + gc_interval: ReadableDuration::minutes(10), + ia_meta_cap: AbsoluteOrPercentSize::Percent(10.0), + schema_file_cache_max_keyspaces: DEFAULT_SCHEMA_FILE_CACHE_MAX_KEYSPACES, log: HubLogConfig::default(), log_level: String::default(), log_file: String::default(), @@ -238,8 +244,10 @@ struct HubConfigSummary<'a> { const STORE_HEARTBEAT_INTERVAL: Duration = Duration::from_secs(10); const HEARTBEAT_SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(200); +const BACKGROUND_GC_SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_secs(1); const STORE_TOMBSTONE_WAIT_TIMEOUT: Duration = Duration::from_secs(30); const STORE_TOMBSTONE_POLL_INTERVAL: Duration = Duration::from_secs(1); + fn init_metrics() { static INIT: Once = Once::new(); @@ -1043,15 +1051,11 @@ fn spawn_store_heartbeat_loop( } } - let mut slept = Duration::from_secs(0); - while slept < STORE_HEARTBEAT_INTERVAL && !shutdown.load(Ordering::SeqCst) { - let step = std::cmp::min( - HEARTBEAT_SHUTDOWN_POLL_INTERVAL, - STORE_HEARTBEAT_INTERVAL - slept, - ); - thread::sleep(step); - slept += step; - } + sleep_with_shutdown( + &shutdown, + STORE_HEARTBEAT_INTERVAL, + HEARTBEAT_SHUTDOWN_POLL_INTERVAL, + ); } info!( @@ -1062,6 +1066,57 @@ fn spawn_store_heartbeat_loop( .unwrap_or_else(|err| panic!("failed to spawn store heartbeat thread: {}", err)) } +fn resolve_ia_meta_cap(meta_cap: &AbsoluteOrPercentSize, data_dir: &Path) -> u64 { + match meta_cap.as_disk_size(data_dir) { + Ok(bytes) => bytes, + Err(err) => { + error!( + "resolve ia meta cap failed"; + "path" => data_dir.display().to_string(), + "err" => ?err, + ); + 0 + } + } +} + +fn sleep_with_shutdown(shutdown: &AtomicBool, duration: Duration, poll_interval: Duration) { + let mut slept = Duration::from_secs(0); + while slept < duration && !shutdown.load(Ordering::SeqCst) { + let step = std::cmp::min(poll_interval, duration - slept); + thread::sleep(step); + slept += step; + } +} + +fn spawn_background_gc_loop( + cloud_helper: CloudHelper, + ia_meta_cap: AbsoluteOrPercentSize, + data_dir: PathBuf, + gc_interval: Duration, + shutdown: Arc, +) -> thread::JoinHandle<()> { + thread::Builder::new() + .name("hub-background-gc".to_owned()) + .spawn(move || { + info!( + "starting TiFlash Columnar Hub background gc loop"; + "interval_secs" => gc_interval.as_secs(), + "data_dir" => data_dir.display().to_string() + ); + + while !shutdown.load(Ordering::SeqCst) { + let meta_cap_bytes = resolve_ia_meta_cap(&ia_meta_cap, &data_dir); + cloud_helper.gc_ia_meta_files(meta_cap_bytes); + cloud_helper.gc_schema_file_cache(); + sleep_with_shutdown(&shutdown, gc_interval, BACKGROUND_GC_SHUTDOWN_POLL_INTERVAL); + } + + info!("stopped TiFlash Columnar Hub background gc loop"); + }) + .unwrap_or_else(|err| panic!("failed to spawn background gc thread: {}", err)) +} + fn store_id_path(data_dir: &Path) -> PathBuf { data_dir.join("tiflash-columnar-hub").join("store_id") } @@ -1402,8 +1457,9 @@ pub unsafe fn run_proxy(argc: c_int, argv: *const *const c_char, helper_ptr: *co let status_config_json = build_status_config_json(&config); let mut store_registration: Option<(metapb::Store, u32)> = None; let mut heartbeat_context: Option<(u64, u32)> = None; - let heartbeat_shutdown = Arc::new(AtomicBool::new(false)); + let service_shutdown = Arc::new(AtomicBool::new(false)); let mut heartbeat_handle: Option> = None; + let mut background_gc_handle: Option> = None; if !init_only { let store_id = ensure_store_id(pd_client.as_ref(), &data_dir); @@ -1426,6 +1482,7 @@ pub unsafe fn run_proxy(argc: c_int, argv: *const *const c_char, helper_ptr: *co config.fts_cache.clone(), config.fts_delta_cache.clone(), config.block_cache_size, + config.schema_file_cache_max_keyspaces, ); let hub_status = Arc::new(AtomicU8::new(RaftProxyStatus::Idle as u8)); let hub = ColumnarHub::new(hub_status.clone(), cloud_helper, hub_config_str); @@ -1472,13 +1529,25 @@ pub unsafe fn run_proxy(argc: c_int, argv: *const *const c_char, helper_ptr: *co } hub.set_status(RaftProxyStatus::Running); + let gc_interval = config.gc_interval.0; + if gc_interval.is_zero() { + info!("background gc task disabled: gc-interval is zero"); + } else { + background_gc_handle = Some(spawn_background_gc_loop( + hub.cloud_helper.clone(), + config.ia_meta_cap, + data_dir.clone(), + gc_interval, + service_shutdown.clone(), + )); + } if let Some((store_id, start_time)) = heartbeat_context { heartbeat_handle = Some(spawn_store_heartbeat_loop( pd_client.clone(), store_id, start_time, data_dir.clone(), - heartbeat_shutdown.clone(), + service_shutdown.clone(), )); } } @@ -1488,7 +1557,7 @@ pub unsafe fn run_proxy(argc: c_int, argv: *const *const c_char, helper_ptr: *co EngineStoreServerStatus::Running => thread::sleep(Duration::from_millis(200)), EngineStoreServerStatus::Stopping => { hub.set_status(RaftProxyStatus::Stopped); - heartbeat_shutdown.store(true, Ordering::SeqCst); + service_shutdown.store(true, Ordering::SeqCst); while !matches!( helper.handle_get_engine_store_server_status(), EngineStoreServerStatus::Terminated @@ -1498,7 +1567,7 @@ pub unsafe fn run_proxy(argc: c_int, argv: *const *const c_char, helper_ptr: *co break; } EngineStoreServerStatus::Terminated => { - heartbeat_shutdown.store(true, Ordering::SeqCst); + service_shutdown.store(true, Ordering::SeqCst); break; } EngineStoreServerStatus::Idle => thread::sleep(Duration::from_millis(200)), @@ -1514,6 +1583,9 @@ pub unsafe fn run_proxy(argc: c_int, argv: *const *const c_char, helper_ptr: *co if let Some(handle) = heartbeat_handle.take() { let _ = handle.join(); } + if let Some(handle) = background_gc_handle.take() { + let _ = handle.join(); + } if let Some(server) = status_server.take() { server.stop(); } @@ -1632,6 +1704,7 @@ log-level = "debug" AbsoluteOrPercentSize::Abs(ReadableSize::mb(100)) ); } + assert_eq!(config.ia_meta_cap, AbsoluteOrPercentSize::Percent(10.0)); } #[test]