feat(foyer): cherry-pick disk cache optimization related PRs (#21844)

Author: MrCroxx

Cargo.lock

@@ -81,8 +81,8 @@ license = "Apache-2.0"
 repository = "https://github.com/risingwavelabs/risingwave"
 
 [workspace.dependencies]
-foyer = { version = "0.15.4", features = ["tracing", "nightly"] }
-mixtrics ={ version = "0.0.4", features = ["prometheus"] }
+foyer = { version = "0.17.2", features = ["serde", "tracing", "nightly"] }
+mixtrics = { version = "0.1", features = ["prometheus_0_13"] }
 apache-avro = { git = "https://github.com/risingwavelabs/avro", rev = "25113ba88234a9ae23296e981d8302c290fdaa4b", features = [
     "snappy",
     "zstandard",

Cargo.toml

@@ -265,7 +265,7 @@ mod tests {
     use std::ops::Bound;
 
     use assert_matches::assert_matches;
-    use foyer::CacheHint;
+    use foyer::Hint;
     use futures::StreamExt;
     use risingwave_common::array::{Array, ArrayImpl, I32Array, StructArray};
     use risingwave_common::catalog::{
@@ -396,7 +396,7 @@ mod tests {
                 epoch,
                 None,
                 ReadOptions {
-                    cache_policy: CachePolicy::Fill(CacheHint::Normal),
+                    cache_policy: CachePolicy::Fill(Hint::Normal),
                     ..Default::default()
                 },
             )

src/batch/executors/src/executor/insert.rs

@@ -24,7 +24,9 @@ use std::num::NonZeroUsize;
 use anyhow::Context;
 use clap::ValueEnum;
 use educe::Educe;
-use foyer::{Compression, LfuConfig, LruConfig, RecoverMode, RuntimeOptions, S3FifoConfig};
+use foyer::{
+    Compression, LfuConfig, LruConfig, RecoverMode, RuntimeOptions, S3FifoConfig, Throttle,
+};
 use risingwave_common_proc_macro::ConfigDoc;
 pub use risingwave_common_proc_macro::OverrideConfig;
 use risingwave_pb::meta::SystemParams;
@@ -994,6 +996,7 @@ pub struct FileCacheConfig {
     #[serde(default = "default::file_cache::recover_concurrency")]
     pub recover_concurrency: usize,
 
+    /// Deprecated soon. Please use `throttle` to do I/O throttling instead.
     #[serde(default = "default::file_cache::insert_rate_limit_mb")]
     pub insert_rate_limit_mb: usize,
 
@@ -1006,6 +1009,12 @@ pub struct FileCacheConfig {
     #[serde(default = "default::file_cache::flush_buffer_threshold_mb")]
     pub flush_buffer_threshold_mb: Option<usize>,
 
+    #[serde(default = "default::file_cache::throttle")]
+    pub throttle: Throttle,
+
+    #[serde(default = "default::file_cache::fifo_probation_ratio")]
+    pub fifo_probation_ratio: f64,
+
     /// Recover mode.
     ///
     /// Options:
@@ -1912,7 +1921,9 @@ pub mod default {
     }
 
     pub mod file_cache {
-        use foyer::{Compression, RecoverMode, RuntimeOptions, TokioRuntimeOptions};
+        use std::num::NonZeroUsize;
+
+        use foyer::{Compression, RecoverMode, RuntimeOptions, Throttle, TokioRuntimeOptions};
 
         pub fn dir() -> String {
             "".to_owned()
@@ -1954,13 +1965,28 @@ pub mod default {
             None
         }
 
+        pub fn fifo_probation_ratio() -> f64 {
+            0.1
+        }
+
         pub fn recover_mode() -> RecoverMode {
             RecoverMode::None
         }
 
         pub fn runtime_config() -> RuntimeOptions {
             RuntimeOptions::Unified(TokioRuntimeOptions::default())
         }
+
+        pub fn throttle() -> Throttle {
+            Throttle::new()
+                .with_iops_counter(foyer::IopsCounter::PerIoSize(
+                    NonZeroUsize::new(128 * 1024).unwrap(),
+                ))
+                .with_read_iops(100000)
+                .with_write_iops(100000)
+                .with_write_throughput(1024 * 1024 * 1024)
+                .with_read_throughput(1024 * 1024 * 1024)
+        }
     }
 
     pub mod cache_refill {

src/common/src/config.rs

@@ -195,15 +195,17 @@ This page is automatically generated by `./risedev generate-example-config`
 | capacity_mb |  | 1024 |
 | compression |  | "None" |
 | dir |  | "" |
+| fifo_probation_ratio |  | 0.1 |
 | file_capacity_mb |  | 64 |
 | flush_buffer_threshold_mb |  |  |
 | flushers |  | 4 |
 | indexer_shards |  | 64 |
-| insert_rate_limit_mb |  | 0 |
+| insert_rate_limit_mb | Deprecated soon. Please use `throttle` to do I/O throttling instead. | 0 |
 | reclaimers |  | 4 |
 | recover_concurrency |  | 8 |
 | recover_mode | Recover mode. Options: - "None": Do not recover disk cache. - "Quiet": Recover disk cache and skip errors. - "Strict": Recover disk cache and panic on errors. More details, see [`RecoverMode::None`], [`RecoverMode::Quiet`] and [`RecoverMode::Strict`], | "None" |
 | runtime_config |  |  |
+| throttle |  |  |
 
 ## storage.meta_file_cache
 
@@ -212,15 +214,17 @@ This page is automatically generated by `./risedev generate-example-config`
 | capacity_mb |  | 1024 |
 | compression |  | "None" |
 | dir |  | "" |
+| fifo_probation_ratio |  | 0.1 |
 | file_capacity_mb |  | 64 |
 | flush_buffer_threshold_mb |  |  |
 | flushers |  | 4 |
 | indexer_shards |  | 64 |
-| insert_rate_limit_mb |  | 0 |
+| insert_rate_limit_mb | Deprecated soon. Please use `throttle` to do I/O throttling instead. | 0 |
 | reclaimers |  | 4 |
 | recover_concurrency |  | 8 |
 | recover_mode | Recover mode. Options: - "None": Do not recover disk cache. - "Quiet": Recover disk cache and skip errors. - "Strict": Recover disk cache and panic on errors. More details, see [`RecoverMode::None`], [`RecoverMode::Quiet`] and [`RecoverMode::Strict`], | "None" |
 | runtime_config |  |  |
+| throttle |  |  |
 
 ## streaming
 

src/config/docs.md

@@ -226,8 +226,18 @@ recover_concurrency = 8
 insert_rate_limit_mb = 0
 indexer_shards = 64
 compression = "None"
+fifo_probation_ratio = 0.1
 recover_mode = "None"
 
+[storage.data_file_cache.throttle]
+write_iops = 100000
+read_iops = 100000
+write_throughput = 1073741824
+read_throughput = 1073741824
+
+[storage.data_file_cache.throttle.iops_counter]
+PerIoSize = 131072
+
 [storage.data_file_cache.runtime_config.Unified]
 worker_threads = 0
 max_blocking_threads = 0
@@ -242,8 +252,18 @@ recover_concurrency = 8
 insert_rate_limit_mb = 0
 indexer_shards = 64
 compression = "None"
+fifo_probation_ratio = 0.1
 recover_mode = "None"
 
+[storage.meta_file_cache.throttle]
+write_iops = 100000
+read_iops = 100000
+write_throughput = 1073741824
+read_throughput = 1073741824
+
+[storage.meta_file_cache.throttle.iops_counter]
+PerIoSize = 131072
+
 [storage.meta_file_cache.runtime_config.Unified]
 worker_threads = 0
 max_blocking_threads = 0

src/config/example.toml

@@ -18,7 +18,7 @@ use std::sync::Arc;
 
 use criterion::async_executor::FuturesExecutor;
 use criterion::{criterion_group, criterion_main, Criterion};
-use foyer::{CacheHint, Engine, HybridCacheBuilder};
+use foyer::{Engine, Hint, HybridCacheBuilder};
 use risingwave_common::catalog::{ColumnDesc, ColumnId, TableId};
 use risingwave_common::config::{MetricLevel, ObjectStoreConfig};
 use risingwave_common::hash::VirtualNode;
@@ -95,7 +95,7 @@ pub fn default_writer_opts() -> SstableWriterOptions {
     SstableWriterOptions {
         capacity_hint: None,
         tracker: None,
-        policy: CachePolicy::Fill(CacheHint::Normal),
+        policy: CachePolicy::Fill(Hint::Normal),
     }
 }
 
@@ -132,7 +132,7 @@ async fn build_table(
         SstableWriterOptions {
             capacity_hint: None,
             tracker: None,
-            policy: CachePolicy::Fill(CacheHint::Normal),
+            policy: CachePolicy::Fill(Hint::Normal),
         },
     );
     let table_id_to_vnode = HashMap::from_iter(vec![(0, VirtualNode::COUNT_FOR_TEST)]);
@@ -181,7 +181,7 @@ async fn build_table_2(
         SstableWriterOptions {
             capacity_hint: None,
             tracker: None,
-            policy: CachePolicy::Fill(CacheHint::Normal),
+            policy: CachePolicy::Fill(Hint::Normal),
         },
     );
 
@@ -345,7 +345,7 @@ fn bench_merge_iterator_compactor(c: &mut Criterion) {
     });
     let level2 = vec![info1, info2];
     let read_options = Arc::new(SstableIteratorReadOptions {
-        cache_policy: CachePolicy::Fill(CacheHint::Normal),
+        cache_policy: CachePolicy::Fill(Hint::Normal),
         prefetch_for_large_query: false,
         must_iterated_end_user_key: None,
         max_preload_retry_times: 0,

src/storage/benches/bench_compactor.rs

@@ -17,7 +17,7 @@ use std::sync::Arc;
 
 use bytes::Bytes;
 use criterion::{criterion_group, criterion_main, Criterion};
-use foyer::CacheHint;
+use foyer::Hint;
 use futures::pin_mut;
 use risingwave_common::util::epoch::test_epoch;
 use risingwave_hummock_sdk::key::TableKey;
@@ -109,7 +109,7 @@ fn criterion_benchmark(c: &mut Criterion) {
                     epoch,
                     ReadOptions {
                         prefetch_options: PrefetchOptions::default(),
-                        cache_policy: CachePolicy::Fill(CacheHint::Normal),
+                        cache_policy: CachePolicy::Fill(Hint::Normal),
                         ..Default::default()
                     },
                 ))

src/storage/hummock_test/benches/bench_hummock_iter.rs

@@ -19,7 +19,7 @@ pub(crate) mod tests {
     use std::sync::Arc;
 
     use bytes::{BufMut, Bytes, BytesMut};
-    use foyer::CacheHint;
+    use foyer::Hint;
     use itertools::Itertools;
     use rand::{Rng, RngCore, SeedableRng};
     use risingwave_common::bitmap::BitmapBuilder;
@@ -338,7 +338,7 @@ pub(crate) mod tests {
                 TableKey(key.clone()),
                 get_epoch,
                 ReadOptions {
-                    cache_policy: CachePolicy::Fill(CacheHint::Normal),
+                    cache_policy: CachePolicy::Fill(Hint::Normal),
                     ..Default::default()
                 },
             )
@@ -668,7 +668,7 @@ pub(crate) mod tests {
                 ReadOptions {
                     table_id: TableId::from(existing_table_id),
                     prefetch_options: PrefetchOptions::default(),
-                    cache_policy: CachePolicy::Fill(CacheHint::Normal),
+                    cache_policy: CachePolicy::Fill(Hint::Normal),
                     ..Default::default()
                 },
             )
@@ -874,7 +874,7 @@ pub(crate) mod tests {
                 ReadOptions {
                     table_id: TableId::from(existing_table_id),
                     prefetch_options: PrefetchOptions::default(),
-                    cache_policy: CachePolicy::Fill(CacheHint::Normal),
+                    cache_policy: CachePolicy::Fill(Hint::Normal),
                     ..Default::default()
                 },
             )
@@ -1088,7 +1088,7 @@ pub(crate) mod tests {
                     prefix_hint: Some(Bytes::from(bloom_filter_key)),
                     table_id: TableId::from(existing_table_id),
                     prefetch_options: PrefetchOptions::default(),
-                    cache_policy: CachePolicy::Fill(CacheHint::Normal),
+                    cache_policy: CachePolicy::Fill(Hint::Normal),
                     ..Default::default()
                 },
             )