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:48 UTC
(iggy) 01/01: feat(io_uring): refactor statistics and implement topic creation
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 {