diff --git a/README.md b/README.md index 846c852a..619a4ce5 100644 --- a/README.md +++ b/README.md @@ -130,6 +130,13 @@ use foyer::{ }; use tempfile::tempdir; +fn nonzero(value: usize) -> NonZeroUsize { + match NonZeroUsize::new(value) { + Some(value) => value, + None => unreachable!("example constants are nonzero"), + } +} + #[tokio::main] async fn main() -> anyhow::Result<()> { let dir = tempdir()?; @@ -142,7 +149,7 @@ async fn main() -> anyhow::Result<()> { .with_write_iops(2000) .with_write_throughput(100 * 1024 * 1024) .with_read_throughput(800 * 1024 * 1024) - .with_iops_counter(IopsCounter::PerIoSize(NonZeroUsize::new(128 * 1024).unwrap())), + .with_iops_counter(IopsCounter::PerIoSize(nonzero(128 * 1024))), ) .build()?; @@ -162,10 +169,10 @@ async fn main() -> anyhow::Result<()> { .with_engine_config( BlockEngineConfig::new(device) .with_block_size(16 * 1024 * 1024) - .with_indexer_shards(64) - .with_recover_concurrency(8) - .with_flushers(2) - .with_reclaimers(2) + .with_indexer_shards(nonzero(64)) + .with_recover_concurrency(nonzero(8)) + .with_flushers(nonzero(2)) + .with_reclaimers(nonzero(2)) .with_buffer_pool_size(256 * 1024 * 1024) .with_clean_block_threshold(4) .with_eviction_pickers(vec![Box::::default()]) diff --git a/examples/hybrid_full.rs b/examples/hybrid_full.rs index b82d55b3..566ad249 100644 --- a/examples/hybrid_full.rs +++ b/examples/hybrid_full.rs @@ -20,6 +20,13 @@ use foyer::{ }; use tempfile::tempdir; +fn nonzero(value: usize) -> NonZeroUsize { + match NonZeroUsize::new(value) { + Some(value) => value, + None => unreachable!("example constants are nonzero"), + } +} + #[tokio::main] async fn main() -> anyhow::Result<()> { let dir = tempdir()?; @@ -32,7 +39,7 @@ async fn main() -> anyhow::Result<()> { .with_write_iops(2000) .with_write_throughput(100 * 1024 * 1024) .with_read_throughput(800 * 1024 * 1024) - .with_iops_counter(IopsCounter::PerIoSize(NonZeroUsize::new(128 * 1024).unwrap())), + .with_iops_counter(IopsCounter::PerIoSize(nonzero(128 * 1024))), ) .build()?; @@ -52,10 +59,10 @@ async fn main() -> anyhow::Result<()> { .with_engine_config( BlockEngineConfig::new(device) .with_block_size(16 * 1024 * 1024) - .with_indexer_shards(64) - .with_recover_concurrency(8) - .with_flushers(2) - .with_reclaimers(2) + .with_indexer_shards(nonzero(64)) + .with_recover_concurrency(nonzero(8)) + .with_flushers(nonzero(2)) + .with_reclaimers(nonzero(2)) .with_buffer_pool_size(256 * 1024 * 1024) .with_clean_block_threshold(4) .with_eviction_pickers(vec![Box::::default()]) diff --git a/foyer-bench/src/main.rs b/foyer-bench/src/main.rs index 431494b1..d6ef2467 100644 --- a/foyer-bench/src/main.rs +++ b/foyer-bench/src/main.rs @@ -25,6 +25,7 @@ use std::{ collections::BTreeMap, fs::create_dir_all, net::SocketAddr, + num::NonZeroUsize, ops::{Deref, Range}, sync::{ Arc, @@ -127,12 +128,12 @@ struct Args { block_size: ByteSize, /// Flusher count. - #[arg(long, default_value_t = 4)] - flushers: usize, + #[arg(long, default_value = "4")] + flushers: NonZeroUsize, /// Reclaimer count. - #[arg(long, default_value_t = 4)] - reclaimers: usize, + #[arg(long, default_value = "4")] + reclaimers: NonZeroUsize, /// Writer count. #[arg(long, default_value_t = 16)] @@ -146,8 +147,8 @@ struct Args { recover_mode: RecoverMode, /// Recover concurrency. - #[arg(long, default_value_t = 16)] - recover_concurrency: usize, + #[arg(long, default_value = "16")] + recover_concurrency: NonZeroUsize, /// Disk write iops throttle. #[arg(long, default_value_t = 0)] @@ -170,8 +171,8 @@ struct Args { clean_block_threshold: usize, /// Shards of both in-memory cache and disk cache indexer. - #[arg(long, default_value_t = 64)] - shards: usize, + #[arg(long, default_value = "64")] + shards: NonZeroUsize, /// weigher to enable metrics exporter #[arg(long, default_value_t = false)] @@ -277,14 +278,14 @@ struct Args { #[arg(long, default_value_t = false)] direct: bool, - #[arg(long, default_value_t = 1)] - io_uring_threads: usize, + #[arg(long, default_value = "1")] + io_uring_threads: NonZeroUsize, #[arg(long, required = false)] io_uring_cpus: Vec, - #[arg(long, default_value_t = 64)] - io_uring_iodepth: usize, + #[arg(long, default_value = "64")] + io_uring_iodepth: NonZeroUsize, #[arg(long, default_value_t = false)] io_uring_sqpoll: bool, @@ -516,9 +517,9 @@ async fn benchmark(args: Args) { builder .with_metrics_registry(Box::new(PrometheusMetricsRegistry::new(registry))) .memory(args.mem.as_u64() as _) - .with_shards(args.shards) + .with_shards(args.shards.get()) } else { - builder.memory(args.mem.as_u64() as _).with_shards(args.shards) + builder.memory(args.mem.as_u64() as _).with_shards(args.shards.get()) }; let builder = match args.eviction.as_str() { diff --git a/foyer-storage/src/engine/block/engine.rs b/foyer-storage/src/engine/block/engine.rs index 27b0e60b..0ccf73dd 100644 --- a/foyer-storage/src/engine/block/engine.rs +++ b/foyer-storage/src/engine/block/engine.rs @@ -16,6 +16,7 @@ use std::{ fmt::Debug, future::Future, marker::PhantomData, + num::NonZeroUsize, sync::{ Arc, atomic::{AtomicBool, AtomicUsize, Ordering}, @@ -84,10 +85,10 @@ where device: Arc, block_size: usize, compression: Compression, - indexer_shards: usize, - recover_concurrency: usize, - flushers: usize, - reclaimers: usize, + indexer_shards: NonZeroUsize, + recover_concurrency: NonZeroUsize, + flushers: NonZeroUsize, + reclaimers: NonZeroUsize, buffer_pool_size: usize, blob_index_size: usize, submit_queue_size_threshold: usize, @@ -103,6 +104,13 @@ where marker: PhantomData<(K, V, P)>, } +fn nonzero(value: usize) -> NonZeroUsize { + match NonZeroUsize::new(value) { + Some(value) => value, + None => unreachable!("default block engine counts are nonzero"), + } +} + impl Debug for BlockEngineConfig where K: StorageKey, @@ -142,10 +150,10 @@ where device, block_size: 16 * 1024 * 1024, // 16 MiB compression: Compression::default(), - indexer_shards: 64, - recover_concurrency: 8, - flushers: 1, - reclaimers: 1, + indexer_shards: nonzero(64), + recover_concurrency: nonzero(8), + flushers: nonzero(1), + reclaimers: nonzero(1), buffer_pool_size: 16 * 1024 * 1024, // 16 MiB blob_index_size: 4 * 1024, // 4 KiB submit_queue_size_threshold: 16 * 1024 * 1024, // 16 MiB @@ -178,7 +186,7 @@ where /// Set the shard num of the indexer. Each shard has its own lock. /// /// Default: `64`. - pub fn with_indexer_shards(mut self, indexer_shards: usize) -> Self { + pub fn with_indexer_shards(mut self, indexer_shards: NonZeroUsize) -> Self { self.indexer_shards = indexer_shards; self } @@ -186,7 +194,7 @@ where /// Set the recover concurrency for the disk cache store. /// /// Default: `8`. - pub fn with_recover_concurrency(mut self, recover_concurrency: usize) -> Self { + pub fn with_recover_concurrency(mut self, recover_concurrency: NonZeroUsize) -> Self { self.recover_concurrency = recover_concurrency; self } @@ -196,7 +204,7 @@ where /// The flusher count limits how many blocks can be concurrently written. /// /// Default: `1`. - pub fn with_flushers(mut self, flushers: usize) -> Self { + pub fn with_flushers(mut self, flushers: NonZeroUsize) -> Self { self.flushers = flushers; self } @@ -216,7 +224,7 @@ where /// The reclaimer count limits how many blocks can be concurrently reclaimed. /// /// Default: `1`. - pub fn with_reclaimers(mut self, reclaimers: usize) -> Self { + pub fn with_reclaimers(mut self, reclaimers: NonZeroUsize) -> Self { self.reclaimers = reclaimers; self } @@ -355,13 +363,14 @@ where None }; - let indexer = Indexer::new(self.indexer_shards); + let indexer = Indexer::new(self.indexer_shards.get()); let submit_queue_size = Arc::::default(); #[expect(clippy::type_complexity)] - let (flushers, rxs): (Vec>, Vec>>) = (0..self.flushers) - .map(|id| Flusher::::new(id, submit_queue_size.clone(), metrics.clone())) - .unzip(); + let (flushers, rxs): (Vec>, Vec>>) = + (0..self.flushers.get()) + .map(|id| Flusher::::new(id, submit_queue_size.clone(), metrics.clone())) + .unzip(); let reclaimer = Reclaimer::new( indexer.clone(), @@ -379,14 +388,14 @@ where block_size, self.eviction_pickers, reclaimer, - self.reclaimers, + self.reclaimers.get(), self.clean_block_threshold, metrics.clone(), runtime.clone(), )?; let blocks = block_manager.blocks(); - if self.flushers + self.clean_block_threshold > blocks / 2 { + if self.flushers.get() + self.clean_block_threshold > blocks / 2 { tracing::warn!( "[block engine]: block-based object disk cache stable blocks count is too small, flusher [{flushers}] + clean block threshold [{clean_block_threshold}] (default = reclaimers) is supposed to be much larger than the block count [{blocks}]", flushers = self.flushers, @@ -397,7 +406,7 @@ where let sequence = AtomicSequence::default(); RecoverRunner::run( - self.recover_concurrency, + self.recover_concurrency.get(), recover_mode, self.blob_index_size, (0..blocks as BlockId).collect_vec(), @@ -410,7 +419,7 @@ where ) .await?; - let io_buffer_size = self.buffer_pool_size / self.flushers; + let io_buffer_size = self.buffer_pool_size / self.flushers.get(); for (flusher, rx) in flushers.iter().zip(rxs) { flusher.run( rx, @@ -905,10 +914,10 @@ mod tests { device, block_size: 16 * 1024, compression: Compression::None, - indexer_shards: 4, - recover_concurrency: 2, - flushers: 1, - reclaimers: 1, + indexer_shards: nonzero(4), + recover_concurrency: nonzero(2), + flushers: nonzero(1), + reclaimers: nonzero(1), clean_block_threshold: 1, admission_filter: StorageFilter::new(), eviction_pickers: vec![Box::::default()], @@ -948,10 +957,10 @@ mod tests { device, block_size: 16 * 1024, compression: Compression::None, - indexer_shards: 4, - recover_concurrency: 2, - flushers: 1, - reclaimers: 1, + indexer_shards: nonzero(4), + recover_concurrency: nonzero(2), + flushers: nonzero(1), + reclaimers: nonzero(1), clean_block_threshold: 1, eviction_pickers: vec![Box::::default()], admission_filter: StorageFilter::new(), diff --git a/foyer-storage/src/io/device/file.rs b/foyer-storage/src/io/device/file.rs index 7d161f9e..19c4e9ea 100644 --- a/foyer-storage/src/io/device/file.rs +++ b/foyer-storage/src/io/device/file.rs @@ -18,7 +18,7 @@ use std::{ sync::{Arc, RwLock}, }; -use foyer_common::error::{Error, Result}; +use foyer_common::error::{Error, ErrorKind, Result}; use fs4::free_space; use crate::{ @@ -73,6 +73,32 @@ impl FileDeviceBuilder { self.direct = direct; self } + + fn free_space_capacity(&self) -> Result { + // Create an empty directory if needed before to get free space. + let dir = self + .path + .parent() + .ok_or_else(|| Error::new(ErrorKind::Config, "file device path must have a parent directory"))?; + create_dir_all(dir).map_err(Error::io_error)?; + Ok(free_space(dir).map_err(Error::io_error)? as usize / 10 * 8) + } + + fn default_capacity(&self) -> Result { + // Try to get the capacity if `path` refer to a raw block device. + #[cfg(unix)] + if let Ok(metadata) = std::fs::metadata(&self.path) { + use std::os::unix::fs::FileTypeExt; + + return if metadata.file_type().is_block_device() { + super::utils::get_dev_capacity(&self.path) + } else { + self.free_space_capacity() + }; + } + + self.free_space_capacity() + } } impl DeviceBuilder for FileDeviceBuilder { @@ -81,27 +107,9 @@ impl DeviceBuilder for FileDeviceBuilder { let align_v = |value: usize, align: usize| value - (value % align); - let capacity = self.capacity.unwrap_or_else(|| { - // Try to get the capacity if `path` refer to a raw block device. - #[cfg(unix)] - if let Ok(metadata) = std::fs::metadata(&self.path) { - let file_type = metadata.file_type(); - - use std::os::unix::fs::FileTypeExt; - if file_type.is_block_device() { - return super::utils::get_dev_capacity(&self.path).unwrap(); - } - } - - // Create an empty directory if needed before to get free space. - let dir = self.path.parent().expect("path must point to a file").to_path_buf(); - create_dir_all(&dir).unwrap(); - free_space(&dir).unwrap() as usize / 10 * 8 - }); + let capacity = self.capacity.map_or_else(|| self.default_capacity(), Ok)?; let capacity = align_v(capacity, PAGE); - println!("==========> {capacity}"); - // Build device. let mut opts = OpenOptions::new(); @@ -114,7 +122,7 @@ impl DeviceBuilder for FileDeviceBuilder { let file = opts.open(&self.path).map_err(Error::io_error)?; - if file.metadata().unwrap().is_file() { + if file.metadata().map_err(Error::io_error)?.is_file() { tracing::warn!( "{} {} {}", "It seems a `DirectFileDevice` is used within a normal file system, which is inefficient.", @@ -138,6 +146,19 @@ impl DeviceBuilder for FileDeviceBuilder { } } +#[cfg(test)] +mod tests { + use foyer_common::error::ErrorKind; + + use super::*; + + #[test] + fn test_file_device_builder_rejects_path_without_parent() { + let err = FileDeviceBuilder::new("").build().unwrap_err(); + assert_eq!(err.kind(), ErrorKind::Config); + } +} + /// A device upon a single file or a raw block device. #[derive(Debug)] pub struct FileDevice { diff --git a/foyer-storage/src/io/device/fs.rs b/foyer-storage/src/io/device/fs.rs index f03d7e0b..fc815c6b 100644 --- a/foyer-storage/src/io/device/fs.rs +++ b/foyer-storage/src/io/device/fs.rs @@ -73,6 +73,12 @@ impl FsDeviceBuilder { self.direct = direct; self } + + fn default_capacity(&self) -> Result { + // Create an empty directory before to get free space. + create_dir_all(&self.dir).map_err(Error::io_error)?; + Ok(free_space(&self.dir).map_err(Error::io_error)? as usize / 10 * 8) + } } impl DeviceBuilder for FsDeviceBuilder { @@ -81,11 +87,7 @@ impl DeviceBuilder for FsDeviceBuilder { let align_v = |value: usize, align: usize| value - value % align; - let capacity = self.capacity.unwrap_or({ - // Create an empty directory before to get free space. - create_dir_all(&self.dir).unwrap(); - free_space(&self.dir).unwrap() as usize / 10 * 8 - }); + let capacity = self.capacity.map_or_else(|| self.default_capacity(), Ok)?; let capacity = align_v(capacity, PAGE); let statistics = Arc::new(Statistics::new(self.throttle)); @@ -177,6 +179,19 @@ impl Device for FsDevice { } } +#[cfg(test)] +mod tests { + use foyer_common::error::ErrorKind; + + use super::*; + + #[test] + fn test_fs_device_builder_propagates_dir_creation_errors() { + let err = FsDeviceBuilder::new("foo\0bar").build().unwrap_err(); + assert_eq!(err.kind(), ErrorKind::Io); + } +} + #[derive(Debug)] pub struct FsPartition { id: PartitionId, diff --git a/foyer-storage/src/io/engine/mod.rs b/foyer-storage/src/io/engine/mod.rs index a67c110c..3b623d05 100644 --- a/foyer-storage/src/io/engine/mod.rs +++ b/foyer-storage/src/io/engine/mod.rs @@ -125,6 +125,9 @@ pub trait IoEngine: Send + Sync + 'static + Debug { #[cfg(test)] mod tests { + #[cfg(not(madsim))] + #[cfg(target_os = "linux")] + use std::num::NonZeroUsize; use std::path::Path; use rand::{Fill, rng}; @@ -143,6 +146,15 @@ mod tests { const KIB: usize = 1024; const MIB: usize = 1024 * 1024; + #[cfg(not(madsim))] + #[cfg(target_os = "linux")] + fn nonzero(value: usize) -> NonZeroUsize { + match NonZeroUsize::new(value) { + Some(value) => value, + None => unreachable!("test constants are nonzero"), + } + } + fn build_test_file_device(path: impl AsRef) -> Result> { let device = FileDeviceBuilder::new(&path).with_capacity(16 * MIB).build()?; for _ in 0..16 { @@ -176,8 +188,8 @@ mod tests { let path = dir.path().join("test_file_1"); let device = build_test_file_device(&path).unwrap(); let engine = UringIoEngineConfig::new() - .with_threads(4) - .with_io_depth(64) + .with_threads(nonzero(4)) + .with_io_depth(nonzero(64)) .boxed() .build(IoEngineBuildContext { spawner: Spawner::current(), diff --git a/foyer-storage/src/io/engine/uring.rs b/foyer-storage/src/io/engine/uring.rs index 21979f2b..6d878b52 100644 --- a/foyer-storage/src/io/engine/uring.rs +++ b/foyer-storage/src/io/engine/uring.rs @@ -14,6 +14,7 @@ use std::{ fmt::Debug, + num::NonZeroUsize, sync::{Arc, mpsc}, }; @@ -35,14 +36,21 @@ use crate::{ }, }; +fn nonzero(value: usize) -> NonZeroUsize { + match NonZeroUsize::new(value) { + Some(value) => value, + None => unreachable!("default io_uring counts are nonzero"), + } +} + /// Config for io_uring based I/O engine. #[derive(Debug)] pub struct UringIoEngineConfig { - threads: usize, - cpus: Vec, - io_depth: usize, + threads: NonZeroUsize, + cpus: Option>, + io_depth: NonZeroUsize, sqpoll: bool, - sqpoll_cpus: Vec, + sqpoll_cpus: Option>, sqpoll_idle: u32, iopoll: bool, weight: f64, @@ -63,11 +71,11 @@ impl UringIoEngineConfig { /// Create a new io_uring based I/O engine config with default configurations. pub fn new() -> Self { Self { - threads: 1, - cpus: vec![], - io_depth: 64, + threads: nonzero(1), + cpus: None, + io_depth: nonzero(64), sqpoll: false, - sqpoll_cpus: vec![], + sqpoll_cpus: None, sqpoll_idle: 10, iopoll: false, weight: 1.0, @@ -79,21 +87,28 @@ impl UringIoEngineConfig { } /// Set the number of threads to use for the I/O engine. - pub fn with_threads(mut self, threads: usize) -> Self { + pub fn with_threads(mut self, threads: NonZeroUsize) -> Self { self.threads = threads; self } /// Bind the engine threads to specific CPUs. /// - /// The length of `cpus` must be equal to the threads. - pub fn with_cpus(mut self, cpus: Vec) -> Self { - self.cpus = cpus; + /// A nonempty CPU list also sets the thread count to the CPU count. An empty CPU list disables pinning. + pub fn with_cpus(mut self, cpus: impl Into>) -> Self { + let cpus = cpus.into(); + match NonZeroUsize::new(cpus.len()) { + Some(threads) => { + self.threads = threads; + self.cpus = Some(cpus); + } + None => self.cpus = None, + } self } /// Set the I/O depth for each thread. - pub fn with_io_depth(mut self, io_depth: usize) -> Self { + pub fn with_io_depth(mut self, io_depth: NonZeroUsize) -> Self { self.io_depth = io_depth; self } @@ -147,9 +162,14 @@ impl UringIoEngineConfig { /// /// This flag is only meaningful when [`Self::with_sqpoll`] is enabled. /// - /// The length of `cpus` must be equal to the number of threads. - pub fn with_sqpoll_cpus(mut self, cpus: Vec) -> Self { - self.sqpoll_cpus = cpus; + /// The length of `cpus` must match the number of threads. An empty CPU list disables pinning. + pub fn with_sqpoll_cpus(mut self, cpus: impl Into>) -> Self { + let cpus = cpus.into(); + if cpus.is_empty() { + self.sqpoll_cpus = None; + } else { + self.sqpoll_cpus = Some(cpus); + } self } @@ -180,19 +200,34 @@ impl UringIoEngineConfig { impl IoEngineConfig for UringIoEngineConfig { fn build(self: Box, _: IoEngineBuildContext) -> BoxFuture<'static, Result>> { async move { - if self.threads == 0 { - return Err(Error::new(ErrorKind::Config, "shards must be greater than 0") - .with_context("threads", self.threads)); + let threads = self.threads.get(); + let io_depth = self.io_depth.get(); + + if let Some(cpus) = self.cpus.as_deref() { + if cpus.len() != threads { + return Err(Error::new(ErrorKind::Config, "cpu count must match threads") + .with_context("threads", threads) + .with_context("cpus", cpus.len())); + } } - let (read_txs, read_rxs): (Vec>, Vec>) = (0..self.threads) + match (self.sqpoll, self.sqpoll_cpus.as_deref()) { + (true, Some(cpus)) if cpus.len() != threads => { + return Err(Error::new(ErrorKind::Config, "sqpoll cpu count must match threads") + .with_context("threads", threads) + .with_context("sqpoll_cpus", cpus.len())); + } + _ => {} + } + + let (read_txs, read_rxs): (Vec>, Vec>) = (0..threads) .map(|_| { let (tx, rx) = mpsc::sync_channel(4096); (tx, rx) }) .unzip(); - let (write_txs, write_rxs): (Vec>, Vec>) = (0..self.threads) + let (write_txs, write_rxs): (Vec>, Vec>) = (0..threads) .map(|_| { let (tx, rx) = mpsc::sync_channel(4096); (tx, rx) @@ -206,18 +241,18 @@ impl IoEngineConfig for UringIoEngineConfig { } if self.sqpoll { builder.setup_sqpoll(self.sqpoll_idle); - if !self.sqpoll_cpus.is_empty() { - let cpu = self.sqpoll_cpus[i]; + if let Some(cpus) = &self.sqpoll_cpus { + let cpu = cpus[i]; builder.setup_sqpoll_cpu(cpu); } } - let cpu = if self.cpus.is_empty() { None } else { Some(self.cpus[i]) }; - let uring = builder.build(self.io_depth as _).map_err(Error::io_error)?; + let cpu = self.cpus.as_ref().map(|cpus| cpus[i]); + let uring = builder.build(io_depth as _).map_err(Error::io_error)?; let shard = UringIoEngineShard { read_rx, write_rx, uring, - io_depth: self.io_depth, + io_depth, weight: self.weight, read_inflight: 0, write_inflight: 0, @@ -454,3 +489,38 @@ impl IoEngine for UringIoEngine { self.write(buf, partition, offset) } } + +#[cfg(test)] +mod tests { + use foyer_common::{error::ErrorKind, spawn::Spawner}; + + use super::*; + + #[test_log::test(tokio::test)] + async fn test_uring_empty_cpu_sets_disable_pinning() { + let config = UringIoEngineConfig::new() + .with_cpus([0, 1]) + .with_sqpoll_cpus([0, 1]) + .with_cpus([]) + .with_sqpoll_cpus([]); + + assert_eq!(config.threads, nonzero(2)); + assert!(config.cpus.is_none()); + assert!(config.sqpoll_cpus.is_none()); + } + + #[test_log::test(tokio::test)] + async fn test_uring_rejects_mismatched_cpu_count() { + let err = UringIoEngineConfig::new() + .with_cpus([0, 1]) + .with_threads(nonzero(3)) + .boxed() + .build(IoEngineBuildContext { + spawner: Spawner::current(), + }) + .await + .unwrap_err(); + + assert_eq!(err.kind(), ErrorKind::Config); + } +} diff --git a/foyer-storage/src/store.rs b/foyer-storage/src/store.rs index 73e1b91d..88862500 100644 --- a/foyer-storage/src/store.rs +++ b/foyer-storage/src/store.rs @@ -512,6 +512,8 @@ where #[cfg(test)] mod tests { + use std::num::NonZeroUsize; + use foyer_common::hasher::ModHasher; use foyer_memory::CacheBuilder; @@ -522,6 +524,13 @@ mod tests { io::{device::fs::FsDeviceBuilder, engine::psync::PsyncIoEngineConfig}, }; + fn nonzero(value: usize) -> NonZeroUsize { + match NonZeroUsize::new(value) { + Some(value) => value, + None => unreachable!("test constants are nonzero"), + } + } + #[tokio::test] async fn test_build_with_unaligned_buffer_pool_size() { let dir = tempfile::tempdir().unwrap(); @@ -536,7 +545,7 @@ mod tests { .build() .unwrap(), ) - .with_flushers(3) + .with_flushers(nonzero(3)) .with_block_size(16 * 1024) .with_buffer_pool_size(128 * 1024 * 1024), ) diff --git a/foyer-storage/tests/storage_fuzzy_test.rs b/foyer-storage/tests/storage_fuzzy_test.rs index ec792814..79b31147 100644 --- a/foyer-storage/tests/storage_fuzzy_test.rs +++ b/foyer-storage/tests/storage_fuzzy_test.rs @@ -16,7 +16,7 @@ #![expect(clippy::identity_op)] -use std::{path::Path, sync::Arc}; +use std::{num::NonZeroUsize, path::Path, sync::Arc}; use foyer_common::{hasher::ModHasher, metrics::Metrics}; use foyer_memory::{Cache, CacheBuilder, FifoConfig, TestProperties}; @@ -30,6 +30,13 @@ const MB: usize = 1024 * 1024; const INSERTS: usize = 100; const LOOPS: usize = 10; +fn nonzero(value: usize) -> NonZeroUsize { + match NonZeroUsize::new(value) { + Some(value) => value, + None => unreachable!("test constants are nonzero"), + } +} + async fn test_store( memory: Cache, ModHasher, TestProperties>, builder: impl Fn( @@ -111,8 +118,8 @@ fn basic( BlockEngineConfig::new(FsDeviceBuilder::new(path).with_capacity(4 * MB).build().unwrap()) .with_admission_filter(StorageFilter::new().with_condition(recorder.admission())) .with_block_size(MB) - .with_recover_concurrency(2) - .with_indexer_shards(4) + .with_recover_concurrency(nonzero(2)) + .with_indexer_shards(nonzero(4)) .with_reinsertion_filter(StorageFilter::new().with_condition(recorder.eviction())), ) }