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/07/31 10:21:47 UTC
(iggy) branch heart_transplant_part2 created (now a450cc14)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch heart_transplant_part2
in repository https://gitbox.apache.org/repos/asf/iggy.git
at a450cc14 feat(io_uring): refactor statistics and implement topic creation
This branch includes the following new commits:
new a450cc14 feat(io_uring): refactor statistics and implement topic creation
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): refactor statistics and implement topic creation
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch heart_transplant_part2
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit a450cc14411bd8199f238b52938542f0e7b90856
Author: numminex <g....@gmail.com>
AuthorDate: Thu Jul 31 12:21:36 2025 +0200
feat(io_uring): refactor statistics and implement topic creation
---
core/common/src/error/iggy_error.rs | 1 +
core/common/src/types/identifier/mod.rs | 2 +
.../handlers/streams/create_stream_handler.rs | 7 +-
.../binary/handlers/topics/create_topic_handler.rs | 33 +++++
core/server/src/http/consumer_offsets.rs | 6 +-
core/server/src/http/mapper.rs | 5 +-
core/server/src/io/storage.rs | 3 +-
core/server/src/shard/mod.rs | 89 ++++++++-----
core/server/src/shard/system/streams.rs | 26 +++-
core/server/src/shard/system/topics.rs | 143 +++++++++++++++++++++
core/server/src/shard/transmission/event.rs | 15 ++-
core/server/src/slab/consumer_groups.rs | 1 +
core/server/src/slab/mod.rs | 22 +---
core/server/src/slab/partitions.rs | 16 +--
core/server/src/slab/streams.rs | 64 +++++----
core/server/src/slab/topics.rs | 48 ++++---
core/server/src/streaming/mod.rs | 1 +
core/server/src/streaming/partitions/partition2.rs | 19 ++-
core/server/src/streaming/segments/segment2.rs | 1 +
core/server/src/streaming/stats/mod.rs | 1 +
core/server/src/streaming/stats/stats.rs | 55 ++++++++
core/server/src/streaming/streams/mod.rs | 1 +
core/server/src/streaming/streams/storage2.rs | 28 ++++
core/server/src/streaming/streams/stream2.rs | 22 +++-
.../server/src/streaming/topics/consumer_group2.rs | 2 +-
core/server/src/streaming/topics/topic2.rs | 54 +++++++-
26 files changed, 542 insertions(+), 123 deletions(-)
diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index 53b4eb5b..5759e864 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -16,6 +16,7 @@
* under the License.
*/
+use crate::Identifier;
use crate::utils::topic_size::MaxTopicSize;
use crate::{IggyMessage, utils::byte_size::IggyByteSize};
use std::sync::Arc;
diff --git a/core/common/src/types/identifier/mod.rs b/core/common/src/types/identifier/mod.rs
index 7a058a4c..1f599648 100644
--- a/core/common/src/types/identifier/mod.rs
+++ b/core/common/src/types/identifier/mod.rs
@@ -157,9 +157,11 @@ impl Identifier {
/// Creates a new identifier from the given numeric value.
pub fn numeric(value: u32) -> Result<Self, IggyError> {
+ /*
if value == 0 {
return Err(IggyError::InvalidIdentifier);
}
+ */
Ok(Self {
kind: IdKind::Numeric,
diff --git a/core/server/src/binary/handlers/streams/create_stream_handler.rs b/core/server/src/binary/handlers/streams/create_stream_handler.rs
index 6f0cf510..4461745e 100644
--- a/core/server/src/binary/handlers/streams/create_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/create_stream_handler.rs
@@ -26,11 +26,13 @@ use crate::shard_info;
use crate::state::command::EntryCommand;
use crate::state::models::CreateStreamWithId;
use crate::streaming::session::Session;
+use crate::streaming::stats::stats::StreamStats;
use anyhow::Result;
use error_set::ErrContext;
use iggy_common::create_stream::CreateStream;
use iggy_common::{Identifier, IggyError};
use std::rc::Rc;
+use std::sync::Arc;
use tracing::{debug, instrument};
impl ServerCommandHandler for CreateStream {
@@ -49,8 +51,10 @@ impl ServerCommandHandler for CreateStream {
debug!("session: {session}, command: {self}");
let stream_id = self.stream_id;
let name = self.name.clone();
+ let stats = Arc::new(StreamStats::new());
+
let new_stream_id = shard
- .create_stream2(session, stream_id, self.name.clone())
+ .create_stream2(session, stream_id, self.name.clone(), stats.clone())
.await?;
shard_info!(
shard.id,
@@ -61,6 +65,7 @@ impl ServerCommandHandler for CreateStream {
let event = ShardEvent::CreatedStream2 {
id: new_stream_id,
name: self.name.clone(),
+ stats,
};
let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
diff --git a/core/server/src/binary/handlers/topics/create_topic_handler.rs b/core/server/src/binary/handlers/topics/create_topic_handler.rs
index 801703a4..db7d409e 100644
--- a/core/server/src/binary/handlers/topics/create_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/create_topic_handler.rs
@@ -26,7 +26,9 @@ use crate::shard::{IggyShard, ShardInfo};
use crate::shard_info;
use crate::state::command::EntryCommand;
use crate::state::models::CreateTopicWithId;
+use crate::state::system::TopicState;
use crate::streaming::session::Session;
+use crate::streaming::stats::stats::TopicStats;
use crate::streaming::topics::topic::Topic;
use anyhow::Result;
use error_set::ErrContext;
@@ -34,6 +36,7 @@ use iggy_common::IggyError;
use iggy_common::create_topic::CreateTopic;
use iggy_common::locking::IggyRwLockFn;
use std::rc::Rc;
+use std::sync::Arc;
use tracing::{debug, instrument};
impl ServerCommandHandler for CreateTopic {
@@ -52,6 +55,36 @@ impl ServerCommandHandler for CreateTopic {
debug!("session: {session}, command: {self}");
let stream_id = self.stream_id.clone();
let maybe_topic_id = self.topic_id;
+ let parent = shard
+ .streams2
+ .with_stats_by_id(&stream_id, |stats| stats.clone());
+ let stats = Arc::new(TopicStats::new(parent));
+ let new_topic_id = shard
+ .create_topic2(
+ session,
+ &stream_id,
+ maybe_topic_id,
+ self.name.clone(),
+ self.partitions_count,
+ self.message_expiry,
+ self.compression_algorithm,
+ self.max_topic_size,
+ self.replication_factor,
+ stats.clone(),
+ )
+ .await?;
+ let event = ShardEvent::CreatedTopic2 {
+ stream_id: self.stream_id.clone(),
+ id: new_topic_id,
+ name: self.name.clone(),
+ partitions_count: self.partitions_count,
+ message_expiry: self.message_expiry,
+ compression_algorithm: self.compression_algorithm,
+ max_topic_size: self.max_topic_size,
+ replication_factor: self.replication_factor,
+ stats,
+ };
+ let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
let (topic_id, partition_ids) = shard
.create_topic(
session,
diff --git a/core/server/src/http/consumer_offsets.rs b/core/server/src/http/consumer_offsets.rs
index f6326ced..72d8747b 100644
--- a/core/server/src/http/consumer_offsets.rs
+++ b/core/server/src/http/consumer_offsets.rs
@@ -63,7 +63,8 @@ async fn get_consumer_offset(
query.validate()?;
let consumer = Consumer::new(query.0.consumer.id);
let session = SendWrapper::new(Session::stateless(identity.user_id, identity.ip_address));
- let Ok(offset) = state.shard
+ let Ok(offset) = state
+ .shard
.get_consumer_offset(
&session,
&consumer,
@@ -118,7 +119,8 @@ async fn delete_consumer_offset(
) -> Result<StatusCode, CustomError> {
let consumer = Consumer::new(consumer_id.try_into()?);
let session = SendWrapper::new(Session::stateless(identity.user_id, identity.ip_address));
- state.shard
+ state
+ .shard
.delete_consumer_offset(
&session,
consumer,
diff --git a/core/server/src/http/mapper.rs b/core/server/src/http/mapper.rs
index 5a03d0eb..4e63282f 100644
--- a/core/server/src/http/mapper.rs
+++ b/core/server/src/http/mapper.rs
@@ -33,7 +33,6 @@ use iggy_common::{ConsumerGroupDetails, ConsumerGroupMember};
use iggy_common::{IdentityInfo, TokenInfo};
use iggy_common::{UserInfo, UserInfoDetails};
-
pub fn map_stream(stream: &Stream) -> StreamDetails {
let topics = map_topics(&stream.get_topics());
let mut stream_details = StreamDetails {
@@ -193,9 +192,7 @@ pub fn map_clients(clients: &[Client]) -> Vec<iggy_common::ClientInfo> {
all_clients
}
-pub fn map_consumer_groups(
- consumer_groups: &[ConsumerGroup],
-) -> Vec<iggy_common::ConsumerGroup> {
+pub fn map_consumer_groups(consumer_groups: &[ConsumerGroup]) -> Vec<iggy_common::ConsumerGroup> {
let mut groups = Vec::new();
for consumer_group in consumer_groups {
let consumer_group = iggy_common::ConsumerGroup {
diff --git a/core/server/src/io/storage.rs b/core/server/src/io/storage.rs
index 079ec5b4..6ec828c5 100644
--- a/core/server/src/io/storage.rs
+++ b/core/server/src/io/storage.rs
@@ -3,7 +3,7 @@ use std::future::Future;
use compio::{
buf::{IoBuf, IoBufMut},
driver::op,
- io::{util::Splittable, AsyncReadAtExt, AsyncWriteAt, AsyncWriteAtExt, BufWriter},
+ io::{AsyncReadAtExt, AsyncWriteAt, AsyncWriteAtExt, BufWriter, util::Splittable},
};
use enum_dispatch::enum_dispatch;
@@ -78,7 +78,6 @@ impl BlockStorage {
let buf = Vec::new();
(&*file).write_all_at(buf, 0).await.unwrap();
}
-
}
impl BlockStorage {
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 74c93dcb..dd0a8376 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -597,10 +597,10 @@ impl IggyShard {
let partition = topic.get_partition(partition_id)?;
let mut partition = partition.write().await;
partition.open().await.with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to open partition with ID: {partition_id} in topic with ID: {topic_id} for stream with ID: {stream_id}"
- )
- })?;
+ format!(
+ "{COMPONENT} (error: {error}) - failed to open partition with ID: {partition_id} in topic with ID: {topic_id} for stream with ID: {stream_id}"
+ )
+ })?;
}
}
self.insert_shard_table_records(records);
@@ -614,11 +614,11 @@ impl IggyShard {
let mut stream = self.get_stream_mut(stream_id)?;
let topic = stream.get_topic_mut(topic_id)?;
topic.add_persisted_partitions(*partitions_count)
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to create partitions for topic with ID: {topic_id} in stream with ID: {stream_id}"
- )
- })?;
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to create partitions for topic with ID: {topic_id} in stream with ID: {stream_id}"
+ )
+ })?;
topic.reassign_consumer_groups();
self.metrics.increment_partitions(*partitions_count);
self.metrics.increment_segments(*partitions_count);
@@ -632,10 +632,10 @@ impl IggyShard {
let mut stream = self.get_stream_mut(stream_id)?;
let topic = stream.get_topic_mut(topic_id)?;
let partitions = topic
- . delete_persisted_partitions_by_ids(partition_ids)
- .with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to delete persisted partitions for topic: {topic}")
- })?;
+ . delete_persisted_partitions_by_ids(partition_ids)
+ .with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to delete persisted partitions for topic: {topic}")
+ })?;
drop(stream);
let mut segments_count = 0;
@@ -679,12 +679,12 @@ impl IggyShard {
let partition = topic.get_partition(ns.partition_id)?;
let mut partition = partition.write().await;
partition.delete().await.with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to delete partition with ID: {} in topic with ID: {}",
- ns.partition_id,
- topic_id
- )
- })?;
+ format!(
+ "{COMPONENT} (error: {error}) - failed to delete partition with ID: {} in topic with ID: {}",
+ ns.partition_id,
+ topic_id
+ )
+ })?;
}
}
Ok(())
@@ -692,10 +692,10 @@ impl IggyShard {
ShardEvent::DeletedStream { stream_id } => {
let shard_id = self.id;
self.delete_stream_bypass_auth(stream_id).with_error_context(|err| {
- format!(
- "{COMPONENT} (error: {err}) - failed to delete, when handling event on shard: {shard_id} stream with ID: {stream_id}",
- )
- })?;
+ format!(
+ "{COMPONENT} (error: {err}) - failed to delete, when handling event on shard: {shard_id} stream with ID: {stream_id}",
+ )
+ })?;
Ok(())
}
ShardEvent::UpdatedStream { stream_id, name } => {
@@ -739,12 +739,12 @@ impl IggyShard {
topic_id,
} => {
self.delete_topic_bypass_auth(stream_id, topic_id)
- .await
- .with_error_context(|err| {
- format!(
- "{COMPONENT} (error: {err}) - failed to delete topic with ID: {topic_id} in stream with ID: {stream_id}"
- )
- })?;
+ .await
+ .with_error_context(|err| {
+ format!(
+ "{COMPONENT} (error: {err}) - failed to delete topic with ID: {topic_id} in stream with ID: {stream_id}"
+ )
+ })?;
Ok(())
}
ShardEvent::CreatedConsumerGroup {
@@ -826,11 +826,37 @@ impl IggyShard {
self.tcp_bound_address.set(Some(*address));
Ok(())
}
- ShardEvent::CreatedStream2 { id, name } => {
- let stream_id = self.create_stream2_bypass_auth(name.to_owned()).await?;
+ ShardEvent::CreatedStream2 { id, name, stats } => {
+ let stream_id = self
+ .create_stream2_bypass_auth(name.to_owned(), stats.clone())
+ .await?;
assert_eq!(stream_id, *id);
Ok(())
}
+ ShardEvent::CreatedTopic2 {
+ id,
+ stream_id,
+ name,
+ partitions_count,
+ message_expiry,
+ compression_algorithm,
+ max_topic_size,
+ replication_factor,
+ stats,
+ } => {
+ let topic_id = self.create_topic2_bypass_auth(
+ stream_id,
+ name.to_owned(),
+ *partitions_count,
+ *replication_factor,
+ *message_expiry,
+ *compression_algorithm,
+ *max_topic_size,
+ stats.clone(),
+ )?;
+ assert_eq!(topic_id, *id);
+ Ok(())
+ }
}
}
@@ -887,6 +913,7 @@ impl IggyShard {
ShardEvent::CreatedShardTableRecords { .. }
| ShardEvent::DeletedShardTableRecords { .. }
| ShardEvent::CreatedStream2 { .. }
+ | ShardEvent::CreatedTopic2 { .. }
) {
let (sender, receiver) = async_channel::bounded(1);
conn.send(ShardFrame::new(event.clone().into(), Some(sender.clone())));
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index ccfae4d8..d05aa343 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -21,6 +21,8 @@ use crate::shard::IggyShard;
use crate::shard::namespace::IggyNamespace;
use crate::streaming::partitions::partition;
use crate::streaming::session::Session;
+use crate::streaming::stats::stats::StreamStats;
+use crate::streaming::streams::storage2::create_stream_file_hierarchy;
use crate::streaming::streams::stream::Stream;
use crate::streaming::streams::stream2;
use error_set::ErrContext;
@@ -28,6 +30,7 @@ use futures::future::try_join_all;
use iggy_common::locking::IggyRwLockFn;
use iggy_common::{IdKind, Identifier, IggyError};
use std::cell::{Ref, RefCell, RefMut};
+use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use tokio::fs;
use tracing::{error, info, warn};
@@ -208,23 +211,30 @@ impl IggyShard {
session: &Session,
stream_id: Option<u32>,
name: String,
+ stats: Arc<StreamStats>,
) -> Result<usize, IggyError> {
self.ensure_authenticated(session)?;
self.permissioner
.borrow()
.create_stream(session.get_user_id())?;
- let stream_id = self.create_stream2_base(name).await?;
- self.streams2
- .create_stream_file_hierarchy(stream_id, &self.config.system)
- .await?;
+ let stream_id = self.create_stream2_base(name, stats).await?;
+ create_stream_file_hierarchy(stream_id, &self.config.system).await?;
Ok(stream_id)
}
- pub async fn create_stream2_bypass_auth(&self, name: String) -> Result<usize, IggyError> {
- self.create_stream2_base(name).await
+ pub async fn create_stream2_bypass_auth(
+ &self,
+ name: String,
+ stats: Arc<StreamStats>,
+ ) -> Result<usize, IggyError> {
+ self.create_stream2_base(name, stats).await
}
- async fn create_stream2_base(&self, name: String) -> Result<usize, IggyError> {
+ async fn create_stream2_base(
+ &self,
+ name: String,
+ stats: Arc<StreamStats>,
+ ) -> Result<usize, IggyError> {
let exists = self.streams2.with(|streams| streams.exists(&name));
if exists {
return Err(IggyError::StreamNameAlreadyExists(name));
@@ -233,6 +243,8 @@ impl IggyShard {
let stream_id = self
.streams2
.with_mut(|streams| stream.insert_into(streams));
+ self.streams2
+ .with_stats_mut(|container| container.insert(stats));
Ok(stream_id)
}
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index ec5b5c0b..a66ad1a9 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -16,13 +16,19 @@
* under the License.
*/
+use std::sync::Arc;
+
use super::COMPONENT;
use crate::shard::namespace::IggyNamespace;
use crate::shard::transmission::event::ShardEvent;
use crate::shard::{IggyShard, ShardInfo};
+use crate::streaming::partitions::partition2;
use crate::streaming::session::Session;
+use crate::streaming::stats::stats::TopicStats;
use crate::streaming::streams::stream::Stream;
use crate::streaming::topics::topic::Topic;
+use crate::streaming::topics::topic2;
+use clap::Id;
use error_set::ErrContext;
use iggy_common::locking::IggyRwLockFn;
use iggy_common::{CompressionAlgorithm, Identifier, IggyError, IggyExpiry, MaxTopicSize};
@@ -131,6 +137,143 @@ impl IggyShard {
Ok(())
}
+ #[allow(clippy::too_many_arguments)]
+ pub async fn create_topic2(
+ &self,
+ session: &Session,
+ stream_id: &Identifier,
+ topic_id: Option<u32>,
+ name: String,
+ partitions_count: u32,
+ message_expiry: IggyExpiry,
+ compression: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ replication_factor: Option<u8>,
+ stats: Arc<TopicStats>,
+ ) -> Result<usize, IggyError> {
+ self.ensure_authenticated(session)?;
+ {
+ let stream_id = self
+ .streams2
+ .with_stream_by_id(stream_id, |stream| stream.id());
+ self.permissioner
+ .borrow()
+ .create_topic(session.get_user_id(), stream_id as u32)
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - permission denied to create topic with name: {name} in stream with ID: {stream_id} for user with ID: {}",
+ session.get_user_id(),
+ )
+ })?;
+ }
+ let topic_id = self.create_topic2_base(
+ stream_id,
+ name,
+ replication_factor.unwrap_or(1),
+ message_expiry,
+ compression,
+ max_topic_size,
+ stats,
+ )?;
+ // TODO: Create dir hierarchy for the topic.
+ let partition_ids = self.create_partitions2(
+ stream_id,
+ &Identifier::numeric(topic_id as u32).unwrap(),
+ partitions_count,
+ )?;
+ // TODO: Create partition files and open descriptors for partitions that belong to _this_ shard.
+ Ok(topic_id)
+ }
+
+ fn create_partitions2(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partitions_count: u32,
+ ) -> Result<Vec<usize>, IggyError> {
+ let partition_ids = self
+ .streams2
+ .with_topic_by_id_mut(stream_id, topic_id, |topic| {
+ // Create partitions...
+ let partition_ids = std::iter::repeat_with(partition2::Partition::new)
+ .map(|partition| {
+ topic
+ .partitions_mut()
+ .with_mut(|partitions| partition.insert_into(partitions))
+ })
+ .take(partitions_count as usize)
+ .collect::<Vec<_>>();
+ partition_ids
+ });
+ Ok(partition_ids)
+ }
+
+ pub fn create_topic2_bypass_auth(
+ &self,
+ stream_id: &Identifier,
+ name: String,
+ partitions_count: u32,
+ replication_factor: Option<u8>,
+ message_expiry: IggyExpiry,
+ compression: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ stats: Arc<TopicStats>,
+ ) -> Result<usize, IggyError> {
+ let topic_id = self.create_topic2_base(
+ stream_id,
+ name,
+ replication_factor.unwrap_or(1),
+ message_expiry,
+ compression,
+ max_topic_size,
+ stats,
+ )?;
+ let partition_ids = self.create_partitions2(
+ stream_id,
+ &Identifier::numeric(topic_id as u32).unwrap(),
+ partitions_count,
+ )?;
+ // TODO: Create partition files and open descriptors for partitions that belong to _this_ shard.
+ Ok(topic_id)
+ }
+
+ fn create_topic2_base(
+ &self,
+ stream_id: &Identifier,
+ name: String,
+ replication_factor: u8,
+ message_expiry: IggyExpiry,
+ compression: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ stats: Arc<TopicStats>,
+ ) -> Result<usize, IggyError> {
+ let topic_id = self.streams2.with_stream_by_id(stream_id, |stream| {
+ let exists = stream.topics().exists(&name);
+ if exists {
+ // TODO: Fixme, replace the second argument with identifier, rather than numeric ID.
+ return Err(IggyError::TopicNameAlreadyExists(name.to_owned(), 0));
+ }
+ let topic = topic2::Topic::new(
+ name,
+ replication_factor,
+ message_expiry,
+ compression,
+ max_topic_size,
+ );
+ let topic_id = stream.topics().with_mut(|topics| {
+ let topic_id = topic.insert_into(topics);
+ topic_id
+ });
+
+ stream.topics().with_stats_mut(|container| {
+ container.insert(stats);
+ });
+ Ok(topic_id)
+ })?;
+
+ Ok(topic_id)
+ }
+
#[allow(clippy::too_many_arguments)]
pub async fn create_topic(
&self,
diff --git a/core/server/src/shard/transmission/event.rs b/core/server/src/shard/transmission/event.rs
index dfc33ce8..ee73ef20 100644
--- a/core/server/src/shard/transmission/event.rs
+++ b/core/server/src/shard/transmission/event.rs
@@ -1,4 +1,4 @@
-use std::net::SocketAddr;
+use std::{net::SocketAddr, sync::Arc};
use iggy_common::{
CompressionAlgorithm, Identifier, IggyExpiry, MaxTopicSize, Permissions, UserStatus,
@@ -10,6 +10,7 @@ use crate::{
clients::client_manager::Transport,
personal_access_tokens::personal_access_token::PersonalAccessToken,
polling_consumer::PollingConsumer,
+ stats::stats::{StreamStats, TopicStats},
},
};
@@ -26,6 +27,7 @@ pub enum ShardEvent {
CreatedStream2 {
id: usize,
name: String,
+ stats: Arc<StreamStats>,
},
CreatedStream {
stream_id: Option<u32>,
@@ -61,6 +63,17 @@ pub enum ShardEvent {
max_topic_size: MaxTopicSize,
replication_factor: Option<u8>,
},
+ CreatedTopic2 {
+ id: usize,
+ stream_id: Identifier,
+ name: String,
+ partitions_count: u32,
+ message_expiry: IggyExpiry,
+ compression_algorithm: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ replication_factor: Option<u8>,
+ stats: Arc<TopicStats>,
+ },
CreatedConsumerGroup {
stream_id: Identifier,
topic_id: Identifier,
diff --git a/core/server/src/slab/consumer_groups.rs b/core/server/src/slab/consumer_groups.rs
index f096054f..e269f212 100644
--- a/core/server/src/slab/consumer_groups.rs
+++ b/core/server/src/slab/consumer_groups.rs
@@ -2,6 +2,7 @@ use crate::{slab::IndexedSlab, streaming::topics::consumer_group2};
const CAPACITY: usize = 1024;
+#[derive(Debug)]
pub struct ConsumerGroups {
container: IndexedSlab<consumer_group2::ConsumerGroup>,
}
diff --git a/core/server/src/slab/mod.rs b/core/server/src/slab/mod.rs
index a90ccdaa..fc489e81 100644
--- a/core/server/src/slab/mod.rs
+++ b/core/server/src/slab/mod.rs
@@ -12,34 +12,24 @@ pub trait Keyed {
fn key(&self) -> &Self::Key;
}
+#[derive(Debug)]
pub struct IndexedSlab<T: Keyed> {
slab: Slab<T>,
index: AHashMap<T::Key, usize>,
}
-impl<T> Default for IndexedSlab<T>
-where
- T: Keyed + Default,
-{
- fn default() -> Self {
- Self {
- slab: Slab::new(),
- index: AHashMap::new(),
- }
- }
-}
-
impl<T: Keyed + Default> IndexedSlab<T> {
pub fn with_capacity(capacity: usize) -> Self {
- let mut slab = Slab::with_capacity(capacity);
- // Advance the slab once, by inserting a default value.
- slab.insert(T::default());
Self {
- slab,
+ slab: Slab::with_capacity(capacity),
index: AHashMap::with_capacity(capacity),
}
}
+ pub fn len(&self) -> usize {
+ self.slab.len()
+ }
+
pub fn exists(&self, key: &T::Key) -> bool {
self.index.contains_key(key)
}
diff --git a/core/server/src/slab/partitions.rs b/core/server/src/slab/partitions.rs
index ec1ab663..5b0cb815 100644
--- a/core/server/src/slab/partitions.rs
+++ b/core/server/src/slab/partitions.rs
@@ -1,17 +1,18 @@
-use std::cell::RefCell;
+use std::{cell::RefCell, sync::Arc};
use iggy_common::Partition;
use slab::Slab;
-use crate::streaming::{partitions::partition2, segments};
+use crate::streaming::{partitions::partition2, segments, stats::stats::PartitionStats};
// TODO: This could be upper limit of partitions per topic, use that value to validate instead of whathever this thing is in `common` crate.
const CAPACITY: usize = 16384;
+
+#[derive(Debug)]
pub struct Partitions {
container: Slab<partition2::Partition>,
- storage: (),
- stats: (),
+ stats: Slab<Arc<PartitionStats>>,
segments: Slab<Vec<segments::Segment2>>,
}
@@ -19,8 +20,7 @@ impl Default for Partitions {
fn default() -> Self {
Self {
container: Slab::with_capacity(CAPACITY),
- storage: (),
- stats: (),
+ stats: Slab::with_capacity(CAPACITY),
segments: Slab::with_capacity(CAPACITY),
}
}
@@ -37,9 +37,9 @@ impl Partitions {
f(&container);
}
- pub fn with_mut(&mut self, f: impl FnOnce(&mut Slab<partition2::Partition>)) {
+ pub fn with_mut<T>(&mut self, f: impl FnOnce(&mut Slab<partition2::Partition>) -> T) -> T {
let mut container = &mut self.container;
- f(&mut container);
+ f(&mut container)
}
pub fn with_partition_id(&self, partition_id: usize, f: impl FnOnce(&partition2::Partition)) {
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index 6635f2d1..1c320d16 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -1,10 +1,11 @@
use compio::fs::create_dir_all;
-use iggy_common::{Identifier, IggyError};
+use iggy_common::{CompressionAlgorithm, Identifier, IggyError, MaxTopicSize};
use slab::Slab;
use std::{
cell::{RefCell, UnsafeCell},
ops::Index,
path::Path,
+ sync::Arc,
};
use tracing::info;
@@ -13,46 +14,46 @@ use crate::{
io::storage::Storage,
shard_info,
slab::{IndexedSlab, partitions::Partitions, topics::Topics},
- streaming::{partitions::partition2, streams::stream2, topics::topic2},
+ streaming::{
+ partitions::partition2, stats::stats::StreamStats, streams::stream2, topics::topic2,
+ },
};
const CAPACITY: usize = 1024;
pub struct Streams {
container: RefCell<IndexedSlab<stream2::Stream>>,
- stats: (),
+ stats: RefCell<Slab<Arc<StreamStats>>>,
}
impl Streams {
pub fn init() -> Self {
Self {
container: RefCell::new(IndexedSlab::with_capacity(CAPACITY)),
- stats: (),
+ stats: RefCell::new(Slab::with_capacity(CAPACITY)),
}
}
- pub async fn create_stream_file_hierarchy(
- &self,
- id: usize,
- config: &SystemConfig,
- ) -> Result<(), IggyError> {
- let id = id as u32;
- let path = config.get_stream_path(id);
- let topics_path = config.get_topics_path(id);
-
- if !Path::new(&path).exists() && create_dir_all(&path).await.is_err() {
- return Err(IggyError::CannotCreateStreamDirectory(id, path.clone()));
- }
+ pub fn with_stats<T>(&self, f: impl FnOnce(&Slab<Arc<StreamStats>>) -> T) -> T {
+ let stats = self.stats.borrow();
+ f(&stats)
+ }
- if !Path::new(&topics_path).exists() && create_dir_all(&topics_path).await.is_err() {
- return Err(IggyError::CannotCreateTopicsDirectory(
- id,
- topics_path.clone(),
- ));
- }
+ pub fn with_stats_by_id<T>(
+ &self,
+ id: &Identifier,
+ f: impl FnOnce(&Arc<StreamStats>) -> T,
+ ) -> T {
+ let stream_id = self.with_stream_by_id(id, |stream| stream.id());
+ self.with_stats(|stats| {
+ let stats = &stats[stream_id];
+ f(stats)
+ })
+ }
- info!("Saved stream with ID: {id}.");
- Ok(())
+ pub fn with_stats_mut<T>(&self, f: impl FnOnce(&mut Slab<Arc<StreamStats>>) -> T) -> T {
+ let mut stats = self.stats.borrow_mut();
+ f(&mut stats)
}
pub async fn with_async(&self, f: impl AsyncFnOnce(&IndexedSlab<stream2::Stream>)) {
@@ -90,7 +91,7 @@ impl Streams {
})
}
- pub fn with_stream_by_id_mut(&self, id: &Identifier, mut f: impl FnMut(&mut stream2::Stream)) {
+ pub fn with_stream_by_id_mut(&self, id: &Identifier, mut f: impl FnOnce(&mut stream2::Stream)) {
self.with_mut(|streams| {
let stream = match id.kind {
iggy_common::IdKind::Numeric => {
@@ -106,7 +107,7 @@ impl Streams {
});
}
- pub fn with_topics(&self, stream_id: &Identifier, f: impl FnOnce(&Topics)) {
+ pub fn with_topics<T>(&self, stream_id: &Identifier, f: impl FnOnce(&Topics) -> T) -> T {
self.with(|streams| {
let stream = match stream_id.kind {
iggy_common::IdKind::Numeric => {
@@ -119,7 +120,7 @@ impl Streams {
}
};
f(stream.topics())
- });
+ })
}
pub fn with_topic_by_id(
@@ -133,6 +134,15 @@ impl Streams {
});
}
+ pub fn with_topic_by_id_mut<T>(
+ &self,
+ id: &Identifier,
+ topic_id: &Identifier,
+ f: impl FnOnce(&mut topic2::Topic) -> T,
+ ) -> T {
+ self.with_topics(id, |topics| topics.with_topic_by_id_mut(topic_id, f))
+ }
+
pub fn with_partitions(
&self,
stream_id: &Identifier,
diff --git a/core/server/src/slab/topics.rs b/core/server/src/slab/topics.rs
index 9fd84f34..8dc8666c 100644
--- a/core/server/src/slab/topics.rs
+++ b/core/server/src/slab/topics.rs
@@ -1,48 +1,58 @@
use iggy_common::Identifier;
-use std::cell::RefCell;
+use slab::Slab;
+use std::{cell::RefCell, sync::Arc};
use crate::{
slab::{IndexedSlab, partitions::Partitions},
- streaming::{partitions::partition2, topics::topic2},
+ streaming::{partitions::partition2, stats::stats::TopicStats, topics::topic2},
};
const CAPACITY: usize = 1024;
pub struct Topics {
container: RefCell<IndexedSlab<topic2::Topic>>,
- stats: (),
-}
-
-impl Default for Topics {
- fn default() -> Self {
- Self {
- container: RefCell::new(IndexedSlab::with_capacity(CAPACITY)),
- stats: (),
- }
- }
+ stats: RefCell<Slab<Arc<TopicStats>>>,
}
impl Topics {
pub fn init() -> Self {
Self {
container: RefCell::new(IndexedSlab::with_capacity(CAPACITY)),
- stats: (),
+ stats: RefCell::new(Slab::with_capacity(CAPACITY)),
}
}
+ pub fn len(&self) -> usize {
+ self.container.borrow().len()
+ }
+
+ pub fn exists(&self, key: &String) -> bool {
+ self.container.borrow().exists(key)
+ }
+
pub async fn with_async(&self, f: impl AsyncFnOnce(&IndexedSlab<topic2::Topic>)) {
let container = self.container.borrow();
f(&container).await;
}
+ pub fn with_stats<T>(&self, f: impl FnOnce(&Slab<Arc<TopicStats>>) -> T) -> T {
+ let stats = self.stats.borrow();
+ f(&stats)
+ }
+
+ pub fn with_stats_mut<T>(&self, f: impl FnOnce(&mut Slab<Arc<TopicStats>>) -> T) -> T {
+ let mut stats = self.stats.borrow_mut();
+ f(&mut stats)
+ }
+
pub fn with(&self, f: impl FnOnce(&IndexedSlab<topic2::Topic>)) {
let container = self.container.borrow();
f(&container);
}
- pub fn with_mut(&self, f: impl FnOnce(&mut IndexedSlab<topic2::Topic>)) {
+ pub fn with_mut<T>(&self, f: impl FnOnce(&mut IndexedSlab<topic2::Topic>) -> T) -> T {
let mut container = self.container.borrow_mut();
- f(&mut container);
+ f(&mut container)
}
pub fn with_topic_by_id(&self, id: &Identifier, f: impl FnOnce(&topic2::Topic)) {
@@ -61,7 +71,11 @@ impl Topics {
});
}
- pub fn with_topic_by_id_mut(&self, id: &Identifier, mut f: impl FnMut(&mut topic2::Topic)) {
+ pub fn with_topic_by_id_mut<T>(
+ &self,
+ id: &Identifier,
+ f: impl FnOnce(&mut topic2::Topic) -> T,
+ ) -> T {
self.with_mut(|topics| {
let topic = match id.kind {
iggy_common::IdKind::Numeric => {
@@ -74,7 +88,7 @@ impl Topics {
}
};
f(topic)
- });
+ })
}
pub fn with_partitions(&self, topic_id: &Identifier, f: impl FnOnce(&Partitions)) {
diff --git a/core/server/src/streaming/mod.rs b/core/server/src/streaming/mod.rs
index f9d40af8..9eef99e6 100644
--- a/core/server/src/streaming/mod.rs
+++ b/core/server/src/streaming/mod.rs
@@ -26,6 +26,7 @@ pub mod personal_access_tokens;
pub mod polling_consumer;
pub mod segments;
pub mod session;
+pub mod stats;
pub mod storage;
pub mod streams;
pub mod topics;
diff --git a/core/server/src/streaming/partitions/partition2.rs b/core/server/src/streaming/partitions/partition2.rs
index 0ca10af0..57c430dd 100644
--- a/core/server/src/streaming/partitions/partition2.rs
+++ b/core/server/src/streaming/partitions/partition2.rs
@@ -1,4 +1,21 @@
-#[derive(Default)]
+use slab::Slab;
+
+use crate::slab::IndexedSlab;
+
+#[derive(Default, Debug)]
pub struct Partition {
id: usize,
}
+
+impl Partition {
+ pub fn new() -> Self {
+ Self { id: 0 }
+ }
+
+ pub fn insert_into(self, container: &mut Slab<Self>) -> usize {
+ let idx = container.insert(self);
+ let partition = &mut container[idx];
+ partition.id = idx;
+ idx
+ }
+}
diff --git a/core/server/src/streaming/segments/segment2.rs b/core/server/src/streaming/segments/segment2.rs
index ae236040..9d2f6e70 100644
--- a/core/server/src/streaming/segments/segment2.rs
+++ b/core/server/src/streaming/segments/segment2.rs
@@ -1,3 +1,4 @@
+#[derive(Debug)]
pub struct Segment2 {
pub id: usize,
}
diff --git a/core/server/src/streaming/stats/mod.rs b/core/server/src/streaming/stats/mod.rs
new file mode 100644
index 00000000..9d34677f
--- /dev/null
+++ b/core/server/src/streaming/stats/mod.rs
@@ -0,0 +1 @@
+pub mod stats;
diff --git a/core/server/src/streaming/stats/stats.rs b/core/server/src/streaming/stats/stats.rs
new file mode 100644
index 00000000..ef69b20d
--- /dev/null
+++ b/core/server/src/streaming/stats/stats.rs
@@ -0,0 +1,55 @@
+use std::sync::{
+ Arc,
+ atomic::{AtomicU32, AtomicU64},
+};
+
+#[derive(Default, Debug)]
+pub struct StreamStats {
+ size_bytes: AtomicU64,
+ messages_count: AtomicU64,
+ segments_count: AtomicU32,
+}
+
+impl StreamStats {
+ pub fn new() -> Self {
+ Self {
+ size_bytes: AtomicU64::new(0),
+ messages_count: AtomicU64::new(0),
+ segments_count: AtomicU32::new(0),
+ }
+ }
+}
+
+#[derive(Default, Debug)]
+pub struct TopicStats {
+ parent: Arc<StreamStats>,
+ size_bytes: AtomicU64,
+ messages_count: AtomicU64,
+}
+
+impl TopicStats {
+ pub fn new(parent: Arc<StreamStats>) -> Self {
+ Self {
+ parent,
+ size_bytes: AtomicU64::new(0),
+ messages_count: AtomicU64::new(0),
+ }
+ }
+}
+
+#[derive(Default, Debug)]
+pub struct PartitionStats {
+ parent: Arc<TopicStats>,
+ messages_count: AtomicU64,
+ size_bytes: AtomicU64,
+}
+
+impl PartitionStats {
+ pub fn new(parent: Arc<TopicStats>) -> Self {
+ Self {
+ parent,
+ messages_count: AtomicU64::new(0),
+ size_bytes: AtomicU64::new(0),
+ }
+ }
+}
diff --git a/core/server/src/streaming/streams/mod.rs b/core/server/src/streaming/streams/mod.rs
index 75468b62..8f78c7ff 100644
--- a/core/server/src/streaming/streams/mod.rs
+++ b/core/server/src/streaming/streams/mod.rs
@@ -21,6 +21,7 @@ pub mod partitions;
pub mod persistence;
pub mod segments;
pub mod storage;
+pub mod storage2;
pub mod stream;
pub mod stream2;
pub mod topics;
diff --git a/core/server/src/streaming/streams/storage2.rs b/core/server/src/streaming/streams/storage2.rs
new file mode 100644
index 00000000..bdbdacb4
--- /dev/null
+++ b/core/server/src/streaming/streams/storage2.rs
@@ -0,0 +1,28 @@
+use compio::fs::create_dir_all;
+use iggy_common::IggyError;
+use std::path::Path;
+
+use crate::configs::system::SystemConfig;
+
+pub async fn create_stream_file_hierarchy(
+ id: usize,
+ config: &SystemConfig,
+) -> Result<(), IggyError> {
+ let id = id as u32;
+ let path = config.get_stream_path(id);
+ let topics_path = config.get_topics_path(id);
+
+ if !Path::new(&path).exists() && create_dir_all(&path).await.is_err() {
+ return Err(IggyError::CannotCreateStreamDirectory(id, path.clone()));
+ }
+
+ if !Path::new(&topics_path).exists() && create_dir_all(&topics_path).await.is_err() {
+ return Err(IggyError::CannotCreateTopicsDirectory(
+ id,
+ topics_path.clone(),
+ ));
+ }
+
+ tracing::info!("Saved stream with ID: {}.", id);
+ Ok(())
+}
diff --git a/core/server/src/streaming/streams/stream2.rs b/core/server/src/streaming/streams/stream2.rs
index 5fb17482..9d065bcf 100644
--- a/core/server/src/streaming/streams/stream2.rs
+++ b/core/server/src/streaming/streams/stream2.rs
@@ -3,7 +3,6 @@ use iggy_common::IggyTimestamp;
use crate::slab::{IndexedSlab, Keyed, topics::Topics};
-#[derive(Default)]
pub struct Stream {
id: usize,
name: String,
@@ -11,6 +10,17 @@ pub struct Stream {
topics: Topics,
}
+impl Default for Stream {
+ fn default() -> Self {
+ Self {
+ id: 0,
+ name: String::new(),
+ created_at: IggyTimestamp::now(),
+ topics: Topics::init(),
+ }
+ }
+}
+
impl Keyed for Stream {
type Key = String;
@@ -26,10 +36,18 @@ impl Stream {
id: 0,
name,
created_at: now,
- topics: Topics::default(),
+ topics: Topics::init(),
}
}
+ pub fn id(&self) -> usize {
+ self.id
+ }
+
+ pub fn topics_count(&self) -> usize {
+ self.topics.len()
+ }
+
pub fn insert_into(self, container: &mut IndexedSlab<Self>) -> usize {
let idx = container.insert(self);
let stream = &mut container[idx];
diff --git a/core/server/src/streaming/topics/consumer_group2.rs b/core/server/src/streaming/topics/consumer_group2.rs
index d39ba62c..6091afa4 100644
--- a/core/server/src/streaming/topics/consumer_group2.rs
+++ b/core/server/src/streaming/topics/consumer_group2.rs
@@ -1,6 +1,6 @@
use crate::slab::Keyed;
-#[derive(Default)]
+#[derive(Default, Debug)]
pub struct ConsumerGroup {
name: String,
}
diff --git a/core/server/src/streaming/topics/topic2.rs b/core/server/src/streaming/topics/topic2.rs
index 3aed3d5c..187df277 100644
--- a/core/server/src/streaming/topics/topic2.rs
+++ b/core/server/src/streaming/topics/topic2.rs
@@ -1,16 +1,64 @@
-use crate::slab::{Keyed, consumer_groups::ConsumerGroups, partitions::Partitions};
+use crate::slab::{IndexedSlab, Keyed, consumer_groups::ConsumerGroups, partitions::Partitions};
+use iggy_common::{CompressionAlgorithm, IggyExpiry, IggyTimestamp, MaxTopicSize};
-#[derive(Default)]
+#[derive(Default, Debug)]
pub struct Topic {
- name: String,
partitions: Partitions,
consumer_groups: ConsumerGroups,
+
+ id: usize,
+ // TODO: This property should be removed, we won't use it in our clustering impl.
+ replication_factor: u8,
+ name: String,
+ created_at: IggyTimestamp,
+ message_expiry: IggyExpiry,
+ compression_algorithm: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
}
impl Topic {
+ pub fn new(
+ name: String,
+ replication_factor: u8,
+ message_expiry: IggyExpiry,
+ compression: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ ) -> Self {
+ Self {
+ id: 0,
+ name,
+ created_at: IggyTimestamp::now(),
+ partitions: Partitions::default(),
+ consumer_groups: ConsumerGroups::default(),
+ replication_factor,
+ message_expiry,
+ compression_algorithm: compression,
+ max_topic_size,
+ }
+ }
+
+ pub fn with(&self, f: impl FnOnce(&Self)) {
+ f(self);
+ }
+
+ pub fn with_mut(&mut self, f: impl FnOnce(&mut Self)) {
+ f(self);
+ }
+
pub fn partitions(&self) -> &Partitions {
&self.partitions
}
+
+ pub fn partitions_mut(&mut self) -> &mut Partitions {
+ &mut self.partitions
+ }
+
+ pub fn insert_into(self, container: &mut IndexedSlab<Self>) -> usize {
+ let idx = container.insert(self);
+ let topic = &mut container[idx];
+ topic.id = idx;
+ idx
+ }
}
impl Keyed for Topic {