feat: support disabled, unified, and separated runtime

examples/hybrid_full.rs

@@ -18,7 +18,7 @@ use anyhow::Result;
 use chrono::Datelike;
 use foyer::{
     DirectFsDeviceOptionsBuilder, FifoPicker, HybridCache, HybridCacheBuilder, LruConfig, RateLimitPicker, RecoverMode,
-    RuntimeConfig, TombstoneLogConfigBuilder,
+    RuntimeConfig, TokioRuntimeConfig, TombstoneLogConfigBuilder,
 };
 use tempfile::tempdir;
 
@@ -59,9 +59,15 @@ async fn main() -> Result<()> {
                 .with_flush(true)
                 .build(),
         )
-        .with_runtime_config(RuntimeConfig {
-            worker_threads: 4,
-            max_blocking_threads: 8,
+        .with_runtime_config(RuntimeConfig::Separated {
+            read_runtime_config: TokioRuntimeConfig {
+                worker_threads: 4,
+                max_blocking_threads: 8,
+            },
+            write_runtime_config: TokioRuntimeConfig {
+                worker_threads: 4,
+                max_blocking_threads: 8,
+            },
         })
         .build()
         .await?;

foyer-bench/src/main.rs

@@ -20,7 +20,7 @@
 use foyer::{
     Compression, DirectFileDeviceOptionsBuilder, DirectFsDeviceOptionsBuilder, FifoConfig, FifoPicker, HybridCache,
     HybridCacheBuilder, InvalidRatioPicker, LfuConfig, LruConfig, RateLimitPicker, RecoverMode, RuntimeConfig,
-    S3FifoConfig, TracingConfig,
+    S3FifoConfig, TokioRuntimeConfig, TracingConfig,
 };
 use metrics_exporter_prometheus::PrometheusBuilder;
 
@@ -37,7 +37,7 @@
 };
 
 use analyze::{analyze, monitor, Metrics};
-use clap::{ArgGroup, Parser};
+use clap::{builder::PossibleValuesParser, ArgGroup, Parser};
 
 use futures::future::join_all;
 use itertools::Itertools;
@@ -156,16 +156,49 @@
     #[arg(long, default_value_t = false)]
     metrics: bool,
 
+    /// Benchmark user runtime worker threads.
     #[arg(long, default_value_t = 0)]
-    benchmark_runtime_worker_threads: usize,
+    user_runtime_worker_threads: usize,
 
-    /// dedicated runtime worker threads
+    /// Dedicated runtime type.
+    #[arg(long, value_parser = PossibleValuesParser::new(["disabled", "unified", "separated"]), default_value = "disabled")]
+    runtime: String,
+
+    /// Dedicated runtime worker threads.
+    ///
+    /// Only valid when using unified dedicated runtime.
     #[arg(long, default_value_t = 0)]
     runtime_worker_threads: usize,
 
-    /// max threads for blocking io
+    /// Max threads for blocking io.
+    ///
+    /// Only valid when using unified dedicated runtime.
+    #[arg(long, default_value_t = 0)]
+    runtime_max_blocking_threads: usize,
+
+    /// Dedicated runtime for writes worker threads.
+    ///
+    /// Only valid when using separated dedicated runtime.
+    #[arg(long, default_value_t = 0)]
+    write_runtime_worker_threads: usize,
+
+    /// Dedicated runtime for writes Max threads for blocking io.
+    ///
+    /// Only valid when using separated dedicated runtime.
+    #[arg(long, default_value_t = 0)]
+    write_runtime_max_blocking_threads: usize,
+
+    /// Dedicated runtime for reads worker threads.
+    ///
+    /// Only valid when using separated dedicated runtime.
+    #[arg(long, default_value_t = 0)]
+    read_runtime_worker_threads: usize,
+
+    /// Dedicated runtime for writes max threads for blocking io.
+    ///
+    /// Only valid when using separated dedicated runtime.
     #[arg(long, default_value_t = 0)]
-    max_blocking_threads: usize,
+    read_runtime_max_blocking_threads: usize,
 
     /// compression algorithm
     #[arg(long, value_enum, default_value_t = Compression::None)]
@@ -379,8 +412,8 @@
     println!("{:#?}", args);
 
     let mut builder = tokio::runtime::Builder::new_multi_thread();
-    if args.benchmark_runtime_worker_threads != 0 {
-        builder.worker_threads(args.benchmark_runtime_worker_threads);
+    if args.user_runtime_worker_threads != 0 {
+        builder.worker_threads(args.user_runtime_worker_threads);
     }
     builder.thread_name("foyer-bench");
     let runtime = builder.enable_all().build().unwrap();
@@ -484,9 +517,23 @@
             Box::<FifoPicker>::default(),
         ])
         .with_compression(args.compression)
