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/24 09:52:14 UTC
(iggy) branch io_uring_tpc updated: temp
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
The following commit(s) were added to refs/heads/io_uring_tpc by this push:
new 26ec032d temp
26ec032d is described below
commit 26ec032da031a3dc91654d0a7db9033211e51ce9
Author: numinex <g....@gmail.com>
AuthorDate: Tue Jun 24 11:52:06 2025 +0200
temp
---
Cargo.lock | 33 ++++-
core/server/Cargo.toml | 3 +-
core/server/src/binary/sender.rs | 14 +-
core/server/src/configs/server.rs | 3 +-
core/server/src/main.rs | 8 +-
core/server/src/quic/listener.rs | 3 +
core/server/src/shard/builder.rs | 16 ++-
core/server/src/shard/gate.rs | 2 +-
core/server/src/shard/mod.rs | 158 ++++++++++++++-------
core/server/src/shard/namespace.rs | 17 ++-
.../{streaming/systems => shard/system}/clients.rs | 2 +-
.../systems => shard/system}/consumer_groups.rs | 2 +-
.../systems => shard/system}/consumer_offsets.rs | 2 +-
.../{streaming/systems => shard/system}/info.rs | 0
.../systems => shard/system}/messages.rs | 5 +-
.../src/{streaming/systems => shard/system}/mod.rs | 2 +-
.../systems => shard/system}/partitions.rs | 2 +-
.../system}/personal_access_tokens.rs | 2 +-
.../systems => shard/system}/segments.rs | 2 +-
.../systems => shard/system}/snapshot/mod.rs | 0
.../systems => shard/system}/snapshot/procdump.rs | 0
.../{streaming/systems => shard/system}/stats.rs | 2 +-
.../{streaming/systems => shard/system}/storage.rs | 2 -
.../{streaming/systems => shard/system}/streams.rs | 142 +-----------------
.../{streaming/systems => shard/system}/system.rs | 0
.../{streaming/systems => shard/system}/topics.rs | 0
.../{streaming/systems => shard/system}/users.rs | 106 +-------------
core/server/src/streaming/mod.rs | 1 -
core/server/src/streaming/storage.rs | 5 +-
core/server/src/streaming/streams/stream.rs | 9 +-
core/server/src/tcp/sender.rs | 16 +--
core/server/src/tcp/tcp_listener.rs | 61 +++-----
core/server/src/tcp/tcp_sender.rs | 3 +-
core/server/src/tcp/tcp_server.rs | 21 ++-
core/server/src/tcp/tcp_socket.rs | 1 -
core/server/src/tcp/tcp_tls_sender.rs | 5 +-
36 files changed, 248 insertions(+), 402 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 54f19c07..678e0c3a 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3299,6 +3299,12 @@ dependencies = [
"byteorder",
]
+[[package]]
+name = "hash32"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8e4e11d13d42fc8d55326b323c63978c75721fbbb695a6e6686765bcb8b33917"
+
[[package]]
name = "hashbrown"
version = "0.12.3"
@@ -3354,7 +3360,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cdc6457c0eb62c71aac4bc17216026d8410337c4126773b9c5daba343f17964f"
dependencies = [
"atomic-polyfill",
- "hash32",
+ "hash32 0.2.1",
"rustc_version",
"serde",
"spin",
@@ -4837,6 +4843,15 @@ dependencies = [
"windows-sys 0.48.0",
]
+[[package]]
+name = "monoio-io-wrapper"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4bcfaa76e5daf87cc4d31b4d1b6bc93c12db59c19df50b9200afdbde42077655"
+dependencies = [
+ "monoio",
+]
+
[[package]]
name = "monoio-macros"
version = "0.1.0"
@@ -4848,6 +4863,19 @@ dependencies = [
"syn 2.0.103",
]
+[[package]]
+name = "monoio-native-tls"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9022f5aaa19f9688f97bfcfa0c4a4318d424851995badb356674ca742652cdb"
+dependencies = [
+ "bytes",
+ "monoio",
+ "monoio-io-wrapper",
+ "native-tls",
+ "thiserror 1.0.69",
+]
+
[[package]]
name = "nanorand"
version = "0.7.0"
@@ -6909,15 +6937,16 @@ dependencies = [
"figment",
"flume",
"futures",
+ "hash32 1.0.0",
"human-repr",
"iggy_common",
- "io-uring",
"jsonwebtoken",
"lending-iterator",
"mimalloc",
"mockall",
"moka",
"monoio",
+ "monoio-native-tls",
"nix 0.30.1",
"once_cell",
"openssl",
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index e612af10..093a143f 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -60,12 +60,13 @@ async-channel = { workspace = true }
futures = { workspace = true }
human-repr = { workspace = true }
iggy_common = { workspace = true }
-io-uring = "0.6"
jsonwebtoken = "9.3.1"
lending-iterator = "0.1.7"
+hash32 = "1.0.0"
mimalloc = { workspace = true, optional = true }
moka = { version = "0.12.10", features = ["future"] }
monoio = { version = "0.2.4", features = ["mkdirat", "unlinkat"] }
+monoio-native-tls = "0.4.0"
nix = { version = "0.30", features = ["fs"] }
once_cell = "1.21.3"
openssl = { workspace = true }
diff --git a/core/server/src/binary/sender.rs b/core/server/src/binary/sender.rs
index 706cb128..a0659c42 100644
--- a/core/server/src/binary/sender.rs
+++ b/core/server/src/binary/sender.rs
@@ -24,7 +24,7 @@ use crate::tcp::tcp_tls_sender::TcpTlsSender;
use crate::{quic::quic_sender::QuicSender, server_error::ServerError};
use iggy_common::IggyError;
use quinn::{RecvStream, SendStream};
-use tokio::net::TcpStream;
+use monoio::net::TcpStream;
use tokio_native_tls::TlsStream;
macro_rules! forward_async_methods {
@@ -48,22 +48,22 @@ macro_rules! forward_async_methods {
}
pub trait Sender {
- fn read(&mut self, buffer: &mut [u8]) -> impl Future<Output = Result<usize, IggyError>> + Send;
- fn send_empty_ok_response(&mut self) -> impl Future<Output = Result<(), IggyError>> + Send;
+ fn read(&mut self, buffer: &mut [u8]) -> impl Future<Output = Result<usize, IggyError>>;
+ fn send_empty_ok_response(&mut self) -> impl Future<Output = Result<(), IggyError>>;
fn send_ok_response(
&mut self,
payload: &[u8],
- ) -> impl Future<Output = Result<(), IggyError>> + Send;
+ ) -> impl Future<Output = Result<(), IggyError>>;
fn send_ok_response_vectored(
&mut self,
length: &[u8],
slices: Vec<IoSlice<'_>>,
- ) -> impl Future<Output = Result<(), IggyError>> + Send;
+ ) -> impl Future<Output = Result<(), IggyError>>;
fn send_error_response(
&mut self,
error: IggyError,
- ) -> impl Future<Output = Result<(), IggyError>> + Send;
- fn shutdown(&mut self) -> impl Future<Output = Result<(), ServerError>> + Send;
+ ) -> impl Future<Output = Result<(), IggyError>>;
+ fn shutdown(&mut self) -> impl Future<Output = Result<(), ServerError>>;
}
pub enum SenderKind {
diff --git a/core/server/src/configs/server.rs b/core/server/src/configs/server.rs
index 252df9a7..059fd1c4 100644
--- a/core/server/src/configs/server.rs
+++ b/core/server/src/configs/server.rs
@@ -31,6 +31,7 @@ use iggy_common::Validatable;
use serde::{Deserialize, Serialize};
use serde_with::DisplayFromStr;
use serde_with::serde_as;
+use std::rc::Rc;
use std::str::FromStr;
use std::sync::Arc;
@@ -40,7 +41,7 @@ pub struct ServerConfig {
pub message_saver: MessageSaverConfig,
pub personal_access_token: PersonalAccessTokenConfig,
pub heartbeat: HeartbeatConfig,
- pub system: Arc<SystemConfig>,
+ pub system: Rc<SystemConfig>,
pub quic: QuicConfig,
pub tcp: TcpConfig,
pub http: HttpConfig,
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index fc2b21f4..06b5f445 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -235,7 +235,7 @@ fn main() -> Result<(), ServerError> {
.await;
let builder = IggyShard::builder();
- let mut shard = builder
+ let shard: Rc<IggyShard> = builder
.id(id)
.connections(connections)
.config(config)
@@ -243,13 +243,13 @@ fn main() -> Result<(), ServerError> {
.version(version)
.state(state)
.build()
- .await;
+ .into();
//TODO: If one of the shards fails to initialize, we should crash the whole program;
- shard.init().await.expect("Failed to initialize shard-{id}: {e}");
- info!("Initiated shard with ID: {id}");
+ shard.run().await.expect("Failed to run shard");
//TODO: If one of the shards fails to initialize, we should crash the whole program;
shard.assert_init();
+ let shard = Rc::new(shard);
})
})
.expect(format!("Failed to spawn thread for shard-{id}").as_str())
diff --git a/core/server/src/quic/listener.rs b/core/server/src/quic/listener.rs
index b5eb859d..e9b329de 100644
--- a/core/server/src/quic/listener.rs
+++ b/core/server/src/quic/listener.rs
@@ -34,6 +34,8 @@ pub fn start(endpoint: Endpoint, system: SharedSystem) {
for _ in 0..LISTENERS_COUNT {
let endpoint = endpoint.clone();
let system = system.clone();
+ //TODO: Fixme
+ /*
tokio::spawn(async move {
while let Some(incoming_connection) = endpoint.accept().await {
info!(
@@ -57,6 +59,7 @@ pub fn start(endpoint: Endpoint, system: SharedSystem) {
});
}
});
+ */
}
}
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index 0fd5250f..0428a483 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -26,8 +26,8 @@ use crate::{
configs::server::ServerConfig,
map_toggle_str,
shard::Shard,
- state::{StateKind, file::FileState},
- streaming::storage::SystemStorage,
+ state::{file::FileState, StateKind},
+ streaming::{diagnostics::metrics::Metrics, storage::SystemStorage},
versioning::SemanticVersion,
};
@@ -75,11 +75,12 @@ impl IggyShardBuilder {
}
// TODO: Too much happens in there, some of those bootstrapping logic should be moved outside.
- pub async fn build(self) -> IggyShard {
+ pub fn build(self) -> IggyShard {
let id = self.id.unwrap();
let config = self.config.unwrap();
let connections = self.connections.unwrap();
let state = self.state.unwrap();
+ let encryptor = self.encryptor;
let version = self.version.unwrap();
let (stop_sender, stop_receiver, frame_receiver) = connections
.iter()
@@ -106,12 +107,21 @@ impl IggyShardBuilder {
shards: shards,
shards_table: Default::default(),
storage: storage,
+ encryptor: encryptor,
state: state,
config: config,
version: version,
stop_receiver: stop_receiver,
stop_sender: stop_sender,
frame_receiver: Cell::new(Some(frame_receiver)),
+ metrics: Metrics::init(),
+
+ users: Default::default(),
+ permissioner: Default::default(),
+ streams: Default::default(),
+ streams_ids: Default::default(),
+ client_manager: Default::default(),
+ active_sessions: Default::default(),
}
}
}
diff --git a/core/server/src/shard/gate.rs b/core/server/src/shard/gate.rs
index a26950a4..2ac20445 100644
--- a/core/server/src/shard/gate.rs
+++ b/core/server/src/shard/gate.rs
@@ -1,4 +1,4 @@
-use std::sync::{Condvar, Mutex};
+use std::sync::Mutex;
#[derive(Default)]
pub struct Gate<T> {
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 23ef1e15..3e7e10f1 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -17,39 +17,37 @@
*/
pub mod builder;
+pub mod system;
pub mod gate;
pub mod namespace;
pub mod transmission;
-use ahash::HashMap;
+use ahash::{AHashMap, AHashSet, HashMap};
use builder::IggyShardBuilder;
use error_set::ErrContext;
-use iggy_common::IggyError;
+use futures::future::try_join_all;
+use iggy_common::{EncryptorKind, IggyError, UserId};
use namespace::IggyNamespace;
use std::{
- cell::{Cell, RefCell},
- rc::Rc,
- str::FromStr,
- sync::Arc,
- time::Instant,
+ cell::{Cell, RefCell}, pin::Pin, rc::Rc, str::FromStr, sync::{atomic::{AtomicU32, Ordering}, Arc, RwLock}, time::Instant
};
-use tracing::info;
+use tracing::{error, info, instrument, trace, warn};
use transmission::connector::{Receiver, ShardConnector, StopReceiver, StopSender};
use crate::{
- bootstrap::create_root_user,
configs::server::ServerConfig,
- shard::transmission::frame::ShardFrame,
+ shard::{system::info::SystemInfo, transmission::frame::ShardFrame},
state::{
- StateKind,
- file::FileState,
- system::{SystemState, UserState},
+ file::FileState, system::{StreamState, SystemState, UserState}, StateKind
},
- streaming::{storage::SystemStorage, systems::info::SystemInfo},
+ streaming::{clients::client_manager::ClientManager, diagnostics::metrics::Metrics, personal_access_tokens::personal_access_token::PersonalAccessToken, session::Session, storage::SystemStorage, streams::stream::Stream, users::{permissioner::Permissioner, user::User}},
versioning::SemanticVersion,
};
pub const COMPONENT: &str = "SHARD";
+static USER_ID: AtomicU32 = AtomicU32::new(1);
+
+type Task = Pin<Box<dyn Future<Output = Result<(), IggyError>>>>;
pub(crate) struct Shard {
id: u16,
@@ -75,19 +73,21 @@ pub struct IggyShard {
shards_table: RefCell<HashMap<IggyNamespace, ShardInfo>>,
version: SemanticVersion,
- //pub(crate) permissioner: RefCell<Permissioner>,
- //pub(crate) streams: RwLock<HashMap<u32, Stream>>,
- //pub(crate) streams_ids: RefCell<HashMap<String, u32>>,
- //pub(crate) users: RefCell<HashMap<UserId, User>>,
+ pub(crate) streams: RefCell<HashMap<u32, Stream>>,
+ pub(crate) streams_ids: RefCell<HashMap<String, u32>>,
// TODO: Refactor.
pub(crate) storage: Rc<SystemStorage>,
pub(crate) state: StateKind,
- //pub(crate) encryptor: Option<Rc<dyn Encryptor>>,
- config: ServerConfig,
- //pub(crate) client_manager: RefCell<ClientManager>,
- //pub(crate) active_sessions: RefCell<Vec<Session>>,
- //pub(crate) metrics: Metrics,
+ pub(crate) encryptor: Option<EncryptorKind>,
+ pub(crate) config: ServerConfig,
+ //TODO: This could be shared.
+ pub(crate) client_manager: RefCell<ClientManager>,
+ pub(crate) active_sessions: RefCell<Vec<Session>>,
+ pub(crate) permissioner: RefCell<Permissioner>,
+ pub(crate) users: RefCell<HashMap<UserId, User>>,
+
+ pub(crate) metrics: Metrics,
pub frame_receiver: Cell<Option<Receiver<ShardFrame>>>,
stop_receiver: StopReceiver,
stop_sender: StopSender,
@@ -98,16 +98,12 @@ impl IggyShard {
Default::default()
}
- pub async fn init(&mut self) -> Result<(), IggyError> {
+ pub async fn init(&self) -> Result<(), IggyError> {
let now = Instant::now();
- //TODO: Fix this either by moving it to main function, or by using `run_once` barrier.
- //let state_entries = self.state.init().await?;
- //let system_state = SystemState::init(state_entries).await?;
- //let user = create_root_user();
+
self.load_version().await?;
let SystemState { users, streams } = self.load_state().await?;
self.load_users(users.into_values().collect()).await;
- // Add default root user.
self.load_streams(streams.into_values().collect()).await;
//TODO: Fix the archiver.
@@ -123,6 +119,27 @@ impl IggyShard {
Ok(())
}
+ pub async fn run(self: &Rc<Self>) -> Result<(), IggyError> {
+ // Workaround to ensure that the statistics are initialized before the server
+ // loads streams and starts accepting connections. This is necessary to
+ // have the correct statistics when the server starts.
+ //self.get_stats().await?;
+ self.init().await?;
+ self.assert_init();
+ info!("Initiated shard with ID: {}", self.id);
+ // Create all tasks (tcp listener, http listener, command processor, in the future also the background jobs).
+ /*
+ let mut tasks: Vec<Task> = vec![Box::pin(spawn_shard_message_task(shard.clone()))];
+ if self.config.tcp.enabled {
+ tasks.push(Box::pin(spawn_tcp_server(self.clone())));
+ }
+ let result = try_join_all(tasks).await;
+ result?;
+ */
+
+ Ok(())
+ }
+
async fn load_version(&self) -> Result<(), IggyError> {
async fn update_system_info(
storage: &Rc<SystemStorage>,
@@ -181,10 +198,8 @@ impl IggyShard {
Ok(system_state)
}
- async fn load_users(&mut self, users: Vec<UserState>) -> Result<(), IggyError> {
+ async fn load_users(&self, users: Vec<UserState>) -> Result<(), IggyError> {
info!("Loading users...");
- /*
-
for user_state in users.into_iter() {
let mut user = User::with_password(
user_state.id,
@@ -210,32 +225,34 @@ impl IggyShard {
)
})
.collect();
- self.users.insert(user_state.id, user);
+ self.users.borrow_mut().insert(user_state.id, user);
}
- let users_count = self.users.len();
- let current_user_id = self.users.keys().max().unwrap_or(&1);
+ let users = self.users.borrow();
+ let users_count = users.len();
+ let current_user_id = users.keys().max().unwrap_or(&1);
USER_ID.store(current_user_id + 1, Ordering::SeqCst);
self.permissioner
- .init(&self.users.values().collect::<Vec<&User>>());
+ .borrow_mut()
+ .init(&users.values().collect::<Vec<_>>());
self.metrics.increment_users(users_count as u32);
info!("Initialized {users_count} user(s).");
- */
Ok(())
}
- async fn load_streams(&mut self, streams: Vec<StreamState>) -> Result<(), IggyError> {
- todo!();
+ async fn load_streams(&self, streams: Vec<StreamState>) -> Result<(), IggyError> {
info!("Loading streams from disk...");
let mut unloaded_streams = Vec::new();
- let mut dir_entries = read_dir(&self.config.get_streams_path())
- .await
+ // Does mononio has api for that ?
+ let mut dir_entries = std::fs::read_dir(&self.config.system.get_streams_path())
.map_err(|error| {
error!("Cannot read streams directory: {error}");
IggyError::CannotReadStreams
})?;
- while let Some(dir_entry) = dir_entries.next_entry().await.unwrap_or(None) {
+ //TODO: User the dir walk impl from main function, once implemented.
+ while let Some(dir_entry) = dir_entries.next() {
+ let dir_entry = dir_entry.unwrap();
let name = dir_entry.file_name().into_string().unwrap();
let stream_id = name.parse::<u32>().map_err(|_| {
error!("Invalid stream ID file with name: '{name}'.");
@@ -246,7 +263,7 @@ impl IggyShard {
error!(
"Stream with ID: '{stream_id}' was not found in state, but exists on disk and will be removed."
);
- if let Err(error) = fs::remove_dir_all(&dir_entry.path()).await {
+ if let Err(error) = std::fs::remove_dir_all(&dir_entry.path()) {
error!("Cannot remove stream directory: {error}");
} else {
warn!("Stream with ID: '{stream_id}' was removed.");
@@ -258,7 +275,7 @@ impl IggyShard {
let mut stream = Stream::empty(
stream_id,
&stream_state.name,
- self.config.clone(),
+ self.config.system.clone(),
self.storage.clone(),
);
stream.created_at = stream_state.created_at;
@@ -281,7 +298,7 @@ impl IggyShard {
info!("All streams found on disk were found in state.");
} else {
warn!("Streams with IDs: '{missing_ids:?}' were not found on disk.");
- if self.config.recovery.recreate_missing_state {
+ if self.config.system.recovery.recreate_missing_state {
info!(
"Recreating missing state in recovery config is enabled, missing streams will be created."
);
@@ -291,7 +308,7 @@ impl IggyShard {
let stream = Stream::create(
stream_id,
&stream_state.name,
- self.config.clone(),
+ self.config.system.clone(),
self.storage.clone(),
);
stream.persist().await?;
@@ -327,12 +344,12 @@ impl IggyShard {
try_join_all(load_stream_tasks).await?;
for stream in loaded_streams.take() {
- if self.streams.contains_key(&stream.stream_id) {
+ if self.streams.borrow().contains_key(&stream.stream_id) {
error!("Stream with ID: '{}' already exists.", &stream.stream_id);
continue;
}
- if self.streams_ids.contains_key(&stream.name) {
+ if self.streams_ids.borrow().contains_key(&stream.name) {
error!("Stream with name: '{}' already exists.", &stream.name);
continue;
}
@@ -345,13 +362,52 @@ impl IggyShard {
self.metrics.increment_messages(stream.get_messages_count());
self.streams_ids
+ .borrow_mut()
.insert(stream.name.clone(), stream.stream_id);
- self.streams.insert(stream.stream_id, stream);
+ self.streams.borrow_mut().insert(stream.stream_id, stream);
}
- info!("Loaded {} stream(s) from disk.", self.streams.len());
+ info!("Loaded {} stream(s) from disk.", self.streams.borrow().len());
Ok(())
}
- pub fn assert_init(&self) {}
+ pub fn assert_init(&self) -> Result<(), IggyError> { Ok(())}
+
+ #[instrument(skip_all, name = "trace_shutdown")]
+ pub async fn shutdown(&mut self) -> Result<(), IggyError> {
+ self.persist_messages().await?;
+ Ok(())
+ }
+
+ #[instrument(skip_all, name = "trace_persist_messages")]
+ pub async fn persist_messages(&self) -> Result<usize, IggyError> {
+ trace!("Saving buffered messages on disk...");
+ let mut saved_messages_number = 0;
+ //TODO: Fixme
+ /*
+ for stream in self.streams.values() {
+ saved_messages_number += stream.persist_messages().await?;
+ }
+ */
+
+ Ok(saved_messages_number)
+ }
+
+ pub fn get_available_shards_count(&self) -> u32 {
+ self.shards.len() as u32
+ }
+
+ pub fn ensure_authenticated(&self, client_id: u32) -> Result<u32, IggyError> {
+ let active_sessions = self.active_sessions.borrow();
+ let session = active_sessions
+ .iter()
+ .find(|s| s.client_id == client_id)
+ .ok_or_else(|| IggyError::Unauthenticated)?;
+ if session.is_authenticated() {
+ Ok(session.get_user_id())
+ } else {
+ error!("{COMPONENT} - unauthenticated access attempt, session: {session}");
+ Err(IggyError::Unauthenticated)
+ }
+ }
}
diff --git a/core/server/src/shard/namespace.rs b/core/server/src/shard/namespace.rs
index d9aedfb7..d9d68ec9 100644
--- a/core/server/src/shard/namespace.rs
+++ b/core/server/src/shard/namespace.rs
@@ -16,15 +16,20 @@
* under the License.
*/
+use std::hash::Hasher as _;
+use iggy_common::Identifier;
+use hash32::{Hasher, Murmur3Hasher};
+
+//TODO: Will probably want to move it to separate crate so we can share it with sdk.
#[derive(Debug, Clone, Eq, PartialEq, Hash)]
pub struct IggyNamespace {
- pub(crate) stream_id: u32,
- pub(crate) topic_id: u32,
+ pub(crate) stream_id: Identifier,
+ pub(crate) topic_id: Identifier,
pub(crate) partition_id: u32,
}
impl IggyNamespace {
- pub fn new(stream_id: u32, topic_id: u32, partition_id: u32) -> Self {
+ pub fn new(stream_id: Identifier, topic_id: Identifier, partition_id: u32) -> Self {
Self {
stream_id,
topic_id,
@@ -33,6 +38,10 @@ impl IggyNamespace {
}
pub fn generate_hash(&self) -> u32 {
- todo!();
+ let mut hasher = Murmur3Hasher::default();
+ hasher.write(&self.stream_id.value);
+ hasher.write(&self.topic_id.value);
+ hasher.write_u32(self.partition_id);
+ hasher.finish32()
}
}
diff --git a/core/server/src/streaming/systems/clients.rs b/core/server/src/shard/system/clients.rs
similarity index 99%
rename from core/server/src/streaming/systems/clients.rs
rename to core/server/src/shard/system/clients.rs
index 050a0b58..2cb0fbe8 100644
--- a/core/server/src/streaming/systems/clients.rs
+++ b/core/server/src/shard/system/clients.rs
@@ -29,7 +29,7 @@ use std::net::SocketAddr;
use std::sync::Arc;
use tracing::{error, info};
-impl System {
+impl IggyShard {
pub async fn add_client(&self, address: &SocketAddr, transport: Transport) -> Arc<Session> {
let mut client_manager = self.client_manager.write().await;
let session = client_manager.add_client(address, transport);
diff --git a/core/server/src/streaming/systems/consumer_groups.rs b/core/server/src/shard/system/consumer_groups.rs
similarity index 99%
rename from core/server/src/streaming/systems/consumer_groups.rs
rename to core/server/src/shard/system/consumer_groups.rs
index 4303dc4d..ae883d9f 100644
--- a/core/server/src/streaming/systems/consumer_groups.rs
+++ b/core/server/src/shard/system/consumer_groups.rs
@@ -26,7 +26,7 @@ use iggy_common::IggyError;
use iggy_common::locking::IggySharedMutFn;
use tokio::sync::RwLock;
-impl System {
+impl IggyShard {
pub fn get_consumer_group(
&self,
session: &Session,
diff --git a/core/server/src/streaming/systems/consumer_offsets.rs b/core/server/src/shard/system/consumer_offsets.rs
similarity index 99%
rename from core/server/src/streaming/systems/consumer_offsets.rs
rename to core/server/src/shard/system/consumer_offsets.rs
index fdac2b0c..e63815ff 100644
--- a/core/server/src/streaming/systems/consumer_offsets.rs
+++ b/core/server/src/shard/system/consumer_offsets.rs
@@ -22,7 +22,7 @@ use crate::streaming::systems::system::System;
use error_set::ErrContext;
use iggy_common::{Consumer, ConsumerOffsetInfo, Identifier, IggyError};
-impl System {
+impl IggyShard {
pub async fn store_consumer_offset(
&self,
session: &Session,
diff --git a/core/server/src/streaming/systems/info.rs b/core/server/src/shard/system/info.rs
similarity index 100%
rename from core/server/src/streaming/systems/info.rs
rename to core/server/src/shard/system/info.rs
diff --git a/core/server/src/streaming/systems/messages.rs b/core/server/src/shard/system/messages.rs
similarity index 99%
rename from core/server/src/streaming/systems/messages.rs
rename to core/server/src/shard/system/messages.rs
index 53ed0432..da9b67c2 100644
--- a/core/server/src/streaming/systems/messages.rs
+++ b/core/server/src/shard/system/messages.rs
@@ -17,10 +17,9 @@
*/
use crate::binary::handlers::messages::poll_messages_handler::IggyPollMetadata;
+use crate::shard::IggyShard;
use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut, IggyMessagesBatchSet};
use crate::streaming::session::Session;
-use crate::streaming::systems::COMPONENT;
-use crate::streaming::systems::system::System;
use crate::streaming::utils::PooledBuffer;
use error_set::ErrContext;
use iggy_common::{
@@ -29,7 +28,7 @@ use iggy_common::{
};
use tracing::{error, trace};
-impl System {
+impl IggyShard {
pub async fn poll_messages(
&self,
session: &Session,
diff --git a/core/server/src/streaming/systems/mod.rs b/core/server/src/shard/system/mod.rs
similarity index 95%
rename from core/server/src/streaming/systems/mod.rs
rename to core/server/src/shard/system/mod.rs
index 7039e919..1419abb9 100644
--- a/core/server/src/streaming/systems/mod.rs
+++ b/core/server/src/shard/system/mod.rs
@@ -32,4 +32,4 @@ pub mod system;
pub mod topics;
pub mod users;
-pub const COMPONENT: &str = "STREAMING_SYSTEMS";
+pub const COMPONENT: &str = "SYSTEM";
diff --git a/core/server/src/streaming/systems/partitions.rs b/core/server/src/shard/system/partitions.rs
similarity index 99%
rename from core/server/src/streaming/systems/partitions.rs
rename to core/server/src/shard/system/partitions.rs
index 5bb72875..7712f498 100644
--- a/core/server/src/streaming/systems/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -23,7 +23,7 @@ use error_set::ErrContext;
use iggy_common::Identifier;
use iggy_common::IggyError;
-impl System {
+impl IggyShard {
pub async fn create_partitions(
&mut self,
session: &Session,
diff --git a/core/server/src/streaming/systems/personal_access_tokens.rs b/core/server/src/shard/system/personal_access_tokens.rs
similarity index 99%
rename from core/server/src/streaming/systems/personal_access_tokens.rs
rename to core/server/src/shard/system/personal_access_tokens.rs
index 945c853b..18125bed 100644
--- a/core/server/src/streaming/systems/personal_access_tokens.rs
+++ b/core/server/src/shard/system/personal_access_tokens.rs
@@ -27,7 +27,7 @@ use iggy_common::IggyExpiry;
use iggy_common::IggyTimestamp;
use tracing::{error, info};
-impl System {
+impl IggyShard {
pub async fn get_personal_access_tokens(
&self,
session: &Session,
diff --git a/core/server/src/streaming/systems/segments.rs b/core/server/src/shard/system/segments.rs
similarity index 99%
rename from core/server/src/streaming/systems/segments.rs
rename to core/server/src/shard/system/segments.rs
index a15952c1..bd4043a2 100644
--- a/core/server/src/streaming/systems/segments.rs
+++ b/core/server/src/shard/system/segments.rs
@@ -23,7 +23,7 @@ use iggy_common::Identifier;
use iggy_common::IggyError;
use iggy_common::locking::IggySharedMutFn;
-impl System {
+impl IggyShard {
pub async fn delete_segments(
&mut self,
session: &Session,
diff --git a/core/server/src/streaming/systems/snapshot/mod.rs b/core/server/src/shard/system/snapshot/mod.rs
similarity index 100%
rename from core/server/src/streaming/systems/snapshot/mod.rs
rename to core/server/src/shard/system/snapshot/mod.rs
diff --git a/core/server/src/streaming/systems/snapshot/procdump.rs b/core/server/src/shard/system/snapshot/procdump.rs
similarity index 100%
rename from core/server/src/streaming/systems/snapshot/procdump.rs
rename to core/server/src/shard/system/snapshot/procdump.rs
diff --git a/core/server/src/streaming/systems/stats.rs b/core/server/src/shard/system/stats.rs
similarity index 99%
rename from core/server/src/streaming/systems/stats.rs
rename to core/server/src/shard/system/stats.rs
index 80c588e6..2d2d2ea7 100644
--- a/core/server/src/streaming/systems/stats.rs
+++ b/core/server/src/shard/system/stats.rs
@@ -34,7 +34,7 @@ fn sysinfo() -> &'static Mutex<SysinfoSystem> {
})
}
-impl System {
+impl IggyShard {
pub async fn get_stats(&self) -> Result<Stats, IggyError> {
let mut sys = sysinfo().lock().await;
let process_id = std::process::id();
diff --git a/core/server/src/streaming/systems/storage.rs b/core/server/src/shard/system/storage.rs
similarity index 97%
rename from core/server/src/streaming/systems/storage.rs
rename to core/server/src/shard/system/storage.rs
index d4363631..9a493751 100644
--- a/core/server/src/streaming/systems/storage.rs
+++ b/core/server/src/shard/system/storage.rs
@@ -18,8 +18,6 @@
use crate::streaming::persistence::persister::PersisterKind;
use crate::streaming::storage::SystemInfoStorage;
-use crate::streaming::systems::COMPONENT;
-use crate::streaming::systems::info::SystemInfo;
use crate::streaming::utils::PooledBuffer;
use crate::streaming::utils::file;
use anyhow::Context;
diff --git a/core/server/src/streaming/systems/streams.rs b/core/server/src/shard/system/streams.rs
similarity index 70%
rename from core/server/src/streaming/systems/streams.rs
rename to core/server/src/shard/system/streams.rs
index 18c6aa83..efcfff2b 100644
--- a/core/server/src/streaming/systems/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -16,156 +16,18 @@
* under the License.
*/
-use crate::state::system::StreamState;
+use crate::shard::IggyShard;
use crate::streaming::session::Session;
use crate::streaming::streams::stream::Stream;
-use crate::streaming::systems::COMPONENT;
-use crate::streaming::systems::system::System;
-use ahash::{AHashMap, AHashSet};
use error_set::ErrContext;
use futures::future::try_join_all;
-use iggy_common::locking::IggySharedMutFn;
-use iggy_common::{IdKind, Identifier, IggyError};
use std::cell::RefCell;
use std::sync::atomic::{AtomicU32, Ordering};
use tokio::fs;
-use tokio::fs::read_dir;
use tracing::{error, info, warn};
-static CURRENT_STREAM_ID: AtomicU32 = AtomicU32::new(1);
-
-impl System {
- pub(crate) async fn load_streams(
- &mut self,
- streams: Vec<StreamState>,
- ) -> Result<(), IggyError> {
- info!("Loading streams from disk...");
- let mut unloaded_streams = Vec::new();
- let mut dir_entries = read_dir(&self.config.get_streams_path())
- .await
- .map_err(|error| {
- error!("Cannot read streams directory: {error}");
- IggyError::CannotReadStreams
- })?;
-
- while let Some(dir_entry) = dir_entries.next_entry().await.unwrap_or(None) {
- let name = dir_entry.file_name().into_string().unwrap();
- let stream_id = name.parse::<u32>().map_err(|_| {
- error!("Invalid stream ID file with name: '{name}'.");
- IggyError::InvalidNumberValue
- })?;
- let stream_state = streams.iter().find(|s| s.id == stream_id);
- if stream_state.is_none() {
- error!(
- "Stream with ID: '{stream_id}' was not found in state, but exists on disk and will be removed."
- );
- if let Err(error) = fs::remove_dir_all(&dir_entry.path()).await {
- error!("Cannot remove stream directory: {error}");
- } else {
- warn!("Stream with ID: '{stream_id}' was removed.");
- }
- continue;
- }
-
- let stream_state = stream_state.unwrap();
- let mut stream = Stream::empty(
- stream_id,
- &stream_state.name,
- self.config.clone(),
- self.storage.clone(),
- );
- stream.created_at = stream_state.created_at;
- unloaded_streams.push(stream);
- }
-
- let state_stream_ids = streams
- .iter()
- .map(|stream| stream.id)
- .collect::<AHashSet<u32>>();
- let unloaded_stream_ids = unloaded_streams
- .iter()
- .map(|stream| stream.stream_id)
- .collect::<AHashSet<u32>>();
- let mut missing_ids = state_stream_ids
- .difference(&unloaded_stream_ids)
- .copied()
- .collect::<AHashSet<u32>>();
- if missing_ids.is_empty() {
- info!("All streams found on disk were found in state.");
- } else {
- warn!("Streams with IDs: '{missing_ids:?}' were not found on disk.");
- if self.config.recovery.recreate_missing_state {
- info!(
- "Recreating missing state in recovery config is enabled, missing streams will be created."
- );
- for stream_id in missing_ids.iter() {
- let stream_id = *stream_id;
- let stream_state = streams.iter().find(|s| s.id == stream_id).unwrap();
- let stream = Stream::create(
- stream_id,
- &stream_state.name,
- self.config.clone(),
- self.storage.clone(),
- );
- stream.persist().await?;
- unloaded_streams.push(stream);
- info!(
- "Missing stream with ID: '{stream_id}', name: {} was recreated.",
- stream_state.name
- );
- }
- missing_ids.clear();
- } else {
- warn!(
- "Recreating missing state in recovery config is disabled, missing streams will not be created."
- );
- }
- }
-
- let mut streams_states = streams
- .into_iter()
- .filter(|s| !missing_ids.contains(&s.id))
- .map(|s| (s.id, s))
- .collect::<AHashMap<_, _>>();
- let loaded_streams = RefCell::new(Vec::new());
- let load_stream_tasks = unloaded_streams.into_iter().map(|mut stream| {
- let state = streams_states.remove(&stream.stream_id).unwrap();
-
- async {
- stream.load(state).await?;
- loaded_streams.borrow_mut().push(stream);
- Result::<(), IggyError>::Ok(())
- }
- });
- try_join_all(load_stream_tasks).await?;
-
- for stream in loaded_streams.take() {
- if self.streams.contains_key(&stream.stream_id) {
- error!("Stream with ID: '{}' already exists.", &stream.stream_id);
- continue;
- }
-
- if self.streams_ids.contains_key(&stream.name) {
- error!("Stream with name: '{}' already exists.", &stream.name);
- continue;
- }
-
- self.metrics.increment_streams(1);
- self.metrics.increment_topics(stream.get_topics_count());
- self.metrics
- .increment_partitions(stream.get_partitions_count());
- self.metrics.increment_segments(stream.get_segments_count());
- self.metrics.increment_messages(stream.get_messages_count());
-
- self.streams_ids
- .insert(stream.name.clone(), stream.stream_id);
- self.streams.insert(stream.stream_id, stream);
- }
-
- info!("Loaded {} stream(s) from disk.", self.streams.len());
- Ok(())
- }
+impl IggyShard {
pub fn get_streams(&self) -> Vec<&Stream> {
self.streams.values().collect()
}
diff --git a/core/server/src/streaming/systems/system.rs b/core/server/src/shard/system/system.rs
similarity index 100%
rename from core/server/src/streaming/systems/system.rs
rename to core/server/src/shard/system/system.rs
diff --git a/core/server/src/streaming/systems/topics.rs b/core/server/src/shard/system/topics.rs
similarity index 100%
rename from core/server/src/streaming/systems/topics.rs
rename to core/server/src/shard/system/topics.rs
diff --git a/core/server/src/streaming/systems/users.rs b/core/server/src/shard/system/users.rs
similarity index 79%
rename from core/server/src/streaming/systems/users.rs
rename to core/server/src/shard/system/users.rs
index cc27fcae..f2405f1f 100644
--- a/core/server/src/streaming/systems/users.rs
+++ b/core/server/src/shard/system/users.rs
@@ -16,13 +16,12 @@
* under the License.
*/
+use crate::shard::IggyShard;
use crate::state::command::EntryCommand;
use crate::state::models::CreateUserWithId;
use crate::state::system::UserState;
use crate::streaming::personal_access_tokens::personal_access_token::PersonalAccessToken;
use crate::streaming::session::Session;
-use crate::streaming::systems::COMPONENT;
-use crate::streaming::systems::system::System;
use crate::streaming::users::user::User;
use crate::streaming::utils::crypto;
use crate::{IGGY_ROOT_PASSWORD_ENV, IGGY_ROOT_USERNAME_ENV};
@@ -42,108 +41,7 @@ use tracing::{error, info, warn};
static USER_ID: AtomicU32 = AtomicU32::new(1);
const MAX_USERS: usize = u32::MAX as usize;
-impl System {
- pub(crate) async fn load_users(&mut self, users: Vec<UserState>) -> Result<(), IggyError> {
- info!("Loading users...");
- if users.is_empty() {
- info!("No users found, creating the root user...");
- let root = Self::create_root_user();
- let command = CreateUser {
- username: root.username.clone(),
- password: root.password.clone(),
- status: root.status,
- permissions: root.permissions.clone(),
- };
- self.state
- .apply(0, &EntryCommand::CreateUser(CreateUserWithId {
- user_id: root.id,
- command
- }))
- .await
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to apply create user command, username: {}",
- root.username
- )
- })?;
-
- self.users.insert(root.id, root);
- info!("Created the root user.");
- }
-
- for user_state in users.into_iter() {
- let mut user = User::with_password(
- user_state.id,
- &user_state.username,
- user_state.password_hash,
- user_state.status,
- user_state.permissions,
- );
-
- user.created_at = user_state.created_at;
- user.personal_access_tokens = user_state
- .personal_access_tokens
- .into_values()
- .map(|token| {
- (
- Arc::new(token.token_hash.clone()),
- PersonalAccessToken::raw(
- user_state.id,
- &token.name,
- &token.token_hash,
- token.expiry_at,
- ),
- )
- })
- .collect();
- self.users.insert(user_state.id, user);
- }
-
- let users_count = self.users.len();
- let current_user_id = self.users.keys().max().unwrap_or(&1);
- USER_ID.store(current_user_id + 1, Ordering::SeqCst);
- self.permissioner
- .init(&self.users.values().collect::<Vec<&User>>());
- self.metrics.increment_users(users_count as u32);
- info!("Initialized {users_count} user(s).");
- Ok(())
- }
-
- 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."
- );
- }
- if username.is_ok() && password.is_ok() {
- info!("Using the custom root user credentials.");
- } else {
- info!("Using the default root user credentials.");
- }
-
- let username = username.unwrap_or(DEFAULT_ROOT_USERNAME.to_string());
- let password = password.unwrap_or(DEFAULT_ROOT_PASSWORD.to_string());
- if username.is_empty() || password.is_empty() {
- panic!("Root user credentials are not set.");
- }
- if username.len() < MIN_USERNAME_LENGTH {
- panic!("Root username is too short.");
- }
- if username.len() > MAX_USERNAME_LENGTH {
- panic!("Root username is too long.");
- }
- if password.len() < MIN_PASSWORD_LENGTH {
- panic!("Root password is too short.");
- }
- if password.len() > MAX_PASSWORD_LENGTH {
- panic!("Root password is too long.");
- }
-
- User::root(&username, &password)
- }
-
+impl IggyShard {
pub fn find_user(
&self,
session: &Session,
diff --git a/core/server/src/streaming/mod.rs b/core/server/src/streaming/mod.rs
index 5009cb9a..f9d40af8 100644
--- a/core/server/src/streaming/mod.rs
+++ b/core/server/src/streaming/mod.rs
@@ -28,7 +28,6 @@ pub mod segments;
pub mod session;
pub mod storage;
pub mod streams;
-pub mod systems;
pub mod topics;
pub mod users;
pub mod utils;
diff --git a/core/server/src/streaming/storage.rs b/core/server/src/streaming/storage.rs
index fc0a752d..f423b842 100644
--- a/core/server/src/streaming/storage.rs
+++ b/core/server/src/streaming/storage.rs
@@ -33,6 +33,7 @@ use iggy_common::IggyError;
use mockall::automock;
use std::fmt::Debug;
use std::future::Future;
+use std::rc::Rc;
use std::sync::Arc;
macro_rules! forward_async_methods {
@@ -143,7 +144,7 @@ pub trait PartitionStorage: Send {
#[derive(Debug)]
pub struct SystemStorage {
- pub info: Arc<SystemInfoStorageKind>,
+ pub info: Rc<SystemInfoStorageKind>,
pub stream: Arc<StreamStorageKind>,
pub topic: Arc<TopicStorageKind>,
pub partition: Arc<PartitionStorageKind>,
@@ -151,7 +152,7 @@ pub struct SystemStorage {
}
impl SystemStorage {
- pub fn new(config: Arc<SystemConfig>, persister: Arc<PersisterKind>) -> Self {
+ pub fn new(config: Rc<SystemConfig>, persister: Arc<PersisterKind>) -> Self {
Self {
info: Arc::new(SystemInfoStorageKind::File(FileSystemInfoStorage::new(
config.get_state_info_path(),
diff --git a/core/server/src/streaming/streams/stream.rs b/core/server/src/streaming/streams/stream.rs
index ef528197..339fc08a 100644
--- a/core/server/src/streaming/streams/stream.rs
+++ b/core/server/src/streaming/streams/stream.rs
@@ -23,6 +23,7 @@ use ahash::AHashMap;
use iggy_common::IggyByteSize;
use iggy_common::IggyTimestamp;
use std::fmt::Display;
+use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
@@ -47,8 +48,8 @@ impl Stream {
pub fn empty(
id: u32,
name: &str,
- config: Arc<SystemConfig>,
- storage: Arc<SystemStorage>,
+ config: Rc<SystemConfig>,
+ storage: Rc<SystemStorage>,
) -> Self {
Stream::create(id, name, config, storage)
}
@@ -56,8 +57,8 @@ impl Stream {
pub fn create(
id: u32,
name: &str,
- config: Arc<SystemConfig>,
- storage: Arc<SystemStorage>,
+ config: Rc<SystemConfig>,
+ storage: Rc<SystemStorage>,
) -> Self {
let path = config.get_stream_path(id);
let topics_path = config.get_topics_path(id);
diff --git a/core/server/src/tcp/sender.rs b/core/server/src/tcp/sender.rs
index 2b6b6d17..4dd11725 100644
--- a/core/server/src/tcp/sender.rs
+++ b/core/server/src/tcp/sender.rs
@@ -17,15 +17,15 @@
*/
use iggy_common::IggyError;
+use monoio::io::{AsyncReadRent, AsyncReadRentExt, AsyncWriteRent, AsyncWriteRentExt};
use std::io::IoSlice;
-use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tracing::debug;
const STATUS_OK: &[u8] = &[0; 4];
pub(crate) async fn read<T>(stream: &mut T, buffer: &mut [u8]) -> Result<usize, IggyError>
where
- T: AsyncRead + AsyncWrite + Unpin,
+ T: AsyncReadRent + AsyncWriteRent + Unpin,
{
match stream.read_exact(buffer).await {
Ok(0) => Err(IggyError::ConnectionClosed),
@@ -42,14 +42,14 @@ where
pub(crate) async fn send_empty_ok_response<T>(stream: &mut T) -> Result<(), IggyError>
where
- T: AsyncRead + AsyncWrite + Unpin,
+ T: AsyncReadRent + AsyncWriteRent + Unpin,
{
send_ok_response(stream, &[]).await
}
pub(crate) async fn send_ok_response<T>(stream: &mut T, payload: &[u8]) -> Result<(), IggyError>
where
- T: AsyncRead + AsyncWrite + Unpin,
+ T: AsyncReadRent + AsyncWriteRent + Unpin,
{
send_response(stream, STATUS_OK, payload).await
}
@@ -60,7 +60,7 @@ pub(crate) async fn send_ok_response_vectored<T>(
slices: Vec<IoSlice<'_>>,
) -> Result<(), IggyError>
where
- T: AsyncRead + AsyncWrite + Unpin,
+ T: AsyncReadRentExt + AsyncWriteRentExt + Unpin,
{
send_response_vectored(stream, STATUS_OK, length, slices).await
}
@@ -70,7 +70,7 @@ pub(crate) async fn send_error_response<T>(
error: IggyError,
) -> Result<(), IggyError>
where
- T: AsyncRead + AsyncWrite + Unpin,
+ T: AsyncReadRent + AsyncWriteRent + Unpin,
{
send_response(stream, &error.as_code().to_le_bytes(), &[]).await
}
@@ -81,7 +81,7 @@ pub(crate) async fn send_response<T>(
payload: &[u8],
) -> Result<(), IggyError>
where
- T: AsyncRead + AsyncWrite + Unpin,
+ T: AsyncReadRent + AsyncWriteRent + Unpin,
{
debug!(
"Sending response of len: {} with status: {:?}...",
@@ -104,7 +104,7 @@ pub(crate) async fn send_response_vectored<T>(
mut slices: Vec<IoSlice<'_>>,
) -> Result<(), IggyError>
where
- T: AsyncReadExt + AsyncWriteExt + Unpin,
+ T: AsyncReadRentExt + AsyncWriteRentExt + Unpin,
{
debug!(
"Sending vectored response of len: {} with status: {:?}...",
diff --git a/core/server/src/tcp/tcp_listener.rs b/core/server/src/tcp/tcp_listener.rs
index 0d84aa10..2e4f25be 100644
--- a/core/server/src/tcp/tcp_listener.rs
+++ b/core/server/src/tcp/tcp_listener.rs
@@ -17,60 +17,44 @@
*/
use crate::binary::sender::SenderKind;
+use crate::shard::IggyShard;
use crate::streaming::clients::client_manager::Transport;
-use crate::streaming::systems::system::SharedSystem;
use crate::tcp::connection_handler::{handle_connection, handle_error};
use std::net::SocketAddr;
+use std::rc::Rc;
+use monoio::net::TcpListener;
+use rustls::pki_types::Ipv4Addr;
use tokio::net::TcpSocket;
use tokio::sync::oneshot;
use tracing::{error, info};
-pub async fn start(address: &str, socket: TcpSocket, system: SharedSystem) -> SocketAddr {
- let address = address.to_string();
- let (tx, rx) = oneshot::channel();
- tokio::spawn(async move {
- let addr = address.parse();
- if addr.is_err() {
- panic!("Unable to parse address {:?}", address);
- }
-
- socket
- .bind(addr.unwrap())
- .expect("Unable to bind socket to address");
-
- let listener = socket.listen(1024).expect("Unable to start TCP server.");
-
- let local_addr = listener
- .local_addr()
- .expect("Failed to get local address for TCP listener");
-
- tx.send(local_addr).unwrap_or_else(|_| {
- panic!(
- "Failed to send the local address {:?} for TCP listener",
- local_addr
- )
- });
-
+pub async fn start(server_name: &str, shard: Rc<IggyShard>) {
+ let addr: SocketAddr = if shard.config.tcp.ipv6 {
+ shard.config.tcp.address.parse().expect("Unable to parse IPv6 address")
+ } else {
+ shard.config.tcp.address.parse().expect("Unable to parse IPv4 address")
+ };
+ monoio::spawn(async move {
+ let listener = TcpListener::bind(addr).expect(format!("Unable to start {server_name}.").as_ref());
loop {
match listener.accept().await {
Ok((stream, address)) => {
+ let shard = shard.clone();
info!("Accepted new TCP connection: {address}");
- let session = system
- .read()
- .await
- .add_client(&address, Transport::Tcp)
- .await;
+ let session = shard
+ .add_client(&address, Transport::Tcp);
let client_id = session.client_id;
info!("Created new session: {session}");
- let system = system.clone();
let mut sender = SenderKind::get_tcp_sender(stream);
- tokio::spawn(async move {
+ monoio::spawn(async move {
if let Err(error) =
- handle_connection(session, &mut sender, system.clone()).await
+ handle_connection(session, &mut sender, shard.clone()).await
{
handle_error(error);
- system.read().await.delete_client(client_id).await;
+ //TODO: Fixme
+ /*
+ //system.read().await.delete_client(client_id).await;
if let Err(error) = sender.shutdown().await {
error!(
"Failed to shutdown TCP stream for client: {client_id}, address: {address}. {error}"
@@ -80,6 +64,7 @@ pub async fn start(address: &str, socket: TcpSocket, system: SharedSystem) -> So
"Successfully closed TCP stream for client: {client_id}, address: {address}."
);
}
+ */
}
});
}
@@ -87,8 +72,4 @@ pub async fn start(address: &str, socket: TcpSocket, system: SharedSystem) -> So
}
}
});
- match rx.await {
- Ok(addr) => addr,
- Err(_) => panic!("Failed to get the local address for TCP listener."),
- }
}
diff --git a/core/server/src/tcp/tcp_sender.rs b/core/server/src/tcp/tcp_sender.rs
index 837c8f06..a157dead 100644
--- a/core/server/src/tcp/tcp_sender.rs
+++ b/core/server/src/tcp/tcp_sender.rs
@@ -21,7 +21,8 @@ use crate::tcp::COMPONENT;
use crate::{server_error::ServerError, tcp::sender};
use error_set::ErrContext;
use iggy_common::IggyError;
-use tokio::{io::AsyncWriteExt, net::TcpStream};
+use tokio::{io::AsyncWriteExt};
+use monoio::net::TcpStream;
#[derive(Debug)]
pub struct TcpSender {
diff --git a/core/server/src/tcp/tcp_server.rs b/core/server/src/tcp/tcp_server.rs
index a57b7c3f..82bbe0c1 100644
--- a/core/server/src/tcp/tcp_server.rs
+++ b/core/server/src/tcp/tcp_server.rs
@@ -16,26 +16,25 @@
* under the License.
*/
-use crate::configs::tcp::TcpConfig;
-use crate::streaming::systems::system::SharedSystem;
-use crate::tcp::{tcp_listener, tcp_socket, tcp_tls_listener};
-use std::net::SocketAddr;
+use crate::shard::IggyShard;
+use crate::tcp::{tcp_listener};
+use std::rc::Rc;
+use iggy_common::IggyError;
use tracing::info;
/// Starts the TCP server.
/// Returns the address the server is listening on.
-pub async fn start(config: TcpConfig, system: SharedSystem) -> SocketAddr {
- let server_name = if config.tls.enabled {
+pub async fn start(shard: Rc<IggyShard>) -> Result<(), IggyError> {
+ let server_name = if shard.config.tcp.tls.enabled {
"Iggy TCP TLS"
} else {
"Iggy TCP"
};
info!("Initializing {server_name} server...");
- let socket = tcp_socket::build(config.ipv6, config.socket);
- let addr = match config.tls.enabled {
- true => tcp_tls_listener::start(&config.address, config.tls, socket, system).await,
- false => tcp_listener::start(&config.address, socket, system).await,
+ let addr = match shard.config.tcp.tls.enabled {
+ true => unimplemented!("TLS support is not implemented yet"),
+ false => tcp_listener::start(server_name, shard).await,
};
info!("{server_name} server has started on: {:?}", addr);
- addr
+ Ok(())
}
diff --git a/core/server/src/tcp/tcp_socket.rs b/core/server/src/tcp/tcp_socket.rs
index 91ed105a..69c1d408 100644
--- a/core/server/src/tcp/tcp_socket.rs
+++ b/core/server/src/tcp/tcp_socket.rs
@@ -18,7 +18,6 @@
use std::num::TryFromIntError;
-use tokio::net::TcpSocket;
use crate::configs::tcp::TcpSocketConfig;
diff --git a/core/server/src/tcp/tcp_tls_sender.rs b/core/server/src/tcp/tcp_tls_sender.rs
index 98aef288..6ba62315 100644
--- a/core/server/src/tcp/tcp_tls_sender.rs
+++ b/core/server/src/tcp/tcp_tls_sender.rs
@@ -21,9 +21,8 @@ use crate::tcp::COMPONENT;
use crate::{server_error::ServerError, tcp::sender};
use error_set::ErrContext;
use iggy_common::IggyError;
-use tokio::io::AsyncWriteExt;
-use tokio::net::TcpStream;
-use tokio_native_tls::TlsStream;
+use monoio::net::TcpStream;
+use monoio_native_tls::TlsStream;
#[derive(Debug)]
pub struct TcpTlsSender {