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/08/25 12:52:23 UTC
(iggy) branch initial_loading_state_from_disk created (now 41f444c8)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch initial_loading_state_from_disk
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 41f444c8 feat(io_uring): initial loading stream/topic/partition state from disk
This branch includes the following new commits:
new 41f444c8 feat(io_uring): initial loading stream/topic/partition state from disk
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: feat(io_uring): initial loading stream/topic/partition state from disk
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch initial_loading_state_from_disk
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 41f444c81bc210850f902ca792afa04f981fb48d
Author: numminex <g....@gmail.com>
AuthorDate: Mon Aug 25 14:52:11 2025 +0200
feat(io_uring): initial loading stream/topic/partition state from disk
---
core/server/src/bootstrap.rs | 190 ++++++++++++++++--
core/server/src/main.rs | 16 +-
core/server/src/shard/builder.rs | 26 +--
core/server/src/shard/mod.rs | 50 +----
core/server/src/shard/system/partitions.rs | 25 +--
core/server/src/shard/system/streams.rs | 8 +-
core/server/src/shard/system/topics.rs | 10 +-
core/server/src/slab/streams.rs | 32 +--
core/server/src/slab/topics.rs | 13 +-
core/server/src/state/system.rs | 6 +
core/server/src/streaming/partitions/helpers.rs | 29 ++-
core/server/src/streaming/partitions/partition2.rs | 5 +-
core/server/src/streaming/partitions/storage.rs | 218 +--------------------
core/server/src/streaming/stats/stats.rs | 8 -
core/server/src/streaming/streams/stream2.rs | 4 +
core/server/src/streaming/topics/storage2.rs | 188 ++++++++++++++++++
core/server/src/streaming/topics/topic2.rs | 4 +
17 files changed, 480 insertions(+), 352 deletions(-)
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index ac8e9f54..75f4d284 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -1,13 +1,3 @@
-use compio::{fs::create_dir_all, runtime::Runtime};
-use iggy_common::{
- IggyError,
- defaults::{
- DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME, MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH,
- MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH,
- },
-};
-use tracing::info;
-
use crate::{
IGGY_ROOT_PASSWORD_ENV, IGGY_ROOT_USERNAME_ENV,
configs::{config_provider::ConfigProviderKind, server::ServerConfig, system::SystemConfig},
@@ -20,15 +10,193 @@ use crate::{
frame::ShardFrame,
},
},
+ slab::{
+ streams::Streams,
+ traits_ext::{
+ EntityComponentSystem, EntityComponentSystemMutCell, InsertCell, IntoComponents,
+ },
+ },
+ state::system::{StreamState, TopicState, UserState},
streaming::{
+ deduplication::message_deduplicator,
+ partitions::{helpers::create_message_deduplicator, partition2},
persistence::persister::{FilePersister, FileWithSyncPersister, PersisterKind},
+ personal_access_tokens::personal_access_token::PersonalAccessToken,
+ stats::stats::{PartitionStats, StreamStats, TopicStats},
storage::SystemStorage,
+ streams::stream2,
+ topics::{consumer_group2, topic2},
users::user::User,
utils::file::overwrite,
},
versioning::SemanticVersion,
};
-use std::{collections::HashSet, env, path::Path, rc::Rc, sync::Arc, time::Duration};
+use ahash::HashMap;
+use compio::{fs::create_dir_all, runtime::Runtime};
+use iggy_common::{
+ IggyError, UserId,
+ defaults::{
+ DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME, MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH,
+ MIN_PASSWORD_LENGTH, MIN_USERNAME_LENGTH,
+ },
+};
+use std::{collections::HashSet, env, path::Path, sync::Arc};
+use tracing::info;
+
+pub fn load_streams(
+ state: impl IntoIterator<Item = StreamState>,
+ config: &SystemConfig,
+) -> Streams {
+ let streams = Streams::default();
+ for StreamState {
+ name,
+ created_at,
+ id,
+ topics,
+ } in state
+ {
+ let stream_id = id;
+ let stats = Arc::new(Default::default());
+ let stream = stream2::Stream::new(name.clone(), stats.clone(), created_at);
+ let new_id = streams.insert(stream);
+
+ assert_eq!(
+ new_id, stream_id as usize,
+ "load_streams: id mismatch when inserting stream, mismatch for stream with ID: {}, name: {}",
+ stream_id, name
+ );
+ let topics = topics.into_values();
+ for TopicState {
+ id,
+ name,
+ created_at,
+ compression_algorithm,
+ message_expiry,
+ max_topic_size,
+ replication_factor,
+ consumer_groups,
+ partitions,
+ } in topics
+ {
+ let topic_id = id;
+ let parent_stats = stats.clone();
+ let stats = Arc::new(TopicStats::new(parent_stats));
+ let topic_id = streams.with_components_by_id(stream_id, move |(root, ..)| {
+ let topic = topic2::Topic::new(
+ name.clone(),
+ stats.clone(),
+ created_at,
+ replication_factor,
+ message_expiry,
+ compression_algorithm,
+ max_topic_size,
+ );
+ let new_id = root.topics_mut().insert(topic);
+ assert_eq!(
+ new_id, topic_id as usize,
+ "load_streams: topic id mismatch when inserting topic, mismatch for topic with ID: {}, name: {}",
+ topic_id, name
+ );
+ new_id
+ });
+
+ let parent_stats = stats.clone();
+ let cgs = consumer_groups.into_values();
+ let partitions = partitions.into_values();
+ for partition_state in partitions {
+ // TODO: Load consumer groups and partitions
+ streams.with_components_by_id(stream_id, move |(root, ..)| {
+ root.topics()
+ .with_components_by_id_mut(topic_id, move |(root, ..)| {
+ let stats = PartitionStats::new(parent_stats);
+ // TODO: This has to be sampled from segments, if there exists more than 0 segements and the segment has more than 0 messages, then we set it to true.
+ let should_increment_offset = todo!();
+ // TODO: This has to be sampled from segments and it's indexes, offset = segment.start_offset + last_index.offset;
+ let offset = todo!();
+ let message_deduplicator = create_message_deduplicator(config);
+ let id = partition_state.id;
+ // TODO: Load the consumer and consumer_group offsets, outside of this closure.
+ let partition = partition2::Partition::new(
+ partition_state.created_at,
+ should_increment_offset,
+ stats,
+ message_deduplicator,
+ offset,
+ Default::default(),
+ Default::default(),
+ );
+ let new_id = root.partitions_mut().insert(partition);
+ assert_eq!(
+ new_id, id as usize,
+ "load_streams: partition id mismatch when inserting partition, mismatch for partition with ID: {}, for topic with ID: {}, for stream with ID: {}",
+ id, topic_id, stream_id
+ );
+
+ })
+ });
+ }
+ let partition_ids = streams.with_components_by_id(stream_id, |(root, ..)| {
+ root.topics().with_components_by_id(topic_id, |(root, ..)| {
+ root.partitions().with_components(|components| {
+ let (root, ..) = components.into_components();
+ root.iter().map(|(_, root)| root.id()).collect::<Vec<_>>()
+ })
+ })
+ });
+
+ for cg_state in cgs {
+ streams.with_components_by_id(stream_id, move |(root, ..)| {
+ root.topics()
+ .with_components_by_id_mut(topic_id, move |(root, ..)| {
+ let id = cg_state.id;
+ let cg = consumer_group2::ConsumerGroup::new(cg_state.name.clone(), Default::default(), partition_ids);
+ let new_id = root.consumer_groups_mut().insert(cg);
+ assert_eq!(
+ new_id, id as usize,
+ "load_streams: consumer group id mismatch when inserting consumer group, mismatch for consumer group with ID: {}, name: {} for topic with ID: {}, for stream with ID: {}",
+ id, cg_state.name, topic_id, stream_id
+ );
+ });
+ });
+ }
+ }
+ }
+ streams
+}
+
+pub fn load_users(state: impl IntoIterator<Item = UserState>) -> HashMap<UserId, User> {
+ let mut users = Default::default();
+ for user_state in state {
+ let UserState {
+ id,
+ username,
+ password_hash,
+ status,
+ created_at,
+ permissions,
+ personal_access_tokens,
+ } = user_state;
+ let mut user = User::with_password(id, &username, password_hash, status, permissions);
+ user.created_at = 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();
+ users.insert(id, user);
+ }
+ users
+}
pub fn create_shard_connections(
shards_set: &HashSet<usize>,
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index 6d72d048..99f33aea 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -30,11 +30,12 @@ use figlet_rs::FIGfont;
use iggy_common::create_user::CreateUser;
use iggy_common::defaults::DEFAULT_ROOT_USER_ID;
use iggy_common::{Aes256GcmEncryptor, EncryptorKind, IggyError};
+use lending_iterator::lending_iterator::constructors::into_lending_iter;
use server::archiver::{ArchiverKind, ArchiverKindType};
use server::args::Args;
+use server::binary::handlers::streams;
use server::bootstrap::{
- create_directories, create_root_user, create_shard_connections, create_shard_executor,
- load_config, resolve_persister, update_system_info,
+ create_directories, create_root_user, create_shard_connections, create_shard_executor, create_streams, create_users, load_config, load_streams, load_users, resolve_persister, update_system_info
};
use server::configs::config_provider::{self};
use server::configs::server::ServerConfig;
@@ -207,7 +208,10 @@ async fn main() -> Result<(), ServerError> {
state_persister,
encryptor.clone(),
));
- let init_state = SystemState::load(state).await?;
+ let state = SystemState::load(state).await?;
+ let (streams_state, users_state) = state.decompose();
+ let streams = load_streams(streams_state.into_values(), &config.system);
+ let users = load_users(users_state.into_values());
// ELEVENTH DISCRETE LOADING STEP.
let shards_set = config.system.sharding.cpu_allocation.to_shard_set();
@@ -238,6 +242,8 @@ async fn main() -> Result<(), ServerError> {
for shard_id in shards_set {
let id = shard_id as u16;
+ let streams = streams.clone();
+ let users = users.clone();
let storage = storage.clone();
let connections = connections.clone();
let config = config.clone();
@@ -262,14 +268,14 @@ async fn main() -> Result<(), ServerError> {
let builder = IggyShard::builder();
let shard = builder
.id(id)
+ .streams(streams)
+ .users(users)
.connections(connections)
.config(config)
.storage(storage)
.archiver(archiver)
.encryptor(encryptor)
.version(current_version)
- .state(state)
- .init_state(init_state)
.metrics(shared_metrics)
.build();
let shard = Rc::new(shard);
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index a287e88d..a4621576 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -22,7 +22,9 @@ use std::{
sync::{Arc, atomic::AtomicBool},
};
-use iggy_common::{Aes256GcmEncryptor, EncryptorKind};
+use ahash::HashMap;
+use iggy_common::{Aes256GcmEncryptor, EncryptorKind, UserId};
+use sysinfo::User;
use tracing::info;
use crate::{
@@ -41,14 +43,14 @@ use super::{IggyShard, transmission::connector::ShardConnector, transmission::fr
#[derive(Default)]
pub struct IggyShardBuilder {
id: Option<u16>,
+ streams: Option<Streams>,
+ users: Option<HashMap<UserId, User>>,
connections: Option<Vec<ShardConnector<ShardFrame>>>,
config: Option<ServerConfig>,
encryptor: Option<EncryptorKind>,
version: Option<SemanticVersion>,
storage: Option<SystemStorage>,
archiver: Option<ArchiverKind>,
- state: Option<StateKind>,
- init_state: Option<SystemState>,
metrics: Option<Metrics>,
}
@@ -78,13 +80,13 @@ impl IggyShardBuilder {
self
}
- pub fn state(mut self, state: StateKind) -> Self {
- self.state = Some(state);
+ pub fn streams(mut self, streams: Streams) -> Self {
+ self.streams = Some(streams);
self
}
- pub fn init_state(mut self, init_state: SystemState) -> Self {
- self.init_state = Some(init_state);
+ pub fn users(mut self, users: HashMap<UserId, User>) -> Self {
+ self.users = Some(users);
self
}
@@ -105,11 +107,11 @@ impl IggyShardBuilder {
// TODO: Too much happens in there, some of those bootstrapping logic should be moved outside.
pub fn build(self) -> IggyShard {
let id = self.id.unwrap();
+ let streams = self.streams.unwrap();
+ let users = self.users.unwrap();
let config = self.config.unwrap();
let connections = self.connections.unwrap();
let storage = self.storage.unwrap();
- let state = self.state.unwrap();
- let init_state = self.init_state;
let encryptor = self.encryptor;
let version = self.version.unwrap();
let (stop_sender, stop_receiver, frame_receiver) = connections
@@ -134,11 +136,11 @@ impl IggyShardBuilder {
id: id,
shards: shards,
shards_table: Default::default(),
+ streams2: streams,
+ users: users,
storage: storage,
encryptor: encryptor,
archiver: archiver,
- state: state,
- init_state: init_state,
config: config,
version: version,
stop_receiver: stop_receiver,
@@ -153,8 +155,6 @@ impl IggyShardBuilder {
streams2: Streams::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/mod.rs b/core/server/src/shard/mod.rs
index 01479fce..557e9c53 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -145,9 +145,7 @@ pub struct IggyShard {
// TODO: Refactor.
pub(crate) storage: Rc<SystemStorage>,
- pub(crate) state: StateKind,
// Temporal...
- pub(crate) init_state: Option<SystemState>,
pub(crate) encryptor: Option<EncryptorKind>,
pub(crate) archiver: Option<Rc<ArchiverKind>>,
pub(crate) config: ServerConfig,
@@ -173,7 +171,6 @@ impl IggyShard {
pub fn default_from_config(server_config: ServerConfig) -> Self {
use crate::bootstrap::resolve_persister;
- use crate::state::file::FileState;
use crate::streaming::storage::SystemStorage;
use crate::versioning::SemanticVersion;
@@ -184,10 +181,6 @@ impl IggyShard {
persister.clone(),
));
- let state_path = server_config.system.get_state_messages_file_path();
- let file_state = FileState::new(&state_path, &version, persister, None);
- let state = crate::state::StateKind::File(file_state);
-
let (stop_sender, stop_receiver) = async_channel::unbounded();
let shard = Self {
@@ -195,11 +188,9 @@ impl IggyShard {
shards: Vec::new(),
shards_table: Default::default(),
version,
- streams2: Streams::init(),
+ streams2: Default::default(),
storage,
- state,
//TODO: Fix
- init_state: None,
encryptor: None,
archiver: None,
config: server_config,
@@ -229,11 +220,7 @@ impl IggyShard {
}
pub async fn init(&self) -> Result<(), IggyError> {
- let system_state = self.init_state.as_ref().unwrap();
- let SystemState { users, streams } = system_state;
- let _ = self.load_users(users.values().cloned().collect()).await;
- let _ = self.load_streams(streams.values().cloned().collect()).await;
-
+ let _ = self.load_users().await;
if let Some(archiver) = self.archiver.as_ref() {
archiver
.init()
@@ -299,36 +286,7 @@ impl IggyShard {
Ok(())
}
- async fn load_users(&self, users: Vec<UserState>) -> Result<(), IggyError> {
- shard_info!(self.id, "Loading users...");
- 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.borrow_mut().insert(user_state.id, user);
- }
-
+ async fn load_users(&self) -> Result<(), IggyError> {
let users = self.users.borrow();
let users_count = users.len();
let current_user_id = users.keys().max().unwrap_or(&1);
@@ -341,6 +299,7 @@ impl IggyShard {
Ok(())
}
+ /*
async fn load_streams(&self, streams: Vec<StreamState>) -> Result<(), IggyError> {
shard_info!(self.id, "Loading streams from disk...");
let mut unloaded_streams = Vec::new();
@@ -487,6 +446,7 @@ impl IggyShard {
);
Ok(())
}
+ */
pub fn assert_init(&self) -> Result<(), IggyError> {
Ok(())
diff --git a/core/server/src/shard/system/partitions.rs b/core/server/src/shard/system/partitions.rs
index a8ddd0c5..9f96e5bf 100644
--- a/core/server/src/shard/system/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -26,7 +26,9 @@ use crate::slab::traits_ext::IntoComponents;
use crate::slab::traits_ext::IntoComponentsById;
use crate::streaming::deduplication::message_deduplicator::MessageDeduplicator;
use crate::streaming::partitions;
+use crate::streaming::partitions::helpers::create_message_deduplicator;
use crate::streaming::partitions::partition2;
+use crate::streaming::partitions::partition2::create_message_deduplicator;
use crate::streaming::partitions::storage2::create_partition_file_hierarchy;
use crate::streaming::partitions::storage2::delete_partitions_from_disk;
use crate::streaming::session::Session;
@@ -127,38 +129,21 @@ impl IggyShard {
parent_stats: Arc<TopicStats>,
partitions_count: u32,
) -> Vec<partition2::Partition> {
- fn create_message_deduplicator(config: &SystemConfig) -> Option<MessageDeduplicator> {
- if !config.message_deduplication.enabled {
- return None;
- }
- let max_entries = if config.message_deduplication.max_entries > 0 {
- Some(config.message_deduplication.max_entries)
- } else {
- None
- };
- let expiry = if !config.message_deduplication.expiry.is_zero() {
- Some(config.message_deduplication.expiry)
- } else {
- None
- };
-
- Some(MessageDeduplicator::new(max_entries, expiry))
- }
-
let range = 0..partitions_count as usize;
let created_at = IggyTimestamp::now();
range
.map(|_| {
// Areczkuuuu.
let stats = Arc::new(PartitionStats::new(parent_stats.clone()));
- let info = partition2::PartitionRoot::new(created_at, false);
+ let should_increment_offset = false;
let deduplicator = create_message_deduplicator(&self.config.system);
let offset = Arc::new(AtomicU64::new(0));
let consumer_offset = Arc::new(papaya::HashMap::with_capacity(2137));
let consumer_group_offset = Arc::new(papaya::HashMap::with_capacity(2137));
let mut partition = partition2::Partition::new(
- info,
+ created_at,
+ should_increment_offset,
stats,
deduplicator,
offset,
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index d308c2b7..19842d5b 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -58,17 +58,13 @@ impl IggyShard {
let now = IggyTimestamp::now();
let stats = Arc::new(StreamStats::new());
let mut stream = stream2::Stream::new(name, stats, now);
- let id = self.insert_stream_mem(stream.clone());
+ let id = self.streams2.insert(stream.clone());
stream.update_id(id);
stream
}
- fn insert_stream_mem(&self, stream: stream2::Stream) -> usize {
- self.streams2.insert(stream)
- }
-
pub fn create_stream2_bypass_auth(&self, stream: stream2::Stream) -> usize {
- self.insert_stream_mem(stream)
+ self.streams2.insert(stream)
}
pub fn update_stream2_bypass_auth(&self, id: &Identifier, name: &str) -> Result<(), IggyError> {
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index 6e1a0207..fd23279b 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -114,20 +114,18 @@ impl IggyShard {
max_topic_size,
);
- let id = self.insert_topic_mem(stream_id, topic.clone());
+ let id = self
+ .streams2
+ .with_topics(stream_id, |topics| topics.insert(topic.clone()));
topic.update_id(id);
topic
}
- fn insert_topic_mem(&self, stream_id: &Identifier, topic: topic2::Topic) -> usize {
+ pub fn create_topic2_bypass_auth(&self, stream_id: &Identifier, topic: topic2::Topic) -> usize {
self.streams2
.with_topics(stream_id, |topics| topics.insert(topic))
}
- pub fn create_topic2_bypass_auth(&self, stream_id: &Identifier, topic: topic2::Topic) -> usize {
- self.insert_topic_mem(stream_id, topic)
- }
-
pub fn update_topic2(
&self,
session: &Session,
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index ef210e32..4a2fe5b6 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -29,12 +29,23 @@ use std::{cell::RefCell, sync::Arc};
const CAPACITY: usize = 1024;
pub type ContainerId = usize;
+#[derive(Debug, Clone)]
pub struct Streams {
index: RefCell<AHashMap<<stream2::StreamRoot as Keyed>::Key, ContainerId>>,
root: RefCell<Slab<stream2::StreamRoot>>,
stats: RefCell<Slab<Arc<StreamStats>>>,
}
+impl Default for Streams {
+ fn default() -> Self {
+ Self {
+ index: RefCell::new(AHashMap::with_capacity(CAPACITY)),
+ root: RefCell::new(Slab::with_capacity(CAPACITY)),
+ stats: RefCell::new(Slab::with_capacity(CAPACITY)),
+ }
+ }
+}
+
impl<'a> From<&'a Streams> for stream2::StreamRef<'a> {
fn from(value: &'a Streams) -> Self {
let root = value.root.borrow();
@@ -57,15 +68,20 @@ impl InsertCell for Streams {
fn insert(&self, item: Self::Item) -> Self::Idx {
let (root, stats) = item.into_components();
- let key = root.key().clone();
+ let mut root_container = self.root.borrow_mut();
+ let mut indexes = self.index.borrow_mut();
+ let mut stats_container = self.stats.borrow_mut();
- let entity_id = self.root.borrow_mut().insert(root);
- let id = self.stats.borrow_mut().insert(stats);
+ let key = root.key().clone();
+ let entity_id = root_container.insert(root);
+ let id = stats_container.insert(stats);
assert_eq!(
entity_id, id,
"stream_insert: id mismatch when inserting stats"
);
- self.index.borrow_mut().insert(key, entity_id);
+ let root = root_container.get_mut(entity_id).unwrap();
+ root.update_id(entity_id);
+ indexes.insert(key, entity_id);
entity_id
}
}
@@ -111,14 +127,6 @@ impl EntityComponentSystemMutCell for Streams {
}
impl Streams {
- pub fn init() -> Self {
- Self {
- index: RefCell::new(AHashMap::with_capacity(CAPACITY)),
- root: RefCell::new(Slab::with_capacity(CAPACITY)),
- stats: RefCell::new(Slab::with_capacity(CAPACITY)),
- }
- }
-
pub fn exists(&self, id: &Identifier) -> bool {
match id.kind {
iggy_common::IdKind::Numeric => {
diff --git a/core/server/src/slab/topics.rs b/core/server/src/slab/topics.rs
index 565efafc..52dac8ad 100644
--- a/core/server/src/slab/topics.rs
+++ b/core/server/src/slab/topics.rs
@@ -40,15 +40,20 @@ impl InsertCell for Topics {
fn insert(&self, item: Self::Item) -> Self::Idx {
let (root, stats) = item.into_components();
- let key = root.key().clone();
+ let mut root_container = self.root.borrow_mut();
+ let mut indexes = self.index.borrow_mut();
+ let mut stats_container = self.stats.borrow_mut();
- let entity_id = self.root.borrow_mut().insert(root);
- let id = self.stats.borrow_mut().insert(stats);
+ let key = root.key().clone();
+ let entity_id = root_container.insert(root);
+ let id = stats_container.insert(stats);
assert_eq!(
entity_id, id,
"topic_insert: id mismatch when inserting stats"
);
- self.index.borrow_mut().insert(key, entity_id);
+ let root = root_container.get_mut(entity_id).unwrap();
+ root.update_id(entity_id);
+ indexes.insert(key, entity_id);
entity_id
}
}
diff --git a/core/server/src/state/system.rs b/core/server/src/state/system.rs
index 2fcfdef7..663a0f75 100644
--- a/core/server/src/state/system.rs
+++ b/core/server/src/state/system.rs
@@ -39,6 +39,12 @@ pub struct SystemState {
pub users: AHashMap<u32, UserState>,
}
+impl SystemState {
+ pub fn decompose(self) -> (AHashMap<u32, StreamState>, AHashMap<u32, UserState>) {
+ (self.streams, self.users)
+ }
+}
+
#[derive(Debug, Clone)]
pub struct StreamState {
pub id: u32,
diff --git a/core/server/src/streaming/partitions/helpers.rs b/core/server/src/streaming/partitions/helpers.rs
index 4a278ca0..0c39aeff 100644
--- a/core/server/src/streaming/partitions/helpers.rs
+++ b/core/server/src/streaming/partitions/helpers.rs
@@ -11,10 +11,13 @@ use crate::{
IntoComponents,
},
},
- streaming::partitions::{
- consumer_offset::ConsumerOffset,
- partition2::{self, PartitionRef, PartitionRefMut},
- storage2,
+ streaming::{
+ deduplication::message_deduplicator::MessageDeduplicator,
+ partitions::{
+ consumer_offset::ConsumerOffset,
+ partition2::{self, PartitionRef, PartitionRefMut},
+ storage2,
+ },
},
};
@@ -214,3 +217,21 @@ pub fn purge_segments_mem() -> impl FnOnce(&mut Partitions) {
//partitions.segments_mut()
}
}
+
+pub fn create_message_deduplicator(config: &SystemConfig) -> Option<MessageDeduplicator> {
+ if !config.message_deduplication.enabled {
+ return None;
+ }
+ let max_entries = if config.message_deduplication.max_entries > 0 {
+ Some(config.message_deduplication.max_entries)
+ } else {
+ None
+ };
+ let expiry = if !config.message_deduplication.expiry.is_zero() {
+ Some(config.message_deduplication.expiry)
+ } else {
+ None
+ };
+
+ Some(MessageDeduplicator::new(max_entries, expiry))
+}
diff --git a/core/server/src/streaming/partitions/partition2.rs b/core/server/src/streaming/partitions/partition2.rs
index 78f7904a..99737f5b 100644
--- a/core/server/src/streaming/partitions/partition2.rs
+++ b/core/server/src/streaming/partitions/partition2.rs
@@ -1,4 +1,5 @@
use crate::{
+ configs::system::SystemConfig,
slab::{
partitions::{self, Partitions},
traits_ext::{EntityMarker, IntoComponents, IntoComponentsById},
@@ -27,13 +28,15 @@ pub struct Partition {
impl Partition {
pub fn new(
- root: PartitionRoot,
+ created_at: IggyTimestamp,
+ should_increment_offset: bool,
stats: Arc<PartitionStats>,
message_deduplicator: Option<MessageDeduplicator>,
offset: Arc<AtomicU64>,
consumer_offset: Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>,
consumer_group_offset: Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>,
) -> Self {
+ let root = PartitionRoot::new(created_at, should_increment_offset);
Self {
root,
stats,
diff --git a/core/server/src/streaming/partitions/storage.rs b/core/server/src/streaming/partitions/storage.rs
index e76205e1..b9f67164 100644
--- a/core/server/src/streaming/partitions/storage.rs
+++ b/core/server/src/streaming/partitions/storage.rs
@@ -21,10 +21,9 @@ use crate::configs::cache_indexes::CacheIndexesConfig;
use crate::io::fs_utils;
use crate::state::system::PartitionState;
use crate::streaming::partitions::COMPONENT;
-use crate::streaming::partitions::partition::{ConsumerOffset, Partition};
+use crate::streaming::partitions::consumer_offset::ConsumerOffset;
use crate::streaming::persistence::persister::PersisterKind;
use crate::streaming::segments::*;
-use crate::streaming::storage::PartitionStorage;
use crate::streaming::utils::file;
use compio::fs;
use compio::fs::create_dir_all;
@@ -261,219 +260,4 @@ impl PartitionStorage for FilePartitionStorage {
Ok(())
}
-
- async fn save(&self, partition: &mut Partition) -> Result<(), IggyError> {
- info!(
- "Saving partition with start ID: {} for stream with ID: {} and topic with ID: {}...",
- partition.partition_id, partition.stream_id, partition.topic_id
- );
- if !Path::new(&partition.partition_path).exists()
- && create_dir_all(&partition.partition_path).await.is_err()
- {
- return Err(IggyError::CannotCreatePartitionDirectory(
- partition.partition_id,
- partition.stream_id,
- partition.topic_id,
- ));
- }
-
- if !Path::new(&partition.offsets_path).exists()
- && create_dir_all(&partition.offsets_path).await.is_err()
- {
- error!(
- "Failed to create offsets directory for partition with ID: {} for stream with ID: {} and topic with ID: {}.",
- partition.partition_id, partition.stream_id, partition.topic_id
- );
- return Err(IggyError::CannotCreatePartition(
- partition.partition_id,
- partition.stream_id,
- partition.topic_id,
- ));
- }
-
- if !Path::new(&partition.consumer_offsets_path).exists()
- && create_dir_all(&partition.consumer_offsets_path)
- .await
- .is_err()
- {
- error!(
- "Failed to create consumer offsets directory for partition with ID: {} for stream with ID: {} and topic with ID: {}.",
- partition.partition_id, partition.stream_id, partition.topic_id
- );
- return Err(IggyError::CannotCreatePartition(
- partition.partition_id,
- partition.stream_id,
- partition.topic_id,
- ));
- }
-
- if !Path::new(&partition.consumer_group_offsets_path).exists()
- && create_dir_all(&partition.consumer_group_offsets_path)
- .await
- .is_err()
- {
- error!(
- "Failed to create consumer group offsets directory for partition with ID: {} for stream with ID: {} and topic with ID: {}.",
- partition.partition_id, partition.stream_id, partition.topic_id
- );
- return Err(IggyError::CannotCreatePartition(
- partition.partition_id,
- partition.stream_id,
- partition.topic_id,
- ));
- }
-
- info!(
- "Saved partition with start ID: {} for stream with ID: {} and topic with ID: {}, path: {}.",
- partition.partition_id,
- partition.stream_id,
- partition.topic_id,
- partition.partition_path
- );
-
- Ok(())
- }
-
- async fn delete(&self, partition: &Partition) -> Result<(), IggyError> {
- info!(
- "Deleting partition with ID: {} for stream with ID: {} and topic with ID: {}...",
- partition.partition_id, partition.stream_id, partition.topic_id,
- );
-
- if let Err(err) = self
- .delete_consumer_offsets(&partition.consumer_offsets_path)
- .await
- {
- error!(
- "Cannot delete consumer offsets for partition with ID: {} for topic with ID: {} for stream with ID: {}. Error: {}",
- partition.partition_id, partition.topic_id, partition.stream_id, err
- );
- return Err(IggyError::CannotDeletePartition(
- partition.partition_id,
- partition.topic_id,
- partition.stream_id,
- ));
- }
-
- if let Err(err) = self
- .delete_consumer_offsets(&partition.consumer_group_offsets_path)
- .await
- {
- error!(
- "Cannot delete consumer group offsets for partition with ID: {} for topic with ID: {} for stream with ID: {}. Error: {}",
- partition.partition_id, partition.topic_id, partition.stream_id, err
- );
- return Err(IggyError::CannotDeletePartition(
- partition.partition_id,
- partition.topic_id,
- partition.stream_id,
- ));
- }
-
- if fs_utils::remove_dir_all(&partition.partition_path)
- .await
- .is_err()
- {
- error!(
- "Cannot delete partition directory: {} for partition with ID: {} for topic with ID: {} for stream with ID: {}.",
- partition.partition_path,
- partition.partition_id,
- partition.topic_id,
- partition.stream_id
- );
- return Err(IggyError::CannotDeletePartitionDirectory(
- partition.partition_id,
- partition.stream_id,
- partition.topic_id,
- ));
- }
- info!(
- "Deleted partition with ID: {} for stream with ID: {} and topic with ID: {}.",
- partition.partition_id, partition.stream_id, partition.topic_id,
- );
- Ok(())
- }
-
- async fn save_consumer_offset(&self, offset: u64, path: &str) -> Result<(), IggyError> {
- self.persister
- .overwrite(path, Box::new(offset.to_le_bytes()))
- .await
- .with_error_context(|error| format!(
- "{COMPONENT} (error: {error}) - failed to overwrite consumer offset with value: {offset}, path: {path}",
- ))?;
- trace!("Stored consumer offset value: {}, path: {}", offset, path);
- Ok(())
- }
-
- async fn load_consumer_offsets(
- &self,
- kind: ConsumerKind,
- path: &str,
- ) -> Result<Vec<ConsumerOffset>, IggyError> {
- trace!("Loading consumer offsets from path: {path}...");
- //TODO: replace with async once its there
- let dir_entries = std::fs::read_dir(&path);
- if dir_entries.is_err() {
- return Err(IggyError::CannotReadConsumerOffsets(path.to_owned()));
- }
-
- let mut consumer_offsets = Vec::new();
- let mut dir_entries = dir_entries.unwrap();
- // TODO: reimplement using the compio dir walk
- while let Some(dir_entry) = dir_entries.next() {
- let dir_entry = dir_entry.unwrap();
- let metadata = dir_entry.metadata();
- if metadata.is_err() {
- break;
- }
-
- if metadata.unwrap().is_dir() {
- continue;
- }
-
- let name = dir_entry.file_name().into_string().unwrap();
- let consumer_id = name.parse::<u32>();
- if consumer_id.is_err() {
- error!("Invalid consumer ID file with name: '{}'.", name);
- continue;
- }
-
- let path = dir_entry.path();
- let path = path.to_str();
- if path.is_none() {
- error!("Invalid consumer ID path for file with name: '{}'.", name);
- continue;
- }
-
- let path = Arc::new(path.unwrap().to_string());
- let consumer_id = consumer_id.unwrap();
- let file = file::open(&path)
- .await
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to open offset file, path: {path}"
- )
- })
- .map_err(|_| IggyError::CannotReadFile)?;
- // TODO: This is like awfull.
- let mut cursor = std::io::Cursor::new(file);
- let offset = cursor
- .read_u64_le()
- .await
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to read consumer offset from file, path: {path}")
- })
- .map_err(|_| IggyError::CannotReadFile)?;
-
- consumer_offsets.push(ConsumerOffset {
- kind,
- consumer_id,
- offset,
- path,
- });
- }
-
- consumer_offsets.sort_by(|a, b| a.consumer_id.cmp(&b.consumer_id));
- Ok(consumer_offsets)
- }
}
diff --git a/core/server/src/streaming/stats/stats.rs b/core/server/src/streaming/stats/stats.rs
index 97047880..20933c42 100644
--- a/core/server/src/streaming/stats/stats.rs
+++ b/core/server/src/streaming/stats/stats.rs
@@ -11,14 +11,6 @@ pub struct StreamStats {
}
impl StreamStats {
- pub fn new() -> Self {
- Self {
- size_bytes: AtomicU64::new(0),
- messages_count: AtomicU64::new(0),
- segments_count: AtomicU32::new(0),
- }
- }
-
pub fn increment_size_bytes(&self, size_bytes: u64) {
self.size_bytes.fetch_add(size_bytes, Ordering::AcqRel);
}
diff --git a/core/server/src/streaming/streams/stream2.rs b/core/server/src/streaming/streams/stream2.rs
index 962259b8..f0c88455 100644
--- a/core/server/src/streaming/streams/stream2.rs
+++ b/core/server/src/streaming/streams/stream2.rs
@@ -40,6 +40,10 @@ impl StreamRoot {
}
}
+ pub fn update_id(&mut self, id: usize) {
+ self.id = id;
+ }
+
pub fn id(&self) -> usize {
self.id
}
diff --git a/core/server/src/streaming/topics/storage2.rs b/core/server/src/streaming/topics/storage2.rs
index 43f12875..c73a2af0 100644
--- a/core/server/src/streaming/topics/storage2.rs
+++ b/core/server/src/streaming/topics/storage2.rs
@@ -83,3 +83,191 @@ pub async fn delete_topic_from_disk(
);
Ok(())
}
+
+// Old implementation kept for reference
+/*
+async fn load(&self, topic: &mut Topic, mut state: TopicState) -> Result<(), IggyError> {
+ info!("Loading topic {} from disk...", topic);
+ if !Path::new(&topic.path).exists() {
+ return Err(IggyError::TopicIdNotFound(topic.topic_id, topic.stream_id));
+ }
+
+ let message_expiry = Topic::get_message_expiry(state.message_expiry, &topic.config);
+ let max_topic_size = Topic::get_max_topic_size(state.max_topic_size, &topic.config)?;
+ topic.created_at = state.created_at;
+ topic.message_expiry = message_expiry;
+ topic.max_topic_size = max_topic_size;
+ topic.compression_algorithm = state.compression_algorithm;
+ topic.replication_factor = state.replication_factor.unwrap_or(1);
+
+ let mut dir_entries = fs::read_dir(&topic.partitions_path).await
+ .with_context(|| format!("Failed to read partition with ID: {} for stream with ID: {} for topic with ID: {} and path: {}",
+ topic.topic_id, topic.stream_id, topic.topic_id, &topic.partitions_path))
+ .map_err(|_| IggyError::CannotReadPartitions)?;
+
+ let mut unloaded_partitions = Vec::new();
+ while let Some(dir_entry) = dir_entries.next_entry().await.unwrap_or(None) {
+ let metadata = dir_entry.metadata().await;
+ if metadata.is_err() || metadata.unwrap().is_file() {
+ continue;
+ }
+
+ let name = dir_entry.file_name().into_string().unwrap();
+ let partition_id = name.parse::<u32>();
+ if partition_id.is_err() {
+ error!("Invalid partition ID file with name: '{}'.", name);
+ continue;
+ }
+
+ let partition_id = partition_id.unwrap();
+ let partition_state = state.partitions.get(&partition_id);
+ if partition_state.is_none() {
+ let stream_id = topic.stream_id;
+ let topic_id = topic.topic_id;
+ error!(
+ "Partition with ID: '{partition_id}' for stream with ID: '{stream_id}' and topic with ID: '{topic_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 partition directory: {error}");
+ } else {
+ warn!(
+ "Partition with ID: '{partition_id}' for stream with ID: '{stream_id}' and topic with ID: '{topic_id}' was removed."
+ );
+ }
+ continue;
+ }
+
+ let partition_state = partition_state.unwrap();
+ let partition = Partition::create(
+ topic.stream_id,
+ topic.topic_id,
+ partition_id,
+ false,
+ topic.config.clone(),
+ topic.storage.clone(),
+ message_expiry,
+ topic.messages_count_of_parent_stream.clone(),
+ topic.messages_count.clone(),
+ topic.size_of_parent_stream.clone(),
+ topic.size_bytes.clone(),
+ topic.segments_count_of_parent_stream.clone(),
+ partition_state.created_at,
+ )
+ .await;
+ unloaded_partitions.push(partition);
+ }
+
+ let state_partition_ids = state.partitions.keys().copied().collect::<AHashSet<u32>>();
+ let unloaded_partition_ids = unloaded_partitions
+ .iter()
+ .map(|partition| partition.partition_id)
+ .collect::<AHashSet<u32>>();
+ let missing_ids = state_partition_ids
+ .difference(&unloaded_partition_ids)
+ .copied()
+ .collect::<AHashSet<u32>>();
+ if missing_ids.is_empty() {
+ info!(
+ "All partitions for topic with ID: '{}' for stream with ID: '{}' found on disk were found in state.",
+ topic.topic_id, topic.stream_id
+ );
+ } else {
+ warn!(
+ "Partitions with IDs: '{missing_ids:?}' for topic with ID: '{topic_id}' for stream with ID: '{stream_id}' were not found on disk.",
+ topic_id = topic.topic_id,
+ stream_id = topic.stream_id
+ );
+ if topic.config.recovery.recreate_missing_state {
+ info!(
+ "Recreating missing state in recovery config is enabled, missing partitions will be created for topic with ID: '{}' for stream with ID: '{}'.",
+ topic.topic_id, topic.stream_id
+ );
+
+ for partition_id in missing_ids {
+ let partition_state = state.partitions.get(&partition_id).unwrap();
+ let mut partition = Partition::create(
+ topic.stream_id,
+ topic.topic_id,
+ partition_id,
+ true,
+ topic.config.clone(),
+ topic.storage.clone(),
+ message_expiry,
+ topic.messages_count_of_parent_stream.clone(),
+ topic.messages_count.clone(),
+ topic.size_of_parent_stream.clone(),
+ topic.size_bytes.clone(),
+ topic.segments_count_of_parent_stream.clone(),
+ partition_state.created_at,
+ )
+ .await;
+ partition.persist().await.with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to persist partition: {partition}"
+ )
+ })?;
+ partition.segments.clear();
+ unloaded_partitions.push(partition);
+ info!(
+ "Created missing partition with ID: '{partition_id}', for topic with ID: '{}' for stream with ID: '{}'.",
+ topic.topic_id, topic.stream_id
+ );
+ }
+ } else {
+ warn!(
+ "Recreating missing state in recovery config is disabled, missing partitions will not be created for topic with ID: '{}' for stream with ID: '{}'.",
+ topic.topic_id, topic.stream_id
+ );
+ }
+ }
+
+ let stream_id = topic.stream_id;
+ let topic_id = topic.topic_id;
+ let loaded_partitions = Arc::new(Mutex::new(Vec::new()));
+ let mut load_partitions = Vec::new();
+ for mut partition in unloaded_partitions {
+ let loaded_partitions = loaded_partitions.clone();
+ let partition_state = state.partitions.remove(&partition.partition_id).unwrap();
+ let load_partition = tokio::spawn(async move {
+ match partition.load(partition_state).await {
+ Ok(_) => {
+ loaded_partitions.lock().await.push(partition);
+ }
+ Err(error) => {
+ error!(
+ "Failed to load partition with ID: {} for stream with ID: {stream_id} and topic with ID: {topic_id}. Error: {error}",
+ partition.partition_id
+ );
+ }
+ }
+ });
+ load_partitions.push(load_partition);
+ }
+
+ join_all(load_partitions).await;
+ for partition in loaded_partitions.lock().await.drain(..) {
+ topic
+ .partitions
+ .insert(partition.partition_id, IggySharedMut::new(partition));
+ }
+
+ for consumer_group in state.consumer_groups.into_values() {
+ let consumer_group = ConsumerGroup::new(
+ topic.topic_id,
+ consumer_group.id,
+ &consumer_group.name,
+ topic.get_partitions_count(),
+ );
+ topic
+ .consumer_groups_ids
+ .insert(consumer_group.name.to_owned(), consumer_group.group_id);
+ topic
+ .consumer_groups
+ .insert(consumer_group.group_id, RwLock::new(consumer_group));
+ }
+
+ info!("Loaded topic {topic}");
+
+ Ok(())
+ }
+ */
diff --git a/core/server/src/streaming/topics/topic2.rs b/core/server/src/streaming/topics/topic2.rs
index 8a99035c..e6533a53 100644
--- a/core/server/src/streaming/topics/topic2.rs
+++ b/core/server/src/streaming/topics/topic2.rs
@@ -187,6 +187,10 @@ impl TopicRoot {
f(self).await
}
+ pub fn update_id(&mut self, id: usize) {
+ self.id = id;
+ }
+
pub fn id(&self) -> topics::ContainerId {
self.id
}