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/11 11:35:27 UTC
(iggy) 02/02: improvements another part
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch improved_tpc_partition_ecs
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 4781c280ac651089b263a77dcaee51e7d85d0748
Author: numminex <g....@gmail.com>
AuthorDate: Mon Aug 11 13:35:20 2025 +0200
improvements another part
---
Cargo.lock | 21 ++
core/server/Cargo.toml | 1 +
.../partitions/create_partitions_handler.rs | 2 +-
.../binary/handlers/topics/delete_topic_handler.rs | 11 +
core/server/src/shard/mod.rs | 4 +-
core/server/src/shard/system/consumer_offsets.rs | 2 +-
core/server/src/shard/system/partitions.rs | 308 +++++++++------------
core/server/src/shard/transmission/event.rs | 1 +
core/server/src/slab/partitions.rs | 147 +++-------
core/server/src/slab/streams.rs | 12 -
core/server/src/slab/topics.rs | 11 -
.../src/streaming/partitions/consumer_offset.rs | 22 ++
core/server/src/streaming/partitions/mod.rs | 1 +
core/server/src/streaming/partitions/partition2.rs | 167 ++++++++++-
core/server/src/streaming/stats/stats.rs | 4 +-
.../server/src/streaming/topics/consumer_group2.rs | 58 ++--
core/server/src/streaming/topics/topic2.rs | 4 +-
17 files changed, 419 insertions(+), 357 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index c566720d..37dbdb8f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5724,6 +5724,16 @@ version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39"
+[[package]]
+name = "papaya"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f92dd0b07c53a0a0c764db2ace8c541dc47320dad97c2200c2a637ab9dd2328f"
+dependencies = [
+ "equivalent",
+ "seize",
+]
+
[[package]]
name = "parking"
version = "2.2.1"
@@ -7248,6 +7258,16 @@ dependencies = [
"libc",
]
+[[package]]
+name = "seize"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e4b8d813387d566f627f3ea1b914c068aac94c40ae27ec43f5f33bde65abefe7"
+dependencies = [
+ "libc",
+ "windows-sys 0.52.0",
+]
+
[[package]]
name = "semver"
version = "1.0.26"
@@ -7491,6 +7511,7 @@ dependencies = [
"opentelemetry-otlp",
"opentelemetry-semantic-conventions",
"opentelemetry_sdk",
+ "papaya",
"prometheus-client",
"quinn",
"reqwest",
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index fc2ae8c4..35693903 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -95,6 +95,7 @@ opentelemetry_sdk = { version = "0.30.0", features = [
"experimental_trace_batch_span_processor_with_async_runtime",
] }
prometheus-client = "0.23.1"
+papaya = "0.2.3"
quinn = { workspace = true }
reqwest = { workspace = true, features = ["rustls-tls-no-provider"] }
ring = "0.17.14"
diff --git a/core/server/src/binary/handlers/partitions/create_partitions_handler.rs b/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
index 8b435943..a208f3ee 100644
--- a/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
+++ b/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
@@ -45,7 +45,7 @@ impl ServerCommandHandler for CreatePartitions {
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
debug!("session: {session}, command: {self}");
- let (shared_partitions, created_at) = shard
+ let partitions = shard
.create_partitions2(
session,
&self.stream_id,
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 c7ce984c..af16cf8c 100644
--- a/core/server/src/binary/handlers/topics/delete_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/delete_topic_handler.rs
@@ -19,6 +19,7 @@
use crate::binary::command::{BinaryServerCommand, ServerCommand, ServerCommandHandler};
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::{handlers::topics::COMPONENT, sender::SenderKind};
+use crate::io::fs_utils::remove_dir_all;
use crate::shard::IggyShard;
use crate::shard::namespace::IggyNamespace;
use crate::shard::transmission::event::ShardEvent;
@@ -26,6 +27,8 @@ use crate::shard_info;
use crate::state::command::EntryCommand;
use crate::streaming::session::Session;
use anyhow::Result;
+use compio::driver::op::OpenFile;
+use compio::fs::File;
use error_set::ErrContext;
use iggy_common::IggyError;
use iggy_common::delete_topic::DeleteTopic;
@@ -47,6 +50,11 @@ impl ServerCommandHandler for DeleteTopic {
session: &Rc<Session>,
shard: &Rc<IggyShard>,
) -> Result<(), IggyError> {
+ // TODO: There is a correctness bug,
+ // We have to first apply the state, then proceed with deleting the topic from the disk.
+ // Otherwise if we would delete the topic from disk first, and then the state would fail
+ // we would end up in a state where the topic is deleted from disk, but during state recreation it would be recreated,
+ // without persisted messages in the partitions.
debug!("session: {session}, command: {self}");
let topic2 = shard
.delete_topic2(session, &self.stream_id, &self.topic_id)
@@ -57,6 +65,9 @@ impl ServerCommandHandler for DeleteTopic {
topic_id: self.topic_id.clone(),
};
let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
+ // Drop the topic to force readers/writers to be dropped.
+ drop(topic2);
+ // Remove all the files and directories.
let topic = shard
.delete_topic(session, &self.stream_id, &self.topic_id)
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 6eba978b..ca1ea4ea 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -627,10 +627,10 @@ impl IggyShard {
ShardEvent::DeletedPartitions2 {
stream_id,
topic_id,
+ partition_count,
partition_ids,
} => {
- self.delete_partitions2_bypass_auth(stream_id, topic_id, &partition_ids)
- .await?;
+ self.delete_partitions2_bypass_auth(stream_id, topic_id, &partition_ids)?;
Ok(())
}
ShardEvent::DeletedPartitions {
diff --git a/core/server/src/shard/system/consumer_offsets.rs b/core/server/src/shard/system/consumer_offsets.rs
index e5504173..f2ed0a84 100644
--- a/core/server/src/shard/system/consumer_offsets.rs
+++ b/core/server/src/shard/system/consumer_offsets.rs
@@ -19,7 +19,7 @@
use super::COMPONENT;
use crate::{shard::IggyShard, streaming::session::Session};
use error_set::ErrContext;
-use iggy_common::{Consumer, ConsumerOffsetInfo, Identifier, IggyError};
+use iggy_common::{Consumer, ConsumerOffsetInfo, Identifier, IggyError, store_consumer_offset};
impl IggyShard {
pub async fn store_consumer_offset(
diff --git a/core/server/src/shard/system/partitions.rs b/core/server/src/shard/system/partitions.rs
index ebdd1384..5ae0e6e8 100644
--- a/core/server/src/shard/system/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -16,11 +16,17 @@
* under the License.
*/
+use std::sync::Arc;
+
use super::COMPONENT;
+use crate::configs::system::SystemConfig;
use crate::shard::IggyShard;
+use crate::slab::traits_ext::Delete;
+use crate::slab::traits_ext::EntityComponentSystem;
+use crate::slab::traits_ext::Insert;
use crate::streaming::deduplication::message_deduplicator::MessageDeduplicator;
+use crate::streaming::partitions::partition::Partition;
use crate::streaming::partitions::partition2;
-use crate::streaming::partitions::partition2::SharedPartition;
use crate::streaming::partitions::storage2::create_partition_file_hierarchy;
use crate::streaming::session::Session;
use crate::streaming::stats::stats::PartitionStats;
@@ -29,50 +35,12 @@ use error_set::ErrContext;
use iggy_common::Identifier;
use iggy_common::IggyError;
use iggy_common::IggyTimestamp;
+use iggy_common::delete_partitions;
use iggy_common::locking::IggyRwLockFn;
-use std::sync::Arc;
+use serde::de;
use std::sync::atomic::AtomicU64;
impl IggyShard {
- fn create_shared_partition(
- &self,
- created_at: IggyTimestamp,
- topic_stats: Arc<TopicStats>,
- ) -> (partition2::Partition, SharedPartition) {
- let partition = partition2::Partition::new(created_at, false);
- let partition_stats = Arc::new(PartitionStats::new(topic_stats));
-
- let shared_partition = SharedPartition {
- id: partition.id(),
- stats: partition_stats,
- current_offset: Arc::new(AtomicU64::new(0)),
- consumer_group_offset: Arc::new(()),
- consumer_offset: Arc::new(()),
- };
-
- (partition, shared_partition)
- }
-
- fn create_message_deduplicator(&self) -> Option<MessageDeduplicator> {
- if !self.config.system.message_deduplication.enabled {
- return None;
- }
-
- let max_entries = if self.config.system.message_deduplication.max_entries > 0 {
- Some(self.config.system.message_deduplication.max_entries)
- } else {
- None
- };
-
- let expiry = if !self.config.system.message_deduplication.expiry.is_zero() {
- Some(self.config.system.message_deduplication.expiry)
- } else {
- None
- };
-
- Some(MessageDeduplicator::new(max_entries, expiry))
- }
-
fn validate_partition_permissions(
&self,
session: &Session,
@@ -97,47 +65,13 @@ impl IggyShard {
})
}
- fn insert_partition_resources(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- partition: partition2::Partition,
- shared_partition: &SharedPartition,
- ) -> usize {
- self.streams2
- .with_topic_by_id_mut(stream_id, topic_id, |topic| {
- let partition_id = topic
- .partitions_mut()
- .with_mut(|partitions| partition.insert_into(partitions));
-
- topic
- .partitions_mut()
- .with_stats_mut(|stats| stats.insert(shared_partition.stats.clone()));
-
- let message_deduplicator = self.create_message_deduplicator();
- topic
- .partitions_mut()
- .with_message_deduplicators_mut(|deduplicators| {
- deduplicators.insert(message_deduplicator);
- });
-
- topic
- .partitions_mut()
- .with_partition_offsets_mut(|offsets| {
- offsets.insert(shared_partition.current_offset.clone());
- });
-
- partition_id
- })
- }
-
pub async fn create_partitions2(
&self,
session: &Session,
stream_id: &Identifier,
topic_id: &Identifier,
partitions_count: u32,
- ) -> Result<(Vec<SharedPartition>, IggyTimestamp), IggyError> {
+ ) -> Result<Vec<partition2::Partition>, IggyError> {
self.ensure_authenticated(session)?;
let numeric_stream_id =
self.streams2
@@ -153,38 +87,18 @@ impl IggyShard {
"create",
)?;
- let topic_stats = self.streams2.with_stream_by_id(stream_id, |stream| {
+ let parent_stats = self.streams2.with_stream_by_id(stream_id, |stream| {
stream
.topics()
.with_topic_stats_by_id(topic_id, |stats| stats)
});
-
- let created_at = IggyTimestamp::now();
- let shared_partitions = (0..partitions_count)
- .map(|_| {
- let partition = partition2::Partition::new(created_at, false);
- let partition_stats = Arc::new(PartitionStats::new(topic_stats.clone()));
-
- let mut shared_partition = SharedPartition {
- id: partition.id(),
- stats: partition_stats,
- current_offset: Arc::new(AtomicU64::new(0)),
- consumer_group_offset: Arc::new(()),
- consumer_offset: Arc::new(()),
- };
- let partition_id = self.insert_partition_resources(
- stream_id,
- topic_id,
- partition,
- &shared_partition,
- );
-
- shared_partition.id = partition_id;
- shared_partition
- })
- .collect::<Vec<SharedPartition>>();
-
- for partition_id in shared_partitions.iter().map(|p| p.id) {
+ let partitions = self.create_and_insert_partitions_mem(
+ stream_id,
+ topic_id,
+ parent_stats,
+ partitions_count,
+ );
+ for partition_id in partitions.iter().map(|p| p.id()) {
create_partition_file_hierarchy(
self.id,
numeric_stream_id as usize,
@@ -194,33 +108,92 @@ impl IggyShard {
)
.await?;
}
-
- Ok((shared_partitions, created_at))
+ Ok(partitions)
}
- pub async fn create_partitions2_bypass_auth(
+ fn create_and_insert_partitions_mem(
&self,
stream_id: &Identifier,
topic_id: &Identifier,
- created_at: IggyTimestamp,
- shared_partitions: &Vec<SharedPartition>,
- ) -> Result<(), IggyError> {
- for shared_partition in shared_partitions {
- let partition = partition2::Partition::new(created_at, false);
- let partition_id =
- self.insert_partition_resources(stream_id, topic_id, partition, shared_partition);
-
- if partition_id != shared_partition.id {
- return Err(IggyError::PartitionNotFound(
- partition_id as u32,
- self.streams2.with_stream_by_id(stream_id, |s| s.id()) as u32,
- self.streams2
- .with_topic_by_id(stream_id, topic_id, |t| t.id())
- as u32,
- ));
+ parent_stats: Arc<TopicStats>,
+ partitions_count: u32,
+ ) -> Vec<partition2::Partition> {
+ fn create_message_deduplicator(config: &SystemConfig) -> Option<MessageDeduplicator> {
+ if !config.message_deduplication.enabled {
+ return None;
}
+
+ let max_entries = if config.message_deduplication.max_entries > 0 {
+ Some(config.message_deduplication.max_entries)
+ } else {
+ None
+ };
+
+ let expiry = if !config.message_deduplication.expiry.is_zero() {
+ Some(config.message_deduplication.expiry)
+ } else {
+ None
+ };
+ Some(MessageDeduplicator::new(max_entries, expiry))
}
+ let range = 0..partitions_count as usize;
+ let created_at = IggyTimestamp::now();
+ range
+ .map(|_| {
+ // Areczkuuuu.
+ let stats = Arc::new(PartitionStats::new(parent_stats.clone()));
+ let info = partition2::PartitionInfo::new(created_at, false);
+ let deduplicator = create_message_deduplicator(&self.config.system);
+ let offset = Arc::new(AtomicU64::new(0));
+ let consumer_offset = Arc::new(papaya::HashMap::with_capacity(2137));
+ let consumer_group_offset = Arc::new(papaya::HashMap::with_capacity(2137));
+
+ let mut partition = partition2::Partition::new(
+ info,
+ stats,
+ deduplicator,
+ offset,
+ consumer_offset,
+ consumer_group_offset,
+ );
+ let id = self.insert_partition_mem(
+ stream_id,
+ topic_id,
+ partition.clone(create_message_deduplicator(&self.config.system)),
+ );
+ partition.update_id(id);
+ partition
+ })
+ .collect()
+ }
+
+ fn insert_partition_mem(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partition: partition2::Partition,
+ ) -> usize {
+ self.streams2
+ .with_topic_by_id_mut(stream_id, topic_id, |topic| {
+ topic.partitions_mut().insert(partition)
+ })
+ }
+
+ pub fn create_partitions2_bypass_auth(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partitions: Vec<partition2::Partition>,
+ ) -> Result<(), IggyError> {
+ for partition in partitions {
+ let actual_id = partition.id();
+ let id = self.insert_partition_mem(stream_id, topic_id, partition);
+ assert_eq!(
+ id, actual_id,
+ "create_partitions_bypass_auth: partition mismatch ID, wrong creation order ?!"
+ );
+ }
Ok(())
}
@@ -306,79 +279,56 @@ impl IggyShard {
numeric_topic_id,
"delete",
)?;
+ let deleted_partition_ids =
+ self.delete_partitions_base2(stream_id, topic_id, partitions_count);
+
+ Ok(deleted_partition_ids)
+ }
+ fn delete_partitions_base2(
+ &self,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partitions_count: u32,
+ ) -> Vec<u32> {
let deleted_partition_ids =
self.streams2
.with_topic_by_id_mut(stream_id, topic_id, |topic| {
- let current_count = topic.partitions().with(|p| p.len()) as u32;
-
- if current_count == 0 || partitions_count == 0 {
- return Vec::new();
- }
-
+ let partitions = topic.partitions_mut();
+ let current_count = partitions.count() as u32;
let partitions_to_delete = partitions_count.min(current_count);
let start_idx = (current_count - partitions_to_delete) as usize;
let mut deleted_ids = Vec::with_capacity(partitions_to_delete as usize);
-
- for idx in (start_idx..current_count as usize).rev() {
- topic
- .partitions_mut()
- .with_mut(|partitions| partitions.try_remove(idx));
-
- topic
- .partitions_mut()
- .with_stats_mut(|stats| stats.try_remove(idx));
-
- topic
- .partitions_mut()
- .with_message_deduplicators_mut(|deduplicators| {
- deduplicators.try_remove(idx)
- });
-
- topic
- .partitions_mut()
- .with_partition_offsets_mut(|offsets| offsets.try_remove(idx));
-
- deleted_ids.push(idx as u32);
+ for idx in (start_idx..current_count as usize) {
+ let partition = topic.partitions_mut().delete(idx);
+ assert_eq!(partition.id(), idx);
+ deleted_ids.push(partition.id() as u32);
}
-
deleted_ids
});
-
- Ok(deleted_partition_ids)
+ deleted_partition_ids
}
- pub async fn delete_partitions2_bypass_auth(
+ pub fn delete_partitions2_bypass_auth(
&self,
stream_id: &Identifier,
topic_id: &Identifier,
- partition_ids: &[u32],
+ partitions_count: u32,
+ partition_ids: Vec<u32>,
) -> Result<(), IggyError> {
- self.streams2
- .with_topic_by_id_mut(stream_id, topic_id, |topic| {
- for &partition_id in partition_ids {
- let idx = partition_id as usize;
-
- topic
- .partitions_mut()
- .with_mut(|partitions| partitions.try_remove(idx));
-
- topic
- .partitions_mut()
- .with_stats_mut(|stats| stats.try_remove(idx));
-
- topic
- .partitions_mut()
- .with_message_deduplicators_mut(|deduplicators| {
- deduplicators.try_remove(idx)
- });
-
- topic
- .partitions_mut()
- .with_partition_offsets_mut(|offsets| offsets.try_remove(idx));
- }
- });
+ assert_eq!(partitions_count as usize, partition_ids.len());
+ let deleted_partition_ids =
+ self.delete_partitions_base2(stream_id, topic_id, partitions_count);
+ for (deleted_partition_id, actual_deleted_partition_id) in deleted_partition_ids
+ .into_iter()
+ .zip(partition_ids.into_iter())
+ {
+ assert_eq!(
+ deleted_partition_id, actual_deleted_partition_id,
+ "delete_partitions2_bypass_auth: partition mismatch ID"
+ );
+ }
Ok(())
}
diff --git a/core/server/src/shard/transmission/event.rs b/core/server/src/shard/transmission/event.rs
index 286bee43..897f8363 100644
--- a/core/server/src/shard/transmission/event.rs
+++ b/core/server/src/shard/transmission/event.rs
@@ -81,6 +81,7 @@ pub enum ShardEvent {
DeletedPartitions2 {
stream_id: Identifier,
topic_id: Identifier,
+ partition_count: u32,
partition_ids: Vec<u32>,
},
CreatedTopic {
diff --git a/core/server/src/slab/partitions.rs b/core/server/src/slab/partitions.rs
index 0cc1425c..3ba5ba52 100644
--- a/core/server/src/slab/partitions.rs
+++ b/core/server/src/slab/partitions.rs
@@ -1,22 +1,30 @@
use crate::{
slab::traits_ext::{
- Borrow, Components, Delete, EntityComponentSystem, EntityMarker, IndexComponents, Insert, IntoComponents,
+ Borrow, Components, Delete, EntityComponentSystem, EntityMarker, IndexComponents, Insert,
+ IntoComponents,
},
streaming::{
- deduplication::message_deduplicator::MessageDeduplicator, partitions::{partition::ConsumerOffset, partition2}, segments,
+ deduplication::message_deduplicator::MessageDeduplicator,
+ partitions::{
+ consumer_offset,
+ partition::ConsumerOffset,
+ partition2::{self, Partition, PartitionRef},
+ },
+ segments,
stats::stats::PartitionStats,
},
};
use ahash::AHashMap;
use slab::Slab;
-use std::{
- ops::Index,
- sync::{Arc, atomic::AtomicU64},
-};
+use std::sync::{Arc, atomic::AtomicU64};
// TODO: This could be upper limit of partitions per topic, use that value to validate instead of whathever this thing is in `common` crate.
pub const PARTITIONS_CAPACITY: usize = 16384;
-type Id = usize;
+pub type SlabId = usize;
+
+struct PartitionOffset {
+ offset: u64,
+}
#[derive(Debug)]
pub struct Partitions {
@@ -26,123 +34,36 @@ pub struct Partitions {
message_deduplicator: Slab<Option<MessageDeduplicator>>,
offset: Slab<Arc<AtomicU64>>,
- consumer_offset: Slab<AHashMap<usize, ConsumerOffset>>,
- consumer_group_offset: Slab<AHashMap<usize, ConsumerOffset>>,
-}
-
-impl Insert<Id> for Partitions {
- type Item = Partition;
-
- fn insert(&mut self, item: Self::Item) -> Id {
- todo!();
- }
-}
-
-impl Delete<Id> for Partitions {
- type Item = Partition;
-
- fn delete(&mut self, id: Id) -> Self::Item {
- todo!()
- }
-}
-
-pub struct Partition {
- info: partition2::PartitionInfo,
- stats: Arc<PartitionStats>,
- message_deduplicator: Option<MessageDeduplicator>,
- offset: Arc<AtomicU64>,
-}
-
-impl EntityMarker for Partition {}
-
-impl IntoComponents for Partition {
- type Components = (
- partition2::PartitionInfo,
- Arc<PartitionStats>,
- Option<MessageDeduplicator>,
- Arc<AtomicU64>,
- );
-
- fn into_components(self) -> Self::Components {
- (
- self.info,
- self.stats,
- self.message_deduplicator,
- self.offset,
- )
- }
-}
-
-pub struct PartitionRef<'a> {
- info: &'a Slab<partition2::PartitionInfo>,
- stats: &'a Slab<Arc<PartitionStats>>,
- message_deduplicator: &'a Slab<Option<MessageDeduplicator>>,
- offset: &'a Slab<Arc<AtomicU64>>,
+ consumer_offset: Slab<Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>>,
+ consumer_group_offset: Slab<Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>>,
}
impl<'a> From<&'a Partitions> for PartitionRef<'a> {
fn from(value: &'a Partitions) -> Self {
- PartitionRef {
- info: &value.info,
- stats: &value.stats,
- message_deduplicator: &value.message_deduplicator,
- offset: &value.offset,
- }
- }
-}
-
-impl<'a> IntoComponents for PartitionRef<'a> {
- type Components = (
- &'a Slab<partition2::PartitionInfo>,
- &'a Slab<Arc<PartitionStats>>,
- &'a Slab<Option<MessageDeduplicator>>,
- &'a Slab<Arc<AtomicU64>>,
- );
-
- fn into_components(self) -> Self::Components {
- (
- self.info,
- self.stats,
- self.message_deduplicator,
- self.offset,
+ PartitionRef::new(
+ &value.info,
+ &value.stats,
+ &value.message_deduplicator,
+ &value.offset,
+ &value.consumer_offset,
+ &value.consumer_group_offset,
)
}
}
-impl<'a> IndexComponents<Id> for PartitionRef<'a> {
- type Output = (
- &'a partition2::PartitionInfo,
- &'a Arc<PartitionStats>,
- &'a Option<MessageDeduplicator>,
- &'a Arc<AtomicU64>,
- );
+impl Insert<SlabId> for Partitions {
+ type Item = Partition;
- fn index(&self, index: Id) -> Self::Output {
- (
- &self.info[index],
- &self.stats[index],
- &self.message_deduplicator[index],
- &self.offset[index],
- )
+ fn insert(&mut self, item: Self::Item) -> SlabId {
+ todo!();
}
}
-impl EntityComponentSystem<Id, Borrow> for Partitions {
- type Entity = Partition;
- type EntityRef<'a> = PartitionRef<'a>;
-
- fn with<O, F>(&self, f: F) -> O
- where
- F: for<'a> FnOnce(Self::EntityRef<'a>) -> O,
- {
- f(self.into())
- }
+impl Delete<SlabId> for Partitions {
+ type Item = Partition;
- async fn with_async<O, F>(&self, f: F) -> O
- where
- F: for<'a> AsyncFnOnce(Self::EntityRef<'a>) -> O,
- {
- f(self.into()).await
+ fn delete(&mut self, id: SlabId) -> Self::Item {
+ todo!()
}
}
@@ -161,6 +82,10 @@ impl Default for Partitions {
}
impl Partitions {
+ pub fn count(&self) -> usize {
+ self.info.len()
+ }
+
pub fn with_stats<T>(&self, f: impl FnOnce(&Slab<Arc<PartitionStats>>) -> T) -> T {
let stats = &self.stats;
f(stats)
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index 6b63c406..14951dae 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -205,18 +205,6 @@ impl Streams {
.await
}
- pub fn with_partition_by_id(
- &self,
- stream_id: &Identifier,
- topic_id: &Identifier,
- partition_id: usize,
- f: impl FnOnce(&partition2::Partition),
- ) {
- self.with_partitions(stream_id, topic_id, |partitions| {
- partitions.with_partition_id(partition_id, f);
- });
- }
-
pub fn len(&self) -> usize {
self.container.borrow().len()
}
diff --git a/core/server/src/slab/topics.rs b/core/server/src/slab/topics.rs
index 5e89c242..fe06668b 100644
--- a/core/server/src/slab/topics.rs
+++ b/core/server/src/slab/topics.rs
@@ -126,17 +126,6 @@ impl Topics {
.await
}
- pub fn with_partition_by_id(
- &self,
- id: &Identifier,
- partition_id: usize,
- f: impl FnOnce(&partition2::Partition),
- ) {
- self.with_partitions(id, |partitions| {
- 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)
diff --git a/core/server/src/streaming/partitions/consumer_offset.rs b/core/server/src/streaming/partitions/consumer_offset.rs
new file mode 100644
index 00000000..78ac0722
--- /dev/null
+++ b/core/server/src/streaming/partitions/consumer_offset.rs
@@ -0,0 +1,22 @@
+use std::sync::atomic::AtomicU64;
+
+use iggy_common::ConsumerKind;
+
+#[derive(Debug)]
+pub struct ConsumerOffset {
+ pub kind: ConsumerKind,
+ pub consumer_id: u32,
+ pub offset: AtomicU64,
+ pub path: String,
+}
+
+impl ConsumerOffset {
+ pub fn new(kind: ConsumerKind, consumer_id: u32, offset: u64, path: &str) -> ConsumerOffset {
+ ConsumerOffset {
+ kind,
+ consumer_id,
+ offset: AtomicU64::new(offset),
+ path: format!("{path}/{consumer_id}"),
+ }
+ }
+}
diff --git a/core/server/src/streaming/partitions/mod.rs b/core/server/src/streaming/partitions/mod.rs
index e9d62716..08e8eaae 100644
--- a/core/server/src/streaming/partitions/mod.rs
+++ b/core/server/src/streaming/partitions/mod.rs
@@ -16,6 +16,7 @@
* under the License.
*/
+pub mod consumer_offset;
pub mod consumer_offsets;
pub mod messages;
pub mod partition;
diff --git a/core/server/src/streaming/partitions/partition2.rs b/core/server/src/streaming/partitions/partition2.rs
index e2baef51..7d4461db 100644
--- a/core/server/src/streaming/partitions/partition2.rs
+++ b/core/server/src/streaming/partitions/partition2.rs
@@ -1,18 +1,94 @@
-use crate::streaming::stats::stats::PartitionStats;
+use crate::{
+ slab::{
+ partitions::{Partitions, SlabId},
+ traits_ext::{EntityMarker, IndexComponents, IntoComponents},
+ },
+ streaming::{
+ deduplication::message_deduplicator::{self, MessageDeduplicator},
+ partitions::consumer_offset,
+ stats::stats::PartitionStats,
+ },
+};
use iggy_common::IggyTimestamp;
use slab::Slab;
use std::sync::{Arc, atomic::AtomicU64};
#[derive(Debug)]
-pub struct SharedPartition {
- pub id: usize,
- pub stats: Arc<PartitionStats>,
- pub current_offset: Arc<AtomicU64>,
- pub consumer_group_offset: Arc<()>,
- pub consumer_offset: Arc<()>,
+pub struct Partition {
+ info: PartitionInfo,
+ stats: Arc<PartitionStats>,
+ message_deduplicator: Option<MessageDeduplicator>,
+ offset: Arc<AtomicU64>,
+ consumer_offset: Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>,
+ consumer_group_offset: Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>,
}
-#[derive(Default, Debug)]
+impl Partition {
+ pub fn new(
+ info: PartitionInfo,
+ stats: Arc<PartitionStats>,
+ message_deduplicator: Option<MessageDeduplicator>,
+ offset: Arc<AtomicU64>,
+ consumer_offset: Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>,
+ consumer_group_offset: Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>,
+ ) -> Self {
+ Self {
+ info,
+ stats,
+ message_deduplicator,
+ offset,
+ consumer_offset,
+ consumer_group_offset,
+ }
+ }
+
+ pub fn update_id(&mut self, id: usize) {
+ self.info.id = id;
+ }
+
+ pub fn id(&self) -> usize {
+ self.info.id
+ }
+}
+
+impl Partition {
+ pub fn clone(&self, message_deduplicator: Option<MessageDeduplicator>) -> Self {
+ Self {
+ info: self.info.clone(),
+ stats: Arc::clone(&self.stats),
+ message_deduplicator,
+ offset: Arc::clone(&self.offset),
+ consumer_offset: Arc::clone(&self.consumer_offset),
+ consumer_group_offset: Arc::clone(&self.consumer_group_offset),
+ }
+ }
+}
+
+impl EntityMarker for Partition {}
+
+impl IntoComponents for Partition {
+ type Components = (
+ PartitionInfo,
+ Arc<PartitionStats>,
+ Option<MessageDeduplicator>,
+ Arc<AtomicU64>,
+ Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>,
+ Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>,
+ );
+
+ fn into_components(self) -> Self::Components {
+ (
+ self.info,
+ self.stats,
+ self.message_deduplicator,
+ self.offset,
+ self.consumer_offset,
+ self.consumer_group_offset,
+ )
+ }
+}
+
+#[derive(Default, Debug, Clone)]
pub struct PartitionInfo {
id: usize,
created_at: IggyTimestamp,
@@ -39,3 +115,78 @@ impl PartitionInfo {
idx
}
}
+
+pub struct PartitionRef<'a> {
+ info: &'a Slab<PartitionInfo>,
+ stats: &'a Slab<Arc<PartitionStats>>,
+ message_deduplicator: &'a Slab<Option<MessageDeduplicator>>,
+ offset: &'a Slab<Arc<AtomicU64>>,
+ consumer_offset: &'a Slab<Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>>,
+ consumer_group_offset: &'a Slab<Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>>,
+}
+
+impl<'a> PartitionRef<'a> {
+ pub fn new(
+ info: &'a Slab<PartitionInfo>,
+ stats: &'a Slab<Arc<PartitionStats>>,
+ message_deduplicator: &'a Slab<Option<MessageDeduplicator>>,
+ offset: &'a Slab<Arc<AtomicU64>>,
+ consumer_offset: &'a Slab<Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>>,
+ consumer_group_offset: &'a Slab<
+ Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>,
+ >,
+ ) -> Self {
+ Self {
+ info,
+ stats,
+ message_deduplicator,
+ offset,
+ consumer_offset,
+ consumer_group_offset,
+ }
+ }
+}
+
+impl<'a> IntoComponents for PartitionRef<'a> {
+ type Components = (
+ &'a Slab<PartitionInfo>,
+ &'a Slab<Arc<PartitionStats>>,
+ &'a Slab<Option<MessageDeduplicator>>,
+ &'a Slab<Arc<AtomicU64>>,
+ &'a Slab<Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>>,
+ &'a Slab<Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>>,
+ );
+
+ fn into_components(self) -> Self::Components {
+ (
+ self.info,
+ self.stats,
+ self.message_deduplicator,
+ self.offset,
+ self.consumer_offset,
+ self.consumer_group_offset,
+ )
+ }
+}
+
+impl<'a> IndexComponents<SlabId> for PartitionRef<'a> {
+ type Output = (
+ &'a PartitionInfo,
+ &'a Arc<PartitionStats>,
+ &'a Option<MessageDeduplicator>,
+ &'a Arc<AtomicU64>,
+ &'a Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>,
+ &'a Arc<papaya::HashMap<usize, consumer_offset::ConsumerOffset>>,
+ );
+
+ fn index(&self, index: SlabId) -> Self::Output {
+ (
+ &self.info[index],
+ &self.stats[index],
+ &self.message_deduplicator[index],
+ &self.offset[index],
+ &self.consumer_offset[index],
+ &self.consumer_group_offset[index],
+ )
+ }
+}
diff --git a/core/server/src/streaming/stats/stats.rs b/core/server/src/streaming/stats/stats.rs
index 86dbdae8..7a834600 100644
--- a/core/server/src/streaming/stats/stats.rs
+++ b/core/server/src/streaming/stats/stats.rs
@@ -57,9 +57,9 @@ pub struct PartitionStats {
}
impl PartitionStats {
- pub fn new(parent: Arc<TopicStats>) -> Self {
+ pub fn new(parent_stats: Arc<TopicStats>) -> Self {
Self {
- parent,
+ parent: parent_stats,
messages_count: AtomicU64::new(0),
size_bytes: AtomicU64::new(0),
}
diff --git a/core/server/src/streaming/topics/consumer_group2.rs b/core/server/src/streaming/topics/consumer_group2.rs
index 277cb3f9..f1da2f85 100644
--- a/core/server/src/streaming/topics/consumer_group2.rs
+++ b/core/server/src/streaming/topics/consumer_group2.rs
@@ -52,9 +52,12 @@ impl ConsumerGroup {
pub fn reassign_partitions(&mut self, partitions: Vec<usize>) {
self.partitions = partitions;
- let mut members = self.mimic_members();
- self.assign_partitions_to_members(&mut members);
- self.members.update(members);
+ self.members.rcu(|members| {
+ let mut members = Self::mimic_members(members);
+ let partitions = self.partitions.clone();
+ Self::assign_partitions_to_members(self.id, &mut members, partitions);
+ members
+ });
}
pub fn calculate_partition_id_unchecked(&self, member_id: usize) -> Option<usize> {
@@ -91,10 +94,13 @@ impl ConsumerGroup {
}
pub fn add_member(&mut self, client_id: u32) {
- let mut members = self.mimic_members();
- Member::new(client_id).insert_into(&mut members);
- self.assign_partitions_to_members(&mut members);
- self.members.update(members);
+ self.members.rcu(|members| {
+ let mut members = Self::mimic_members(members);
+ Member::new(client_id).insert_into(&mut members);
+ let partitions = self.partitions.clone();
+ Self::assign_partitions_to_members(self.id, &mut members, partitions);
+ members
+ });
}
pub fn delete_member(&mut self, client_id: u32) {
@@ -104,38 +110,39 @@ impl ConsumerGroup {
.iter()
.find_map(|(_, member)| (member.client_id == client_id).then_some(member.id))
.expect("delete_member: find member in consumer group slab");
- let mut members = self.mimic_members();
- members.remove(member_id);
- members.compact(|entry, _, idx| {
- entry.id = idx;
- true
+ self.members.rcu(|members| {
+ let mut members = Self::mimic_members(members);
+ let partitions = self.partitions.clone();
+ members.remove(member_id);
+ members.compact(|entry, _, idx| {
+ entry.id = idx;
+ true
+ });
+ Self::assign_partitions_to_members(self.id, &mut members, partitions);
+ members
});
- self.assign_partitions_to_members(&mut members);
- self.members.update(members);
}
- fn mimic_members(&self) -> Slab<Member> {
- let mut container = Slab::with_capacity(self.members.shared_get().len());
- self.with_members(|members| {
- for (_, member) in members {
- Member::new(member.client_id).insert_into(&mut container);
- }
- });
+ fn mimic_members(members: &Slab<Member>) -> Slab<Member> {
+ let mut container = Slab::with_capacity(members.len());
+ for (_, member) in members {
+ Member::new(member.client_id).insert_into(&mut container);
+ }
container
}
- fn assign_partitions_to_members(&self, members: &mut Slab<Member>) {
+ fn assign_partitions_to_members(id: usize, members: &mut Slab<Member>, partitions: Vec<usize>) {
members
.iter_mut()
.for_each(|(_, member)| member.partitions.clear());
let count = members.len();
- for (idx, partition) in self.partitions.iter().enumerate() {
+ for (idx, partition) in partitions.iter().enumerate() {
let position = idx % count;
let member = &mut members[position];
member.partitions.push(*partition);
trace!(
"Assigned partition ID: {} to member with ID: {} in consumer group: {}",
- partition, member.id, self.id
+ partition, member.id, id
);
}
}
@@ -206,7 +213,6 @@ mod tests {
}
#[test_matrix([1, 10, 25, 34, 100, 255, 512, 1024, 4096, 8192, 16384])]
- #[test]
fn test_initial_creation_with_few_members(members_count: usize) {
let consumer_group = create_test_consumer_group("test_group", members_count);
@@ -227,7 +233,6 @@ mod tests {
#[test_matrix([10, 20, 25, 34, 100, 255, 512, 1024, 4096, 8192, 16384],
[1, 2, 3, 4 , 5, 6, 7, 8, 9, 12])]
- #[test]
fn test_consumer_group_partition_assignment(members_count: usize, multiplier: usize) {
let partitions = (0..(members_count * multiplier)).collect::<Vec<_>>();
let mut consumer_group = create_test_consumer_group("test_group", members_count);
@@ -245,7 +250,6 @@ mod tests {
}
#[test_matrix([10, 20, 26, 30, 60, 100], [10, 25, 32, 45, 96, 100, 128, 256, 321], [1, 2, 3, 4, 5, 6, 7, 8, 9])]
- #[test]
fn test_consumer_group_join_members(
mut members_count: usize,
partitions_count: usize,
diff --git a/core/server/src/streaming/topics/topic2.rs b/core/server/src/streaming/topics/topic2.rs
index 738d2d89..f66cfda5 100644
--- a/core/server/src/streaming/topics/topic2.rs
+++ b/core/server/src/streaming/topics/topic2.rs
@@ -5,10 +5,8 @@ use crate::{
consumer_groups::ConsumerGroups,
partitions::{PARTITIONS_CAPACITY, Partitions},
},
- streaming::partitions::partition::ConsumerOffset,
+ streaming::partitions::consumer_offset,
};
-use ahash::AHashMap;
-use dashmap::DashMap;
use iggy_common::{CompressionAlgorithm, IggyExpiry, IggyTimestamp, MaxTopicSize};
use slab::Slab;
use std::sync::Arc;