-        .with_runtime_config(RuntimeConfig {
-            worker_threads: args.runtime_worker_threads,
-            max_blocking_threads: args.max_blocking_threads,
+        .with_runtime_config(match args.runtime.as_str() {
+            "disabled" => RuntimeConfig::Disabled,
+            "unified" => RuntimeConfig::Unified(TokioRuntimeConfig {
+                worker_threads: args.runtime_worker_threads,
+                max_blocking_threads: args.runtime_max_blocking_threads,
+            }),
+            "separated" => RuntimeConfig::Separated {
+                read_runtime_config: TokioRuntimeConfig {
+                    worker_threads: args.read_runtime_worker_threads,
+                    max_blocking_threads: args.read_runtime_max_blocking_threads,
+                },
+                write_runtime_config: TokioRuntimeConfig {
+                    worker_threads: args.write_runtime_worker_threads,
+                    max_blocking_threads: args.write_runtime_max_blocking_threads,
+                },
+            },
+            _ => unreachable!(),
         });
 
     if args.admission_rate_limit > 0 {

foyer-storage/src/engine.rs

@@ -304,13 +304,4 @@ where
             Engine::Combined(storage) => StoreFuture::Combined(storage.wait()),
         }
     }
-
-    fn runtime(&self) -> &tokio::runtime::Handle {
-        match self {
-            Engine::Noop(storage) => storage.runtime(),
-            Engine::Large(storage) => storage.runtime(),
-            Engine::Small(storage) => storage.runtime(),
-            Engine::Combined(storage) => storage.runtime(),
-        }
-    }
 }

foyer-storage/src/large/generic.rs

@@ -83,6 +83,9 @@
     pub reinsertion_picker: Arc<dyn ReinsertionPicker<Key = K>>,
     pub tombstone_log_config: Option<TombstoneLogConfig>,
     pub statistics: Arc<Statistics>,
+    pub read_runtime_handle: Handle,
+    pub write_runtime_handle: Handle,
+    pub user_runtime_handle: Handle,
     pub marker: PhantomData<(V, S)>,
 }
 
@@ -109,6 +112,8 @@
             .field("reinsertion_pickers", &self.reinsertion_picker)
             .field("tombstone_log_config", &self.tombstone_log_config)
             .field("statistics", &self.statistics)
+            .field("read_runtime_handle", &self.read_runtime_handle)
+            .field("write_runtime_handle", &self.write_runtime_handle)
             .finish()
     }
 }
@@ -152,7 +157,9 @@
 
     sequence: AtomicSequence,
 
-    runtime: Handle,
+    _read_runtime_handle: Handle,
+    write_runtime_handle: Handle,
+    _user_runtime_handle: Handle,
 
     active: AtomicBool,
 
@@ -179,8 +186,6 @@
     S: HashBuilder + Debug,
 {
     async fn open(mut config: GenericLargeStorageConfig<K, V, S>) -> Result<Self> {
-        let runtime = Handle::current();
-
         let stats = config.statistics.clone();
 
         let device = config.device.clone();
@@ -223,7 +228,7 @@
             &indexer,
             &region_manager,
             &tombstones,
-            runtime.clone(),
+            config.user_runtime_handle.clone(),
         )
         .await?;
 
@@ -236,7 +241,7 @@
                 tombstone_log.clone(),
                 stats.clone(),
                 metrics.clone(),
-                runtime.clone(),
+                config.write_runtime_handle.clone(),
             )
             .await
         }))
@@ -251,7 +256,7 @@
                 flushers.clone(),
                 stats.clone(),
                 config.flush,
-                runtime.clone(),
+                config.write_runtime_handle.clone(),
             )
             .await
         }))
@@ -267,7 +272,9 @@
                 statistics: stats,
                 flush: config.flush,
                 sequence,
-                runtime,
+                _read_runtime_handle: config.read_runtime_handle,
+                write_runtime_handle: config.write_runtime_handle,
+                _user_runtime_handle: config.user_runtime_handle,
                 active: AtomicBool::new(true),
                 metrics,
             }),
@@ -375,7 +382,7 @@
         });
 
         let this = self.clone();
