Introduce bifrost benchpress

We introduce bifrost benchpress, a binary that's used to run simulation perf tests focused on bifrost loglets performance. Initial code contains tests to reliably measure local loglet's critical-path single log append latency and write-to-read latency with a given configuration.

The benchmark utility is built to give full control over how the benchmark runs, in contrast to criterion which is designed for micro benches.

Cargo.toml

@@ -6,6 +6,7 @@ members = [
     "server",
     "tools/service-protocol-wireshark-dissector",
     "tools/xtask",
+    "tools/bifrost-benchpress",
 ]
 default-members = [
     "cli",

crates/bifrost/src/error.rs

@@ -21,9 +21,9 @@ use crate::types::SealReason;
 pub enum Error {
     #[error("log '{0}' is sealed")]
     LogSealed(LogId, SealReason),
-    #[error("unknown log '{0}")]
+    #[error("unknown log '{0}'")]
     UnknownLogId(LogId),
-    #[error("invalid log sequence number '{0}")]
+    #[error("invalid log sequence number '{0}'")]
     InvalidLsn(Lsn),
     #[error("operation failed due to an ongoing shutdown")]
     Shutdown(#[from] ShutdownError),

crates/types/src/config/admin.rs

@@ -37,10 +37,6 @@ pub struct AdminOptions {
     concurrent_api_requests_limit: Option<NonZeroUsize>,
     pub query_engine: QueryEngineOptions,
 
-    #[cfg(any(test, feature = "test-util"))]
-    #[serde(skip, default = "super::default_arc_tmp")]
-    data_dir: std::sync::Arc<tempfile::TempDir>,
-
     /// # Controller heartbeats
     ///
     /// Controls the interval at which cluster controller polls nodes of the cluster.
@@ -50,16 +46,10 @@ pub struct AdminOptions {
 }
 
 impl AdminOptions {
-    #[cfg(not(any(test, feature = "test-util")))]
     pub fn data_dir(&self) -> PathBuf {
         super::data_dir("registry")
     }
 
-    #[cfg(any(test, feature = "test-util"))]
-    pub fn data_dir(&self) -> PathBuf {
-        self.data_dir.path().join("registry")
-    }
-
     pub fn concurrent_api_requests_limit(&self) -> usize {
         std::cmp::min(
             self.concurrent_api_requests_limit
@@ -77,8 +67,6 @@ impl Default for AdminOptions {
             // max is limited by Tower's LoadShedLayer.
             concurrent_api_requests_limit: None,
             query_engine: Default::default(),
-            #[cfg(any(test, feature = "test-util"))]
-            data_dir: super::default_arc_tmp(),
             heartbeat_interval: Duration::from_millis(1500).into(),
         }
     }

crates/types/src/config/bifrost.rs

@@ -75,22 +75,12 @@ pub struct LocalLogletOptions {
     ///
     /// Default: True.
     pub batch_wal_flushes: bool,
-
-    #[cfg(any(test, feature = "test-util"))]
-    #[serde(skip, default = "super::default_arc_tmp")]
-    data_dir: std::sync::Arc<tempfile::TempDir>,
 }
 
 impl LocalLogletOptions {
-    #[cfg(not(any(test, feature = "test-util")))]
     pub fn data_dir(&self) -> PathBuf {
         super::data_dir("local-loglet")
     }
-
-    #[cfg(any(test, feature = "test-util"))]
-    pub fn data_dir(&self) -> PathBuf {
-        self.data_dir.path().join("local-loglet")
-    }
 }
 
 impl Default for LocalLogletOptions {
@@ -106,8 +96,6 @@ impl Default for LocalLogletOptions {
             sync_wal_before_ack: true,
             writer_batch_commit_count: 500,
             writer_batch_commit_duration: Duration::ZERO.into(),
-            #[cfg(any(test, feature = "test-util"))]
-            data_dir: super::default_arc_tmp(),
         }
     }
 }

crates/types/src/config/common.rs

@@ -225,6 +225,12 @@ impl CommonOptions {
     pub fn cluster_name(&self) -> &str {
         &self.cluster_name
     }
+
+    #[cfg(feature = "test-util")]
+    pub fn set_base_dir(&mut self, path: PathBuf) {
+        self.base_dir = Some(path);
+    }
+
     pub fn base_dir(&self) -> PathBuf {
         self.base_dir.clone().unwrap_or_else(|| {
             std::env::current_dir()

crates/types/src/config/mod.rs

@@ -52,17 +52,18 @@ use crate::errors::GenericError;
 use crate::nodes_config::Role;
 
 #[cfg(any(test, feature = "test-util"))]
-pub(crate) fn default_arc_tmp() -> std::sync::Arc<tempfile::TempDir> {
-    std::sync::Arc::new(tempfile::TempDir::new().unwrap())
+enum TempOrPath {
+    Temp(tempfile::TempDir),
+    Path(PathBuf),
 }
 
 static CONFIGURATION: Lazy<Arc<ArcSwap<Configuration>>> = Lazy::new(Arc::default);
 #[cfg(not(any(test, feature = "test-util")))]
 static NODE_BASE_DIR: std::sync::OnceLock<PathBuf> = std::sync::OnceLock::new();
 
 #[cfg(any(test, feature = "test-util"))]
-static NODE_BASE_DIR: Lazy<std::sync::RwLock<tempfile::TempDir>> =
-    Lazy::new(|| std::sync::RwLock::new(tempfile::TempDir::new().unwrap()));
+static NODE_BASE_DIR: Lazy<std::sync::RwLock<TempOrPath>> =
+    Lazy::new(|| std::sync::RwLock::new(TempOrPath::Temp(tempfile::TempDir::new().unwrap())));
 
 pub type UpdateableConfiguration = Arc<ArcSwap<Configuration>>;
 
@@ -77,7 +78,10 @@ fn data_dir(dir: &str) -> PathBuf {
 #[cfg(any(test, feature = "test-util"))]
 pub fn data_dir(dir: &str) -> PathBuf {
     let guard = NODE_BASE_DIR.read().unwrap();
-    guard.path().join(dir)
+    match &*guard {
+        TempOrPath::Temp(temp) => temp.path().join(dir),
+        TempOrPath::Path(path) => path.join(dir),
+    }
 }
 
 pub fn node_filepath(filename: &str) -> PathBuf {
@@ -88,8 +92,17 @@ pub fn node_filepath(filename: &str) -> PathBuf {
 pub fn reset_base_temp_dir() -> PathBuf {
     let mut guard = NODE_BASE_DIR.write().unwrap();
     let new = tempfile::TempDir::new().unwrap();
-    *guard = new;
-    PathBuf::from(guard.path())
+    let path = PathBuf::from(new.path());
+    *guard = TempOrPath::Temp(new);
+    path
+}
+
+#[cfg(any(test, feature = "test-util"))]
+pub fn reset_base_temp_dir_and_retain() -> PathBuf {
+    let mut guard = NODE_BASE_DIR.write().unwrap();
+    let path = tempfile::TempDir::new().unwrap().into_path();
+    *guard = TempOrPath::Path(path.clone());
+    path
 }
 
 /// Set the current configuration, this is temporary until we have a dedicated configuration loader

crates/types/src/config/worker.rs

@@ -196,22 +196,12 @@ impl Default for InvokerOptions {
 pub struct StorageOptions {
     #[serde(flatten)]
     pub rocksdb: RocksDbOptions,
-
-    #[cfg(any(test, feature = "test-util"))]
-    #[serde(skip, default = "super::default_arc_tmp")]
-    data_dir: std::sync::Arc<tempfile::TempDir>,
 }
 
 impl StorageOptions {
-    #[cfg(not(any(test, feature = "test-util")))]
     pub fn data_dir(&self) -> PathBuf {
         super::data_dir("db")
     }
-
-    #[cfg(any(test, feature = "test-util"))]
-    pub fn data_dir(&self) -> PathBuf {
-        self.data_dir.path().join("db")
-    }
 }
 
 impl Default for StorageOptions {
@@ -221,10 +211,6 @@ impl Default for StorageOptions {
             .build()
             .expect("valid RocksDbOptions");
 
-        StorageOptions {
-            rocksdb,
-            #[cfg(any(test, feature = "test-util"))]
-            data_dir: super::default_arc_tmp(),
-        }
+        StorageOptions { rocksdb }
     }
 }

server/Cargo.toml

@@ -32,7 +32,6 @@ restate-errors = { workspace = true, features = ["include_doc"] }
 restate-fs-util = { workspace = true }
 restate-node = { workspace = true }
 restate-rocksdb = { workspace = true }
-#restate-partition-store = { workspace = true }
 restate-tracing-instrumentation = { workspace = true, features = ["rt-tokio"] }
 restate-types = { workspace = true, features = ["clap"] }
 restate-worker = { workspace = true }

tools/bifrost-benchpress/Cargo.toml

@@ -0,0 +1,50 @@
+[package]
+name = "bifrost-benchpress"
+version.workspace = true
+authors.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+publish = false
+
+[dependencies]
+restate-bifrost = { workspace = true, features = ["test-util"] }
+restate-core = { workspace = true, features = ["test-util"] }
+restate-errors = { workspace = true }
+restate-metadata-store = { workspace = true }
+restate-rocksdb = { workspace = true }
+restate-server = { workspace = true }
+restate-test-util = { workspace = true }
+restate-tracing-instrumentation = { workspace = true, features = ["console-subscriber"] }
+restate-types = { workspace = true, features = ["test-util"] }
+
+anyhow = { workspace = true }
+bytes = { workspace = true }
+bytestring = { workspace = true, features = ["serde"] }
+clap = { workspace = true, features = ["derive", "env", "color", "help", "wrap_help", "usage", "suggestions", "error-context", "std"] }
+codederror = { workspace = true }
+figment = { version = "0.10.8", features = ["env", "toml"] }
+futures = { workspace = true }
+hdrhistogram = { version = "7.5.4" }
+metrics = { workspace = true }
+metrics-exporter-prometheus = { version = "0.14", default-features = false, features = ["async-runtime"] }
+once_cell = { workspace = true }
+quanta = { version = "0.12.3" }
+rocksdb = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+serde_with = { workspace = true }
+smallvec = { workspace = true }
+static_assertions = { workspace = true }
+strum = { workspace = true }
+strum_macros = { workspace = true }
+tempfile = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["test-util"] }
+tokio-stream = { workspace = true }
+toml = { version = "0.8" }
+tracing = { workspace = true }
+tracing-subscriber = { workspace = true }
+
+[target.'cfg(not(target_env = "msvc"))'.dependencies]
+tikv-jemallocator = { workspace = true, features = ["unprefixed_malloc_on_supported_platforms"] }

tools/bifrost-benchpress/README.md

@@ -0,0 +1,8 @@
+# Bifrost Bench Press
+
+A simple tool for benchmarking bifrost loglet implementations
+
+### How to run?
+```sh
+RUST_LOG=info cargo run --profile=bench --bin bifrost-benchpress -- --config-file=restate.toml --retain-test-dir write-to-read
+```