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/09/05 09:17:23 UTC
(iggy) 01/01: feat(io_uring): replace shards_table with dashmap
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch replace_shards_table_with_dashmap
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 959b9a864a91968bda2574dcba914332b63442c7
Author: numminex <g....@gmail.com>
AuthorDate: Fri Sep 5 11:17:11 2025 +0200
feat(io_uring): replace shards_table with dashmap
---
.../bench/src/actors/consumer/client/high_level.rs | 2 +-
core/bench/src/actors/consumer/client/low_level.rs | 2 +-
core/bench/src/actors/producer/client/low_level.rs | 2 +-
.../handlers/messages/poll_messages_handler.rs | 1 -
.../handlers/messages/send_messages_handler.rs | 1 -
.../partitions/create_partitions_handler.rs | 2 +-
.../partitions/delete_partitions_handler.rs | 1 -
core/server/src/http/consumer_groups.rs | 129 +++++++++++++--------
core/server/src/http/http_shard_wrapper.rs | 10 +-
core/server/src/http/mapper.rs | 39 ++++---
core/server/src/http/partitions.rs | 15 ++-
core/server/src/http/streams.rs | 105 ++++++++++-------
core/server/src/http/topics.rs | 58 +++++----
core/server/src/main.rs | 17 ++-
core/server/src/shard/builder.rs | 20 +++-
core/server/src/shard/mod.rs | 72 +++++++-----
core/server/src/shard/namespace.rs | 63 +++++++---
core/server/src/shard/system/messages.rs | 1 -
core/server/src/shard/system/partitions.rs | 21 +++-
core/server/src/shard/transmission/event.rs | 21 ++--
core/server/src/streaming/partitions/storage2.rs | 25 +++-
core/server/src/streaming/streams/messages.rs | 5 +-
core/server/src/streaming/utils/mod.rs | 1 +
core/server/src/streaming/utils/ptr.rs | 53 +++++++++
24 files changed, 442 insertions(+), 224 deletions(-)
diff --git a/core/bench/src/actors/consumer/client/high_level.rs b/core/bench/src/actors/consumer/client/high_level.rs
index 235eacfa..e34f7b12 100644
--- a/core/bench/src/actors/consumer/client/high_level.rs
+++ b/core/bench/src/actors/consumer/client/high_level.rs
@@ -126,7 +126,7 @@ impl BenchmarkInit for HighLevelConsumerClient {
&format!("hl_consumer_{}", self.config.consumer_id),
&stream_id_str,
&topic_id_str,
- 1,
+ 0,
)?
.polling_strategy(PollingStrategy::offset(0))
.batch_length(self.config.messages_per_batch.get())
diff --git a/core/bench/src/actors/consumer/client/low_level.rs b/core/bench/src/actors/consumer/client/low_level.rs
index 5a809f1a..139dae24 100644
--- a/core/bench/src/actors/consumer/client/low_level.rs
+++ b/core/bench/src/actors/consumer/client/low_level.rs
@@ -114,7 +114,7 @@ impl ConsumerClient for LowLevelConsumerClient {
impl BenchmarkInit for LowLevelConsumerClient {
async fn setup(&mut self) -> Result<(), IggyError> {
let topic_id_str = "topic-1";
- let default_partition_id = 1u32;
+ let default_partition_id = 0u32;
let client = self.client_factory.create_client().await;
let client = IggyClient::create(client, None, None);
diff --git a/core/bench/src/actors/producer/client/low_level.rs b/core/bench/src/actors/producer/client/low_level.rs
index c0ea4b4c..86d02264 100644
--- a/core/bench/src/actors/producer/client/low_level.rs
+++ b/core/bench/src/actors/producer/client/low_level.rs
@@ -87,7 +87,7 @@ impl ProducerClient for LowLevelProducerClient {
impl BenchmarkInit for LowLevelProducerClient {
async fn setup(&mut self) -> Result<(), IggyError> {
- let default_partition_id = 1;
+ let default_partition_id = 0u32;
let partitions = self.config.partitions;
let client = self.client_factory.create_client().await;
diff --git a/core/server/src/binary/handlers/messages/poll_messages_handler.rs b/core/server/src/binary/handlers/messages/poll_messages_handler.rs
index 2f02a039..37aaac20 100644
--- a/core/server/src/binary/handlers/messages/poll_messages_handler.rs
+++ b/core/server/src/binary/handlers/messages/poll_messages_handler.rs
@@ -21,7 +21,6 @@ use crate::binary::handlers::messages::COMPONENT;
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::sender::SenderKind;
use crate::shard::IggyShard;
-use crate::shard::namespace::IggyNamespace;
use crate::shard::system::messages::PollingArgs;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardMessage, ShardRequest, ShardRequestPayload};
diff --git a/core/server/src/binary/handlers/messages/send_messages_handler.rs b/core/server/src/binary/handlers/messages/send_messages_handler.rs
index bb536c61..6b998c31 100644
--- a/core/server/src/binary/handlers/messages/send_messages_handler.rs
+++ b/core/server/src/binary/handlers/messages/send_messages_handler.rs
@@ -20,7 +20,6 @@ use super::COMPONENT;
use crate::binary::command::{BinaryServerCommand, ServerCommandHandler};
use crate::binary::sender::SenderKind;
use crate::shard::IggyShard;
-use crate::shard::namespace::IggyNamespace;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{ShardMessage, ShardRequest, ShardRequestPayload};
use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut};
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 3522d5bb..721f063f 100644
--- a/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
+++ b/core/server/src/binary/handlers/partitions/create_partitions_handler.rs
@@ -21,6 +21,7 @@ use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::{handlers::partitions::COMPONENT, sender::SenderKind};
use crate::shard::IggyShard;
use crate::shard::transmission::event::ShardEvent;
+use crate::slab::traits_ext::EntityMarker;
use crate::state::command::EntryCommand;
use crate::streaming::session::Session;
use crate::streaming::{streams, topics};
@@ -59,7 +60,6 @@ impl ServerCommandHandler for CreatePartitions {
partitions,
};
let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
- // TODO: Create shard table records.
// TODO: Rebalance the consumer group.
let stream_id = shard
diff --git a/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs b/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs
index 425a06f5..c88a35b8 100644
--- a/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs
+++ b/core/server/src/binary/handlers/partitions/delete_partitions_handler.rs
@@ -64,7 +64,6 @@ impl ServerCommandHandler for DeletePartitions {
partition_ids: deleted_partition_ids,
};
let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
- // TODO: Delete shard table records.
// TODO: Rebalance the consumer group.
shard
diff --git a/core/server/src/http/consumer_groups.rs b/core/server/src/http/consumer_groups.rs
index 65cf5cab..6c5a49af 100644
--- a/core/server/src/http/consumer_groups.rs
+++ b/core/server/src/http/consumer_groups.rs
@@ -61,23 +61,30 @@ async fn get_consumer_group(
let identifier_stream_id = Identifier::from_str_value(&stream_id)?;
let identifier_topic_id = Identifier::from_str_value(&topic_id)?;
let identifier_group_id = Identifier::from_str_value(&group_id)?;
-
+
let session = Session::stateless(identity.user_id, identity.ip_address);
// Check permissions and existence
state.shard.shard().ensure_authenticated(&session)?;
- state.shard.shard().ensure_consumer_group_exists(&identifier_stream_id, &identifier_topic_id, &identifier_group_id)?;
-
+ state.shard.shard().ensure_consumer_group_exists(
+ &identifier_stream_id,
+ &identifier_topic_id,
+ &identifier_group_id,
+ )?;
+
let numeric_topic_id = state.shard.shard().streams2.with_topic_by_id(
&identifier_stream_id,
&identifier_topic_id,
crate::streaming::topics::helpers::get_topic_id(),
);
- let numeric_stream_id = state.shard.shard()
- .streams2
- .with_stream_by_id(&identifier_stream_id, crate::streaming::streams::helpers::get_stream_id());
-
- state.shard.shard()
+ let numeric_stream_id = state.shard.shard().streams2.with_stream_by_id(
+ &identifier_stream_id,
+ crate::streaming::streams::helpers::get_stream_id(),
+ );
+
+ state
+ .shard
+ .shard()
.permissioner
.borrow()
.get_consumer_group(
@@ -87,16 +94,18 @@ async fn get_consumer_group(
)?;
let consumer_group = {
- let future = SendWrapper::new(state.shard.shard()
- .streams2
- .with_consumer_group_by_id_async(
- &identifier_stream_id,
- &identifier_topic_id,
- &identifier_group_id,
- async |(root, members)| {
- mapper::map_consumer_group(root, members)
- },
- ));
+ let future = SendWrapper::new(
+ state
+ .shard
+ .shard()
+ .streams2
+ .with_consumer_group_by_id_async(
+ &identifier_stream_id,
+ &identifier_topic_id,
+ &identifier_group_id,
+ async |(root, members)| mapper::map_consumer_group(root, members),
+ ),
+ );
future.await
};
@@ -115,18 +124,24 @@ async fn get_consumer_groups(
// Check permissions and existence
state.shard.shard().ensure_authenticated(&session)?;
- state.shard.shard().ensure_topic_exists(&identifier_stream_id, &identifier_topic_id)?;
-
+ state
+ .shard
+ .shard()
+ .ensure_topic_exists(&identifier_stream_id, &identifier_topic_id)?;
+
let numeric_topic_id = state.shard.shard().streams2.with_topic_by_id(
&identifier_stream_id,
&identifier_topic_id,
crate::streaming::topics::helpers::get_topic_id(),
);
- let numeric_stream_id = state.shard.shard()
- .streams2
- .with_stream_by_id(&identifier_stream_id, crate::streaming::streams::helpers::get_stream_id());
-
- state.shard.shard()
+ let numeric_stream_id = state.shard.shard().streams2.with_stream_by_id(
+ &identifier_stream_id,
+ crate::streaming::streams::helpers::get_stream_id(),
+ );
+
+ state
+ .shard
+ .shard()
.permissioner
.borrow()
.get_consumer_groups(
@@ -136,15 +151,17 @@ async fn get_consumer_groups(
)?;
let consumer_groups = {
- let future = SendWrapper::new(state.shard.shard()
- .streams2
- .with_consumer_groups_async(&identifier_stream_id, &identifier_topic_id, async |cgs| {
+ let future = SendWrapper::new(state.shard.shard().streams2.with_consumer_groups_async(
+ &identifier_stream_id,
+ &identifier_topic_id,
+ async |cgs| {
cgs.with_components_async(async |cgs| {
let (roots, members) = cgs.into_components();
mapper::map_consumer_groups(roots, members)
})
.await
- }));
+ },
+ ));
future.await
};
@@ -175,17 +192,21 @@ async fn create_consumer_group(
.with_error_context(|error| format!("{COMPONENT} (error: {error}) - failed to create consumer group, stream ID: {}, topic ID: {}, name: {}", stream_id, topic_id, command.name))?;
let group_id = consumer_group.id();
-
+
// Send event for consumer group creation
{
let broadcast_future = SendWrapper::new(async {
use crate::shard::transmission::event::ShardEvent;
- let event = ShardEvent::CreatedConsumerGroup2 {
- stream_id: command.stream_id.clone(),
+ let event = ShardEvent::CreatedConsumerGroup2 {
+ stream_id: command.stream_id.clone(),
topic_id: command.topic_id.clone(),
- cg: consumer_group.clone()
+ cg: consumer_group.clone(),
};
- let _responses = state.shard.shard().broadcast_event_to_all_shards(event).await;
+ let _responses = state
+ .shard
+ .shard()
+ .broadcast_event_to_all_shards(event)
+ .await;
});
broadcast_future.await;
}
@@ -193,23 +214,25 @@ async fn create_consumer_group(
// Get the created consumer group details
let group_id_identifier = Identifier::numeric(group_id as u32).unwrap();
let consumer_group_details = {
- let future = SendWrapper::new(state.shard.shard()
- .streams2
- .with_consumer_group_by_id_async(
- &command.stream_id,
- &command.topic_id,
- &group_id_identifier,
- async |(root, members)| {
- mapper::map_consumer_group(root, members)
- },
- ));
+ let future = SendWrapper::new(
+ state
+ .shard
+ .shard()
+ .streams2
+ .with_consumer_group_by_id_async(
+ &command.stream_id,
+ &command.topic_id,
+ &group_id_identifier,
+ async |(root, members)| mapper::map_consumer_group(root, members),
+ ),
+ );
future.await
};
// Apply state change
- let entry_command = EntryCommand::CreateConsumerGroup(CreateConsumerGroupWithId {
- group_id: group_id as u32,
- command
+ let entry_command = EntryCommand::CreateConsumerGroup(CreateConsumerGroupWithId {
+ group_id: group_id as u32,
+ command,
});
let state_future = SendWrapper::new(
state
@@ -236,7 +259,7 @@ async fn delete_consumer_group(
let identifier_group_id = Identifier::from_str_value(&group_id)?;
let session = Session::stateless(identity.user_id, identity.ip_address);
-
+
// Delete using the new API
let consumer_group = state.shard.shard().delete_consumer_group2(
&session,
@@ -245,9 +268,9 @@ async fn delete_consumer_group(
&identifier_group_id
)
.with_error_context(|error| format!("{COMPONENT} (error: {error}) - failed to delete consumer group with ID: {group_id} for topic with ID: {topic_id} in stream with ID: {stream_id}"))?;
-
+
let cg_id = consumer_group.id();
-
+
// Send event for consumer group deletion
{
let broadcast_future = SendWrapper::new(async {
@@ -258,7 +281,11 @@ async fn delete_consumer_group(
topic_id: identifier_topic_id.clone(),
group_id: identifier_group_id.clone(),
};
- let _responses = state.shard.shard().broadcast_event_to_all_shards(event).await;
+ let _responses = state
+ .shard
+ .shard()
+ .broadcast_event_to_all_shards(event)
+ .await;
});
broadcast_future.await;
}
diff --git a/core/server/src/http/http_shard_wrapper.rs b/core/server/src/http/http_shard_wrapper.rs
index da1accc4..e6c29276 100644
--- a/core/server/src/http/http_shard_wrapper.rs
+++ b/core/server/src/http/http_shard_wrapper.rs
@@ -1,8 +1,8 @@
use std::rc::Rc;
use iggy_common::{
- Consumer, ConsumerOffsetInfo, Identifier, IggyError, IggyExpiry,
- Partitioning, Permissions, Stats, UserId, UserStatus,
+ Consumer, ConsumerOffsetInfo, Identifier, IggyError, IggyExpiry, Partitioning, Permissions,
+ Stats, UserId, UserStatus,
};
use send_wrapper::SendWrapper;
@@ -141,7 +141,11 @@ impl HttpSafeShard {
future.await
}
- pub async fn apply_state(&self, user_id: UserId, command: &EntryCommand) -> Result<(), IggyError> {
+ pub async fn apply_state(
+ &self,
+ user_id: UserId,
+ command: &EntryCommand,
+ ) -> Result<(), IggyError> {
self.shard().state.apply(user_id, command).await
}
diff --git a/core/server/src/http/mapper.rs b/core/server/src/http/mapper.rs
index c9e2d639..df93990a 100644
--- a/core/server/src/http/mapper.rs
+++ b/core/server/src/http/mapper.rs
@@ -22,11 +22,11 @@ use crate::slab::traits_ext::{EntityComponentSystem, IntoComponents};
use crate::streaming::clients::client_manager::Client;
use crate::streaming::personal_access_tokens::personal_access_token::PersonalAccessToken;
use crate::streaming::stats::stats::TopicStats;
-use crate::streaming::topics::consumer_group2::{ConsumerGroupRoot, ConsumerGroupMembers};
+use crate::streaming::topics::consumer_group2::{ConsumerGroupMembers, ConsumerGroupRoot};
use crate::streaming::topics::topic2::TopicRoot;
use crate::streaming::users::user::User;
use iggy_common::{ConsumerGroupDetails, ConsumerGroupInfo, ConsumerGroupMember, IggyByteSize};
-use iggy_common::{IdentityInfo, PersonalAccessTokenInfo, TopicDetails, TokenInfo};
+use iggy_common::{IdentityInfo, PersonalAccessTokenInfo, TokenInfo, TopicDetails};
use iggy_common::{UserInfo, UserInfoDetails};
use slab::Slab;
use std::sync::Arc;
@@ -34,10 +34,11 @@ use std::sync::Arc;
/// Map TopicRoot with partitions to TopicDetails for HTTP responses
pub fn map_topic_details(root: &TopicRoot, stats: &TopicStats) -> TopicDetails {
let mut partitions = Vec::new();
-
+
// Get partition details similar to binary mapper
root.partitions().with_components(|partition_components| {
- let (partition_roots, partition_stats, _, offsets, _, _, _) = partition_components.into_components();
+ let (partition_roots, partition_stats, _, offsets, _, _, _) =
+ partition_components.into_components();
for (partition_root, partition_stat, offset) in partition_roots
.iter()
.map(|(_, val)| val)
@@ -118,7 +119,7 @@ pub fn map_topics_from_components(
.zip(stats.iter().map(|(_, stat)| stat))
.map(|(root, stat)| map_topic(root, stat))
.collect::<Vec<_>>();
-
+
topics.sort_by(|a, b| a.id.cmp(&b.id));
topics
}
@@ -222,7 +223,10 @@ pub fn map_consumer_groups(
groups
}
-pub fn map_consumer_group(root: &ConsumerGroupRoot, members: &ConsumerGroupMembers) -> ConsumerGroupDetails {
+pub fn map_consumer_group(
+ root: &ConsumerGroupRoot,
+ members: &ConsumerGroupMembers,
+) -> ConsumerGroupDetails {
let members_guard = members.inner().shared_get();
let mut consumer_group_details = ConsumerGroupDetails {
id: root.id() as u32,
@@ -231,7 +235,7 @@ pub fn map_consumer_group(root: &ConsumerGroupRoot, members: &ConsumerGroupMembe
members_count: members_guard.len() as u32,
members: Vec::new(),
};
-
+
for (_, member) in members_guard.iter() {
consumer_group_details.members.push(ConsumerGroupMember {
id: member.id as u32,
@@ -253,12 +257,15 @@ pub fn map_generated_access_token_to_identity_info(token: GeneratedToken) -> Ide
}
/// Map StreamRoot and StreamStats to StreamDetails for HTTP responses
-pub fn map_stream_details(root: &crate::streaming::streams::stream2::StreamRoot, stats: &crate::streaming::stats::stats::StreamStats) -> iggy_common::StreamDetails {
+pub fn map_stream_details(
+ root: &crate::streaming::streams::stream2::StreamRoot,
+ stats: &crate::streaming::stats::stats::StreamStats,
+) -> iggy_common::StreamDetails {
// Get topics using the new slab-based API
let topics = root.topics().with_components(|topic_ref| {
let (topic_roots, _topic_auxiliaries, topic_stats) = topic_ref.into_components();
let mut topics_vec = Vec::new();
-
+
// Iterate over topics in the stream
for (topic_root, topic_stat) in topic_roots
.iter()
@@ -267,12 +274,12 @@ pub fn map_stream_details(root: &crate::streaming::streams::stream2::StreamRoot,
{
topics_vec.push(map_topic(topic_root, topic_stat));
}
-
+
// Sort topics by ID for consistent ordering
topics_vec.sort_by(|a, b| a.id.cmp(&b.id));
topics_vec
});
-
+
iggy_common::StreamDetails {
id: root.id() as u32,
created_at: root.created_at(),
@@ -285,7 +292,10 @@ pub fn map_stream_details(root: &crate::streaming::streams::stream2::StreamRoot,
}
/// Map StreamRoot and StreamStats to Stream for HTTP responses
-pub fn map_stream(root: &crate::streaming::streams::stream2::StreamRoot, stats: &crate::streaming::stats::stats::StreamStats) -> iggy_common::Stream {
+pub fn map_stream(
+ root: &crate::streaming::streams::stream2::StreamRoot,
+ stats: &crate::streaming::stats::stats::StreamStats,
+) -> iggy_common::Stream {
iggy_common::Stream {
id: root.id() as u32,
created_at: root.created_at(),
@@ -297,7 +307,10 @@ pub fn map_stream(root: &crate::streaming::streams::stream2::StreamRoot, stats:
}
/// Map multiple streams from slabs
-pub fn map_streams_from_slabs(roots: &slab::Slab<crate::streaming::streams::stream2::StreamRoot>, stats: &slab::Slab<Arc<crate::streaming::stats::stats::StreamStats>>) -> Vec<iggy_common::Stream> {
+pub fn map_streams_from_slabs(
+ roots: &slab::Slab<crate::streaming::streams::stream2::StreamRoot>,
+ stats: &slab::Slab<Arc<crate::streaming::stats::stats::StreamStats>>,
+) -> Vec<iggy_common::Stream> {
let mut streams = Vec::new();
for (root, stat) in roots
.iter()
diff --git a/core/server/src/http/partitions.rs b/core/server/src/http/partitions.rs
index 46923416..11a1c4a1 100644
--- a/core/server/src/http/partitions.rs
+++ b/core/server/src/http/partitions.rs
@@ -65,20 +65,27 @@ async fn create_partitions(
&command.stream_id,
&command.topic_id,
command.partitions_count,
- )).await?;
+ ))
+ .await?;
let broadcast_future = SendWrapper::new(async {
let shard = state.shard.shard();
- let event = ShardEvent::CreatedPartitions2 {
+ let event = ShardEvent::CreatedPartitions2 {
stream_id: command.stream_id.clone(),
topic_id: command.topic_id.clone(),
partitions,
};
let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
- let numeric_stream_id = shard.streams2.with_stream_by_id(&command.stream_id, streams::helpers::get_stream_id());
- let numeric_topic_id = shard.streams2.with_topic_by_id(&command.stream_id, &command.topic_id, topics::helpers::get_topic_id());
+ let numeric_stream_id = shard
+ .streams2
+ .with_stream_by_id(&command.stream_id, streams::helpers::get_stream_id());
+ let numeric_topic_id = shard.streams2.with_topic_by_id(
+ &command.stream_id,
+ &command.topic_id,
+ topics::helpers::get_topic_id(),
+ );
// TODO: Replace with new mechanism
/*
diff --git a/core/server/src/http/streams.rs b/core/server/src/http/streams.rs
index 94588047..0e61f657 100644
--- a/core/server/src/http/streams.rs
+++ b/core/server/src/http/streams.rs
@@ -58,21 +58,23 @@ async fn get_stream(
Path(stream_id): Path<String>,
) -> Result<Json<StreamDetails>, CustomError> {
let stream_id = Identifier::from_str_value(&stream_id)?;
-
+
// Use direct slab access for thread-safe stream retrieval
let stream_details = SendWrapper::new(|| {
- state.shard.shard().streams2.with_stream_by_id(&stream_id, |(root, stats)| {
- crate::http::mapper::map_stream_details(&*root, &**stats)
- })
+ state
+ .shard
+ .shard()
+ .streams2
+ .with_stream_by_id(&stream_id, |(root, stats)| {
+ crate::http::mapper::map_stream_details(&*root, &**stats)
+ })
})();
Ok(Json(stream_details))
}
#[debug_handler]
-async fn get_streams(
- State(state): State<Arc<AppState>>,
-) -> Result<Json<Vec<Stream>>, CustomError> {
+async fn get_streams(State(state): State<Arc<AppState>>) -> Result<Json<Vec<Stream>>, CustomError> {
// Use direct slab access for thread-safe streams retrieval
let streams = SendWrapper::new(|| {
state.shard.shard().streams2.with_components(|stream_ref| {
@@ -95,13 +97,12 @@ async fn create_stream(
let result = SendWrapper::new(async move {
let session = Session::stateless(identity.user_id, identity.ip_address);
-
+
// Create stream using wrapper method
- let stream = state.shard.create_stream(
- &session,
- command.stream_id,
- command.name.clone(),
- ).await
+ let stream = state
+ .shard
+ .create_stream(&session, command.stream_id, command.name.clone())
+ .await
.with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - failed to create stream, stream ID: {:?}",
@@ -118,7 +119,11 @@ async fn create_stream(
id: created_stream_id,
stream,
};
- let _responses = state.shard.shard().broadcast_event_to_all_shards(event).await;
+ let _responses = state
+ .shard
+ .shard()
+ .broadcast_event_to_all_shards(event)
+ .await;
}
// Apply state change using wrapper method
@@ -126,25 +131,32 @@ async fn create_stream(
stream_id: created_stream_id as u32,
command,
});
-
- state.shard.apply_state(identity.user_id, &entry_command).await
+
+ state
+ .shard
+ .apply_state(identity.user_id, &entry_command)
+ .await
.with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - failed to apply create stream for id: {:?}",
created_stream_id
)
})?;
-
+
// Get the created stream details using direct slab access
let response = SendWrapper::new(|| {
- state.shard.shard().streams2.with_components_by_id(created_stream_id, |(root, stats)| {
- crate::http::mapper::map_stream_details(&*root, &**stats)
- })
+ state
+ .shard
+ .shard()
+ .streams2
+ .with_components_by_id(created_stream_id, |(root, stats)| {
+ crate::http::mapper::map_stream_details(&*root, &**stats)
+ })
})();
-
+
Ok::<Json<StreamDetails>, CustomError>(Json(response))
});
-
+
result.await
}
@@ -161,15 +173,16 @@ async fn update_stream(
let result = SendWrapper::new(async move {
let session = Session::stateless(identity.user_id, identity.ip_address);
-
+
// Update stream using wrapper method
- state.shard.update_stream(
- &session,
- &command.stream_id,
- command.name.clone(),
- ).with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to update stream, stream ID: {stream_id}")
- })?;
+ state
+ .shard
+ .update_stream(&session, &command.stream_id, command.name.clone())
+ .with_error_context(|error| {
+ format!(
+ "{COMPONENT} (error: {error}) - failed to update stream, stream ID: {stream_id}"
+ )
+ })?;
// Apply state change using wrapper method
let entry_command = EntryCommand::UpdateStream(command);
@@ -180,7 +193,7 @@ async fn update_stream(
})?;
Ok::<StatusCode, CustomError>(StatusCode::NO_CONTENT)
});
-
+
result.await
}
@@ -192,14 +205,19 @@ async fn delete_stream(
Path(stream_id): Path<String>,
) -> Result<StatusCode, CustomError> {
let identifier_stream_id = Identifier::from_str_value(&stream_id)?;
-
+
let result = SendWrapper::new(async move {
let session = Session::stateless(identity.user_id, identity.ip_address);
-
+
// Delete stream using wrapper method
- state.shard.delete_stream(&session, &identifier_stream_id).await
+ state
+ .shard
+ .delete_stream(&session, &identifier_stream_id)
+ .await
.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to delete stream with ID: {stream_id}",)
+ format!(
+ "{COMPONENT} (error: {error}) - failed to delete stream with ID: {stream_id}",
+ )
})?;
// Apply state change using wrapper method
@@ -214,7 +232,7 @@ async fn delete_stream(
})?;
Ok::<StatusCode, CustomError>(StatusCode::NO_CONTENT)
});
-
+
result.await
}
@@ -226,14 +244,19 @@ async fn purge_stream(
Path(stream_id): Path<String>,
) -> Result<StatusCode, CustomError> {
let identifier_stream_id = Identifier::from_str_value(&stream_id)?;
-
+
let result = SendWrapper::new(async move {
let session = Session::stateless(identity.user_id, identity.ip_address);
-
+
// Purge stream using wrapper method
- state.shard.purge_stream(&session, &identifier_stream_id).await
+ state
+ .shard
+ .purge_stream(&session, &identifier_stream_id)
+ .await
.with_error_context(|error| {
- format!("{COMPONENT} (error: {error}) - failed to purge stream, stream ID: {stream_id}")
+ format!(
+ "{COMPONENT} (error: {error}) - failed to purge stream, stream ID: {stream_id}"
+ )
})?;
// Apply state change using wrapper method
@@ -248,6 +271,6 @@ async fn purge_stream(
})?;
Ok::<StatusCode, CustomError>(StatusCode::NO_CONTENT)
});
-
+
result.await
}
diff --git a/core/server/src/http/topics.rs b/core/server/src/http/topics.rs
index 4de7e426..b57a0bb4 100644
--- a/core/server/src/http/topics.rs
+++ b/core/server/src/http/topics.rs
@@ -20,7 +20,7 @@ use crate::http::COMPONENT;
use crate::http::error::CustomError;
use crate::http::jwt::json_web_token::Identity;
use crate::http::shared::AppState;
-use crate::slab::traits_ext::{EntityMarker, EntityComponentSystem, IntoComponents};
+use crate::slab::traits_ext::{EntityComponentSystem, EntityMarker, IntoComponents};
use crate::state::command::EntryCommand;
use crate::state::models::CreateTopicWithId;
use crate::streaming::session::Session;
@@ -30,12 +30,12 @@ use axum::http::StatusCode;
use axum::routing::{delete, get};
use axum::{Extension, Json, Router, debug_handler};
use error_set::ErrContext;
+use iggy_common::Identifier;
use iggy_common::Validatable;
use iggy_common::create_topic::CreateTopic;
use iggy_common::delete_topic::DeleteTopic;
use iggy_common::purge_topic::PurgeTopic;
use iggy_common::update_topic::UpdateTopic;
-use iggy_common::Identifier;
use iggy_common::{Topic, TopicDetails};
use send_wrapper::SendWrapper;
use std::sync::Arc;
@@ -66,21 +66,31 @@ async fn get_topic(
) -> Result<Json<TopicDetails>, CustomError> {
let identity_stream_id = Identifier::from_str_value(&stream_id)?;
let identity_topic_id = Identifier::from_str_value(&topic_id)?;
-
+
let session = Session::stateless(identity.user_id, identity.ip_address);
// Check permissions and stream existence
state.shard.shard().ensure_authenticated(&session)?;
- state.shard.shard().ensure_stream_exists(&identity_stream_id)?;
- state.shard.shard().ensure_topic_exists(&identity_stream_id, &identity_topic_id)?;
-
- let numeric_stream_id = state.shard.shard()
+ state
+ .shard
+ .shard()
+ .ensure_stream_exists(&identity_stream_id)?;
+ state
+ .shard
+ .shard()
+ .ensure_topic_exists(&identity_stream_id, &identity_topic_id)?;
+
+ let numeric_stream_id = state
+ .shard
+ .shard()
.streams2
.with_stream_by_id(&identity_stream_id, streams::helpers::get_stream_id());
- let numeric_topic_id = state.shard.shard()
- .streams2
- .with_topic_by_id(&identity_stream_id, &identity_topic_id, topics::helpers::get_topic_id());
-
+ let numeric_topic_id = state.shard.shard().streams2.with_topic_by_id(
+ &identity_stream_id,
+ &identity_topic_id,
+ topics::helpers::get_topic_id(),
+ );
+
state.shard.shard()
.permissioner
.borrow()
@@ -96,9 +106,7 @@ async fn get_topic(
let topic_details = state.shard.shard().streams2.with_topic_by_id(
&identity_stream_id,
&identity_topic_id,
- |(root, _, stats)| {
- crate::http::mapper::map_topic_details(&root, &stats)
- }
+ |(root, _, stats)| crate::http::mapper::map_topic_details(&root, &stats),
);
Ok(Json(topic_details))
@@ -116,11 +124,13 @@ async fn get_topics(
// Check permissions and stream existence
state.shard.shard().ensure_authenticated(&session)?;
state.shard.shard().ensure_stream_exists(&stream_id)?;
-
- let numeric_stream_id = state.shard.shard()
+
+ let numeric_stream_id = state
+ .shard
+ .shard()
.streams2
.with_stream_by_id(&stream_id, streams::helpers::get_stream_id());
-
+
state.shard.shard()
.permissioner
.borrow()
@@ -133,7 +143,9 @@ async fn get_topics(
})?;
// Get topics using the new API
- let topics = state.shard.shard()
+ let topics = state
+ .shard
+ .shard()
.streams2
.with_topics(&stream_id, |topics| {
topics.with_components(|topics| {
@@ -179,15 +191,15 @@ async fn create_topic(
command.max_topic_size = topic.root().max_topic_size();
let topic_id = topic.id();
-
+
// Send events for topic creation
let broadcast_future = SendWrapper::new(async {
use crate::shard::transmission::event::ShardEvent;
let shard = state.shard.shard();
- let event = ShardEvent::CreatedTopic2 {
- stream_id: command.stream_id.clone(),
+ let event = ShardEvent::CreatedTopic2 {
+ stream_id: command.stream_id.clone(),
topic,
};
let _responses = shard.broadcast_event_to_all_shards(event).await;
@@ -227,7 +239,7 @@ async fn create_topic(
&topic_identifier,
|(root, _, stats)| {
crate::http::mapper::map_topic_details_empty_partitions(&root, &stats)
- }
+ },
);
Json(topic_response)
};
@@ -289,7 +301,7 @@ async fn update_topic(
let (message_expiry, max_topic_size) = state.shard.shard().streams2.with_topic_by_id(
&command.stream_id,
&command.topic_id,
- |(root, _, _)| (root.message_expiry(), root.max_topic_size())
+ |(root, _, _)| (root.message_expiry(), root.max_topic_size()),
);
command.message_expiry = message_expiry;
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index 1598e9ec..e5ba519c 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -24,6 +24,7 @@ use std::sync::atomic::AtomicBool;
use anyhow::Result;
use clap::Parser;
+use dashmap::DashMap;
use dotenvy::dotenv;
use error_set::ErrContext;
use figlet_rs::FIGfont;
@@ -35,8 +36,7 @@ use server::args::Args;
use server::binary::handlers::streams;
use server::bootstrap::{
create_directories, create_root_user, create_shard_connections, create_shard_executor,
- load_config, load_streams, load_users, resolve_persister,
- update_system_info,
+ load_config, load_streams, load_users, resolve_persister, update_system_info,
};
use server::configs::config_provider::{self};
use server::configs::server::ServerConfig;
@@ -47,8 +47,9 @@ use server::log::logger::Logging;
#[cfg(feature = "tokio-console")]
use server::log::tokio_console::Logging;
use server::server_error::{ConfigError, ServerError};
-use server::shard::IggyShard;
+use server::shard::namespace::IggyNamespace;
use server::shard::system::info::SystemInfo;
+use server::shard::{IggyShard, ShardInfo};
use server::state::StateKind;
use server::state::command::EntryCommand;
use server::state::file::FileState;
@@ -57,12 +58,14 @@ use server::state::system::SystemState;
use server::streaming::diagnostics::metrics::Metrics;
use server::streaming::storage::SystemStorage;
use server::streaming::utils::MemoryPool;
+use server::streaming::utils::ptr::EternalPtr;
use server::versioning::SemanticVersion;
use server::{IGGY_ROOT_PASSWORD_ENV, IGGY_ROOT_USERNAME_ENV, map_toggle_str, shard_info};
use tokio::time::Instant;
use tracing::{error, info, instrument};
const COMPONENT: &str = "MAIN";
+const SHARDS_TABLE_CAPACITY: usize = 16384;
#[instrument(skip_all, name = "trace_start_server")]
#[compio::main]
@@ -218,9 +221,16 @@ async fn main() -> Result<(), ServerError> {
let (connections, shutdown_handles) = create_shard_connections(&shards_set);
let mut handles = Vec::with_capacity(shards_set.len());
+ // THIRTEENTH DISCRETE LOADING STEP.
+ // Shared resources bootstrap.
+ let shards_table = Box::new(DashMap::with_capacity(SHARDS_TABLE_CAPACITY));
+ let shards_table = Box::leak(shards_table);
+ let shards_table: EternalPtr<DashMap<IggyNamespace, ShardInfo>> = shards_table.into();
+
for shard_id in shards_set {
let id = shard_id as u16;
let streams = streams.clone();
+ let shards_table = shards_table.clone();
let users = users.clone();
let storage = storage.clone();
let connections = connections.clone();
@@ -247,6 +257,7 @@ async fn main() -> Result<(), ServerError> {
.streams(streams)
.state(state)
.users(users)
+ .shards_table(shards_table)
.connections(connections)
.config(config)
.storage(storage)
diff --git a/core/server/src/shard/builder.rs b/core/server/src/shard/builder.rs
index e4bac27f..615a5f71 100644
--- a/core/server/src/shard/builder.rs
+++ b/core/server/src/shard/builder.rs
@@ -23,16 +23,20 @@ use std::{
};
use ahash::HashMap;
+use dashmap::DashMap;
use iggy_common::{Aes256GcmEncryptor, EncryptorKind, UserId};
use tracing::info;
use crate::{
configs::server::ServerConfig,
io::storage::Storage,
- shard::{Shard, task_registry::TaskRegistry},
+ shard::{Shard, ShardInfo, namespace::IggyNamespace, task_registry::TaskRegistry},
slab::streams::Streams,
state::{StateKind, system::SystemState},
- streaming::{diagnostics::metrics::Metrics, storage::SystemStorage, users::user::User},
+ streaming::{
+ diagnostics::metrics::Metrics, storage::SystemStorage, users::user::User,
+ utils::ptr::EternalPtr,
+ },
versioning::SemanticVersion,
};
@@ -42,6 +46,7 @@ use super::{IggyShard, transmission::connector::ShardConnector, transmission::fr
pub struct IggyShardBuilder {
id: Option<u16>,
streams: Option<Streams>,
+ shards_table: Option<EternalPtr<DashMap<IggyNamespace, ShardInfo>>>,
state: Option<StateKind>,
users: Option<HashMap<UserId, User>>,
connections: Option<Vec<ShardConnector<ShardFrame>>>,
@@ -68,6 +73,14 @@ impl IggyShardBuilder {
self
}
+ pub fn shards_table(
+ mut self,
+ shards_table: EternalPtr<DashMap<IggyNamespace, ShardInfo>>,
+ ) -> Self {
+ self.shards_table = Some(shards_table);
+ self
+ }
+
pub fn encryptor(mut self, encryptor: Option<EncryptorKind>) -> Self {
self.encryptor = encryptor;
self
@@ -107,6 +120,7 @@ impl IggyShardBuilder {
pub fn build(self) -> IggyShard {
let id = self.id.unwrap();
let streams = self.streams.unwrap();
+ let shards_table = self.shards_table.unwrap();
let state = self.state.unwrap();
let users = self.users.unwrap();
let config = self.config.unwrap();
@@ -134,7 +148,7 @@ impl IggyShardBuilder {
IggyShard {
id: id,
shards: shards,
- shards_table: Default::default(),
+ shards_table,
//streams2: streams, // TODO: Fixme
streams2: Default::default(),
users: RefCell::new(users),
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 614d32e3..2a5ec7eb 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -1,5 +1,5 @@
/* Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
+inner() * or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
@@ -27,18 +27,21 @@ pub mod transmission;
use ahash::{AHashMap, AHashSet, HashMap};
use builder::IggyShardBuilder;
+use dashmap::DashMap;
use error_set::ErrContext;
use futures::future::try_join_all;
+use hash32::{Hasher, Murmur3Hasher};
use iggy_common::{
EncryptorKind, Identifier, IggyError, Permissions, UserId, UserStatus,
defaults::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME},
locking::IggyRwLockFn,
};
-use namespace::IggyNamespace;
+use std::hash::Hasher as _;
use std::{
cell::{Cell, RefCell},
future::Future,
net::SocketAddr,
+ ops::Deref,
pin::Pin,
rc::Rc,
sync::{
@@ -55,6 +58,7 @@ use crate::{
http::http_server,
io::fs_utils,
shard::{
+ namespace::IggyNamespace,
task_registry::TaskRegistry,
tasks::messages::spawn_shard_message_task,
transmission::{
@@ -77,6 +81,7 @@ use crate::{
session::Session,
storage::SystemStorage,
users::{permissioner::Permissioner, user::User},
+ utils::ptr::EternalPtr,
},
tcp::tcp_server::spawn_tcp_server,
versioning::SemanticVersion,
@@ -116,7 +121,8 @@ impl Shard {
}
}
-#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
+// TODO: Maybe pad to cache line size?
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub struct ShardInfo {
id: u16,
}
@@ -134,11 +140,11 @@ impl ShardInfo {
pub struct IggyShard {
pub id: u16,
shards: Vec<Shard>,
- shards_table: RefCell<HashMap<IggyNamespace, ShardInfo>>,
version: SemanticVersion,
// Heart transplant of the old streams structure.
pub(crate) streams2: Streams,
+ shards_table: EternalPtr<DashMap<IggyNamespace, ShardInfo>>,
// TODO: Refactor.
pub(crate) storage: Rc<SystemStorage>,
pub(crate) state: StateKind,
@@ -183,11 +189,13 @@ impl IggyShard {
let state_path = server_config.system.get_state_messages_file_path();
let file_state = FileState::new(&state_path, &version, persister, None);
let state = crate::state::StateKind::File(file_state);
+ let shards_table = Box::new(DashMap::new());
+ let shards_table = Box::leak(shards_table);
let shard = Self {
id: 0,
shards: Vec::new(),
- shards_table: Default::default(),
+ shards_table: shards_table.into(),
version,
streams2: Default::default(),
state,
@@ -316,6 +324,12 @@ impl IggyShard {
self.shards.len() as u32
}
+ pub fn calculate_shard_assignment(&self, ns: &IggyNamespace) -> u16 {
+ let mut hasher = Murmur3Hasher::default();
+ hasher.write_u64(ns.inner());
+ (hasher.finish32() % self.get_available_shards_count()) as u16
+ }
+
pub async fn handle_shard_message(&self, message: ShardMessage) -> Option<ShardResponse> {
match message {
ShardMessage::Request(request) => match self.handle_request(request).await {
@@ -469,7 +483,8 @@ impl IggyShard {
topic_id,
partitions,
} => {
- self.create_partitions2_bypass_auth(&stream_id, &topic_id, partitions)?;
+ self.create_partitions2_bypass_auth(&stream_id, &topic_id, partitions)
+ .await?;
Ok(())
}
ShardEvent::DeletedTopic2 {
@@ -554,6 +569,7 @@ impl IggyShard {
}
}
+ /*
pub async fn send_request_to_shard_or_recoil(
&self,
namespace: &IggyNamespace,
@@ -583,6 +599,7 @@ impl IggyShard {
))
}
}
+ */
pub async fn broadcast_event_to_all_shards(&self, event: ShardEvent) -> Vec<ShardResponse> {
let mut responses = Vec::with_capacity(self.get_available_shards_count() as usize);
@@ -634,9 +651,12 @@ impl IggyShard {
responses
}
+ pub fn add_active_session(&self, session: Rc<Session>) {
+ self.active_sessions.borrow_mut().push(session);
+ }
+
fn find_shard(&self, namespace: &IggyNamespace) -> Option<&Shard> {
- let shards_table = self.shards_table.borrow();
- shards_table.get(namespace).map(|shard_info| {
+ self.shards_table.get(namespace).map(|shard_info| {
self.shards
.iter()
.find(|shard| shard.id == shard_info.id)
@@ -645,49 +665,37 @@ impl IggyShard {
}
pub fn find_shard_table_record(&self, namespace: &IggyNamespace) -> Option<ShardInfo> {
- let shards_table = self.shards_table.borrow();
- shards_table.get(namespace).cloned()
+ self.shards_table.get(namespace).map(|entry| *entry)
+ }
+
+ pub fn remove_shard_table_record(&self, namespace: &IggyNamespace) -> ShardInfo {
+ self.shards_table.remove(namespace).map(|(_, shard_info)| shard_info).expect("remove_shard_table_record: namespace not found")
}
pub fn remove_shard_table_records(
&self,
namespaces: &[IggyNamespace],
) -> Vec<(IggyNamespace, ShardInfo)> {
- let mut shards_table = self.shards_table.borrow_mut();
namespaces
.iter()
.map(|ns| {
- let shard_info = shards_table.remove(ns).unwrap();
- (*ns, shard_info)
+ let (ns, shard_info) = self.shards_table.remove(ns).unwrap();
+ (ns, shard_info)
})
.collect()
}
- pub fn create_shard_table_records(
- &self,
- partition_ids: &[usize],
- stream_id: usize,
- topic_id: usize,
- ) -> impl Iterator<Item = (IggyNamespace, ShardInfo)> {
- let records = partition_ids.iter().map(move |partition_id| {
- let namespace = IggyNamespace::new(stream_id, topic_id, *partition_id);
- let hash = namespace.generate_hash();
- let shard_id = hash % self.get_available_shards_count();
- let shard_info = ShardInfo::new(shard_id as u16);
- (namespace, shard_info)
- });
- records
+ pub fn insert_shard_table_record(&self, ns: IggyNamespace, shard_info: ShardInfo) {
+ self.shards_table.insert(ns, shard_info);
}
pub fn insert_shard_table_records(
&self,
records: impl IntoIterator<Item = (IggyNamespace, ShardInfo)>,
) {
- self.shards_table.borrow_mut().extend(records);
- }
-
- pub fn add_active_session(&self, session: Rc<Session>) {
- self.active_sessions.borrow_mut().push(session);
+ for (ns, shard_info) in records {
+ self.shards_table.insert(ns, shard_info);
+ }
}
pub fn remove_active_session(&self, user_id: u32) {
diff --git a/core/server/src/shard/namespace.rs b/core/server/src/shard/namespace.rs
index 5b5502f9..127be415 100644
--- a/core/server/src/shard/namespace.rs
+++ b/core/server/src/shard/namespace.rs
@@ -19,28 +19,55 @@
use hash32::{Hasher, Murmur3Hasher};
use std::hash::Hasher as _;
-//TODO: Will probably want to move it to separate crate so we can share it with sdk.
-#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
-pub struct IggyNamespace {
- pub(crate) stream_id: usize,
- pub(crate) topic_id: usize,
- pub(crate) partition_id: usize,
+// Packed namespace layout (works only on 64bit platforms, but we won't support 32bit anyway)
+// +----------------+----------------+----------------+----------------+
+// | stream_id | topic_id | partition_id | unused |
+// | STREAM_BITS | TOPIC_BITS | PARTITION_BITS | (64 - total) |
+// +----------------+----------------+----------------+----------------+
+
+// TODO Use consts from the `slab` module.
+pub const MAX_STREAMS: usize = 4096;
+pub const MAX_TOPICS: usize = 4096;
+pub const MAX_PARTITIONS: usize = 1_000_000;
+
+const fn bits_required(mut n: u64) -> u32 {
+ if n == 0 {
+ return 1;
+ }
+ let mut b = 0;
+ while n > 0 {
+ b += 1;
+ n >>= 1;
+ }
+ b
}
+pub const STREAM_BITS: u32 = bits_required((MAX_STREAMS - 1) as u64);
+pub const TOPIC_BITS: u32 = bits_required((MAX_TOPICS - 1) as u64);
+pub const PARTITION_BITS: u32 = bits_required((MAX_PARTITIONS - 1) as u64);
+
+pub const PARTITION_SHIFT: u32 = 0;
+pub const TOPIC_SHIFT: u32 = PARTITION_SHIFT + PARTITION_BITS;
+pub const STREAM_SHIFT: u32 = TOPIC_SHIFT + TOPIC_BITS;
+
+pub const PARTITION_MASK: u64 = (1u64 << PARTITION_BITS) - 1;
+pub const TOPIC_MASK: u64 = (1u64 << TOPIC_BITS) - 1;
+pub const STREAM_MASK: u64 = (1u64 << STREAM_BITS) - 1;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct IggyNamespace(u64);
+
impl IggyNamespace {
- pub fn new(stream_id: usize, topic_id: usize, partition_id: usize) -> Self {
- Self {
- stream_id,
- topic_id,
- partition_id,
- }
+ #[inline]
+ pub fn inner(&self) -> u64 {
+ self.0
}
- pub fn generate_hash(&self) -> u32 {
- let mut hasher = Murmur3Hasher::default();
- hasher.write_usize(self.stream_id);
- hasher.write_usize(self.topic_id);
- hasher.write_usize(self.partition_id);
- hasher.finish32()
+ #[inline]
+ pub fn new(stream: usize, topic: usize, partition: usize) -> Self {
+ let value = ((stream as u64) & STREAM_MASK) << STREAM_SHIFT
+ | ((topic as u64) & TOPIC_MASK) << TOPIC_SHIFT
+ | ((partition as u64) & PARTITION_MASK) << PARTITION_SHIFT;
+ Self(value)
}
}
diff --git a/core/server/src/shard/system/messages.rs b/core/server/src/shard/system/messages.rs
index 35b57342..0c898b26 100644
--- a/core/server/src/shard/system/messages.rs
+++ b/core/server/src/shard/system/messages.rs
@@ -22,7 +22,6 @@ use super::COMPONENT;
use crate::binary::handlers::messages::poll_messages_handler::IggyPollMetadata;
use crate::configs::cache_indexes::CacheIndexesConfig;
use crate::shard::IggyShard;
-use crate::shard::namespace::IggyNamespace;
use crate::shard::transmission::frame::ShardResponse;
use crate::shard::transmission::message::{
ShardMessage, ShardRequest, ShardRequestPayload, ShardSendRequestResult,
diff --git a/core/server/src/shard/system/partitions.rs b/core/server/src/shard/system/partitions.rs
index 19c30cec..892110ab 100644
--- a/core/server/src/shard/system/partitions.rs
+++ b/core/server/src/shard/system/partitions.rs
@@ -18,6 +18,8 @@
use super::COMPONENT;
use crate::shard::IggyShard;
+use crate::shard::ShardInfo;
+use crate::shard::namespace::IggyNamespace;
use crate::shard_info;
use crate::slab::traits_ext::EntityComponentSystem;
use crate::slab::traits_ext::EntityComponentSystemMutCell;
@@ -33,7 +35,6 @@ use crate::streaming::partitions::storage2::create_partition_file_hierarchy;
use crate::streaming::partitions::storage2::delete_partitions_from_disk;
use crate::streaming::segments::Segment2;
use crate::streaming::segments::storage::create_segment_storage;
-
use crate::streaming::session::Session;
use crate::streaming::stats::stats::PartitionStats;
use crate::streaming::stats::stats::TopicStats;
@@ -114,6 +115,13 @@ impl IggyShard {
// TODO: Figure out how to do this operation in a batch.
for partition_id in partitions.iter().map(|p| p.id()) {
+ // TODO: Create shard table recordsj.
+ let ns = IggyNamespace::new(numeric_stream_id, numeric_topic_id, partition_id);
+ let shard_id = self.calculate_shard_assignment(&ns);
+ let shard_info = ShardInfo::new(shard_id);
+ let is_current_shard = self.id == shard_info.id;
+ self.insert_shard_table_record(ns, shard_info);
+
create_partition_file_hierarchy(
self.id,
numeric_stream_id as usize,
@@ -122,7 +130,9 @@ impl IggyShard {
&self.config.system,
)
.await?;
- self.init_log(stream_id, topic_id, partition_id).await?;
+ if is_current_shard {
+ self.init_log(stream_id, topic_id, partition_id).await?;
+ }
}
Ok(partitions)
}
@@ -243,7 +253,7 @@ impl IggyShard {
Ok(())
}
- pub fn create_partitions2_bypass_auth(
+ pub async fn create_partitions2_bypass_auth(
&self,
stream_id: &Identifier,
topic_id: &Identifier,
@@ -256,6 +266,7 @@ impl IggyShard {
id, actual_id,
"create_partitions_bypass_auth: partition mismatch ID, wrong creation order ?!"
);
+ self.init_log(stream_id, topic_id, id).await?;
}
Ok(())
}
@@ -298,9 +309,11 @@ impl IggyShard {
for partition in partitions {
let (root, stats, _, _, _, _, mut log) = partition.into_components();
let partition_id = root.id();
+ let ns = IggyNamespace::new(numeric_stream_id, numeric_topic_id, partition_id);
+ self.remove_shard_table_record(&ns);
+
self.delete_partition_dir(numeric_stream_id, numeric_topic_id, partition_id, &mut log)
.await?;
-
let segments_count = stats.segments_count_inconsistent();
let messages_count = stats.messages_count_inconsistent();
let size_bytes = stats.size_bytes_inconsistent();
diff --git a/core/server/src/shard/transmission/event.rs b/core/server/src/shard/transmission/event.rs
index c4217ae9..2f6cef89 100644
--- a/core/server/src/shard/transmission/event.rs
+++ b/core/server/src/shard/transmission/event.rs
@@ -1,15 +1,12 @@
-use crate::{
- shard::namespace::IggyNamespace,
- streaming::{
- clients::client_manager::Transport,
- partitions::partition2,
- personal_access_tokens::personal_access_token::PersonalAccessToken,
- polling_consumer::PollingConsumer,
- streams::stream2,
- topics::{
- consumer_group2::{self},
- topic2,
- },
+use crate::streaming::{
+ clients::client_manager::Transport,
+ partitions::partition2,
+ personal_access_tokens::personal_access_token::PersonalAccessToken,
+ polling_consumer::PollingConsumer,
+ streams::stream2,
+ topics::{
+ consumer_group2::{self},
+ topic2,
},
};
use iggy_common::{
diff --git a/core/server/src/streaming/partitions/storage2.rs b/core/server/src/streaming/partitions/storage2.rs
index 360b300a..5de830f9 100644
--- a/core/server/src/streaming/partitions/storage2.rs
+++ b/core/server/src/streaming/partitions/storage2.rs
@@ -18,7 +18,12 @@ use compio::{
};
use error_set::ErrContext;
use iggy_common::{ConsumerKind, IggyError};
-use std::{io::Read, ops::Deref, path::Path, sync::{atomic::AtomicU64, Arc}};
+use std::{
+ io::Read,
+ ops::Deref,
+ path::Path,
+ sync::{Arc, atomic::AtomicU64},
+};
use tracing::{error, trace};
pub async fn create_partition_file_hierarchy(
@@ -121,7 +126,13 @@ pub async fn delete_partitions_from_disk(
//log.close().await;
let partition_path = config.get_partition_path(stream_id, topic_id, partition_id);
- remove_dir_all(&partition_path).await.map_err(|_| IggyError::CannotDeletePartitionDirectory(stream_id as u32, topic_id as u32, partition_id as u32))?;
+ remove_dir_all(&partition_path).await.map_err(|_| {
+ IggyError::CannotDeletePartitionDirectory(
+ stream_id as u32,
+ topic_id as u32,
+ partition_id as u32,
+ )
+ })?;
shard_info!(
shard_id,
"Deleted partition files for partition with ID: {} stream with ID: {} and topic with ID: {}.",
@@ -150,9 +161,13 @@ pub async fn persist_offset(shard_id: u16, path: &str, offset: u64) -> Result<()
.write(true)
.create(true)
.open(path)
- .await.map_err(|_| IggyError::CannotOpenConsumerOffsetsFile(path.to_owned()))?;
+ .await
+ .map_err(|_| IggyError::CannotOpenConsumerOffsetsFile(path.to_owned()))?;
let buf = offset.to_le_bytes();
- file.write_all_at(buf, 0).await.0.map_err(|_| IggyError::CannotWriteToFile)?;
+ file.write_all_at(buf, 0)
+ .await
+ .0
+ .map_err(|_| IggyError::CannotWriteToFile)?;
shard_trace!(
shard_id,
"Stored consumer offset value: {}, path: {}",
@@ -226,4 +241,4 @@ pub fn load_consumer_offsets(
consumer_offsets.sort_by(|a, b| a.consumer_id.cmp(&b.consumer_id));
Ok(consumer_offsets)
-}
\ No newline at end of file
+}
diff --git a/core/server/src/streaming/streams/messages.rs b/core/server/src/streaming/streams/messages.rs
index 4efb544a..679c9938 100644
--- a/core/server/src/streaming/streams/messages.rs
+++ b/core/server/src/streaming/streams/messages.rs
@@ -23,10 +23,7 @@ use tracing::trace;
use super::COMPONENT;
use crate::{
binary::handlers::messages::poll_messages_handler::IggyPollMetadata,
- shard::{
- namespace::IggyNamespace,
- transmission::message::{ShardMessage, ShardRequest, ShardRequestPayload},
- },
+ shard::transmission::message::{ShardMessage, ShardRequest, ShardRequestPayload},
streaming::{
polling_consumer::PollingConsumer,
segments::{IggyMessagesBatchMut, IggyMessagesBatchSet},
diff --git a/core/server/src/streaming/utils/mod.rs b/core/server/src/streaming/utils/mod.rs
index e441a051..834119bb 100644
--- a/core/server/src/streaming/utils/mod.rs
+++ b/core/server/src/streaming/utils/mod.rs
@@ -20,6 +20,7 @@ pub mod crypto;
pub mod file;
pub mod hash;
pub mod head_tail_buf;
+pub mod ptr;
pub mod random_id;
mod memory_pool;
diff --git a/core/server/src/streaming/utils/ptr.rs b/core/server/src/streaming/utils/ptr.rs
new file mode 100644
index 00000000..0fa9d540
--- /dev/null
+++ b/core/server/src/streaming/utils/ptr.rs
@@ -0,0 +1,53 @@
+use std::{ops::Deref, ptr::NonNull};
+
+// Wrapper around an imutable pointer to a 'static value, that can be cloned and sent across threads.
+pub struct EternalPtr<T: ?Sized + 'static> {
+ ptr: NonNull<T>,
+ _marker: std::marker::PhantomData<&'static T>,
+}
+
+impl<T> From<NonNull<T>> for EternalPtr<T> {
+ fn from(value: NonNull<T>) -> Self {
+ Self {
+ ptr: value,
+ _marker: std::marker::PhantomData,
+ }
+ }
+}
+
+impl<T> From<&'static T> for EternalPtr<T> {
+ fn from(value: &'static T) -> Self {
+ Self {
+ ptr: value.into(),
+ _marker: std::marker::PhantomData,
+ }
+ }
+}
+
+impl<T> From<&'static mut T> for EternalPtr<T> {
+ fn from(value: &'static mut T) -> Self {
+ Self {
+ ptr: value.into(),
+ _marker: std::marker::PhantomData,
+ }
+ }
+}
+
+impl<T: ?Sized + 'static> Clone for EternalPtr<T> {
+ fn clone(&self) -> Self {
+ Self {
+ ptr: self.ptr,
+ _marker: std::marker::PhantomData,
+ }
+ }
+}
+
+impl<T: ?Sized + 'static> Deref for EternalPtr<T> {
+ type Target = T;
+
+ fn deref(&self) -> &Self::Target {
+ unsafe { self.ptr.as_ref() }
+ }
+}
+
+unsafe impl<T: Send> Send for EternalPtr<T> {}