-        self.inner.runtime.spawn(async move {
+        self.inner.write_runtime_handle.spawn(async move {
             let sequence = this.inner.sequence.fetch_add(1, Ordering::Relaxed);
             this.inner.flushers[sequence as usize % this.inner.flushers.len()].submit(Submission::Tombstone {
                 tombstone: Tombstone { hash, sequence },
@@ -424,10 +431,6 @@
 
         Ok(())
     }
-
-    fn runtime(&self) -> &Handle {
-        &self.inner.runtime
-    }
 }
 
 impl<K, V, S> Storage for GenericLargeStorage<K, V, S>
@@ -476,10 +479,6 @@
     fn wait(&self) -> impl Future<Output = ()> + Send + 'static {
         self.wait()
     }
-
-    fn runtime(&self) -> &Handle {
-        self.runtime()
-    }
 }
 
 #[cfg(test)]
@@ -554,6 +553,9 @@
             tombstone_log_config: None,
             buffer_threshold: 16 * 1024 * 1024,
             statistics: Arc::<Statistics>::default(),
+            read_runtime_handle: Handle::current(),
+            write_runtime_handle: Handle::current(),
+            user_runtime_handle: Handle::current(),
             marker: PhantomData,
         };
         GenericLargeStorage::open(config).await.unwrap()
@@ -582,6 +584,9 @@
             tombstone_log_config: Some(TombstoneLogConfigBuilder::new(path).with_flush(true).build()),
             buffer_threshold: 16 * 1024 * 1024,
             statistics: Arc::<Statistics>::default(),
+            read_runtime_handle: Handle::current(),
+            write_runtime_handle: Handle::current(),
+            user_runtime_handle: Handle::current(),
             marker: PhantomData,
         };
         GenericLargeStorage::open(config).await.unwrap()

foyer-storage/src/prelude.rs

@@ -32,5 +32,5 @@ pub use crate::{
     },
     statistics::Statistics,
     storage::{either::Order, Storage},
-    store::{CombinedConfig, DeviceConfig, RuntimeConfig, Store, StoreBuilder},
+    store::{CombinedConfig, DeviceConfig, RuntimeConfig, RuntimeHandles, Store, StoreBuilder, TokioRuntimeConfig},
 };

foyer-storage/src/small/generic.rs

@@ -121,8 +121,4 @@ where
     fn wait(&self) -> impl Future<Output = ()> + Send + 'static {
         async { todo!() }
     }
-
-    fn runtime(&self) -> &tokio::runtime::Handle {
-        todo!()
-    }
 }

foyer-storage/src/storage/either.rs

@@ -18,7 +18,6 @@ use futures::{
     future::{join, ready, select, try_join, Either as EitherFuture},
     pin_mut, Future, FutureExt,
 };
-use tokio::runtime::Handle;
 
 use crate::{error::Result, serde::KvInfo, storage::Storage, DeviceStats, IoBytes};
 
@@ -290,15 +289,4 @@ where
     fn wait(&self) -> impl Future<Output = ()> + Send + 'static {
         join(self.left.wait(), self.right.wait()).map(|_| ())
     }
-
-    fn runtime(&self) -> &Handle {
-        if cfg!(debug_assertions) {
-            let hleft = self.left.runtime();
-            let hright = self.right.runtime();
-            assert_eq!(hleft.id(), hright.id());
-            hleft
-        } else {
-            self.left.runtime()
-        }
-    }
 }

foyer-storage/src/storage/mod.rs

@@ -19,7 +19,6 @@ use std::{fmt::Debug, future::Future, sync::Arc};
 
 use foyer_common::code::{HashBuilder, StorageKey, StorageValue};
 use foyer_memory::CacheEntry;
-use tokio::runtime::Handle;
 
 use crate::{device::monitor::DeviceStats, error::Result, serde::KvInfo, IoBytes};
 
@@ -74,9 +73,4 @@ pub trait Storage: Send + Sync + 'static + Clone + Debug {
     /// Wait for the ongoing flush and reclaim tasks to finish.
     #[must_use]
     fn wait(&self) -> impl Future<Output = ()> + Send + 'static;
-
-    /// Get disk cache runtime handle.
-    ///
-    /// The runtime is determined during the opening phase.
-    fn runtime(&self) -> &Handle;
 }

foyer-storage/src/storage/noop.rs

@@ -19,7 +19,6 @@
 use foyer_memory::CacheEntry;
 
 use futures::future::ready;
-use tokio::runtime::Handle;
 
 use crate::device::monitor::DeviceStats;
 use crate::serde::KvInfo;
@@ -34,7 +33,6 @@
     V: StorageValue,
     S: HashBuilder + Debug,
 {
-    runtime: Handle,
     _marker: PhantomData<(K, V, S)>,
 }
 
@@ -56,10 +54,7 @@
     S: HashBuilder + Debug,
 {
     fn clone(&self) -> Self {
-        Self {
-            runtime: self.runtime.clone(),
-            _marker: PhantomData,
-        }
+        Self { _marker: PhantomData }
     }
 }
 
@@ -75,10 +70,7 @@
     type Config = ();
 
     async fn open(_: Self::Config) -> Result<Self> {
-        Ok(Self {
-            runtime: Handle::current(),
-            _marker: PhantomData,
-        })
+        Ok(Self { _marker: PhantomData })
     }
 
     async fn close(&self) -> Result<()> {
@@ -108,10 +100,6 @@
     fn wait(&self) -> impl Future<Output = ()> + Send + 'static {
         ready(())
     }
-
-    fn runtime(&self) -> &Handle {
-        &self.runtime
-    }
 }
 
 #[cfg(test)]