feat(storage): filecache lazy load, unit-level refill, reduce insert …

…latency (#12714)

Signed-off-by: MrCroxx <[email protected]>

src/common/src/config.rs

@@ -586,15 +586,36 @@ pub struct StorageConfig {
 
 #[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde)]
 pub struct CacheRefillConfig {
+    /// SSTable levels to refill.
     #[serde(default = "default::cache_refill::data_refill_levels")]
     pub data_refill_levels: Vec<u32>,
 
+    /// Cache refill maximum timeout to apply version delta.
     #[serde(default = "default::cache_refill::timeout_ms")]
     pub timeout_ms: u64,
 
+    /// Inflight data cache refill tasks.
     #[serde(default = "default::cache_refill::concurrency")]
     pub concurrency: usize,
 
+    /// Block count that a data cache refill request fetches.
+    #[serde(default = "default::cache_refill::unit")]
+    pub unit: usize,
+
+    /// Data cache refill unit admission ratio.
+    ///
+    /// Only unit whose blocks are admitted above the ratio will be refilled.
+    #[serde(default = "default::cache_refill::threshold")]
+    pub threshold: f64,
+
+    /// Recent filter layer count.
+    #[serde(default = "default::cache_refill::recent_filter_layers")]
+    pub recent_filter_layers: usize,
+
+    /// Recent filter layer rotate interval.
+    #[serde(default = "default::cache_refill::recent_filter_rotate_interval_ms")]
+    pub recent_filter_rotate_interval_ms: usize,
+
     #[serde(default, flatten)]
     pub unrecognized: Unrecognized<Self>,
 }
@@ -637,15 +658,21 @@ pub struct FileCacheConfig {
     #[serde(default = "default::file_cache::lfu_tiny_lru_capacity_ratio")]
     pub lfu_tiny_lru_capacity_ratio: f64,
 
-    #[serde(default = "default::file_cache::rated_random_rate_mb")]
-    pub rated_random_rate_mb: usize,
+    #[serde(default = "default::file_cache::insert_rate_limit_mb")]
+    pub insert_rate_limit_mb: usize,
 
     #[serde(default = "default::file_cache::flush_rate_limit_mb")]
     pub flush_rate_limit_mb: usize,
 
     #[serde(default = "default::file_cache::reclaim_rate_limit_mb")]
     pub reclaim_rate_limit_mb: usize,
 
+    #[serde(default = "default::file_cache::allocation_bits")]
+    pub allocation_bits: usize,
+
+    #[serde(default = "default::file_cache::allocation_timeout_ms")]
+    pub allocation_timeout_ms: usize,
+
     #[serde(default, flatten)]
     pub unrecognized: Unrecognized<Self>,
 }
@@ -1132,7 +1159,7 @@ pub mod default {
             0.01
         }
 
-        pub fn rated_random_rate_mb() -> usize {
+        pub fn insert_rate_limit_mb() -> usize {
             0
         }
 
@@ -1143,6 +1170,14 @@ pub mod default {
         pub fn reclaim_rate_limit_mb() -> usize {
             0
         }
+
+        pub fn allocation_bits() -> usize {
+            0
+        }
+
+        pub fn allocation_timeout_ms() -> usize {
+            10
+        }
     }
 
     pub mod cache_refill {
@@ -1157,6 +1192,22 @@ pub mod default {
         pub fn concurrency() -> usize {
             10
         }
+
+        pub fn unit() -> usize {
+            64
+        }
+
+        pub fn threshold() -> f64 {
+            0.5
+        }
+
+        pub fn recent_filter_layers() -> usize {
+            6
+        }
+
+        pub fn recent_filter_rotate_interval_ms() -> usize {
+            10000
+        }
     }
 
     pub mod heap_profiling {

src/compute/src/memory_management/mod.rs

@@ -152,20 +152,34 @@ pub fn storage_memory_config(
             .ceil() as usize)
             >> 20,
     );
+
+    // `foyer` uses a buffer pool to manage dirty buffers.
+    //
+    // buffer size = region size (single file size with fs device)
+    //
+    // writing buffer + flushing buffer + free buffer = buffer pool buffers
+    //
+    // To utilize flushers and allocators, buffers should >= allocators + flushers.
+    //
+    // Adding more buffers can prevent allocators from waiting for buffers to be freed by flushers.
+
     let data_file_cache_buffer_pool_capacity_mb = storage_config
         .data_file_cache
         .buffer_pool_size_mb
         .unwrap_or(
             storage_config.data_file_cache.file_capacity_mb
-                * storage_config.data_file_cache.flushers,
+                * (storage_config.data_file_cache.flushers
+                    + 2 * (1 << storage_config.data_file_cache.allocation_bits)),
         );
     let meta_file_cache_buffer_pool_capacity_mb = storage_config
         .meta_file_cache
         .buffer_pool_size_mb
         .unwrap_or(
             storage_config.meta_file_cache.file_capacity_mb
-                * storage_config.meta_file_cache.flushers,
+                * (storage_config.meta_file_cache.flushers
+                    + 2 * (1 << storage_config.meta_file_cache.allocation_bits)),
         );
+
     let compactor_memory_limit_mb = storage_config.compactor_memory_limit_mb.unwrap_or(
         ((non_reserved_memory_bytes as f64 * compactor_memory_proportion).ceil() as usize) >> 20,
     );
@@ -229,8 +243,8 @@ mod tests {
         assert_eq!(memory_config.block_cache_capacity_mb, 737);
         assert_eq!(memory_config.meta_cache_capacity_mb, 860);
         assert_eq!(memory_config.shared_buffer_capacity_mb, 737);
-        assert_eq!(memory_config.data_file_cache_buffer_pool_capacity_mb, 256);
-        assert_eq!(memory_config.meta_file_cache_buffer_pool_capacity_mb, 256);
+        assert_eq!(memory_config.data_file_cache_buffer_pool_capacity_mb, 384);
+        assert_eq!(memory_config.meta_file_cache_buffer_pool_capacity_mb, 384);
         assert_eq!(memory_config.compactor_memory_limit_mb, 819);
 
         storage_config.block_cache_capacity_mb = Some(512);

src/config/example.toml

@@ -123,9 +123,11 @@ reclaimers = 4
 recover_concurrency = 8
 lfu_window_to_cache_size_ratio = 1
 lfu_tiny_lru_capacity_ratio = 0.01
-rated_random_rate_mb = 0
+insert_rate_limit_mb = 0
 flush_rate_limit_mb = 0
 reclaim_rate_limit_mb = 0
+allocation_bits = 0
+allocation_timeout_ms = 10
 
 [storage.meta_file_cache]
 dir = ""
@@ -138,14 +140,20 @@ reclaimers = 4
 recover_concurrency = 8
 lfu_window_to_cache_size_ratio = 1
 lfu_tiny_lru_capacity_ratio = 0.01
-rated_random_rate_mb = 0
+insert_rate_limit_mb = 0
 flush_rate_limit_mb = 0
 reclaim_rate_limit_mb = 0
+allocation_bits = 0
+allocation_timeout_ms = 10
 
 [storage.cache_refill]
 data_refill_levels = []
 timeout_ms = 6000
 concurrency = 10
+unit = 64
+threshold = 0.5
+recent_filter_layers = 6
+recent_filter_rotate_interval_ms = 10000
 
 [system]
 barrier_interval_ms = 1000

src/ctl/src/common/hummock_service.rs

@@ -169,6 +169,7 @@ For `./risedev apply-compose-deploy` users,
             0,
             FileCache::none(),
             FileCache::none(),
+            None,
         )))
     }
 }

