You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/06/26 19:57:07 UTC
(iggy) 03/12: feat(io_uring): load server_config (#1798)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch io_uring_tpc
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 4d17283057f5e22a6bb25dde2aae8203e4133fc7
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Fri May 16 21:37:23 2025 +0200
feat(io_uring): load server_config (#1798)
---
Cargo.lock | 31 +++++++++++---
core/server/src/bootstrap.rs | 13 +++++-
core/server/src/main.rs | 87 ++++++++++++++++++++++++++--------------
core/server/src/shard/builder.rs | 9 +----
core/server/src/shard/mod.rs | 3 +-
5 files changed, 96 insertions(+), 47 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 2fa04bba..3a3711a5 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -142,7 +142,7 @@ dependencies = [
"actix-utils",
"futures-core",
"futures-util",
- "mio 1.0.3",
+ "mio 1.0.4",
"socket2",
"tokio",
"tracing",
@@ -639,7 +639,7 @@ checksum = "fd73835ad7deb4bd2b389e6f10333b143f025d607c55ca04c66a0bcc6bb2fc6d"
dependencies = [
"proc-macro2",
"quote",
- "syn 2.0.101",
+ "syn 2.0.104",
]
[[package]]
@@ -4626,6 +4626,15 @@ version = "2.7.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32a282da65faaf38286cf3be983213fcf1d2e2a58700e808f83f4ea9a4804bc0"
+[[package]]
+name = "memoffset"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5de893c32cde5f383baa4c04c5d6dbdd735cfd4a794b0debdb2bb1b421da5ff4"
+dependencies = [
+ "autocfg",
+]
+
[[package]]
name = "mimalloc"
version = "0.1.47"
@@ -4675,6 +4684,18 @@ dependencies = [
"adler2",
]
+[[package]]
+name = "mio"
+version = "0.8.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a4a650543ca06a924e8b371db273b2756685faae30f8487da1b56505a8f78b0c"
+dependencies = [
+ "libc",
+ "log",
+ "wasi 0.11.1+wasi-snapshot-preview1",
+ "windows-sys 0.48.0",
+]
+
[[package]]
name = "mio"
version = "1.0.4"
@@ -4763,7 +4784,7 @@ checksum = "176a5f5e69613d9e88337cf2a65e11135332b4efbcc628404a7c555e4452084c"
dependencies = [
"proc-macro2",
"quote",
- "syn 2.0.101",
+ "syn 2.0.104",
]
[[package]]
@@ -4858,7 +4879,7 @@ dependencies = [
"kqueue",
"libc",
"log",
- "mio 1.0.3",
+ "mio 1.0.4",
"notify-types",
"walkdir",
"windows-sys 0.59.0",
@@ -7404,7 +7425,7 @@ dependencies = [
"backtrace",
"bytes",
"libc",
- "mio",
+ "mio 1.0.4",
"parking_lot 0.12.4",
"pin-project-lite",
"signal-hook-registry",
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 8eacf223..fd6a5aa0 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -1,3 +1,5 @@
+use monoio::{Buildable, Driver, Runtime};
+
use crate::shard::{connector::ShardConnector, frame::ShardFrame};
use std::ops::Range;
@@ -12,9 +14,18 @@ pub fn create_shard_connections(shards_set: Range<usize>) -> Vec<ShardConnector<
}
pub async fn create_directories() {
-
+ todo!();
}
pub async fn create_root_user() {
+ todo!();
+}
+pub fn create_default_executor<D>() -> Runtime<D>
+where
+ D: Driver + Buildable,
+{
+ let builder = monoio::RuntimeBuilder::<D>::new();
+ let rt = Buildable::build(builder).expect("Failed to create default runtime");
+ rt
}
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index 381ac578..3b38700f 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -24,7 +24,9 @@ use dotenvy::dotenv;
use figlet_rs::FIGfont;
use monoio::Buildable;
use server::args::Args;
-use server::bootstrap::{create_directories, create_root_user, create_shard_connections};
+use server::bootstrap::{
+ create_default_executor, create_directories, create_root_user, create_shard_connections,
+};
use server::channels::commands::archive_state::ArchiveStateExecutor;
use server::channels::commands::clean_personal_access_tokens::CleanPersonalAccessTokensExecutor;
use server::channels::commands::maintain_messages::MaintainMessagesExecutor;
@@ -32,7 +34,7 @@ use server::channels::commands::print_sysinfo::SysInfoPrintExecutor;
use server::channels::commands::save_messages::SaveMessagesExecutor;
use server::channels::commands::verify_heartbeats::VerifyHeartbeatsExecutor;
use server::channels::handler::BackgroundServerCommandHandler;
-use server::configs::config_provider;
+use server::configs::config_provider::{self, ConfigProviderKind};
use server::configs::server::ServerConfig;
use server::http::http_server;
#[cfg(not(feature = "tokio-console"))]
@@ -72,37 +74,60 @@ fn main() -> Result<(), ServerError> {
let args = Args::parse();
let config_provider = config_provider::resolve(&args.config_provider)?;
- let config = ServerConfig::default();
- //TODO: Load config.
- /*
- let xd = std::thread::scope(|scope| {
- let config = scope.spawn(move || {
- let mut rt = monoio::RuntimeBuilder::<monoio::IoUringDriver>::new().build().unwrap();
- let config: Result<ServerConfig, ServerError> = rt.block_on(async {
- let config = ServerConfig::load(&config_provider).await?;
- create_directories().await;
- create_root_user().await;
- Ok(config)
- });
- config
- }).join().unwrap();
+ let config = std::thread::scope(|scope| {
+ let config = scope
+ .spawn(move || {
+ async fn load_config(
+ config_provider: &ConfigProviderKind,
+ ) -> Result<ServerConfig, ServerError> {
+ let config = ServerConfig::load(config_provider).await?;
+ Ok(config)
+ }
+
+ let mut rt = create_default_executor::<monoio::IoUringDriver>();
+ rt.block_on(load_config(&config_provider))
+ })
+ .join()
+ .expect("Failed to load config");
config
+ })?;
+
+ // Create directories and root user.
+ // Remove `local_data` directory if run with `--fresh` flag.
+ std::thread::scope(|scope| {
+ scope
+ .spawn(|| {
+ let mut rt = create_default_executor::<monoio::IoUringDriver>();
+ rt.block_on(async {
+ if args.fresh {
+ let system_path = config.system.get_system_path();
+ if monoio::fs::metadata(&system_path).await.is_ok() {
+ println!(
+ "Removing system path at: {} because `--fresh` flag was set",
+ system_path
+ );
+ //TODO: Impl dir walk and remove the files
+ /*
+ if let Err(e) = tokio::fs::remove_dir_all(&system_path).await {
+ eprintln!("Failed to remove system path at {}: {}", system_path, e);
+ }
+ */
+ }
+ }
+
+ // Create directories and root user
+ create_directories().await;
+ create_root_user().await;
+ });
+ })
+ .join()
+ .expect("Failed to create directories and root user");
});
- */
- /*
- let config = ServerConfig::load(&config_provider).await?;
- if args.fresh {
- let system_path = config.system.get_system_path();
- if tokio::fs::metadata(&system_path).await.is_ok() {
- println!("Removing system path at: {system_path} because `--fresh` flag was set");
- if let Err(e) = tokio::fs::remove_dir_all(&system_path).await {
- eprintln!("Failed to remove system path at {system_path}: {e}");
- }
- }
- }
+
+ // Initialize logging
let mut logging = Logging::new(config.telemetry.clone());
logging.early_init();
- */
+
// TODO: Make this configurable from config as a range
// for example this instance of Iggy will use cores from 0..4
let available_cpus = available_parallelism()
@@ -149,11 +174,11 @@ fn main() -> Result<(), ServerError> {
.join()
.expect(format!("Failed to join thread for shard-{id}").as_str());
}
- // From this point on, we can use tracing macros to log messages.
- /*
+ // From this point on, we can use tracing macros to log messages.
logging.late_init(config.system.get_system_path(), &config.system.logging)?;
+ /*
#[cfg(feature = "disable-mimalloc")]
tracing::warn!(
"Using default system allocator because code was build with `disable-mimalloc` feature"
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index 08a02e54..750e2b71 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -61,13 +61,6 @@ impl IggyShardBuilder {
.expect("Failed to find connection with the specified ID");
let shards = connections.into_iter().map(Shard::new).collect();
- IggyShard::new(
- id,
- shards,
- config,
- stop_receiver,
- stop_sender,
- receiver,
- )
+ IggyShard::new(id, shards, config, stop_receiver, stop_sender, receiver)
}
}
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index bffaca45..dcb089d9 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -91,6 +91,5 @@ impl IggyShard {
}
}
- pub fn assert_init(&self) {
- }
+ pub fn assert_init(&self) {}
}