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/04 09:49:14 UTC
(iggy) branch move_topic_to_new_layout created (now 66bfc5ae)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch move_topic_to_new_layout
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 66bfc5ae feat(io_uring): move topic to new layout
This branch includes the following new commits:
new 66bfc5ae feat(io_uring): move topic to new layout
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): move topic to new layout
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch move_topic_to_new_layout
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 66bfc5ae31de09c9c79633d1371bcb1087fbe0d6
Author: numminex <g....@gmail.com>
AuthorDate: Mon Aug 4 11:49:03 2025 +0200
feat(io_uring): move topic to new layout
---
Cargo.lock | 7 +
core/common/src/error/iggy_error.rs | 1 +
core/server/Cargo.toml | 1 +
.../binary/handlers/topics/delete_topic_handler.rs | 10 +
.../binary/handlers/topics/update_topic_handler.rs | 22 ++
core/server/src/shard/mod.rs | 89 +++++---
core/server/src/shard/system/streams.rs | 2 +-
core/server/src/shard/system/topics.rs | 225 ++++++++++++++++++++-
core/server/src/shard/transmission/event.rs | 14 ++
core/server/src/slab/consumer_groups.rs | 8 +-
core/server/src/slab/mod.rs | 38 +++-
core/server/src/slab/partitions.rs | 37 ++--
core/server/src/slab/streams.rs | 81 ++++++--
core/server/src/slab/topics.rs | 135 ++++++++-----
core/server/src/streaming/streams/stream2.rs | 9 +
core/server/src/streaming/streams/topics.rs | 13 +-
.../server/src/streaming/topics/consumer_group2.rs | 28 +++
core/server/src/streaming/topics/topic2.rs | 40 ++++
18 files changed, 626 insertions(+), 134 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 3aa162b0..9ecc5a17 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -420,6 +420,12 @@ version = "1.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "69f7f8c3906b62b754cd5326047894316021dcfe5a194c8ea52bdd94934a3457"
+[[package]]
+name = "arcshift"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "95bcd258b3b56d33539bfd0c0bd423b74ee05abe628386a3a000a88c0147748f"
+
[[package]]
name = "arraydeque"
version = "0.5.1"
@@ -7443,6 +7449,7 @@ version = "0.5.0"
dependencies = [
"ahash 0.8.12",
"anyhow",
+ "arcshift",
"async-channel",
"async_zip",
"axum 0.8.4",
diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index 5759e864..559ae3f2 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -193,6 +193,7 @@ pub enum IggyError {
CannotDeleteStream(u32) = 1007,
#[error("Failed to delete stream directory with ID: {0}")]
CannotDeleteStreamDirectory(u32) = 1008,
+ //TODO: Fix the stream not found errors.
#[error("Stream with ID: {0} was not found.")]
StreamIdNotFound(u32) = 1009,
#[error("Stream with name: {0} was not found.")]
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index 0bb6da58..80e263c4 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -36,6 +36,7 @@ mimalloc = ["dep:mimalloc"]
[dependencies]
ahash = { workspace = true }
+arcshift = "0.4.0"
anyhow = { workspace = true }
async_zip = { workspace = true }
axum = { workspace = true }
diff --git a/core/server/src/binary/handlers/topics/delete_topic_handler.rs b/core/server/src/binary/handlers/topics/delete_topic_handler.rs
index 2f79935c..c7ce984c 100644
--- a/core/server/src/binary/handlers/topics/delete_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/delete_topic_handler.rs
@@ -48,6 +48,16 @@ impl ServerCommandHandler for DeleteTopic {
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
+ let topic2 = shard
+ .delete_topic2(session, &self.stream_id, &self.topic_id)
+ .await?;
+ let event = ShardEvent::DeletedTopic2 {
+ id: topic2.id(),
+ stream_id: self.stream_id.clone(),
+ topic_id: self.topic_id.clone(),
+ };
+ let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
+
let topic = shard
.delete_topic(session, &self.stream_id, &self.topic_id)
.await
diff --git a/core/server/src/binary/handlers/topics/update_topic_handler.rs b/core/server/src/binary/handlers/topics/update_topic_handler.rs
index 55a158a4..e9f2a3b7 100644
--- a/core/server/src/binary/handlers/topics/update_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/update_topic_handler.rs
@@ -44,7 +44,28 @@ impl ServerCommandHandler for UpdateTopic {
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
+ shard.update_topic2(
+ session,
+ &self.stream_id,
+ &self.topic_id,
+ self.name.clone(),
+ self.message_expiry,
+ self.compression_algorithm,
+ self.max_topic_size,
+ self.replication_factor,
+ );
+ let event = ShardEvent::UpdatedTopic2 {
+ stream_id: self.stream_id.clone(),
+ topic_id: self.topic_id.clone(),
+ name: self.name.clone(),
+ message_expiry: self.message_expiry,
+ compression_algorithm: self.compression_algorithm,
+ max_topic_size: self.max_topic_size,
+ replication_factor: self.replication_factor,
+ };
+ let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
+ /*
shard
.update_topic(
session,
@@ -96,6 +117,7 @@ impl ServerCommandHandler for UpdateTopic {
.with_error_context(|error| format!(
"{COMPONENT} (error: {error}) - failed to apply update topic with id: {topic_id}, stream_id: {stream_id}, session: {session}"
))?;
+ */
sender.send_empty_ok_response().await?;
Ok(())
}
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 19af5fc5..f05eff13 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 {
@@ -866,6 +866,35 @@ impl IggyShard {
.await?;
Ok(())
}
+ ShardEvent::DeletedTopic2 {
+ id,
+ stream_id,
+ topic_id,
+ } => {
+ let topic = self.delete_topic_bypass_auth2(stream_id, topic_id).await?;
+ assert_eq!(topic.id(), *id);
+ Ok(())
+ }
+ ShardEvent::UpdatedTopic2 {
+ stream_id,
+ topic_id,
+ name,
+ message_expiry,
+ compression_algorithm,
+ max_topic_size,
+ replication_factor,
+ } => {
+ self.update_topic_bypass_auth2(
+ stream_id,
+ topic_id,
+ name.clone(),
+ *message_expiry,
+ *compression_algorithm,
+ *max_topic_size,
+ *replication_factor,
+ );
+ Ok(())
+ }
}
}
@@ -923,6 +952,8 @@ impl IggyShard {
| ShardEvent::DeletedShardTableRecords { .. }
| ShardEvent::CreatedStream2 { .. }
| ShardEvent::CreatedTopic2 { .. }
+ | ShardEvent::DeletedTopic2 { .. }
+ | ShardEvent::UpdatedTopic2 { .. }
| ShardEvent::CreatedPartitions2 { .. }
) {
let (sender, receiver) = async_channel::bounded(1);
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index d05aa343..dc7cc8b9 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -235,7 +235,7 @@ impl IggyShard {
name: String,
stats: Arc<StreamStats>,
) -> Result<usize, IggyError> {
- let exists = self.streams2.with(|streams| streams.exists(&name));
+ let exists = self.streams2.exists(&Identifier::named(&name).unwrap());
if exists {
return Err(IggyError::StreamNameAlreadyExists(name));
}
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index fa6c216a..91cd62c8 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -16,12 +16,14 @@
* under the License.
*/
+use std::str::FromStr;
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::shard_info;
use crate::streaming::partitions::partition2;
use crate::streaming::session::Session;
use crate::streaming::stats::stats::TopicStats;
@@ -175,6 +177,19 @@ impl IggyShard {
max_topic_size,
stats,
)?;
+
+ self.streams2.with_topic_by_id(
+ stream_id,
+ &Identifier::numeric(topic_id as u32).unwrap(),
+ |topic| {
+ let message_expiry = match topic.message_expiry() {
+ IggyExpiry::ServerDefault => self.config.system.segment.message_expiry,
+ _ => message_expiry,
+ };
+ shard_info!(self.id, "Topic message expiry: {}", message_expiry);
+ },
+ );
+
// TODO: Create dir hierarchy for the topic.
let partition_ids = self.create_partitions2_base(
stream_id,
@@ -225,7 +240,7 @@ impl IggyShard {
stats: Arc<TopicStats>,
) -> Result<usize, IggyError> {
let topic_id = self.streams2.with_stream_by_id(stream_id, |stream| {
- let exists = stream.topics().exists(&name);
+ let exists = stream.topics().exists(&Identifier::named(&name).unwrap());
if exists {
// TODO: Fixme, replace the second argument with identifier, rather than numeric ID.
return Err(IggyError::TopicNameAlreadyExists(name.to_owned(), 0));
@@ -364,6 +379,100 @@ impl IggyShard {
.await
}
+ pub fn update_topic2(
+ &self,
+ session: &Session,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ name: String,
+ message_expiry: IggyExpiry,
+ compression_algorithm: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ replication_factor: Option<u8>,
+ ) -> Result<(), IggyError> {
+ self.ensure_authenticated(session)?;
+ {
+ let topic_id = self
+ .streams2
+ .with_topic_by_id(stream_id, topic_id, |topic| topic.id());
+ let stream_id = self
+ .streams2
+ .with_stream_by_id(stream_id, |stream| stream.id());
+ self.permissioner.borrow().update_topic(
+ session.get_user_id(),
+ stream_id as u32,
+ topic_id as u32
+ ).with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - permission denied to update topic for user with id: {}, stream ID: {}, topic ID: {}",
+ session.get_user_id(),
+ stream_id,
+ topic_id,
+ )
+ })?;
+ }
+ self.update_topic_base2(
+ stream_id,
+ topic_id,
+ name,
+ message_expiry,
+ compression_algorithm,
+ max_topic_size,
+ replication_factor.unwrap_or(1),
+ );
+ Ok(())
+ }
+
+ pub fn update_topic_bypass_auth2(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ name: String,
+ message_expiry: IggyExpiry,
+ compression_algorithm: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ replication_factor: Option<u8>,
+ ) {
+ self.update_topic_base2(
+ stream_id,
+ topic_id,
+ name,
+ message_expiry,
+ compression_algorithm,
+ max_topic_size,
+ replication_factor.unwrap_or(1),
+ );
+ }
+
+ pub fn update_topic_base2(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ name: String,
+ message_expiry: IggyExpiry,
+ compression_algorithm: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ replication_factor: u8,
+ ) {
+ let (old_name, new_name) =
+ self.streams2
+ .with_topic_by_id_mut(stream_id, topic_id, |topic| {
+ let old_name = topic.name().clone();
+ topic.set_name(name.clone());
+ topic.set_message_expiry(message_expiry);
+ topic.set_compression(compression_algorithm);
+ topic.set_max_topic_size(max_topic_size);
+ topic.set_replication_factor(replication_factor);
+ (old_name, name)
+ // TODO: Set message expiry for all partitions and segments.
+ });
+ self.streams2.with_topics(stream_id, |topics| {
+ topics.with_mut(|container| {
+ container.rename_unchecked(&old_name, new_name);
+ })
+ });
+ }
+
#[allow(clippy::too_many_arguments)]
pub async fn update_topic(
&self,
@@ -417,11 +526,6 @@ impl IggyShard {
"{COMPONENT} (error: {error}) - failed to update topic with ID: {topic_id} in stream with ID: {stream_id}",
)
})?;
- let stream = self.get_stream(stream_id)?;
- let topic = stream.get_topic(topic_id)?;
- topic.persist().await.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to persist topic: {topic}")
- })?;
// TODO: if message_expiry is changed, we need to check if we need to purge messages based on the new expiry
// TODO: if max_size_bytes is changed, we need to check if we need to purge messages based on the new size
@@ -439,7 +543,7 @@ impl IggyShard {
max_topic_size: MaxTopicSize,
replication_factor: Option<u8>,
) -> Result<(), IggyError> {
- self.get_stream_mut(stream_id)?
+ let new_name = self.get_stream_mut(stream_id)?
.update_topic(
topic_id,
name,
@@ -457,7 +561,7 @@ impl IggyShard {
format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}")
})?;
let topic = stream
- .get_topic(topic_id)
+ .get_topic(&Identifier::from_str(new_name.as_str()).unwrap())
.with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - failed to get topic with ID: {topic_id} in stream with ID: {stream_id}",
@@ -484,6 +588,80 @@ impl IggyShard {
Ok(topic)
}
+ pub async fn delete_topic2(
+ &self,
+ session: &Session,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ ) -> Result<topic2::Topic, IggyError> {
+ self.ensure_authenticated(session)?;
+ {
+ let topic_id = self
+ .streams2
+ .with_topic_by_id(stream_id, topic_id, |topic| topic.id());
+ let stream_id = self
+ .streams2
+ .with_stream_by_id(stream_id, |stream| stream.id());
+ self.permissioner
+ .borrow()
+ .delete_topic(session.get_user_id(), stream_id as u32, topic_id as u32)
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - permission denied to delete topic with ID: {topic_id} in stream with ID: {stream_id} for user with ID: {}",
+ session.get_user_id(),
+ )
+ })?;
+ }
+ let topic = self.delete_topic_base2(stream_id, topic_id).with_error_context(|error| {
+ format!("{COMPONENT} (error: {error}) - failed to delete topic with ID: {topic_id} in stream with ID: {stream_id}")
+ })?;
+ // TODO: Remove partitions.
+ Ok(topic)
+ }
+
+ pub async fn delete_topic_bypass_auth2(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ ) -> Result<topic2::Topic, IggyError> {
+ let topic = self.delete_topic_base2(stream_id, topic_id)?;
+ Ok(topic)
+ }
+
+ pub fn delete_topic_base2(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ ) -> Result<topic2::Topic, IggyError> {
+ let stream_exists = self.streams2.exists(stream_id);
+ if !stream_exists {
+ return Err(IggyError::StreamIdNotFound(0));
+ }
+
+ let topic = self.streams2.with_stream_by_id(stream_id, |stream| {
+ let id = stream
+ .topics()
+ .with_topic_by_id(topic_id, |topic| topic.id());
+ stream.topics().with_mut(|container| {
+ container.try_remove(id).ok_or_else(|| {
+ let topic_name = stream
+ .topics()
+ .with_topic_by_id(topic_id, |topic| topic.name().clone());
+ IggyError::TopicNameNotFound(topic_name, stream.name().clone())
+ })
+ })
+ })?;
+ let id = topic.id();
+ self.streams2.with_topics(stream_id, |topics| {
+ topics.with_stats_mut(|container| {
+ container
+ .try_remove(id)
+ .expect("Topic delete: topic stats not found");
+ });
+ });
+ Ok(topic)
+ }
+
pub async fn delete_topic(
&self,
session: &Session,
@@ -542,6 +720,37 @@ impl IggyShard {
.delete_consumer_groups_for_topic(stream_id_value, topic.topic_id);
Ok(topic)
}
+ pub async fn purge_topic2(&self, session: &Session, stream_id: &Identifier, topic_id: &Identifier) -> Result<(), IggyError> {
+ self.ensure_authenticated(session)?;
+ {
+ let topic_id = self
+ .streams2
+ .with_topic_by_id(stream_id, topic_id, |topic| topic.id());
+ let stream_id = self
+ .streams2
+ .with_stream_by_id(stream_id, |stream| stream.id());
+ self.permissioner.borrow().purge_topic(
+ session.get_user_id(),
+ stream_id as u32,
+ topic_id as u32
+ ).with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - permission denied to purge topic with ID: {topic_id} in stream with ID: {stream_id} for user with ID: {}",
+ session.get_user_id(),
+ )
+ })?;
+ }
+
+ Ok(())
+ }
+
+ async fn purge_topic_base2(&self, stream_id: &Identifier, topic_id: &Identifier) -> Result<(), IggyError> {
+ self.streams2.with_partitions(stream_id, topic_id, |partitions| {
+ //partitions
+ });
+
+ Ok(())
+ }
pub async fn purge_topic(
&self,
diff --git a/core/server/src/shard/transmission/event.rs b/core/server/src/shard/transmission/event.rs
index 117b0ebc..2edfe68e 100644
--- a/core/server/src/shard/transmission/event.rs
+++ b/core/server/src/shard/transmission/event.rs
@@ -90,6 +90,15 @@ pub enum ShardEvent {
topic_id: Identifier,
consumer_group_id: Identifier,
},
+ UpdatedTopic2 {
+ stream_id: Identifier,
+ topic_id: Identifier,
+ name: String,
+ message_expiry: IggyExpiry,
+ compression_algorithm: CompressionAlgorithm,
+ max_topic_size: MaxTopicSize,
+ replication_factor: Option<u8>,
+ },
UpdatedTopic {
stream_id: Identifier,
topic_id: Identifier,
@@ -107,6 +116,11 @@ pub enum ShardEvent {
stream_id: Identifier,
topic_id: Identifier,
},
+ DeletedTopic2 {
+ id: usize,
+ stream_id: Identifier,
+ topic_id: Identifier,
+ },
CreatedUser {
username: String,
password: String,
diff --git a/core/server/src/slab/consumer_groups.rs b/core/server/src/slab/consumer_groups.rs
index e269f212..2e6ddb45 100644
--- a/core/server/src/slab/consumer_groups.rs
+++ b/core/server/src/slab/consumer_groups.rs
@@ -1,4 +1,8 @@
use crate::{slab::IndexedSlab, streaming::topics::consumer_group2};
+use ahash::AHashMap;
+use arcshift::ArcShift;
+use slab::Slab;
+use std::sync::{Arc, atomic::AtomicUsize};
const CAPACITY: usize = 1024;
@@ -10,7 +14,9 @@ pub struct ConsumerGroups {
impl Default for ConsumerGroups {
fn default() -> Self {
Self {
- container: IndexedSlab::with_capacity(1024),
+ container: IndexedSlab::with_capacity(CAPACITY),
}
}
}
+
+
diff --git a/core/server/src/slab/mod.rs b/core/server/src/slab/mod.rs
index fc489e81..39d0c943 100644
--- a/core/server/src/slab/mod.rs
+++ b/core/server/src/slab/mod.rs
@@ -5,10 +5,13 @@ pub mod topics;
use ahash::AHashMap;
use slab::Slab;
-use std::ops::{Index, IndexMut};
+use std::{
+ fmt::{Debug, Display},
+ ops::{Index, IndexMut},
+};
pub trait Keyed {
- type Key: Eq + std::hash::Hash + Clone;
+ type Key: Eq + std::hash::Hash + Clone + Debug + Display;
fn key(&self) -> &Self::Key;
}
@@ -18,7 +21,7 @@ pub struct IndexedSlab<T: Keyed> {
index: AHashMap<T::Key, usize>,
}
-impl<T: Keyed + Default> IndexedSlab<T> {
+impl<T: Keyed + Default + Debug> IndexedSlab<T> {
pub fn with_capacity(capacity: usize) -> Self {
Self {
slab: Slab::with_capacity(capacity),
@@ -30,10 +33,6 @@ impl<T: Keyed + Default> IndexedSlab<T> {
self.slab.len()
}
- pub fn exists(&self, key: &T::Key) -> bool {
- self.index.contains_key(key)
- }
-
pub fn insert(&mut self, value: T) -> usize {
let key = value.key().clone();
let index = self.slab.insert(value);
@@ -41,6 +40,24 @@ impl<T: Keyed + Default> IndexedSlab<T> {
index
}
+ pub fn try_remove(&mut self, id: usize) -> Option<T> {
+ self.slab.try_remove(id).and_then(|value| {
+ self.index.remove(value.key());
+ Some(value)
+ })
+ }
+
+ pub fn try_remove_by_key(&mut self, key: &T::Key) -> Option<T> {
+ self.index
+ .remove(key)
+ .and_then(|index| Some(self.slab.remove(index)))
+ }
+
+ pub fn rename_unchecked(&mut self, old: &T::Key, new_key: T::Key) {
+ let idx = self.index.remove(old).expect("Rename key: key not found");
+ self.index.insert(new_key, idx);
+ }
+
pub fn get(&self, index: usize) -> Option<&T> {
self.slab.get(index)
}
@@ -49,11 +66,16 @@ impl<T: Keyed + Default> IndexedSlab<T> {
self.slab.get_mut(index)
}
- fn get_by_key(&self, key: &T::Key) -> Option<&T> {
+ pub fn get_by_key(&self, key: &T::Key) -> Option<&T> {
let index = self.index.get(key)?;
self.slab.get(*index)
}
+ pub fn get_by_key_mut(&mut self, key: &T::Key) -> Option<&mut T> {
+ let index = self.index.get(key)?;
+ self.slab.get_mut(*index)
+ }
+
unsafe fn get_by_key_unchecked(&self, key: &T::Key) -> &T {
let index = self.index.get(key).unwrap();
unsafe { self.slab.get_unchecked(*index) }
diff --git a/core/server/src/slab/partitions.rs b/core/server/src/slab/partitions.rs
index f4f1ff35..d98f88d1 100644
--- a/core/server/src/slab/partitions.rs
+++ b/core/server/src/slab/partitions.rs
@@ -1,7 +1,5 @@
-use std::{cell::RefCell, sync::Arc};
-
-use iggy_common::Partition;
use slab::Slab;
+use std::sync::Arc;
use crate::streaming::{partitions::partition2, segments, stats::stats::PartitionStats};
@@ -36,14 +34,14 @@ impl Partitions {
f(&mut stats)
}
- pub async fn with_async(&self, f: impl AsyncFnOnce(&Slab<partition2::Partition>)) {
+ pub async fn with_async<T>(&self, f: impl AsyncFnOnce(&Slab<partition2::Partition>) -> T) -> T {
let container = &self.container;
- f(&container).await;
+ f(&container).await
}
- pub fn with(&self, f: impl FnOnce(&Slab<partition2::Partition>)) {
+ pub fn with<T>(&self, f: impl FnOnce(&Slab<partition2::Partition>) -> T) -> T {
let container = &self.container;
- f(&container);
+ f(&container)
}
pub fn with_mut<T>(&mut self, f: impl FnOnce(&mut Slab<partition2::Partition>) -> T) -> T {
@@ -51,12 +49,26 @@ impl Partitions {
f(&mut container)
}
- pub fn with_partition_id(&self, partition_id: usize, f: impl FnOnce(&partition2::Partition)) {
+ pub fn with_partition_id<T>(
+ &self,
+ partition_id: usize,
+ f: impl FnOnce(&partition2::Partition) -> T,
+ ) -> T {
self.with(|partitions| {
- if let Some(partition) = partitions.get(partition_id) {
- f(partition);
- }
- });
+ let partition = &partitions[partition_id];
+ f(partition)
+ })
+ }
+
+ pub fn with_partition_by_id_mut<T>(
+ &mut self,
+ partition_id: usize,
+ f: impl FnOnce(&mut partition2::Partition) -> T,
+ ) -> T {
+ self.with_mut(|partitions| {
+ let partition = &mut partitions[partition_id];
+ f(partition)
+ })
}
pub fn with_segments(&self, partition_id: usize, f: impl FnOnce(&Vec<segments::Segment2>)) {
@@ -71,6 +83,7 @@ impl Partitions {
f: impl FnOnce(&segments::Segment2),
) {
self.with_segments(partition_id, |segments| {
+ // we could binary search for that segment technically, but this is fine for now.
if let Some(segment) = segments.iter().find(|s| s.id == segment_id) {
f(segment);
}
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index 8d6082cf..448ec602 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -1,18 +1,8 @@
-use compio::fs::create_dir_all;
-use iggy_common::{CompressionAlgorithm, Identifier, IggyError, MaxTopicSize};
+use iggy_common::Identifier;
use slab::Slab;
-use std::{
- cell::{RefCell, UnsafeCell},
- ops::Index,
- path::Path,
- sync::Arc,
-};
-use tracing::info;
+use std::{cell::RefCell, sync::Arc};
use crate::{
- configs::system::SystemConfig,
- io::storage::Storage,
- shard_info,
slab::{IndexedSlab, partitions::Partitions, topics::Topics},
streaming::{
partitions::partition2, stats::stats::StreamStats, streams::stream2, topics::topic2,
@@ -34,6 +24,19 @@ impl Streams {
}
}
+ pub fn exists(&self, id: &Identifier) -> bool {
+ match id.kind {
+ iggy_common::IdKind::Numeric => {
+ let id = id.get_u32_value().unwrap() as usize;
+ self.container.borrow().slab.contains(id)
+ }
+ iggy_common::IdKind::String => {
+ let key = id.get_string_value().unwrap();
+ self.container.borrow().index.contains_key(&key)
+ }
+ }
+ }
+
pub fn with_stats<T>(&self, f: impl FnOnce(&Slab<Arc<StreamStats>>) -> T) -> T {
let stats = self.stats.borrow();
f(&stats)
@@ -56,9 +59,12 @@ impl Streams {
f(&mut stats)
}
- pub async fn with_async(&self, f: impl AsyncFnOnce(&IndexedSlab<stream2::Stream>)) {
+ pub async fn with_async<T>(
+ &self,
+ f: impl AsyncFnOnce(&IndexedSlab<stream2::Stream>) -> T,
+ ) -> T {
let container = self.container.borrow();
- f(&container).await;
+ f(&container).await
}
pub fn with<T>(&self, f: impl FnOnce(&IndexedSlab<stream2::Stream>) -> T) -> T {
@@ -91,7 +97,7 @@ impl Streams {
})
}
- pub fn with_stream_by_id_mut(&self, id: &Identifier, mut f: impl FnOnce(&mut stream2::Stream)) {
+ pub fn with_stream_by_id_mut(&self, id: &Identifier, f: impl FnOnce(&mut stream2::Stream)) {
self.with_mut(|streams| {
let stream = match id.kind {
iggy_common::IdKind::Numeric => {
@@ -108,7 +114,15 @@ impl Streams {
}
pub fn with_topics<T>(&self, stream_id: &Identifier, f: impl FnOnce(&Topics) -> T) -> T {
- self.with(|streams| {
+ self.with_stream_by_id(stream_id, |stream| f(stream.topics()))
+ }
+
+ pub async fn with_topics_async<T>(
+ &self,
+ stream_id: &Identifier,
+ f: impl AsyncFnOnce(&Topics) -> T,
+ ) -> T {
+ self.with_async(async |streams: &IndexedSlab<stream2::Stream>| {
let stream = match stream_id.kind {
iggy_common::IdKind::Numeric => {
let id = stream_id.get_u32_value().unwrap() as usize;
@@ -119,7 +133,28 @@ impl Streams {
unsafe { streams.get_by_key_unchecked(&key) }
}
};
- f(stream.topics())
+ f(stream.topics()).await
+ })
+ .await
+ }
+
+ pub fn with_topics_mut<T>(
+ &self,
+ stream_id: &Identifier,
+ f: impl FnOnce(&mut Topics) -> T,
+ ) -> T {
+ self.with_mut(|streams| {
+ let stream = match stream_id.kind {
+ iggy_common::IdKind::Numeric => {
+ let id = stream_id.get_u32_value().unwrap() as usize;
+ &mut streams[id]
+ }
+ iggy_common::IdKind::String => {
+ let key = stream_id.get_string_value().unwrap();
+ unsafe { streams.get_by_key_mut_unchecked(&key) }
+ }
+ };
+ f(stream.topics_mut())
})
}
@@ -152,6 +187,18 @@ impl Streams {
});
}
+ pub async fn with_partitions_async<T>(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ f: impl AsyncFnOnce(&Partitions) -> T,
+ ) -> T {
+ self.with_topics_async(stream_id, async |topics| {
+ topics.with_partitions_async(topic_id, f).await
+ })
+ .await
+ }
+
pub fn with_partition_by_id(
&self,
stream_id: &Identifier,
diff --git a/core/server/src/slab/topics.rs b/core/server/src/slab/topics.rs
index e9254a88..d1df1ee4 100644
--- a/core/server/src/slab/topics.rs
+++ b/core/server/src/slab/topics.rs
@@ -3,12 +3,13 @@ use slab::Slab;
use std::{cell::RefCell, sync::Arc};
use crate::{
- slab::{IndexedSlab, partitions::Partitions},
+ slab::{IndexedSlab, Keyed, partitions::Partitions},
streaming::{partitions::partition2, stats::stats::TopicStats, topics::topic2},
};
const CAPACITY: usize = 1024;
+#[derive(Debug)]
pub struct Topics {
container: RefCell<IndexedSlab<topic2::Topic>>,
stats: RefCell<Slab<Arc<TopicStats>>>,
@@ -22,27 +23,59 @@ impl Topics {
}
}
- pub fn len(&self) -> usize {
- self.container.borrow().len()
+ pub fn exists(&self, id: &Identifier) -> bool {
+ match id.kind {
+ iggy_common::IdKind::Numeric => {
+ let id = id.get_u32_value().unwrap() as usize;
+ self.container.borrow().slab.contains(id)
+ }
+ iggy_common::IdKind::String => {
+ let key = id.get_string_value().unwrap();
+ self.container.borrow().index.contains_key(&key)
+ }
+ }
}
- pub fn exists(&self, key: &String) -> bool {
- self.container.borrow().exists(key)
+ fn get_topic_ref<'topics>(
+ id: &Identifier,
+ slab: &'topics IndexedSlab<topic2::Topic>,
+ ) -> &'topics topic2::Topic {
+ match id.kind {
+ iggy_common::IdKind::Numeric => {
+ let idx = id.get_u32_value().unwrap() as usize;
+ &slab[idx]
+ }
+ iggy_common::IdKind::String => {
+ let key = id.get_string_value().unwrap();
+ unsafe { slab.get_by_key_unchecked(&key) }
+ }
+ }
}
- pub async fn with_async(&self, f: impl AsyncFnOnce(&IndexedSlab<topic2::Topic>)) {
- let container = self.container.borrow();
- f(&container).await;
+ fn get_topic_mut<'topics>(
+ id: &Identifier,
+ slab: &'topics mut IndexedSlab<topic2::Topic>,
+ ) -> &'topics mut topic2::Topic {
+ match id.kind {
+ iggy_common::IdKind::Numeric => {
+ let idx = id.get_u32_value().unwrap() as usize;
+ &mut slab[idx]
+ }
+ iggy_common::IdKind::String => {
+ let key = id.get_string_value().unwrap();
+ unsafe { slab.get_by_key_mut_unchecked(&key) }
+ }
+ }
}
- pub fn with_stats<T>(&self, f: impl FnOnce(&Slab<Arc<TopicStats>>) -> T) -> T {
- let stats = self.stats.borrow();
- f(&stats)
+ pub fn len(&self) -> usize {
+ self.container.borrow().len()
}
- 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 async fn with_async<T>(&self, f: impl AsyncFnOnce(&IndexedSlab<topic2::Topic>) -> T) -> T {
+ let container = self.container.borrow();
+ f(&container).await
}
pub fn with<T>(&self, f: impl FnOnce(&IndexedSlab<topic2::Topic>) -> T) -> T {
@@ -55,20 +88,17 @@ impl Topics {
f(&mut container)
}
+ pub async fn with_topic_by_id_async<T>(
+ &self,
+ id: &Identifier,
+ f: impl AsyncFnOnce(&topic2::Topic) -> T,
+ ) -> T {
+ self.with_async(async |topics| Self::get_topic_ref(id, topics).invoke_async(f).await)
+ .await
+ }
+
pub fn with_topic_by_id<T>(&self, id: &Identifier, f: impl FnOnce(&topic2::Topic) -> T) -> T {
- self.with(|topics| {
- let topic = match id.kind {
- iggy_common::IdKind::Numeric => {
- let id = id.get_u32_value().unwrap() as usize;
- &topics[id]
- }
- iggy_common::IdKind::String => {
- let key = id.get_string_value().unwrap();
- unsafe { topics.get_by_key_unchecked(&key) }
- }
- };
- f(topic)
- })
+ self.with(|topics| Self::get_topic_ref(id, topics).invoke(f))
}
pub fn with_topic_by_id_mut<T>(
@@ -76,35 +106,24 @@ impl Topics {
id: &Identifier,
f: impl FnOnce(&mut topic2::Topic) -> T,
) -> T {
- self.with_mut(|topics| {
- let topic = match id.kind {
- iggy_common::IdKind::Numeric => {
- let id = id.get_u32_value().unwrap() as usize;
- &mut topics[id]
- }
- iggy_common::IdKind::String => {
- let key = id.get_string_value().unwrap();
- unsafe { topics.get_by_key_mut_unchecked(&key) }
- }
- };
- f(topic)
- })
+ self.with_mut(|topics| Self::get_topic_mut(id, topics).invoke_mut(f))
}
pub fn with_partitions(&self, topic_id: &Identifier, f: impl FnOnce(&Partitions)) {
- self.with(|topics| {
- let topic = match topic_id.kind {
- iggy_common::IdKind::Numeric => {
- let id = topic_id.get_u32_value().unwrap() as usize;
- &topics[id]
- }
- iggy_common::IdKind::String => {
- let key = topic_id.get_string_value().unwrap();
- unsafe { topics.get_by_key_unchecked(&key) }
- }
- };
- f(topic.partitions());
- })
+ self.with_topic_by_id(topic_id, |topic| f(topic.partitions()));
+ }
+
+ pub fn with_partitions_mut(&self, topic_id: &Identifier, f: impl FnOnce(&mut Partitions)) {
+ self.with_topic_by_id_mut(topic_id, |topic| f(topic.partitions_mut()));
+ }
+
+ pub async fn with_partitions_async<T>(
+ &self,
+ topic_id: &Identifier,
+ f: impl AsyncFnOnce(&Partitions) -> T,
+ ) -> T {
+ self.with_topic_by_id_async(topic_id, async |topic| f(topic.partitions()).await)
+ .await
}
pub fn with_partition_by_id(
@@ -117,4 +136,14 @@ impl Topics {
partitions.with_partition_id(partition_id, f);
});
}
+
+ 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)
+ }
}
diff --git a/core/server/src/streaming/streams/stream2.rs b/core/server/src/streaming/streams/stream2.rs
index 9d065bcf..88b4e64b 100644
--- a/core/server/src/streaming/streams/stream2.rs
+++ b/core/server/src/streaming/streams/stream2.rs
@@ -3,6 +3,7 @@ use iggy_common::IggyTimestamp;
use crate::slab::{IndexedSlab, Keyed, topics::Topics};
+#[derive(Debug)]
pub struct Stream {
id: usize,
name: String,
@@ -44,6 +45,10 @@ impl Stream {
self.id
}
+ pub fn name(&self) -> &String {
+ &self.name
+ }
+
pub fn topics_count(&self) -> usize {
self.topics.len()
}
@@ -58,4 +63,8 @@ impl Stream {
pub fn topics(&self) -> &Topics {
&self.topics
}
+
+ pub fn topics_mut(&mut self) -> &mut Topics {
+ &mut self.topics
+ }
}
diff --git a/core/server/src/streaming/streams/topics.rs b/core/server/src/streaming/streams/topics.rs
index cc227515..cec40c33 100644
--- a/core/server/src/streaming/streams/topics.rs
+++ b/core/server/src/streaming/streams/topics.rs
@@ -107,7 +107,7 @@ impl Stream {
compression_algorithm: CompressionAlgorithm,
max_topic_size: MaxTopicSize,
replication_factor: u8,
- ) -> Result<(), IggyError> {
+ ) -> Result<String, IggyError> {
let message_expiry = Topic::get_message_expiry(message_expiry, &self.config);
let max_topic_size = Topic::get_max_topic_size(max_topic_size, &self.config)?;
let topic_id;
@@ -148,10 +148,11 @@ impl Stream {
topic.replication_factor = replication_factor;
}
+ let temp = name.to_owned();
self.topics_ids.remove(&old_topic_name.clone());
- self.topics_ids.insert(name.to_owned(), topic_id);
+ self.topics_ids.insert(temp.clone(), topic_id);
- Ok(())
+ Ok(temp)
}
pub fn remove_topic(&mut self, identifier: &Identifier) -> Result<Topic, IggyError> {
@@ -197,10 +198,12 @@ impl Stream {
}
fn get_topic_by_name(&self, name: &str) -> Result<&Topic, IggyError> {
- self.topics_ids
+ let result = self
+ .topics_ids
.get(name)
.map(|topic_id| self.get_topic_by_id(*topic_id))
- .ok_or_else(|| IggyError::TopicNameNotFound(name.to_string(), self.name.to_owned()))?
+ .ok_or_else(|| IggyError::TopicNameNotFound(name.to_string(), self.name.to_owned()))?;
+ result
}
fn get_topic_by_id_mut(&mut self, id: u32) -> Result<&mut Topic, IggyError> {
diff --git a/core/server/src/streaming/topics/consumer_group2.rs b/core/server/src/streaming/topics/consumer_group2.rs
index 6091afa4..22177109 100644
--- a/core/server/src/streaming/topics/consumer_group2.rs
+++ b/core/server/src/streaming/topics/consumer_group2.rs
@@ -1,10 +1,38 @@
+use std::sync::{atomic::AtomicUsize, Arc};
+use ahash::AHashMap;
+use arcshift::ArcShift;
+use slab::Slab;
use crate::slab::Keyed;
+const CAPACITY: usize = 128;
+
#[derive(Default, Debug)]
pub struct ConsumerGroup {
+ id: usize,
name: String,
+ membership: AHashMap<u32, usize>,
+ members: ArcShift<Slab<Member>>
+}
+
+impl ConsumerGroup {
+ pub fn new(id: usize, name: String) -> Self {
+ ConsumerGroup {
+ id,
+ name,
+ members: ArcShift::new(Slab::with_capacity(CAPACITY)),
+ membership: AHashMap::with_capacity(CAPACITY),
+ }
+ }
}
+#[derive(Debug)]
+struct Member {
+ id: usize,
+ partitions: Vec<u32>,
+ current_partition_idx: Arc<AtomicUsize>,
+}
+
+
impl Keyed for ConsumerGroup {
type Key = String;
diff --git a/core/server/src/streaming/topics/topic2.rs b/core/server/src/streaming/topics/topic2.rs
index c409a908..855b574e 100644
--- a/core/server/src/streaming/topics/topic2.rs
+++ b/core/server/src/streaming/topics/topic2.rs
@@ -37,10 +37,50 @@ impl Topic {
}
}
+ pub fn invoke<T>(&self, f: impl FnOnce(&Self) -> T) -> T {
+ f(self)
+ }
+
+ pub fn invoke_mut<T>(&mut self, f: impl FnOnce(&mut Self) -> T) -> T {
+ f(self)
+ }
+
+ pub async fn invoke_async<T>(&self, f: impl AsyncFnOnce(&Self) -> T) -> T {
+ f(self).await
+ }
+
+ pub fn message_expiry(&self) -> IggyExpiry {
+ self.message_expiry
+ }
+
pub fn id(&self) -> usize {
self.id
}
+ pub fn name(&self) -> &String {
+ &self.name
+ }
+
+ pub fn set_name(&mut self, name: String) {
+ self.name = name;
+ }
+
+ pub fn set_compression(&mut self, compression: CompressionAlgorithm) {
+ self.compression_algorithm = compression;
+ }
+
+ pub fn set_message_expiry(&mut self, message_expiry: IggyExpiry) {
+ self.message_expiry = message_expiry;
+ }
+
+ pub fn set_max_topic_size(&mut self, max_topic_size: MaxTopicSize) {
+ self.max_topic_size = max_topic_size;
+ }
+
+ pub fn set_replication_factor(&mut self, replication_factor: u8) {
+ self.replication_factor = replication_factor;
+ }
+
pub fn partitions(&self) -> &Partitions {
&self.partitions
}