perf: split large io by flush_io_size in the new model

foyer-storage/src/large/flusher.rs

@@ -28,15 +28,16 @@
     bits,
     code::{StorageKey, StorageValue},
     metrics::Metrics,
+    rate::RateLimiter,
 };
 use foyer_memory::Piece;
 use futures_core::future::BoxFuture;
 use futures_util::{
-    future::{try_join, try_join_all},
+    future::{join_all, try_join, try_join_all},
     FutureExt,
 };
 use itertools::Itertools;
-use tokio::sync::oneshot;
+use tokio::sync::{oneshot, Semaphore};
 
 use super::{
     batch::{Batch, BatchWriter, EntryWriter, Op},
@@ -152,6 +153,8 @@
         stats: Arc<Statistics>,
         metrics: Arc<Metrics>,
         runtime: &Runtime,
+        flush_io_depth_limiter: Arc<Semaphore>,
+        flush_io_throughput_limiter: Option<Arc<RateLimiter>>,
         #[cfg(test)] flush_holder: FlushHolder,
     ) -> Result<Self> {
         let (tx, rx) = flume::unbounded();
@@ -167,6 +170,8 @@
         let current_region_handle = region_manager.get_clean_region();
         let remain = device.region_size();
 
+        let flush_io_size = bits::align_up(PAGE, config.flush_io_size);
+
         let runner = Runner {
             rx: Some(rx),
             writer,
@@ -175,6 +180,9 @@
             rotate_buffer,
             queue_init: None,
             submit_queue_size: submit_queue_size.clone(),
+            flush_io_size,
+            flush_io_depth_limiter,
+            flush_io_throughput_limiter,
             region_manager,
             device,
             indexer,
@@ -229,7 +237,6 @@
     current_region_remain: usize,
     metrics: Arc<Metrics>,
 ) -> BatchWriter {
-    // TODO(MrCroxx): optimize buffer allocation.
     BatchWriter::new(buffer, region_size, current_region_remain, metrics)
 }
 
@@ -271,6 +278,9 @@
     rotate_buffer: Option<IoBuffer>,
 
     submit_queue_size: Arc<AtomicUsize>,
+    flush_io_size: usize,
+    flush_io_depth_limiter: Arc<Semaphore>,
+    flush_io_throughput_limiter: Option<Arc<RateLimiter>>,
 
     current_region_handle: GetCleanRegionHandle,
     remain: usize,
@@ -425,7 +435,6 @@
         waiters: Vec<oneshot::Sender<()>>,
         init: Instant,
     ) -> BoxFuture<'static, IoTaskCtx> {
