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/10 18:31:51 UTC
(iggy) branch improved_tpc_partition_ecs updated (ff74233b -> 56c5aca8)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch improved_tpc_partition_ecs
in repository https://gitbox.apache.org/repos/asf/iggy.git
from ff74233b pt1 improved ecs
new 119adf34 temp
new 56c5aca8 improvements pt2
The 2 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.
Summary of changes:
core/server/src/shard/system/consumer_groups.rs | 31 ++++----
core/server/src/shard/system/topics.rs | 6 +-
core/server/src/slab/consumer_groups.rs | 23 ++++--
core/server/src/slab/mod.rs | 90 +---------------------
core/server/src/slab/partitions.rs | 47 +++++------
core/server/src/slab/streams.rs | 73 +++++++-----------
core/server/src/slab/topics.rs | 64 +++++++--------
core/server/src/slab/traits_ext.rs | 85 ++++++++++----------
core/server/src/streaming/streams/stream2.rs | 5 +-
.../server/src/streaming/topics/consumer_group2.rs | 4 +-
core/server/src/streaming/topics/topic2.rs | 4 +-
11 files changed, 159 insertions(+), 273 deletions(-)
(iggy) 02/02: improvements pt2
Posted by gk...@apache.org.
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 56c5aca8c794ce150ddc2b8433a489440e414818
Author: numminex <g....@gmail.com>
AuthorDate: Sun Aug 10 20:31:46 2025 +0200
improvements pt2
---
core/server/src/shard/system/consumer_groups.rs | 31 +++----
core/server/src/shard/system/topics.rs | 6 +-
core/server/src/slab/consumer_groups.rs | 14 +++-
core/server/src/slab/partitions.rs | 38 +++++----
core/server/src/slab/streams.rs | 55 ++++---------
core/server/src/slab/topics.rs | 49 +++++------
core/server/src/slab/traits_ext.rs | 95 +++++++---------------
core/server/src/streaming/streams/stream2.rs | 5 +-
.../server/src/streaming/topics/consumer_group2.rs | 4 +-
core/server/src/streaming/topics/topic2.rs | 4 +-
10 files changed, 124 insertions(+), 177 deletions(-)
diff --git a/core/server/src/shard/system/consumer_groups.rs b/core/server/src/shard/system/consumer_groups.rs
index 79ea516e..959feb49 100644
--- a/core/server/src/shard/system/consumer_groups.rs
+++ b/core/server/src/shard/system/consumer_groups.rs
@@ -262,22 +262,25 @@ impl IggyShard {
let cg = self
.streams2
.with_topic_by_id_mut(stream_id, topic_id, |topic| {
- topic.consumer_groups_mut().with_mut(|container| {
- match group_id.kind {
- iggy_common::IdKind::Numeric => {
+ match group_id.kind {
+ iggy_common::IdKind::Numeric => {
+ topic.consumer_groups_mut().with_mut(|container| {
container.try_remove(group_id.get_u32_value().unwrap() as usize)
- }
- iggy_common::IdKind::String => {
- let key = group_id.get_cow_str_value().unwrap().to_string();
- container.try_remove_by_key(&key)
- }
+ })
}
- .ok_or_else(|| {
- //TODO: Fix the not found erros to accept Identifier instead of u32
- IggyError::ConsumerGroupIdNotFound(0, 0)
- })
- })
- })?;
+ iggy_common::IdKind::String => {
+ let key = group_id.get_string_value().unwrap();
+ let id = topic
+ .consumer_groups()
+ .with_index(|index| *(index.get(&key).unwrap()));
+ topic
+ .consumer_groups_mut()
+ .with_mut(|container| container.try_remove(id))
+ }
+ }
+ // TODO: Fix
+ })
+ .ok_or_else(|| IggyError::ConsumerGroupIdNotFound(0, 0))?;
Ok(cg)
}
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index e7573ed3..1ff9912b 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -459,8 +459,10 @@ impl IggyShard {
// 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);
+ topics.with_mut_index(|index| {
+ // Rename the key inside of hashmap
+ let idx = index.remove(&old_name).expect("Rename key: key not found");
+ index.insert(new_name, idx);
})
});
}
diff --git a/core/server/src/slab/consumer_groups.rs b/core/server/src/slab/consumer_groups.rs
index 6ec1cdb4..fe352696 100644
--- a/core/server/src/slab/consumer_groups.rs
+++ b/core/server/src/slab/consumer_groups.rs
@@ -25,15 +25,22 @@ impl ConsumerGroups {
f(&mut self.container)
}
+ pub fn with_index<T>(
+ &self,
+ f: impl FnOnce(&AHashMap<<consumer_group2::ConsumerGroup as Keyed>::Key, usize>) -> T,
+ ) -> T {
+ f(&self.index)
+ }
+
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.slab.contains(id)
+ self.container.contains(id)
}
iggy_common::IdKind::String => {
let key = id.get_string_value().unwrap();
- self.container.index.contains_key(&key)
+ self.index.contains_key(&key)
}
}
}
@@ -42,7 +49,8 @@ impl ConsumerGroups {
impl Default for ConsumerGroups {
fn default() -> Self {
Self {
- container: IndexedSlab::with_capacity(CAPACITY),
+ index: AHashMap::with_capacity(CAPACITY),
+ container: Slab::with_capacity(CAPACITY),
}
}
}
diff --git a/core/server/src/slab/partitions.rs b/core/server/src/slab/partitions.rs
index c9a6dee2..10bf0f66 100644
--- a/core/server/src/slab/partitions.rs
+++ b/core/server/src/slab/partitions.rs
@@ -1,7 +1,5 @@
use crate::{
- slab::traits_ext::{
- Borrow, Components, ComponentsAsRefs, EntityComponentSystem, IntoComponents,
- },
+ slab::traits_ext::{Borrow, Components, EntityComponentSystem, IntoComponents},
streaming::{
deduplication::message_deduplicator::MessageDeduplicator, partitions::partition2, segments,
stats::stats::PartitionStats,
@@ -24,40 +22,44 @@ pub struct Partitions {
pub struct Part {
partition: partition2::Partition,
- bless: (),
}
impl IntoComponents for Part {
- type Components = (partition2::Partition, ());
+ type Components = (partition2::Partition,);
fn into_components(self) -> Self::Components {
- (self.partition, self.bless)
+ (self.partition,)
}
}
pub struct PartRef<'a> {
- partition: &'a partition2::Partition,
- bless: &'a (),
+ partition: &'a Slab<partition2::Partition>,
}
impl<'a> IntoComponents for PartRef<'a> {
- type Components = (&'a partition2::Partition, &'a ());
+ type Components = (&'a Slab<partition2::Partition>,);
fn into_components(self) -> Self::Components {
- (self.partition, self.bless)
+ (self.partition,)
}
}
-pub struct PartRefMut<'a> {
- partition: &'a mut partition2::Partition,
- bless: &'a mut (),
-}
+impl EntityComponentSystem<Borrow> for Partitions {
+ type Entity = Part;
+ type EntityRef<'a> = PartRef<'a>;
-impl<'a> IntoComponents for PartRefMut<'a> {
- type Components = (&'a mut partition2::Partition, &'a mut ());
+ fn with<O, F>(&self, f: F) -> O
+ where
+ F: for<'a> FnOnce(Components<Self::EntityRef<'a>>) -> O,
+ {
+ todo!()
+ }
- fn into_components(self) -> Self::Components {
- (self.partition, self.bless)
+ async fn with_async<O, F>(&self, f: F) -> O
+ where
+ F: for<'a> FnOnce(Components<Self::EntityRef<'a>>) -> O,
+ {
+ todo!()
}
}
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index d9cff414..1ede79c9 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -31,11 +31,21 @@ impl Streams {
match id.kind {
iggy_common::IdKind::Numeric => {
let id = id.get_u32_value().unwrap() as usize;
- self.container.borrow().slab.contains(id)
+ self.container.borrow().contains(id)
}
iggy_common::IdKind::String => {
let key = id.get_string_value().unwrap();
- self.container.borrow().index.contains_key(&key)
+ self.index.borrow().contains_key(&key)
+ }
+ }
+ }
+
+ fn get_index(&self, id: &Identifier) -> usize {
+ match id.kind {
+ iggy_common::IdKind::Numeric => id.get_u32_value().unwrap() as usize,
+ iggy_common::IdKind::String => {
+ let key = id.get_string_value().unwrap();
+ *self.index.borrow().get(&key).expect("Topic not found")
}
}
}
@@ -82,7 +92,8 @@ impl Streams {
id: &Identifier,
f: impl FnOnce(&stream2::Stream) -> T,
) -> T {
- self.with(|streams| Self::get_stream_ref(id, streams).invoke(f))
+ let id = self.get_index(id);
+ self.with(|streams| streams[id].invoke(f))
}
pub async fn with_stream_by_id_async<T>(
@@ -90,7 +101,8 @@ impl Streams {
id: &Identifier,
f: impl AsyncFnOnce(&stream2::Stream) -> T,
) -> T {
- self.with_async(async |streams| Self::get_stream_ref(id, streams).invoke_async(f).await)
+ let id = self.get_index(id);
+ self.with_async(async |streams| streams[id].invoke_async(f).await)
.await
}
@@ -99,7 +111,8 @@ impl Streams {
id: &Identifier,
f: impl FnOnce(&mut stream2::Stream) -> T,
) -> T {
- self.with_mut(|streams| Self::get_stream_mut(id, streams).invoke_mut(f))
+ let id = self.get_index(id);
+ self.with_mut(|streams| streams[id].invoke_mut(f))
}
pub fn with_topics<T>(&self, stream_id: &Identifier, f: impl FnOnce(&Topics) -> T) -> T {
@@ -188,38 +201,6 @@ impl Streams {
});
}
- fn get_stream_ref<'streams>(
- id: &Identifier,
- slab: &'streams IndexedSlab<stream2::Stream>,
- ) -> &'streams stream2::Stream {
- 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) }
- }
- }
- }
-
- fn get_stream_mut<'streams>(
- id: &Identifier,
- slab: &'streams mut IndexedSlab<stream2::Stream>,
- ) -> &'streams mut stream2::Stream {
- 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 len(&self) -> usize {
self.container.borrow().len()
}
diff --git a/core/server/src/slab/topics.rs b/core/server/src/slab/topics.rs
index b4e5251c..5e89c242 100644
--- a/core/server/src/slab/topics.rs
+++ b/core/server/src/slab/topics.rs
@@ -30,43 +30,21 @@ impl Topics {
match id.kind {
iggy_common::IdKind::Numeric => {
let id = id.get_u32_value().unwrap() as usize;
- self.container.borrow().slab.contains(id)
+ self.container.borrow().contains(id)
}
iggy_common::IdKind::String => {
let key = id.get_string_value().unwrap();
- self.container.borrow().index.contains_key(&key)
+ self.index.borrow().contains_key(&key)
}
}
}
- fn get_topic_ref<'topics>(
- id: &Identifier,
- slab: &'topics Slab<topic2::Topic>,
- ) -> &'topics topic2::Topic {
+ fn get_index(&self, id: &Identifier) -> usize {
match id.kind {
- iggy_common::IdKind::Numeric => {
- let idx = id.get_u32_value().unwrap() as usize;
- &slab[idx]
- }
+ iggy_common::IdKind::Numeric => id.get_u32_value().unwrap() as usize,
iggy_common::IdKind::String => {
let key = id.get_string_value().unwrap();
- unsafe { slab.get_by_key_unchecked(&key) }
- }
- }
- }
-
- fn get_topic_mut<'topics>(
- id: &Identifier,
- slab: &'topics mut Slab<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) }
+ *self.index.borrow().get(&key).expect("Topic not found")
}
}
}
@@ -90,12 +68,21 @@ impl Topics {
f(&mut container)
}
+ pub fn with_mut_index<T>(
+ &self,
+ f: impl FnOnce(&mut AHashMap<<topic2::Topic as Keyed>::Key, usize>) -> T,
+ ) -> T {
+ let mut index = self.index.borrow_mut();
+ f(&mut index)
+ }
+
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)
+ let id = self.get_index(id);
+ self.with_async(async |topics| topics[id].invoke_async(f).await)
.await
}
@@ -109,7 +96,8 @@ impl Topics {
}
pub fn with_topic_by_id<T>(&self, id: &Identifier, f: impl FnOnce(&topic2::Topic) -> T) -> T {
- self.with(|topics| Self::get_topic_ref(id, topics).invoke(f))
+ let id = self.get_index(id);
+ self.with(|topics| topics[id].invoke(f))
}
pub fn with_topic_by_id_mut<T>(
@@ -117,7 +105,8 @@ impl Topics {
id: &Identifier,
f: impl FnOnce(&mut topic2::Topic) -> T,
) -> T {
- self.with_mut(|topics| Self::get_topic_mut(id, topics).invoke_mut(f))
+ let id = self.get_index(id);
+ self.with_mut(|topics| topics[id].invoke_mut(f))
}
pub fn with_partitions(&self, topic_id: &Identifier, f: impl FnOnce(&Partitions)) {
diff --git a/core/server/src/slab/traits_ext.rs b/core/server/src/slab/traits_ext.rs
index 4056116a..15cc161b 100644
--- a/core/server/src/slab/traits_ext.rs
+++ b/core/server/src/slab/traits_ext.rs
@@ -1,5 +1,3 @@
-use std::ops::{Deref, DerefMut};
-
pub trait IntoComponents {
type Components;
fn into_components(self) -> Self::Components;
@@ -13,97 +11,66 @@ mod private {
}
//TODO: Maybe two seperate traits for Ref and RefMut.
-pub trait ComponentsAsRefs<T, C>: private::Sealed {
- type RefMapping<'a>;
- type RefMutMapping<'a>;
+pub trait ComponentsMapping<T>: private::Sealed {
+ type Ref<'a>;
+ type RefMut<'a>;
}
-macro_rules! gen_ref_for {
- ($field:ident) => {
- &'a C
- };
-}
-macro_rules! gen_ref_for_mut {
- ($field:ident) => {
- &'a mut C
- };
-}
-macro_rules! gen_cell_ref_for {
- ($field:ident) => {
- std::cell::Ref<'a, C>
- };
-}
-macro_rules! gen_cell_ref_for_mut {
- ($field:ident) => {
- std::cell::RefMut<'a, C>
- };
-}
macro_rules! impl_components_for_slab_as_refs {
($T:ident) => {
impl<$T> private::Sealed for ($T,) {}
- impl<$T, C> ComponentsAsRefs<Borrow, C> for ($T,)
- where
- for<'a> C: IntoIterator<Item = (usize, ($T,))> + 'a,
- for<'a> $T: 'a,
+ impl<$T> ComponentsMapping<Borrow> for ($T,)
+ where for<'a> $T: 'a
{
- type RefMapping<'a> = &'a C;
- type RefMutMapping<'a> = &'a mut C;
+ type Ref<'a> = (&'a ::slab::Slab<$T>,);
+ type RefMut<'a> = (&'a mut ::slab::Slab<$T>,);
}
- impl<$T, C> ComponentsAsRefs<RefCell, C> for ($T,)
- where
- for<'a> C: IntoIterator<Item = (usize, ($T,))> + 'a,
- for<'a> $T: 'a,
+ impl<$T> ComponentsMapping<RefCell> for ($T,)
+ where for<'a> $T: 'a
{
- type RefMapping<'a> = std::cell::Ref<'a, C>;
- type RefMutMapping<'a> = std::cell::RefMut<'a, C>;
+ type Ref<'a> = (::std::cell::Ref<'a, ::slab::Slab<$T>>,);
+ type RefMut<'a> = (::std::cell::RefMut<'a, ::slab::Slab<$T>>,);
}
};
($T:ident, $($rest:ident),+) => {
impl<$T, $($rest),+> private::Sealed for ($T, $($rest),+) {}
- impl<$T, $($rest),+, C> ComponentsAsRefs<Borrow, C> for ($T, $($rest),+)
+ impl<$T, $($rest),+> ComponentsMapping<Borrow> for ($T, $($rest),+)
where
- for<'a> C: IntoIterator<Item = (usize, ($T,))> + 'a,
- //$(for<'a> C: IntoIterator<Item = (usize, ($rest,))> + 'a),+,
for<'a> $T: 'a,
- $(for<'a> $rest: 'a),+,
+ $(for<'a> $rest: 'a),+
{
- type RefMapping<'a> = (&'a C, $(gen_ref_for!($rest)),+);
- type RefMutMapping<'a> = (&'a mut C, $(gen_ref_for_mut!($rest)),+);
+ type Ref<'a> = (&'a ::slab::Slab<$T>, $(&'a ::slab::Slab<$rest>),+);
+ type RefMut<'a> = (&'a mut ::slab::Slab<$T>, $(&'a mut ::slab::Slab<$rest>),+);
}
- impl<$T, $($rest),+, C> ComponentsAsRefs<RefCell, C> for ($T, $($rest),+)
+ impl<$T, $($rest),+> ComponentsMapping<RefCell> for ($T, $($rest),+)
where
- for<'a> C: IntoIterator<Item = (usize, ($T, $($rest),+))> + 'a,
for<'a> $T: 'a,
- $(for<'a> $rest: 'a),+,
+ $(for<'a> $rest: 'a),+
{
- type RefMapping<'a> = (std::cell::Ref<'a, C>, $(gen_cell_ref_for!($rest)),+);
- type RefMutMapping<'a> = (std::cell::RefMut<'a, C>, $(gen_cell_ref_for_mut!($rest)),+);
+ type Ref<'a> = (std::cell::Ref<'a, ::slab::Slab<$T>>, $(::std::cell::Ref<'a, ::slab::Slab<$rest>>),+);
+ type RefMut<'a> = (std::cell::RefMut<'a, ::slab::Slab<$T>>, $(::std::cell::RefMut<'a, ::slab::Slab<$rest>>),+);
}
-
impl_components_for_slab_as_refs!($($rest),+);
};
}
impl_components_for_slab_as_refs!(T1, T2, T3, T4, T5, T6, T7, T8);
-type Mapping<'a, E, T, C> =
- <<E as IntoComponents>::Components as ComponentsAsRefs<T, C>>::RefMapping<'a>;
-type MappingMut<'a, E, T, C> =
- <<E as IntoComponents>::Components as ComponentsAsRefs<T, C>>::RefMutMapping<'a>;
+type Mapping<'a, E, T> = <<E as IntoComponents>::Components as ComponentsMapping<T>>::Ref<'a>;
+type MappingMut<'a, E, T> = <<E as IntoComponents>::Components as ComponentsMapping<T>>::RefMut<'a>;
pub type Components<T> = <T as IntoComponents>::Components;
-pub trait EntityComponentSystem<T, C>
+pub trait EntityComponentSystem<T>
where
- C: IntoIterator,
- <Self::Entity as IntoComponents>::Components: ComponentsAsRefs<T, C>,
+ <Self::Entity as IntoComponents>::Components: ComponentsMapping<T>,
{
type Entity: IntoComponents;
- type EntityRef<'a>: IntoComponents<Components = Mapping<'a, Self::Entity, T, C>>
+ type EntityRef<'a>: IntoComponents<Components = Mapping<'a, Self::Entity, T>>
where
Self: 'a;
fn with<O, F>(&self, f: F) -> O
@@ -115,11 +82,8 @@ where
F: for<'a> FnOnce(Components<Self::EntityRef<'a>>) -> O;
}
-pub trait EntityComponentSystemMut<C>: EntityComponentSystem<Borrow, C>
-where
- C: IntoIterator,
-{
- type EntityRefMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, Borrow, C>>
+pub trait EntityComponentSystemMut: EntityComponentSystem<Borrow> {
+ type EntityRefMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, Borrow>>
where
Self: 'a;
@@ -128,11 +92,8 @@ where
F: for<'a> FnOnce(Components<Self::EntityRefMut<'a>>) -> O;
}
-pub trait EntityComponentSystemMutCell<C>: EntityComponentSystem<RefCell, C>
-where
- C: IntoIterator,
-{
- type EntityRefMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, RefCell, C>>
+pub trait EntityComponentSystemMutCell: EntityComponentSystem<RefCell> {
+ type EntityRefMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, RefCell>>
where
Self: 'a;
diff --git a/core/server/src/streaming/streams/stream2.rs b/core/server/src/streaming/streams/stream2.rs
index 2dc6743e..10d2b1e5 100644
--- a/core/server/src/streaming/streams/stream2.rs
+++ b/core/server/src/streaming/streams/stream2.rs
@@ -1,7 +1,8 @@
use bytes::{BufMut, BytesMut};
use iggy_common::IggyTimestamp;
+use slab::Slab;
-use crate::slab::{IndexedSlab, Keyed, topics::Topics};
+use crate::slab::{Keyed, topics::Topics};
#[derive(Debug)]
pub struct Stream {
@@ -65,7 +66,7 @@ impl Stream {
self.topics.len()
}
- pub fn insert_into(self, container: &mut IndexedSlab<Self>) -> usize {
+ pub fn insert_into(self, container: &mut Slab<Self>) -> usize {
let idx = container.insert(self);
let stream = &mut container[idx];
stream.id = idx;
diff --git a/core/server/src/streaming/topics/consumer_group2.rs b/core/server/src/streaming/topics/consumer_group2.rs
index f5239c4a..277cb3f9 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::{
binary::handlers::partitions,
- slab::{IndexedSlab, Keyed, partitions::PARTITIONS_CAPACITY},
+ slab::{Keyed, partitions::PARTITIONS_CAPACITY},
};
use ahash::AHashMap;
use arcshift::ArcShift;
@@ -39,7 +39,7 @@ impl ConsumerGroup {
self.id
}
- pub fn insert_into(self, container: &mut IndexedSlab<Self>) -> usize {
+ pub fn insert_into(self, container: &mut Slab<Self>) -> usize {
let idx = container.insert(self);
let group = &mut container[idx];
group.id = idx;
diff --git a/core/server/src/streaming/topics/topic2.rs b/core/server/src/streaming/topics/topic2.rs
index fa8ab8b0..d13fada1 100644
--- a/core/server/src/streaming/topics/topic2.rs
+++ b/core/server/src/streaming/topics/topic2.rs
@@ -1,7 +1,7 @@
use crate::streaming::stats::stats::PartitionStats;
use crate::{
slab::{
- IndexedSlab, Keyed,
+ Keyed,
consumer_groups::ConsumerGroups,
partitions::{PARTITIONS_CAPACITY, Partitions},
},
@@ -137,7 +137,7 @@ impl Topic {
&mut self.consumer_group_offsets
}
- pub fn insert_into(self, container: &mut IndexedSlab<Self>) -> usize {
+ pub fn insert_into(self, container: &mut Slab<Self>) -> usize {
let idx = container.insert(self);
let topic = &mut container[idx];
topic.id = idx;
(iggy) 01/02: temp
Posted by gk...@apache.org.
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 119adf343cfbfcfafccb2ee0d5a4312f42195546
Author: numminex <g....@gmail.com>
AuthorDate: Sun Aug 10 17:40:41 2025 +0200
temp
---
core/server/src/slab/consumer_groups.rs | 9 ++-
core/server/src/slab/mod.rs | 90 +---------------------
core/server/src/slab/partitions.rs | 33 +++-----
core/server/src/slab/streams.rs | 18 ++---
core/server/src/slab/topics.rs | 19 +++--
core/server/src/slab/traits_ext.rs | 132 ++++++++++++++++++++------------
6 files changed, 120 insertions(+), 181 deletions(-)
diff --git a/core/server/src/slab/consumer_groups.rs b/core/server/src/slab/consumer_groups.rs
index 75429792..6ec1cdb4 100644
--- a/core/server/src/slab/consumer_groups.rs
+++ b/core/server/src/slab/consumer_groups.rs
@@ -1,4 +1,4 @@
-use crate::{slab::IndexedSlab, streaming::topics::consumer_group2};
+use crate::{slab::Keyed, streaming::topics::consumer_group2};
use ahash::AHashMap;
use arcshift::ArcShift;
use iggy_common::Identifier;
@@ -9,17 +9,18 @@ const CAPACITY: usize = 1024;
#[derive(Debug)]
pub struct ConsumerGroups {
- container: IndexedSlab<consumer_group2::ConsumerGroup>,
+ index: AHashMap<<consumer_group2::ConsumerGroup as Keyed>::Key, usize>,
+ container: Slab<consumer_group2::ConsumerGroup>,
}
impl ConsumerGroups {
- pub fn with<T>(&self, f: impl FnOnce(&IndexedSlab<consumer_group2::ConsumerGroup>) -> T) -> T {
+ pub fn with<T>(&self, f: impl FnOnce(&Slab<consumer_group2::ConsumerGroup>) -> T) -> T {
f(&self.container)
}
pub fn with_mut<T>(
&mut self,
- f: impl FnOnce(&mut IndexedSlab<consumer_group2::ConsumerGroup>) -> T,
+ f: impl FnOnce(&mut Slab<consumer_group2::ConsumerGroup>) -> T,
) -> T {
f(&mut self.container)
}
diff --git a/core/server/src/slab/mod.rs b/core/server/src/slab/mod.rs
index 006205cc..840de3d3 100644
--- a/core/server/src/slab/mod.rs
+++ b/core/server/src/slab/mod.rs
@@ -16,92 +16,4 @@ 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: Keyed + Debug> IndexedSlab<T> {
- pub fn with_capacity(capacity: usize) -> Self {
- Self {
- slab: Slab::with_capacity(capacity),
- index: AHashMap::with_capacity(capacity),
- }
- }
-
- pub fn len(&self) -> usize {
- self.slab.len()
- }
-
- pub fn insert(&mut self, value: T) -> usize {
- let key = value.key().clone();
- let index = self.slab.insert(value);
- self.index.insert(key, index);
- 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)
- }
-
- pub fn get_mut(&mut self, index: usize) -> Option<&mut T> {
- self.slab.get_mut(index)
- }
-
- 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) }
- }
-
- unsafe fn get_by_key_mut_unchecked(&mut self, key: &T::Key) -> &mut T {
- let index = self.index.get(key).unwrap();
- unsafe { self.slab.get_unchecked_mut(*index) }
- }
-}
-
-impl<T: Keyed> Index<usize> for IndexedSlab<T> {
- type Output = T;
-
- fn index(&self, index: usize) -> &Self::Output {
- // Safety: The access pattern for resources that we store in the `IndexedSlab` is similar to one of a `Vec`,
- // we expect the index to be valid, and we ensure that the index is always valid
- unsafe { self.slab.get_unchecked(index) }
- }
-}
-
-impl<T: Keyed> IndexMut<usize> for IndexedSlab<T> {
- fn index_mut(&mut self, index: usize) -> &mut Self::Output {
- // Safety: The access pattern for resources that we store in the `IndexedSlab` is similar to one of a `Vec`,
- // we expect the index to be valid, and we ensure that the index is always valid
- unsafe { self.slab.get_unchecked_mut(index) }
- }
-}
+//index: AHashMap<T::Key, usize>,
diff --git a/core/server/src/slab/partitions.rs b/core/server/src/slab/partitions.rs
index fb0a1af1..c9a6dee2 100644
--- a/core/server/src/slab/partitions.rs
+++ b/core/server/src/slab/partitions.rs
@@ -1,7 +1,12 @@
-use crate::{slab::traits_ext::{Borrow, ComponentsAsRefs, EntityComponentSystem, IntoComponents}, streaming::{
- deduplication::message_deduplicator::MessageDeduplicator, partitions::partition2, segments,
- stats::stats::PartitionStats,
-}};
+use crate::{
+ slab::traits_ext::{
+ Borrow, Components, ComponentsAsRefs, EntityComponentSystem, IntoComponents,
+ },
+ streaming::{
+ deduplication::message_deduplicator::MessageDeduplicator, partitions::partition2, segments,
+ stats::stats::PartitionStats,
+ },
+};
use slab::Slab;
use std::sync::{Arc, atomic::AtomicU64};
@@ -49,31 +54,13 @@ pub struct PartRefMut<'a> {
}
impl<'a> IntoComponents for PartRefMut<'a> {
- type Components = (&'a mut partition2::Partition, &'a mut () );
+ type Components = (&'a mut partition2::Partition, &'a mut ());
fn into_components(self) -> Self::Components {
(self.partition, self.bless)
}
}
-impl EntityComponentSystem<Borrow> for Partitions {
- type Entity = Part;
- type EntityRef<'a> = PartRef<'a>;
-
- fn with_new<T, F>(&self, f: F) -> T
- where
- F: FnOnce(Self::EntityRef<'_>) -> T {
- todo!()
- }
-
- async fn with_async_new<T, F>(&self, f: F) -> T
- where
- F: FnOnce(Self::EntityRef<'_>) -> T {
- todo!()
- }
-}
-
-
impl Default for Partitions {
fn default() -> Self {
Self {
diff --git a/core/server/src/slab/streams.rs b/core/server/src/slab/streams.rs
index 67112e12..d9cff414 100644
--- a/core/server/src/slab/streams.rs
+++ b/core/server/src/slab/streams.rs
@@ -1,9 +1,10 @@
+use ahash::AHashMap;
use iggy_common::Identifier;
use slab::Slab;
use std::{cell::RefCell, sync::Arc};
use crate::{
- slab::{IndexedSlab, partitions::Partitions, topics::Topics},
+ slab::{Keyed, partitions::Partitions, topics::Topics},
streaming::{
partitions::partition2, stats::stats::StreamStats, streams::stream2, topics::topic2,
},
@@ -12,14 +13,16 @@ use crate::{
const CAPACITY: usize = 1024;
pub struct Streams {
- container: RefCell<IndexedSlab<stream2::Stream>>,
+ index: RefCell<AHashMap<<stream2::Stream as Keyed>::Key, usize>>,
+ container: RefCell<Slab<stream2::Stream>>,
stats: RefCell<Slab<Arc<StreamStats>>>,
}
impl Streams {
pub fn init() -> Self {
Self {
- container: RefCell::new(IndexedSlab::with_capacity(CAPACITY)),
+ index: RefCell::new(AHashMap::with_capacity(CAPACITY)),
+ container: RefCell::new(Slab::with_capacity(CAPACITY)),
stats: RefCell::new(Slab::with_capacity(CAPACITY)),
}
}
@@ -59,20 +62,17 @@ impl Streams {
f(&mut stats)
}
- pub async fn with_async<T>(
- &self,
- f: impl AsyncFnOnce(&IndexedSlab<stream2::Stream>) -> T,
- ) -> T {
+ pub async fn with_async<T>(&self, f: impl AsyncFnOnce(&Slab<stream2::Stream>) -> T) -> T {
let container = self.container.borrow();
f(&container).await
}
- pub fn with<T>(&self, f: impl FnOnce(&IndexedSlab<stream2::Stream>) -> T) -> T {
+ pub fn with<T>(&self, f: impl FnOnce(&Slab<stream2::Stream>) -> T) -> T {
let container = self.container.borrow();
f(&container)
}
- pub fn with_mut<T>(&self, f: impl FnOnce(&mut IndexedSlab<stream2::Stream>) -> T) -> T {
+ pub fn with_mut<T>(&self, f: impl FnOnce(&mut Slab<stream2::Stream>) -> T) -> T {
let mut container = self.container.borrow_mut();
f(&mut container)
}
diff --git a/core/server/src/slab/topics.rs b/core/server/src/slab/topics.rs
index 675600f3..b4e5251c 100644
--- a/core/server/src/slab/topics.rs
+++ b/core/server/src/slab/topics.rs
@@ -1,9 +1,10 @@
+use ahash::AHashMap;
use iggy_common::Identifier;
use slab::Slab;
use std::{cell::RefCell, sync::Arc};
use crate::{
- slab::{IndexedSlab, Keyed, partitions::Partitions},
+ slab::{Keyed, partitions::Partitions},
streaming::{partitions::partition2, stats::stats::TopicStats, topics::topic2},
};
@@ -11,14 +12,16 @@ const CAPACITY: usize = 1024;
#[derive(Debug)]
pub struct Topics {
- container: RefCell<IndexedSlab<topic2::Topic>>,
+ index: RefCell<AHashMap<<topic2::Topic as Keyed>::Key, usize>>,
+ container: RefCell<Slab<topic2::Topic>>,
stats: RefCell<Slab<Arc<TopicStats>>>,
}
impl Topics {
pub fn init() -> Self {
Self {
- container: RefCell::new(IndexedSlab::with_capacity(CAPACITY)),
+ index: RefCell::new(AHashMap::with_capacity(CAPACITY)),
+ container: RefCell::new(Slab::with_capacity(CAPACITY)),
stats: RefCell::new(Slab::with_capacity(CAPACITY)),
}
}
@@ -38,7 +41,7 @@ impl Topics {
fn get_topic_ref<'topics>(
id: &Identifier,
- slab: &'topics IndexedSlab<topic2::Topic>,
+ slab: &'topics Slab<topic2::Topic>,
) -> &'topics topic2::Topic {
match id.kind {
iggy_common::IdKind::Numeric => {
@@ -54,7 +57,7 @@ impl Topics {
fn get_topic_mut<'topics>(
id: &Identifier,
- slab: &'topics mut IndexedSlab<topic2::Topic>,
+ slab: &'topics mut Slab<topic2::Topic>,
) -> &'topics mut topic2::Topic {
match id.kind {
iggy_common::IdKind::Numeric => {
@@ -72,17 +75,17 @@ impl Topics {
self.container.borrow().len()
}
- pub async fn with_async<T>(&self, f: impl AsyncFnOnce(&IndexedSlab<topic2::Topic>) -> T) -> T {
+ pub async fn with_async<T>(&self, f: impl AsyncFnOnce(&Slab<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 {
+ pub fn with<T>(&self, f: impl FnOnce(&Slab<topic2::Topic>) -> T) -> T {
let container = self.container.borrow();
f(&container)
}
- pub fn with_mut<T>(&self, f: impl FnOnce(&mut IndexedSlab<topic2::Topic>) -> T) -> T {
+ pub fn with_mut<T>(&self, f: impl FnOnce(&mut Slab<topic2::Topic>) -> T) -> T {
let mut container = self.container.borrow_mut();
f(&mut container)
}
diff --git a/core/server/src/slab/traits_ext.rs b/core/server/src/slab/traits_ext.rs
index 79abce3f..4056116a 100644
--- a/core/server/src/slab/traits_ext.rs
+++ b/core/server/src/slab/traits_ext.rs
@@ -1,3 +1,5 @@
+use std::ops::{Deref, DerefMut};
+
pub trait IntoComponents {
type Components;
fn into_components(self) -> Self::Components;
@@ -10,97 +12,131 @@ mod private {
pub trait Sealed {}
}
-pub trait ComponentsAsRefs<RefType>: private::Sealed {
+//TODO: Maybe two seperate traits for Ref and RefMut.
+pub trait ComponentsAsRefs<T, C>: private::Sealed {
type RefMapping<'a>;
type RefMutMapping<'a>;
}
-macro_rules! impl_components_as_refs {
+
+macro_rules! gen_ref_for {
+ ($field:ident) => {
+ &'a C
+ };
+}
+macro_rules! gen_ref_for_mut {
+ ($field:ident) => {
+ &'a mut C
+ };
+}
+macro_rules! gen_cell_ref_for {
+ ($field:ident) => {
+ std::cell::Ref<'a, C>
+ };
+}
+macro_rules! gen_cell_ref_for_mut {
+ ($field:ident) => {
+ std::cell::RefMut<'a, C>
+ };
+}
+macro_rules! impl_components_for_slab_as_refs {
($T:ident) => {
impl<$T> private::Sealed for ($T,) {}
-
- impl<$T> ComponentsAsRefs<Borrow> for ($T,)
- where for<'a> $T: 'a
+
+ impl<$T, C> ComponentsAsRefs<Borrow, C> for ($T,)
+ where
+ for<'a> C: IntoIterator<Item = (usize, ($T,))> + 'a,
+ for<'a> $T: 'a,
{
- type RefMapping<'a> = (&'a $T,);
- type RefMutMapping<'a> = (&'a mut $T,);
+ type RefMapping<'a> = &'a C;
+ type RefMutMapping<'a> = &'a mut C;
}
- impl<$T> ComponentsAsRefs<RefCell> for ($T,)
- where for<'a> $T: 'a
+ impl<$T, C> ComponentsAsRefs<RefCell, C> for ($T,)
+ where
+ for<'a> C: IntoIterator<Item = (usize, ($T,))> + 'a,
+ for<'a> $T: 'a,
{
- type RefMapping<'a> = (std::cell::Ref<'a, $T>,);
- type RefMutMapping<'a> = (std::cell::RefMut<'a, $T>,);
+ type RefMapping<'a> = std::cell::Ref<'a, C>;
+ type RefMutMapping<'a> = std::cell::RefMut<'a, C>;
}
};
($T:ident, $($rest:ident),+) => {
impl<$T, $($rest),+> private::Sealed for ($T, $($rest),+) {}
-
- impl<$T, $($rest),+> ComponentsAsRefs<Borrow> for ($T, $($rest),+)
+
+ impl<$T, $($rest),+, C> ComponentsAsRefs<Borrow, C> for ($T, $($rest),+)
where
+ for<'a> C: IntoIterator<Item = (usize, ($T,))> + 'a,
+ //$(for<'a> C: IntoIterator<Item = (usize, ($rest,))> + 'a),+,
for<'a> $T: 'a,
- $(for<'a> $rest: 'a),+
+ $(for<'a> $rest: 'a),+,
{
- type RefMapping<'a> = (&'a $T, $(&'a $rest),+);
- type RefMutMapping<'a> = (&'a mut $T, $(&'a mut $rest),+);
+ type RefMapping<'a> = (&'a C, $(gen_ref_for!($rest)),+);
+ type RefMutMapping<'a> = (&'a mut C, $(gen_ref_for_mut!($rest)),+);
}
- impl<$T, $($rest),+> ComponentsAsRefs<RefCell> for ($T, $($rest),+)
+ impl<$T, $($rest),+, C> ComponentsAsRefs<RefCell, C> for ($T, $($rest),+)
where
+ for<'a> C: IntoIterator<Item = (usize, ($T, $($rest),+))> + 'a,
for<'a> $T: 'a,
- $(for<'a> $rest: 'a),+
+ $(for<'a> $rest: 'a),+,
{
- type RefMapping<'a> = (std::cell::Ref<'a, $T>, $(std::cell::Ref<'a, $rest>),+);
- type RefMutMapping<'a> = (std::cell::RefMut<'a, $T>, $(std::cell::RefMut<'a, $rest>),+);
+ type RefMapping<'a> = (std::cell::Ref<'a, C>, $(gen_cell_ref_for!($rest)),+);
+ type RefMutMapping<'a> = (std::cell::RefMut<'a, C>, $(gen_cell_ref_for_mut!($rest)),+);
}
- impl_components_as_refs!($($rest),+);
+
+ impl_components_for_slab_as_refs!($($rest),+);
};
}
-impl_components_as_refs!(T1, T2, T3, T4, T5, T6, T7, T8);
+impl_components_for_slab_as_refs!(T1, T2, T3, T4, T5, T6, T7, T8);
-type ComponentsTuple<'a, T, RefType> =
- <<T as IntoComponents>::Components as ComponentsAsRefs<RefType>>::RefMapping<'a>;
-type ComponentsTupleMut<'a, T, RefType> =
- <<T as IntoComponents>::Components as ComponentsAsRefs<RefType>>::RefMutMapping<'a>;
+type Mapping<'a, E, T, C> =
+ <<E as IntoComponents>::Components as ComponentsAsRefs<T, C>>::RefMapping<'a>;
+type MappingMut<'a, E, T, C> =
+ <<E as IntoComponents>::Components as ComponentsAsRefs<T, C>>::RefMutMapping<'a>;
-// TODO: extend it with a method `with_by_id`, this one actually will use the `EntityRef`
-// those `with` methods need to work on `impl Iterator<Item = ComponentsAsRefs individual components>`
-pub trait EntityComponentSystem<RefType> {
+pub type Components<T> = <T as IntoComponents>::Components;
+
+pub trait EntityComponentSystem<T, C>
+where
+ C: IntoIterator,
+ <Self::Entity as IntoComponents>::Components: ComponentsAsRefs<T, C>,
+{
type Entity: IntoComponents;
- type EntityRef<'a>: IntoComponents<Components = ComponentsTuple<'a, Self::Entity, RefType>>
+ type EntityRef<'a>: IntoComponents<Components = Mapping<'a, Self::Entity, T, C>>
where
- <Self::Entity as IntoComponents>::Components: ComponentsAsRefs<RefType>,
Self: 'a;
-
- fn with_new<T, F>(&self, f: F) -> T
+ fn with<O, F>(&self, f: F) -> O
where
- F: FnOnce(Self::EntityRef<'_>) -> T;
+ F: for<'a> FnOnce(Components<Self::EntityRef<'a>>) -> O;
- fn with_async_new<T, F>(&self, f: F) -> impl Future<Output = T>
+ fn with_async<O, F>(&self, f: F) -> impl Future<Output = O>
where
- F: FnOnce(Self::EntityRef<'_>) -> T;
+ F: for<'a> FnOnce(Components<Self::EntityRef<'a>>) -> O;
}
-pub trait EntityComponentSystemMut: EntityComponentSystem<Borrow> {
- type EntityRefMut<'a>: IntoComponents<Components = ComponentsTupleMut<'a, Self::Entity, Borrow>>
+pub trait EntityComponentSystemMut<C>: EntityComponentSystem<Borrow, C>
+where
+ C: IntoIterator,
+{
+ type EntityRefMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, Borrow, C>>
where
- <Self::Entity as IntoComponents>::Components: ComponentsAsRefs<Borrow>,
Self: 'a;
- fn with_mut_new<T, F>(&mut self, f: F) -> T
+ fn with_mut<O, F>(&mut self, f: F) -> O
where
- F: FnOnce(Self::EntityRefMut<'_>) -> T;
+ F: for<'a> FnOnce(Components<Self::EntityRefMut<'a>>) -> O;
}
-pub trait EntityComponentSystemMutCell: EntityComponentSystem<RefCell> {
- type EntityRefMut<'a>: IntoComponents<
- Components = ComponentsTupleMut<'a, Self::Entity, RefCell>,
- >
+pub trait EntityComponentSystemMutCell<C>: EntityComponentSystem<RefCell, C>
+where
+ C: IntoIterator,
+{
+ type EntityRefMut<'a>: IntoComponents<Components = MappingMut<'a, Self::Entity, RefCell, C>>
where
- <Self::Entity as IntoComponents>::Components: ComponentsAsRefs<RefCell>,
Self: 'a;
- fn with_mut_new<T, F>(&mut self, f: F) -> T
+ fn with_mut<O, F>(&mut self, f: F) -> O
where
- F: FnOnce(Self::EntityRefMut<'_>) -> T;
+ F: for<'a> FnOnce(Components<Self::EntityRefMut<'a>>) -> O;
}