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/19 08:26:47 UTC
(iggy) 05/05: todos
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 370d1a3e7f2d498e71e5e0026a084aff7af7e7d3
Author: numinex <g....@gmail.com>
AuthorDate: Thu Jun 19 10:17:42 2025 +0200
todos
---
core/sdk/src/tcp/tcp_client.rs | 1 +
core/server/src/binary/command.rs | 1 +
core/server/src/bootstrap.rs | 27 +++++++++++++++++++-------
core/server/src/main.rs | 17 +++--------------
core/server/src/shard/builder.rs | 39 +++++++++++++++++++++++++-------------
core/server/src/shard/connector.rs | 4 ++--
core/server/src/shard/mod.rs | 19 ++++++++++++++-----
7 files changed, 67 insertions(+), 41 deletions(-)
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index 3ea37500..b995e402 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -17,6 +17,7 @@
*/
use crate::prelude::Client;
+use crate::prelude::IggyConsumer;
use crate::prelude::TcpClientConfig;
use crate::tcp::tcp_connection_stream::TcpConnectionStream;
use crate::tcp::tcp_connection_stream_kind::ConnectionStreamKind;
diff --git a/core/server/src/binary/command.rs b/core/server/src/binary/command.rs
index c7802be9..0b997941 100644
--- a/core/server/src/binary/command.rs
+++ b/core/server/src/binary/command.rs
@@ -66,6 +66,7 @@ use iggy_common::update_stream::UpdateStream;
use iggy_common::update_topic::UpdateTopic;
use iggy_common::update_user::UpdateUser;
use iggy_common::*;
+use rustls::crypto::hash::Output;
use strum::EnumString;
use tracing::error;
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index d51096a2..d70d0317 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -1,22 +1,19 @@
use iggy_common::{
+ IggyError,
defaults::{
DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME, MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH,
MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH,
},
- IggyError,
-};
-use monoio::{
- fs::create_dir_all,
- Buildable, Driver, Runtime,
};
+use monoio::{fs::create_dir_all, time::TimeDriver, Buildable, Driver, Runtime};
use tracing::info;
use crate::{
+ IGGY_ROOT_PASSWORD_ENV, IGGY_ROOT_USERNAME_ENV,
configs::{config_provider::ConfigProviderKind, server::ServerConfig, system::SystemConfig},
server_error::ServerError,
shard::{connector::ShardConnector, frame::ShardFrame},
streaming::users::user::User,
- IGGY_ROOT_PASSWORD_ENV, IGGY_ROOT_USERNAME_ENV,
};
use std::{env, fs::remove_dir_all, ops::Range, path::Path};
@@ -109,7 +106,9 @@ pub fn create_root_user() -> User {
let username = env::var(IGGY_ROOT_USERNAME_ENV);
let password = env::var(IGGY_ROOT_PASSWORD_ENV);
if (username.is_ok() && password.is_err()) || (username.is_err() && password.is_ok()) {
- panic!("When providing the custom root user credentials, both username and password must be set.");
+ panic!(
+ "When providing the custom root user credentials, both username and password must be set."
+ );
}
if username.is_ok() && password.is_ok() {
info!("Using the custom root user credentials.");
@@ -146,3 +145,17 @@ where
let rt = Buildable::build(builder).expect("Failed to create default runtime");
rt
}
+
+pub fn create_shard_executor() -> Runtime<TimeDriver<monoio::IoUringDriver>>
+{
+ // TODO: Figure out what else we could tweak there
+ // We for sure want to disable the userspace interrupts on new cq entry (set_coop_taskrun)
+ // let urb = io_uring::IoUring::builder();
+ // TODO: Shall we make the size of ring be configureable ?
+ let builder = monoio::RuntimeBuilder::<monoio::IoUringDriver>::new()
+ //.uring_builder(urb.setup_coop_taskrun()) // broken shit.
+ .with_entries(1024) // Default size
+ .enable_timer();
+ 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 cd79cedc..6ac7d63b 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -23,11 +23,9 @@ use clap::Parser;
use dotenvy::dotenv;
use error_set::ErrContext;
use figlet_rs::FIGfont;
-use monoio::Buildable;
use server::args::Args;
use server::bootstrap::{
- create_default_executor, create_directories, create_root_user, create_shard_connections,
- load_config,
+ create_default_executor, create_directories, create_root_user, create_shard_connections, create_shard_executor, load_config
};
use server::channels::commands::archive_state::ArchiveStateExecutor;
use server::channels::commands::clean_personal_access_tokens::CleanPersonalAccessTokensExecutor;
@@ -112,7 +110,7 @@ fn main() -> Result<(), ServerError> {
}
}
- // Create directories.
+ // Create directories.
create_directories(&config.system).await?;
Ok::<(), ServerError>(())
})
@@ -146,16 +144,7 @@ fn main() -> Result<(), ServerError> {
monoio::utils::bind_to_cpu_set(Some(shard_id))
.expect(format!("Failed to set CPU affinity for shard-{id}").as_str());
- // TODO: Figure out what else we could tweak there
- // We for sure want to disable the userspace interrupts on new cq entry (set_coop_taskrun)
- // let urb = io_uring::IoUring::builder();
- // TODO: Shall we make the size of ring be configureable ?
- let mut rt = monoio::RuntimeBuilder::<monoio::IoUringDriver>::new()
- //.uring_builder(urb.setup_coop_taskrun()) // broken shit.
- .with_entries(1024) // Default size
- .enable_timer()
- .build()
- .expect(format!("Failed to build monoio runtime for shard-{id}").as_str());
+ let mut rt = create_shard_executor();
rt.block_on(async move {
let builder = IggyShard::builder();
let mut shard = builder
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index 640ffb57..72546e63 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -21,9 +21,19 @@ use std::{cell::Cell, rc::Rc, sync::Arc};
use iggy_common::{Aes256GcmEncryptor, EncryptorKind};
use tracing::info;
-use crate::{configs::server::ServerConfig, map_toggle_str, shard::Shard, state::{file::FileState, StateKind}, streaming::{persistence::persister::{FilePersister, FileWithSyncPersister, PersisterKind}, storage::SystemStorage}, versioning::SemanticVersion};
+use crate::{
+ configs::server::ServerConfig,
+ map_toggle_str,
+ shard::Shard,
+ state::{StateKind, file::FileState},
+ streaming::{
+ persistence::persister::{FilePersister, FileWithSyncPersister, PersisterKind},
+ storage::SystemStorage,
+ },
+ versioning::SemanticVersion,
+};
-use super::{connector::ShardConnector, frame::ShardFrame, IggyShard};
+use super::{IggyShard, connector::ShardConnector, frame::ShardFrame};
#[derive(Default)]
pub struct IggyShardBuilder {
@@ -68,6 +78,7 @@ impl IggyShardBuilder {
let shards = connections.into_iter().map(Shard::new).collect();
let version = SemanticVersion::current().expect("Invalid version");
+ //TODO: This can be discrete step in the builder bootstrapped from main function.
info!(
"Server-side encryption is {}.",
map_toggle_str(config.system.encryption.enabled)
@@ -79,6 +90,7 @@ impl IggyShardBuilder {
false => None,
};
+ //TODO: This can be discrete step in the builder bootstrapped from main function.
let state_persister = Self::resolve_persister(config.system.state.enforce_fsync);
let state = Rc::new(StateKind::File(FileState::new(
&config.system.get_state_messages_file_path(),
@@ -87,20 +99,21 @@ impl IggyShardBuilder {
encryptor.clone(),
)));
+ //TODO: This can be discrete step in the builder bootstrapped from main function.
let partition_persister = Self::resolve_persister(config.system.partition.enforce_fsync);
- let storage = SystemStorage::new(config.system, partition_persister);
+ let storage = Rc::new(SystemStorage::new(config.system.clone(), partition_persister));
IggyShard {
- id: id,
- shards: shards,
- shards_table: Default::default(),
- storage: storage,
- state: state,
- config: config,
- stop_receiver: stop_receiver,
- stop_sender: stop_sender,
- frame_receiver: Cell::new(Some(frame_receiver)),
- }
+ id: id,
+ shards: shards,
+ shards_table: Default::default(),
+ storage: storage,
+ state: state,
+ config: config,
+ stop_receiver: stop_receiver,
+ stop_sender: stop_sender,
+ frame_receiver: Cell::new(Some(frame_receiver)),
+ }
}
fn resolve_persister(enforce_fsync: bool) -> Arc<PersisterKind> {
diff --git a/core/server/src/shard/connector.rs b/core/server/src/shard/connector.rs
index 1d762d68..95aedc73 100644
--- a/core/server/src/shard/connector.rs
+++ b/core/server/src/shard/connector.rs
@@ -15,10 +15,10 @@
* specific language governing permissions and limitations
* under the License.
*/
-use futures::{task::AtomicWaker, Stream};
+use futures::{Stream, task::AtomicWaker};
use sharded_queue::ShardedQueue;
use std::{
- sync::{atomic::AtomicUsize, Arc},
+ sync::{Arc, atomic::AtomicUsize},
task::Poll,
};
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 54b88fb1..c3cb9727 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -26,9 +26,16 @@ use builder::IggyShardBuilder;
use connector::{Receiver, ShardConnector, StopReceiver, StopSender};
use frame::ShardFrame;
use namespace::IggyNamespace;
-use std::{cell::{Cell, RefCell}, rc::Rc, sync::Arc};
+use std::{
+ cell::{Cell, RefCell},
+ rc::Rc,
+ sync::Arc,
+};
-use crate::{bootstrap::create_root_user, configs::server::ServerConfig, state::file::FileState, streaming::storage::SystemStorage};
+use crate::{
+ bootstrap::create_root_user, configs::server::ServerConfig, state::{file::FileState, StateKind},
+ streaming::storage::SystemStorage,
+};
pub(crate) struct Shard {
id: u16,
connection: ShardConnector<ShardFrame>,
@@ -57,10 +64,10 @@ pub struct IggyShard {
//pub(crate) streams_ids: RefCell<HashMap<String, u32>>,
//pub(crate) users: RefCell<HashMap<UserId, User>>,
// TODO: Refactor.
- pub(crate) storage: Arc<SystemStorage>,
+ pub(crate) storage: Rc<SystemStorage>,
// TODO - get rid of this dynamic dispatch.
- pub(crate) state: Rc<FileState>,
+ pub(crate) state: Rc<StateKind>,
//pub(crate) encryptor: Option<Rc<dyn Encryptor>>,
config: ServerConfig,
//pub(crate) client_manager: RefCell<ClientManager>,
@@ -77,13 +84,15 @@ impl IggyShard {
}
pub async fn init(&mut self) {
+ let now = Instant::now();
let user = create_root_user();
self.load_state().await;
self.load_users().await;
// Add default root user.
todo!();
self.load_streams().await;
-
+ info!("Initialized system in {} ms.", now.elapsed().as_millis());
+ Ok(())
}
async fn load_state(&self) {