Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix(storage): add disk cache buffer threshold to mitigate OOM (#17595) #17620

Merged
merged 1 commit into from
Jul 9, 2024
Merged

fix(storage): add disk cache buffer threshold to mitigate OOM (#17595) #17620

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 10 additions & 10 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ license = "Apache-2.0"
repository = "https://github.com/risingwavelabs/risingwave"

[workspace.dependencies]
foyer = { version = "0.10.0", features = ["nightly"] }
foyer = { version = "0.10.1", features = ["nightly", "mtrace"] }
apache-avro = { git = "https://github.com/risingwavelabs/avro", rev = "25113ba88234a9ae23296e981d8302c290fdaa4b", features = [
"snappy",
"zstandard",
Expand Down
34 changes: 34 additions & 0 deletions src/common/src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -844,6 +844,9 @@ pub struct FileCacheConfig {
#[serde(default = "default::file_cache::compression")]
pub compression: String,

#[serde(default = "default::file_cache::flush_buffer_threshold_mb")]
pub flush_buffer_threshold_mb: Option<usize>,

#[serde(default, flatten)]
#[config_doc(omitted)]
pub unrecognized: Unrecognized<Self>,
Expand Down Expand Up @@ -1574,6 +1577,18 @@ pub mod default {
pub fn compactor_concurrent_uploading_sst_count() -> Option<usize> {
None
}

pub fn table_info_statistic_history_times() -> usize {
240
}

pub fn block_file_cache_flush_buffer_threshold_mb() -> usize {
256
}

pub fn meta_file_cache_flush_buffer_threshold_mb() -> usize {
64
}
}

pub mod streaming {
Expand Down Expand Up @@ -1634,6 +1649,10 @@ pub mod default {
pub fn compression() -> String {
"none".to_string()
}

pub fn flush_buffer_threshold_mb() -> Option<usize> {
None
}
}

pub mod cache_refill {
Expand Down Expand Up @@ -2102,6 +2121,8 @@ pub struct StorageMemoryConfig {
pub prefetch_buffer_capacity_mb: usize,
pub block_cache_eviction_config: EvictionConfig,
pub meta_cache_eviction_config: EvictionConfig,
pub block_file_cache_flush_buffer_threshold_mb: usize,
pub meta_file_cache_flush_buffer_threshold_mb: usize,
}

pub const MAX_META_CACHE_SHARD_BITS: usize = 4;
Expand Down Expand Up @@ -2213,6 +2234,17 @@ pub fn extract_storage_memory_config(s: &RwConfig) -> StorageMemoryConfig {
}
});

let block_file_cache_flush_buffer_threshold_mb = s
.storage
.data_file_cache
.flush_buffer_threshold_mb
.unwrap_or(default::storage::block_file_cache_flush_buffer_threshold_mb());
let meta_file_cache_flush_buffer_threshold_mb = s
.storage
.meta_file_cache
.flush_buffer_threshold_mb
.unwrap_or(default::storage::block_file_cache_flush_buffer_threshold_mb());

StorageMemoryConfig {
block_cache_capacity_mb,
block_cache_shard_num,
Expand All @@ -2223,6 +2255,8 @@ pub fn extract_storage_memory_config(s: &RwConfig) -> StorageMemoryConfig {
prefetch_buffer_capacity_mb,
block_cache_eviction_config,
meta_cache_eviction_config,
block_file_cache_flush_buffer_threshold_mb,
meta_file_cache_flush_buffer_threshold_mb,
}
}

Expand Down
20 changes: 20 additions & 0 deletions src/compute/src/memory/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,24 @@ pub fn storage_memory_config(
((non_reserved_memory_bytes as f64 * compactor_memory_proportion).ceil() as usize) >> 20,
);

// The file cache flush buffer threshold is used as a emergency limitation.
// On most cases the flush buffer is not supposed to be as large as the threshold.
// So, the file cache flush buffer threshold size is not calculated in the memory usage.
let block_file_cache_flush_buffer_threshold_mb = storage_config
.data_file_cache
.flush_buffer_threshold_mb
.unwrap_or(
risingwave_common::config::default::storage::block_file_cache_flush_buffer_threshold_mb(
),
);
let meta_file_cache_flush_buffer_threshold_mb = storage_config
.meta_file_cache
.flush_buffer_threshold_mb
.unwrap_or(
risingwave_common::config::default::storage::meta_file_cache_flush_buffer_threshold_mb(
),
);

let total_calculated_mb = block_cache_capacity_mb
+ meta_cache_capacity_mb
+ shared_buffer_capacity_mb
Expand Down Expand Up @@ -276,6 +294,8 @@ pub fn storage_memory_config(
prefetch_buffer_capacity_mb,
block_cache_eviction_config,
meta_cache_eviction_config,
block_file_cache_flush_buffer_threshold_mb,
meta_file_cache_flush_buffer_threshold_mb,
}
}

Expand Down
4 changes: 4 additions & 0 deletions src/storage/src/opts.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,7 @@ pub struct StorageOpts {
pub data_file_cache_insert_rate_limit_mb: usize,
pub data_file_cache_indexer_shards: usize,
pub data_file_cache_compression: String,
pub data_file_cache_flush_buffer_threshold_mb: usize,

pub cache_refill_data_refill_levels: Vec<u32>,
pub cache_refill_timeout_ms: u64,
Expand All @@ -108,6 +109,7 @@ pub struct StorageOpts {
pub meta_file_cache_insert_rate_limit_mb: usize,
pub meta_file_cache_indexer_shards: usize,
pub meta_file_cache_compression: String,
pub meta_file_cache_flush_buffer_threshold_mb: usize,

/// The storage url for storing backups.
pub backup_storage_url: String,
Expand Down Expand Up @@ -183,6 +185,7 @@ impl From<(&RwConfig, &SystemParamsReader, &StorageMemoryConfig)> for StorageOpt
data_file_cache_insert_rate_limit_mb: c.storage.data_file_cache.insert_rate_limit_mb,
data_file_cache_indexer_shards: c.storage.data_file_cache.indexer_shards,
data_file_cache_compression: c.storage.data_file_cache.compression.clone(),
data_file_cache_flush_buffer_threshold_mb: s.block_file_cache_flush_buffer_threshold_mb,
meta_file_cache_dir: c.storage.meta_file_cache.dir.clone(),
meta_file_cache_capacity_mb: c.storage.meta_file_cache.capacity_mb,
meta_file_cache_file_capacity_mb: c.storage.meta_file_cache.file_capacity_mb,
Expand All @@ -192,6 +195,7 @@ impl From<(&RwConfig, &SystemParamsReader, &StorageMemoryConfig)> for StorageOpt
meta_file_cache_insert_rate_limit_mb: c.storage.meta_file_cache.insert_rate_limit_mb,
meta_file_cache_indexer_shards: c.storage.meta_file_cache.indexer_shards,
meta_file_cache_compression: c.storage.meta_file_cache.compression.clone(),
meta_file_cache_flush_buffer_threshold_mb: s.meta_file_cache_flush_buffer_threshold_mb,
cache_refill_data_refill_levels: c.storage.cache_refill.data_refill_levels.clone(),
cache_refill_timeout_ms: c.storage.cache_refill.timeout_ms,
cache_refill_concurrency: c.storage.cache_refill.concurrency,
Expand Down
10 changes: 6 additions & 4 deletions src/storage/src/store_impl.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -279,7 +279,7 @@ pub mod verify {

// TODO: may avoid manual async fn when the bug of rust compiler is fixed. Currently it will
// fail to compile.
#[allow(clippy::manual_async_fn)]
#[expect(clippy::manual_async_fn)]
fn iter(
&self,
key_range: TableKeyRange,
Expand All @@ -301,7 +301,7 @@ pub mod verify {
}
}

#[allow(clippy::manual_async_fn)]
#[expect(clippy::manual_async_fn)]
fn rev_iter(
&self,
key_range: TableKeyRange,
Expand Down Expand Up @@ -430,7 +430,7 @@ pub mod verify {
actual
}

#[allow(clippy::manual_async_fn)]
#[expect(clippy::manual_async_fn)]
fn iter(
&self,
key_range: TableKeyRange,
Expand All @@ -451,7 +451,7 @@ pub mod verify {
}
}

#[allow(clippy::manual_async_fn)]
#[expect(clippy::manual_async_fn)]
fn rev_iter(
&self,
key_range: TableKeyRange,
Expand Down Expand Up @@ -649,6 +649,7 @@ impl StateStoreImpl {
.with_indexer_shards(opts.meta_file_cache_indexer_shards)
.with_flushers(opts.meta_file_cache_flushers)
.with_reclaimers(opts.meta_file_cache_reclaimers)
.with_buffer_threshold(opts.meta_file_cache_flush_buffer_threshold_mb * MB) // 128 MiB
.with_clean_region_threshold(
opts.meta_file_cache_reclaimers + opts.meta_file_cache_reclaimers / 2,
)
Expand Down Expand Up @@ -701,6 +702,7 @@ impl StateStoreImpl {
.with_indexer_shards(opts.data_file_cache_indexer_shards)
.with_flushers(opts.data_file_cache_flushers)
.with_reclaimers(opts.data_file_cache_reclaimers)
.with_buffer_threshold(opts.data_file_cache_flush_buffer_threshold_mb * MB) // 128 MiB
.with_clean_region_threshold(
opts.data_file_cache_reclaimers + opts.data_file_cache_reclaimers / 2,
)
Expand Down
Loading