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..2fab221dda7 100644 --- a/contrib/tiflash-columnar-hub/hub-runtime/src/cloud_helper.rs +++ b/contrib/tiflash-columnar-hub/hub-runtime/src/cloud_helper.rs @@ -13,17 +13,17 @@ // limitations under the License. use std::{ - fs, + collections::HashMap, ops::Deref, - path::{Path, PathBuf}, + path::PathBuf, sync::{ atomic::{AtomicBool, Ordering}, Arc, Mutex, Weak, }, - time::{Duration, UNIX_EPOCH}, + time::Duration, }; -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; @@ -32,7 +32,11 @@ use keys::next_key; use kvengine::{ context::{new_meta_file_cache, IaCtx, MetaFileCacheWeighter, PrepareType, SnapCtx}, dfs::{self, Dfs}, - ia::{ia_file::parse_table_meta_filename, manager::IaManager, util::IaConfig}, + ia::{ + gc::{IaGcConfig, IaGcRunner}, + manager::IaManager, + util::IaConfig, + }, table::{ columnar::{ filter::TableScanCtx, ColumnarFileCache, ColumnarFileCacheConfig, ColumnarMetaCache, @@ -76,6 +80,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 +212,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 +353,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 +371,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 +421,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, @@ -403,78 +431,20 @@ impl CloudHelper { } } -#[derive(Debug)] -struct IaMetaFileEntry { - file_id: u64, - path: PathBuf, - size: u64, - modified: Duration, -} - impl CloudHelper { - pub fn gc_ia_meta_files(&self, meta_cap: u64) { + pub fn new_ia_gc_runner(&self, config: IaGcConfig) -> Option { let IaCtx::Enabled(ia_mgr, meta_paths) = &self.ia_ctx else { - return; - }; - - if meta_cap == 0 { - return; - } - - let mut entries = match collect_ia_meta_files(meta_paths) { - Ok(entries) => entries, - Err(err) => { - error!("ia meta gc scan failed"; "err" => ?err); - return; - } + return None; }; - let total_size: u64 = entries.iter().map(|entry| entry.size).sum(); - if total_size <= meta_cap { - return; - } - - entries.sort_by(|a, b| { - a.modified - .cmp(&b.modified) - .then_with(|| a.file_id.cmp(&b.file_id)) - .then_with(|| a.path.cmp(&b.path)) - }); - let mut current_size = total_size; - let mut removed_files = 0; - let mut removed_bytes = 0; - for entry in entries { - if current_size <= meta_cap { - break; - } - - match remove_ia_meta_file(ia_mgr, &entry.path, entry.file_id) { - Ok(true) => { - current_size = current_size.saturating_sub(entry.size); - removed_files += 1; - removed_bytes += entry.size; - } - Ok(false) => {} - Err(err) => { - error!( - "ia meta gc remove failed"; - "file_id" => entry.file_id, - "path" => ?entry.path, - "err" => ?err, - ); - } - } - } + Some(IaGcRunner::new(config, ia_mgr.clone(), meta_paths.clone())) + } - if removed_files > 0 { - info!( - "ia meta gc reclaimed files to satisfy cap"; - "meta_cap" => meta_cap, - "removed_files" => removed_files, - "removed_bytes" => removed_bytes, - "size_before" => total_size, - "size_after" => current_size, - ); - } + 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( @@ -492,6 +462,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 +500,7 @@ impl CloudHelper { snap_cache_capable_stores, meta_file_cache, schema_files, + schema_file_cache_state, txn_mgr, block_cache, fts_cache, @@ -602,56 +574,6 @@ impl CloudHelper { } } -fn collect_ia_meta_files(meta_paths: &[PathBuf]) -> std::io::Result> { - let mut entries = Vec::new(); - for meta_path in meta_paths { - if !meta_path.is_dir() { - continue; - } - for entry in fs::read_dir(meta_path)? { - let entry = entry?; - let path = entry.path(); - if !path.is_file() { - continue; - } - let filename = path - .file_name() - .and_then(|name| name.to_str()) - .unwrap_or(""); - let Some((file_id, _)) = parse_table_meta_filename(filename) else { - continue; - }; - let metadata = match entry.metadata() { - Ok(m) => m, - Err(err) => { - error!("ia meta gc get metadata failed"; "path" => ?path, "err" => ?err); - continue; - } - }; - entries.push(IaMetaFileEntry { - file_id, - path, - size: metadata.len(), - modified: metadata - .modified() - .ok() - .and_then(|mtime| mtime.duration_since(UNIX_EPOCH).ok()) - .unwrap_or(Duration::ZERO), - }); - } - } - Ok(entries) -} - -fn remove_ia_meta_file(ia_mgr: &IaManager, path: &Path, file_id: u64) -> std::io::Result { - ia_mgr.remove_table_meta(file_id); - match fs::remove_file(path) { - Ok(()) => Ok(true), - Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(false), - Err(err) => Err(err), - } -} - fn build_ia_mgr( dfs: Arc, runtime: Arc, @@ -682,6 +604,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 +770,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 +789,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 +1156,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 +1216,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 +1237,180 @@ 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 +1455,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..117500aaad5 100644 --- a/contrib/tiflash-columnar-hub/hub-runtime/src/run.rs +++ b/contrib/tiflash-columnar-hub/hub-runtime/src/run.rs @@ -33,7 +33,10 @@ use cloud_encryption::MasterKey; use grpcio::EnvBuilder; use kvengine::{ dfs::{DFSConfig, Dfs, InMemFs, S3Fs}, - ia::util::IaConfig, + ia::{ + gc::{IaFileGcState, IaGcConfig}, + util::IaConfig, + }, table::{ columnar::ColumnarFileCacheConfig, fts::{FtsCacheConfig, FtsDeltaCacheConfig}, @@ -47,13 +50,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 +166,8 @@ struct ConfigFile { fts_cache: FtsCacheConfig, fts_delta_cache: FtsDeltaCacheConfig, block_cache_size: AbsoluteOrPercentSize, + local_gc: LocalGcConfig, + schema_file_cache_max_keyspaces: usize, log: HubLogConfig, log_level: String, log_file: String, @@ -200,6 +205,8 @@ impl Default for ConfigFile { fts_cache: FtsCacheConfig::default(), fts_delta_cache: FtsDeltaCacheConfig::default(), block_cache_size: AbsoluteOrPercentSize::default(), + local_gc: LocalGcConfig::default(), + schema_file_cache_max_keyspaces: DEFAULT_SCHEMA_FILE_CACHE_MAX_KEYSPACES, log: HubLogConfig::default(), log_level: String::default(), log_file: String::default(), @@ -236,10 +243,29 @@ struct HubConfigSummary<'a> { init_only: bool, } +// Keep this config surface identical to cloud_worker::local_gc::LocalGcConfig. +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq)] +#[serde(default, rename_all = "kebab-case")] +struct LocalGcConfig { + interval: ReadableDuration, + ia: IaGcConfig, +} + +impl Default for LocalGcConfig { + fn default() -> Self { + Self { + interval: ReadableDuration::hours(1), + ia: IaGcConfig::default(), + } + } +} + 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 +1069,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 +1084,46 @@ fn spawn_store_heartbeat_loop( .unwrap_or_else(|err| panic!("failed to spawn store heartbeat thread: {}", err)) } +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, + local_gc_config: LocalGcConfig, + data_dir: PathBuf, + shutdown: Arc, +) -> thread::JoinHandle<()> { + thread::Builder::new() + .name("hub-background-gc".to_owned()) + .spawn(move || { + let gc_interval = local_gc_config.interval.0; + let mut ia_gc_runner = cloud_helper.new_ia_gc_runner(local_gc_config.ia.clone()); + info!( + "starting TiFlash Columnar Hub background gc loop"; + "config" => ?local_gc_config, + "data_dir" => data_dir.display().to_string() + ); + + while !shutdown.load(Ordering::SeqCst) { + if let Some(runner) = ia_gc_runner.as_mut() { + // This is the same state callback used by LocalGcRunner. + runner.run(|_| IaFileGcState::NotLive); + } + 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 +1464,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 +1489,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 +1536,24 @@ 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.local_gc.interval.0; + if gc_interval.is_zero() { + info!("background gc task disabled: local-gc.interval is zero"); + } else { + background_gc_handle = Some(spawn_background_gc_loop( + hub.cloud_helper.clone(), + config.local_gc.clone(), + data_dir.clone(), + 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 +1563,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 +1573,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 +1589,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 +1710,8 @@ log-level = "debug" AbsoluteOrPercentSize::Abs(ReadableSize::mb(100)) ); } + assert_eq!(config.local_gc.interval, ReadableDuration::hours(1)); + assert_eq!(config.local_gc.ia, IaGcConfig::default()); } #[test] @@ -1661,6 +1741,33 @@ engine-addr = "1.2.3.4:3930" } } + #[test] + fn test_local_gc_config_matches_cloud_worker_config() { + let config: ConfigFile = toml::from_str( + r#" +[local-gc] +interval = "10m" + +[local-gc.ia] +meta-lifetime = "2h" +segment-interval = "12h" +tmp-lifetime = "5m" +"#, + ) + .unwrap(); + + assert_eq!(config.local_gc.interval, ReadableDuration::minutes(10)); + assert_eq!(config.local_gc.ia.meta_lifetime, ReadableDuration::hours(2)); + assert_eq!( + config.local_gc.ia.segment_interval, + ReadableDuration::hours(12) + ); + assert_eq!( + config.local_gc.ia.tmp_lifetime, + ReadableDuration::minutes(5) + ); + } + #[test] fn test_resolve_hub_log_config_prefers_nested_log_config() { let config: ConfigFile = toml::from_str(