src/jni_core/src/hummock_iterator.rs

@@ -69,6 +69,7 @@ impl HummockJavaBindingIterator {
             0,
             FileCache::none(),
             FileCache::none(),
+            None,
         ));
         let reader =
             HummockVersionReader::new(sstable_store, Arc::new(HummockStateStoreMetrics::unused()));

src/storage/Cargo.toml

@@ -26,7 +26,7 @@ dyn-clone = "1.0.14"
 either = "1"
 enum-as-inner = "0.6"
 fail = "0.5"
-foyer = { git = "https://github.com/mrcroxx/foyer", rev = "41b1d39" }
+foyer = { git = "https://github.com/mrcroxx/foyer", rev = "438eec8" }
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
 futures-async-stream = { workspace = true }
 hex = "0.4"

src/storage/benches/bench_compactor.rs

@@ -56,6 +56,7 @@ pub fn mock_sstable_store() -> SstableStoreRef {
         0,
         FileCache::none(),
         FileCache::none(),
+        None,
     ))
 }
 

src/storage/benches/bench_multi_builder.rs

@@ -144,6 +144,7 @@ fn bench_builder(
         0,
         FileCache::none(),
         FileCache::none(),
+        None,
     ));
 
     let mut group = c.benchmark_group("bench_multi_builder");

src/storage/hummock_test/src/bin/replay/main.rs

@@ -111,6 +111,7 @@ async fn create_replay_hummock(r: Record, args: &Args) -> Result<impl GlobalRepl
             storage_opts.high_priority_ratio,
             FileCache::none(),
             FileCache::none(),
+            None,
         ))
     };