-        // ) {
         tracing::trace!(
             ?batch,
             ?tombstone_infos,
@@ -481,6 +490,9 @@
                 let flush = self.flush;
                 let slice = shared.absolute_slice(window.absolute_dirty_range.clone());
                 let metrics = self.metrics.clone();
+                let flush_io_size = self.flush_io_size;
+                let flush_io_depth_limiter = self.flush_io_depth_limiter.clone();
+                let flush_io_throughput_limiter = self.flush_io_throughput_limiter.clone();
 
                 async move {
                     // Wait for region is clean.
@@ -495,8 +507,36 @@
                     tracing::trace!(region = region.id(), offset, len, "[flusher]: prepare to write region");
 
                     if !window.is_empty() {
-                        let (_, res) = region.write(slice, offset as _).await;
-                        res?;
+                        let ios = len / flush_io_size + if len % flush_io_size == 0 { 0 } else { 1 };
+                        let ios = (0..ios).map(|i| {
+                            let start = i * flush_io_size;
+                            let end = std::cmp::min(start + flush_io_size, len);
+                            let io_slice = slice.slice(start..end);
+                            let region = region.clone();
+                            let flush_io_depth_limiter = flush_io_depth_limiter.clone();
+                            let flush_io_throughput_limiter = flush_io_throughput_limiter.clone();
+                            async move {
+                                let guard = flush_io_depth_limiter.acquire_owned().await.unwrap();
+                                if let Some(limiter) = flush_io_throughput_limiter {
+                                    if let Some(duration) = limiter.consume(io_slice.len() as _) {
+                                        tokio::time::sleep(duration).await;
+                                    }
+                                }
+                                let res = region.write(io_slice, (offset + start) as _).await;
+                                drop(guard);
+                                res
+                            }
+                        });
+                        let res = join_all(ios).await;
+                        let mut errs = vec![];
+                        for (_, r) in res {
+                            if let Err(e) = r {
+                                errs.push(e);
+                            }
+                        }
+                        if !errs.is_empty() {
+                            return Err(Error::multiple(errs));
+                        }
 
                         if flush {
                             region.flush().await?;

foyer-storage/src/large/generic.rs

@@ -29,6 +29,7 @@
     bits,
     code::{StorageKey, StorageValue},
     metrics::Metrics,
+    rate::RateLimiter,
 };
 use foyer_memory::Piece;
 use futures_util::future::{join_all, try_join_all};
@@ -76,6 +77,9 @@
     pub reclaimers: usize,
     pub buffer_pool_size: usize,
     pub submit_queue_size_threshold: usize,
+    pub flush_io_size: usize,
+    pub flush_io_depth: usize,
+    pub flush_io_throughput: Option<usize>,
     pub clean_region_threshold: usize,
     pub eviction_pickers: Vec<Box<dyn EvictionPicker>>,
     pub reinsertion_picker: Arc<dyn ReinsertionPicker>,
@@ -102,6 +106,9 @@
             .field("reclaimers", &self.reclaimers)
             .field("buffer_pool_size", &self.buffer_pool_size)
             .field("submit_queue_size_threshold", &self.submit_queue_size_threshold)
+            .field("flush_io_size", &self.flush_io_size)
+            .field("flush_io_depth", &self.flush_io_depth)
+            .field("flush_io_throughput", &self.flush_io_throughput)
             .field("clean_region_threshold", &self.clean_region_threshold)
             .field("eviction_pickers", &self.eviction_pickers)
             .field("reinsertion_pickers", &self.reinsertion_picker)
@@ -238,6 +245,8 @@
         #[cfg(test)]
         let flush_holder = FlushHolder::default();
 
+        let flush_io_depth_limiter = Arc::new(Semaphore::new(config.flush_io_depth));
+        let flush_io_throughput_limiter = config.flush_io_throughput.map(|v| Arc::new(RateLimiter::new(v as _)));
         let flushers = try_join_all((0..config.flushers).map(|_| async {
             Flusher::open(
                 &config,
@@ -249,6 +258,8 @@
                 stats.clone(),
                 metrics.clone(),
                 &config.runtime,
+                flush_io_depth_limiter.clone(),
+                flush_io_throughput_limiter.clone(),
                 #[cfg(test)]
                 flush_holder.clone(),
             )
@@ -597,6 +608,9 @@
             tombstone_log_config: None,
             buffer_pool_size: 16 * 1024 * 1024,
             submit_queue_size_threshold: 16 * 1024 * 1024 * 2,
+            flush_io_size: 128 * 1024,
+            flush_io_depth: 256,
+            flush_io_throughput: None,
             statistics: Arc::<Statistics>::default(),
             runtime: Runtime::new(None, None, Handle::current()),
             marker: PhantomData,
@@ -626,6 +640,9 @@
             tombstone_log_config: Some(TombstoneLogConfigBuilder::new(path).with_flush(true).build()),
             buffer_pool_size: 16 * 1024 * 1024,
             submit_queue_size_threshold: 16 * 1024 * 1024 * 2,
+            flush_io_size: 128 * 1024,
+            flush_io_depth: 256,
+            flush_io_throughput: None,
             statistics: Arc::<Statistics>::default(),
             runtime: Runtime::new(None, None, Handle::current()),
             marker: PhantomData,

foyer-storage/src/store.rs

@@ -563,6 +563,9 @@
                                 tombstone_log_config: self.large.tombstone_log_config,
                                 buffer_pool_size: self.large.buffer_pool_size,
                                 submit_queue_size_threshold: self.large.submit_queue_size_threshold.unwrap_or(self.large.buffer_pool_size * 2),
+                                flush_io_size: self.large.flush_io_size,
+                                flush_io_depth: self.large.flush_io_depth,
+                                flush_io_throughput: self.large.flush_io_throughput,
                                 statistics: statistics.clone(),
                                 runtime,
                                 marker: PhantomData,
@@ -621,6 +624,9 @@
                                     tombstone_log_config: self.large.tombstone_log_config,
                                     buffer_pool_size: self.large.buffer_pool_size,
                                     submit_queue_size_threshold: self.large.submit_queue_size_threshold.unwrap_or(self.large.buffer_pool_size * 2),
+                                    flush_io_size: self.large.flush_io_size,
+                                    flush_io_depth: self.large.flush_io_depth,
+                                    flush_io_throughput: self.large.flush_io_throughput,
                                     statistics: statistics.clone(),
                                     runtime,
                                     marker: PhantomData,
@@ -670,6 +676,9 @@
     reclaimers: usize,
     buffer_pool_size: usize,
     submit_queue_size_threshold: Option<usize>,
+    flush_io_size: usize,
+    flush_io_depth: usize,
+    flush_io_throughput: Option<usize>,
     clean_region_threshold: Option<usize>,
     eviction_pickers: Vec<Box<dyn EvictionPicker>>,
     reinsertion_picker: Arc<dyn ReinsertionPicker>,
@@ -704,6 +713,9 @@
             reclaimers: 1,
             buffer_pool_size: 16 * 1024 * 1024, // 16 MiB
             submit_queue_size_threshold: None,
+            flush_io_size: 128 * 1024, // 128 KiB
+            flush_io_depth: 256,
+            flush_io_throughput: None,
             clean_region_threshold: None,
             eviction_pickers: vec![Box::new(InvalidRatioPicker::new(0.8)), Box::<FifoPicker>::default()],
             reinsertion_picker: Arc::<RejectAllPicker>::default(),
@@ -818,6 +830,38 @@
         self.tombstone_log_config = Some(tombstone_log_config);
         self
     }
+
+    /// Limit the maximum flush io size.
+    ///
+    /// One larger disk write while flushing will be split into multiple disk writes based on this value.
+    ///
+    /// The value will be align to 4K if it is not.
+    ///
+    /// Default: 128 * 1024
+    pub fn with_flush_io_size(mut self, flush_io_size: usize) -> Self {
+        self.flush_io_size = flush_io_size;
+        self
+    }
+
+    /// Limit the flush io depth.
+    ///
+    /// Limit the disk writes submitted to the disk based on this value.
+    ///
+    /// Default: 256.
+    pub fn with_flush_io_depth(mut self, flush_io_depth: usize) -> Self {
+        self.flush_io_depth = flush_io_depth;
+        self
+    }
+
+    /// Limit the flush io throughput.
+    ///
+    /// Limit the disk write throughput submitted to the disk based on this value.
+    ///
+    /// Default: Unlimited.
+    pub fn with_flush_io_throughput(mut self, flush_io_throughput: usize) -> Self {
+        self.flush_io_throughput = Some(flush_io_throughput);
+        self
+    }
 }
 
 /// Small object disk cache engine default options.