You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2025/07/14 10:08:00 UTC
(iggy) 44/45: feat(io_uring): fix segment integration tests (#2005)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch merge_master
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 0a5f0b5c5973e0b66973edcdfb432979b2c93f1b
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Mon Jul 14 11:24:08 2025 +0200
feat(io_uring): fix segment integration tests (#2005)
---
Cargo.lock | 1 +
Cargo.toml | 1 +
core/integration/Cargo.toml | 1 +
.../tests/streaming/common/test_setup.rs | 5 +-
core/integration/tests/streaming/segment.rs | 56 +++++++++--------
core/server/Cargo.toml | 2 +-
.../handlers/streams/purge_stream_handler.rs | 6 +-
.../binary/handlers/topics/purge_topic_handler.rs | 4 +-
core/server/src/shard/mod.rs | 70 ++++++++++++++++++++--
core/server/src/shard/system/segments.rs | 4 +-
core/server/src/shard/system/streams.rs | 5 +-
core/server/src/shard/system/topics.rs | 11 ++--
.../src/streaming/segments/indexes/index_reader.rs | 13 ++--
.../streaming/segments/messages/messages_reader.rs | 6 +-
.../src/streaming/segments/reading_messages.rs | 2 +-
core/server/src/streaming/segments/segment.rs | 1 +
16 files changed, 132 insertions(+), 56 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 27300d31..fce0c066 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4394,6 +4394,7 @@ dependencies = [
"async-trait",
"bytes",
"chrono",
+ "compio",
"ctor",
"derive_more 2.0.1",
"env_logger",
diff --git a/Cargo.toml b/Cargo.toml
index 50912629..411d75b9 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -119,6 +119,7 @@ sysinfo = "0.35.2"
tempfile = "3.20.0"
thiserror = "2.0.12"
tokio = { version = "1.45.1", features = ["full"] }
+compio = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be", features = ["runtime", "macros", "io-uring", "time", "rustls"] }
tokio-rustls = "0.26.2"
toml = "0.8.23"
tracing = "0.1.41"
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index b17d3d0b..7c241429 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -51,6 +51,7 @@ server = { workspace = true }
tempfile = { workspace = true }
test-case = { workspace = true }
tokio = { workspace = true }
+compio = { workspace = true }
twox-hash = { workspace = true }
uuid = { workspace = true }
zip = { workspace = true }
diff --git a/core/integration/tests/streaming/common/test_setup.rs b/core/integration/tests/streaming/common/test_setup.rs
index 19363249..88c32194 100644
--- a/core/integration/tests/streaming/common/test_setup.rs
+++ b/core/integration/tests/streaming/common/test_setup.rs
@@ -16,13 +16,15 @@
* under the License.
*/
+use compio::fs;
+use log::error;
+use server::bootstrap::create_directories;
use server::configs::system::SystemConfig;
use server::streaming::persistence::persister::{FileWithSyncPersister, PersisterKind};
use server::streaming::storage::SystemStorage;
use server::streaming::utils::MemoryPool;
use std::rc::Rc;
use std::sync::Arc;
-use tokio::fs;
use uuid::Uuid;
pub struct TestSetup {
@@ -42,6 +44,7 @@ impl TestSetup {
let config = Arc::new(config);
fs::create_dir(config.get_system_path()).await.unwrap();
+ create_directories(&config).await.unwrap();
let persister = PersisterKind::FileWithSync(FileWithSyncPersister {});
let storage = Rc::new(SystemStorage::new(config.clone(), Arc::new(persister)));
MemoryPool::init_pool(config.clone());
diff --git a/core/integration/tests/streaming/segment.rs b/core/integration/tests/streaming/segment.rs
index 04155cb8..cc01eb71 100644
--- a/core/integration/tests/streaming/segment.rs
+++ b/core/integration/tests/streaming/segment.rs
@@ -19,22 +19,25 @@
use crate::streaming::common::test_setup::TestSetup;
use crate::streaming::create_message;
use bytes::Bytes;
+use compio::fs;
+use compio::time::sleep;
use iggy::prelude::locking::IggyRwLockFn;
use iggy::prelude::*;
-use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig};
+use log::error;
+use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig, ServerConfig};
use server::configs::system::{PartitionConfig, SegmentConfig, SystemConfig};
+use server::shard::IggyShard;
use server::streaming::segments::*;
use server::streaming::session::Session;
use std::fs::DirEntry;
use std::net::{Ipv4Addr, SocketAddr};
+use std::rc::Rc;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::time::Duration;
-use tokio::fs;
-use tokio::time::sleep;
-#[tokio::test]
+#[compio::test]
async fn should_persist_segment() {
let setup = TestSetup::init().await;
let stream_id = 1;
@@ -72,7 +75,7 @@ async fn should_persist_segment() {
}
}
-#[tokio::test]
+#[compio::test]
async fn should_load_existing_segment_from_disk() {
let setup = TestSetup::init().await;
let stream_id = 1;
@@ -142,7 +145,7 @@ async fn should_load_existing_segment_from_disk() {
}
}
-#[tokio::test]
+#[compio::test]
async fn should_persist_and_load_segment_with_messages() {
let setup = TestSetup::init().await;
let stream_id = 1;
@@ -215,7 +218,7 @@ async fn should_persist_and_load_segment_with_messages() {
assert_eq!(messages.count(), messages_count);
}
-#[tokio::test]
+#[compio::test]
async fn given_at_least_one_not_expired_message_segment_should_not_be_expired() {
let config = SystemConfig {
partition: PartitionConfig {
@@ -307,9 +310,7 @@ async fn given_at_least_one_not_expired_message_segment_should_not_be_expired()
);
}
-/*
-//TODO: Fixme use shard instead of system
-#[tokio::test]
+#[compio::test]
async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Error>> {
let config = SystemConfig {
segment: SegmentConfig {
@@ -319,11 +320,12 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
..Default::default()
};
let setup = TestSetup::init_with_config(config).await;
- let mut system = System::new(
- setup.config.clone(),
- DataMaintenanceConfig::default(),
- PersonalAccessTokenConfig::default(),
- );
+ let server_config = ServerConfig {
+ system: setup.config.clone(),
+ ..Default::default()
+ };
+ let shard = IggyShard::default_from_config(server_config);
+ shard.init().await.unwrap();
// Properties
let stream_id = Identifier::numeric(1)?;
@@ -341,14 +343,16 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
.await;
let session = Session::new(1, 1, SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 1234));
- system.init().await.unwrap();
+ let session = Rc::new(session);
+ shard.init().await.unwrap();
+ shard.add_active_session(session.clone());
- system
+ shard
.create_stream(&session, Some(stream_id.get_u32_value()?), stream_name)
.await
.unwrap();
- system
+ let result = shard
.create_topic(
&session,
&stream_id,
@@ -360,9 +364,13 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
MaxTopicSize::default(),
None,
)
- .await?;
+ .await;
+ if let Some(error) = result.err() {
+ error!("Failed to create topic: {}", error);
+ }
- let topic = system.find_topic(&session, &stream_id, &topic_id)?;
+ let stream = shard.find_stream(&session, &stream_id)?;
+ let topic = shard.find_topic(&session, &stream, &topic_id)?;
let partitions = topic.get_partitions();
let partition = partitions
.first()
@@ -384,12 +392,13 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
.sum();
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition_lock.append_messages(batch, None).await.unwrap();
+ partition_lock.append_messages(batch).await.unwrap();
drop(partition_lock);
sleep(Duration::from_millis(100)).await;
}
+ drop(stream);
let partition_path = setup.config.get_partition_path(
stream_id.get_u32_value()?,
@@ -418,7 +427,7 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
println!("Initial segment offsets (sorted): {initial_offsets:?}");
let first_keep_count = 3usize;
- system
+ shard
.delete_segments(
&session,
&stream_id,
@@ -449,7 +458,7 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
// Second attempt to delete segments - keeping only 1 closed segment + 1 open
let second_keep_count = 1usize;
- system
+ shard
.delete_segments(
&session,
&stream_id,
@@ -497,7 +506,6 @@ async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Er
Ok(())
}
-*/
// Helper function to extract segment offsets from DirEntry list
fn get_segment_offsets(segments: &[DirEntry]) -> Vec<u64> {
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index 47738185..b171652d 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -67,7 +67,6 @@ lending-iterator = "0.1.7"
hash32 = "1.0.0"
mimalloc = { workspace = true, optional = true }
moka = { version = "0.12.10", features = ["future"] }
-compio = { git = "https://github.com/compio-rs/compio.git", rev = "fe4243f0b6811ebc325afd081c9b087b4d9817be", features = ["runtime", "macros", "io-uring", "time", "rustls"] }
nix = { version = "0.30", features = ["fs"] }
once_cell = "1.21.3"
opentelemetry = { version = "0.30.0", features = ["trace", "logs"] }
@@ -107,6 +106,7 @@ sysinfo = { workspace = true }
tempfile = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true }
+compio = { workspace = true }
tokio-rustls = { workspace = true }
tokio-util = { workspace = true }
toml = { workspace = true }
diff --git a/core/server/src/binary/handlers/streams/purge_stream_handler.rs b/core/server/src/binary/handlers/streams/purge_stream_handler.rs
index 6f8f06e8..1bb5fb39 100644
--- a/core/server/src/binary/handlers/streams/purge_stream_handler.rs
+++ b/core/server/src/binary/handlers/streams/purge_stream_handler.rs
@@ -19,8 +19,8 @@
use crate::binary::command::{BinaryServerCommand, ServerCommand, ServerCommandHandler};
use crate::binary::handlers::utils::receive_and_validate;
use crate::binary::{handlers::streams::COMPONENT, sender::SenderKind};
-use crate::shard::transmission::event::ShardEvent;
use crate::shard::IggyShard;
+use crate::shard::transmission::event::ShardEvent;
use crate::state::command::EntryCommand;
use crate::streaming::session::Session;
use anyhow::Result;
@@ -53,7 +53,9 @@ impl ServerCommandHandler for PurgeStream {
format!("{COMPONENT} (error: {error}) - failed to purge stream with id: {stream_id}, session: {session}")
})?;
- let event = ShardEvent::PurgedStream { stream_id: self.stream_id.clone() };
+ let event = ShardEvent::PurgedStream {
+ stream_id: self.stream_id.clone(),
+ };
let _responses = shard.broadcast_event_to_all_shards(event.into()).await;
shard
diff --git a/core/server/src/binary/handlers/topics/purge_topic_handler.rs b/core/server/src/binary/handlers/topics/purge_topic_handler.rs
index 160fac9a..101c7973 100644
--- a/core/server/src/binary/handlers/topics/purge_topic_handler.rs
+++ b/core/server/src/binary/handlers/topics/purge_topic_handler.rs
@@ -45,7 +45,7 @@ impl ServerCommandHandler for PurgeTopic {
debug!("session: {session}, command: {self}");
let topic_id = self.topic_id.clone();
let stream_id = self.stream_id.clone();
-
+
shard
.purge_topic(session, &self.stream_id, &self.topic_id)
.await
@@ -56,7 +56,7 @@ impl ServerCommandHandler for PurgeTopic {
)
})?;
- let event = crate::shard::transmission::event::ShardEvent::PurgedTopic {
+ let event = crate::shard::transmission::event::ShardEvent::PurgedTopic {
stream_id: self.stream_id.clone(),
topic_id: self.topic_id.clone(),
};
diff --git a/core/server/src/shard/mod.rs b/core/server/src/shard/mod.rs
index 4d0f089e..b8ba22b7 100644
--- a/core/server/src/shard/mod.rs
+++ b/core/server/src/shard/mod.rs
@@ -28,10 +28,15 @@ use ahash::{AHashMap, AHashSet, HashMap};
use builder::IggyShardBuilder;
use error_set::ErrContext;
use futures::future::try_join_all;
-use iggy_common::{EncryptorKind, Identifier, IggyError, UserId, locking::IggyRwLockFn};
+use iggy_common::{
+ EncryptorKind, Identifier, IggyError, Permissions, UserId, UserStatus,
+ defaults::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME},
+ locking::IggyRwLockFn,
+};
use namespace::IggyNamespace;
use std::{
cell::{Cell, RefCell},
+ future::Future,
pin::Pin,
rc::Rc,
str::FromStr,
@@ -156,13 +161,67 @@ impl IggyShard {
Default::default()
}
+ pub fn default_from_config(server_config: ServerConfig) -> Self {
+ use crate::bootstrap::resolve_persister;
+ use crate::state::file::FileState;
+ use crate::streaming::diagnostics::metrics::Metrics;
+ use crate::streaming::storage::SystemStorage;
+ use crate::versioning::SemanticVersion;
+
+ let version = SemanticVersion::current().expect("Invalid version");
+ let persister = resolve_persister(server_config.system.partition.enforce_fsync);
+ let storage = Rc::new(SystemStorage::new(
+ server_config.system.clone(),
+ persister.clone(),
+ ));
+
+ 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 (stop_sender, stop_receiver) = async_channel::unbounded();
+
+ let shard = Self {
+ id: 0, // Default shard ID
+ shards: Vec::new(), // No other shards in default config
+ shards_table: Default::default(),
+ version,
+ streams: Default::default(),
+ streams_ids: Default::default(),
+ storage,
+ state,
+ encryptor: None,
+ config: server_config,
+ client_manager: Default::default(),
+ active_sessions: Default::default(),
+ permissioner: Default::default(),
+ users: Default::default(),
+ metrics: Metrics::init(),
+ messages_receiver: Cell::new(None),
+ stop_receiver,
+ stop_sender,
+ task_registry: TaskRegistry::new(),
+ is_shutting_down: AtomicBool::new(false),
+ };
+ let user = User::root(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD);
+ shard
+ .create_user_bypass_auth(
+ &user.username,
+ &user.password,
+ UserStatus::Active,
+ Some(Permissions::root()),
+ )
+ .unwrap();
+ shard
+ }
+
pub async fn init(&self) -> Result<(), IggyError> {
let now = Instant::now();
self.load_version().await?;
let SystemState { users, streams } = self.load_state().await?;
- self.load_users(users.into_values().collect()).await;
- self.load_streams(streams.into_values().collect()).await;
+ let _ = self.load_users(users.into_values().collect()).await;
+ let _ = self.load_streams(streams.into_values().collect()).await;
//TODO: Fix the archiver.
/*
@@ -534,7 +593,6 @@ impl IggyShard {
*max_topic_size,
*replication_factor,
)
- .await
}
ShardEvent::LoginUser {
client_id,
@@ -689,7 +747,7 @@ impl IggyShard {
.await?;
Ok(())
}
- ShardEvent::PurgedStream { stream_id,} => {
+ ShardEvent::PurgedStream { stream_id } => {
self.purge_stream_bypass_auth(stream_id).await?;
Ok(())
}
@@ -699,7 +757,7 @@ impl IggyShard {
} => {
self.purge_topic_bypass_auth(stream_id, topic_id).await?;
Ok(())
- },
+ }
ShardEvent::DeletedTopic {
stream_id,
topic_id,
diff --git a/core/server/src/shard/system/segments.rs b/core/server/src/shard/system/segments.rs
index 5b907f05..0c7b853b 100644
--- a/core/server/src/shard/system/segments.rs
+++ b/core/server/src/shard/system/segments.rs
@@ -87,6 +87,7 @@ impl IggyShard {
.map(|segment| (segment.start_offset(), segment.get_messages_count()))
.collect::<Vec<_>>();
+
// Delete the segments in sequence.
let (deleted_segments_count, deleted_messages_count) = {
let mut segments_count = 0;
@@ -103,9 +104,8 @@ impl IggyShard {
(segments_count, messages_count)
};
- drop(partition);
- drop(topic);
drop(stream);
+ drop(partition);
let mut stream = self
.get_stream_mut(stream_id)
diff --git a/core/server/src/shard/system/streams.rs b/core/server/src/shard/system/streams.rs
index f5e823c1..b4b1b63a 100644
--- a/core/server/src/shard/system/streams.rs
+++ b/core/server/src/shard/system/streams.rs
@@ -414,10 +414,7 @@ impl IggyShard {
Ok(())
}
- pub async fn purge_stream_bypass_auth(
- &self,
- stream_id: &Identifier,
- ) -> Result<(), IggyError> {
+ pub async fn purge_stream_bypass_auth(&self, stream_id: &Identifier) -> Result<(), IggyError> {
let stream = self.get_stream(stream_id).with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to get stream with ID: {stream_id}")
})?;
diff --git a/core/server/src/shard/system/topics.rs b/core/server/src/shard/system/topics.rs
index d71add8e..bfe974cf 100644
--- a/core/server/src/shard/system/topics.rs
+++ b/core/server/src/shard/system/topics.rs
@@ -102,7 +102,7 @@ impl IggyShard {
Ok(Some(topic))
}
- pub async fn create_topic_bypass_auth(
+ pub fn create_topic_bypass_auth(
&self,
stream_id: &Identifier,
topic_id: Option<u32>,
@@ -445,7 +445,7 @@ impl IggyShard {
session.get_user_id(),
)
})?;
-
+
self.purge_topic_base(topic.stream_id, topic.topic_id).await
}
@@ -462,14 +462,15 @@ impl IggyShard {
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to get topic with ID: {topic_id} in stream with ID: {stream_id}")
})?;
-
- self.purge_topic_base(stream.stream_id, topic.topic_id).await
+
+ self.purge_topic_base(stream.stream_id, topic.topic_id)
+ .await
}
async fn purge_topic_base(&self, stream_id: u32, topic_id: u32) -> Result<(), IggyError> {
let stream = self.get_stream(&Identifier::numeric(stream_id)?)?;
let topic = stream.get_topic(&Identifier::numeric(topic_id)?)?;
-
+
for partition in topic.get_partitions() {
let mut partition = partition.write().await;
let partition_id = partition.partition_id;
diff --git a/core/server/src/streaming/segments/indexes/index_reader.rs b/core/server/src/streaming/segments/indexes/index_reader.rs
index ce235ef4..3275f567 100644
--- a/core/server/src/streaming/segments/indexes/index_reader.rs
+++ b/core/server/src/streaming/segments/indexes/index_reader.rs
@@ -20,12 +20,11 @@ use super::IggyIndexesMut;
use crate::streaming::utils::PooledBuffer;
use bytes::BytesMut;
use compio::{
- BufResult,
- fs::{File, OpenOptions},
- io::AsyncReadAtExt,
+ buf::{IntoInner, IoBuf}, fs::{File, OpenOptions}, io::AsyncReadAtExt, BufResult
};
use error_set::ErrContext;
use iggy_common::{INDEX_SIZE, IggyError, IggyIndex, IggyIndexView};
+use ring::error;
use std::{
io::ErrorKind,
os::unix::fs::FileExt,
@@ -150,6 +149,7 @@ impl IndexReader {
{
Ok(buf) => buf,
Err(error) if error.kind() == ErrorKind::UnexpectedEof => {
+ error!("Unexpected EOF while reading indexes");
return Ok(None);
}
Err(error) => {
@@ -337,9 +337,10 @@ impl IndexReader {
use_pool: bool,
) -> Result<PooledBuffer, std::io::Error> {
if use_pool {
- let mut buf = PooledBuffer::with_capacity(len as usize);
- unsafe { buf.set_len(len as usize) };
- let (result, buf) = self.file.read_exact_at(buf, offset as u64).await.into();
+ let len = len as usize;
+ let buf = PooledBuffer::with_capacity(len as usize);
+ let (result, buf) = self.file.read_exact_at(buf.slice(..len), offset as u64).await.into();
+ let buf = buf.into_inner();
result?;
Ok(buf)
} else {
diff --git a/core/server/src/streaming/segments/messages/messages_reader.rs b/core/server/src/streaming/segments/messages/messages_reader.rs
index e81eb793..fea3ca37 100644
--- a/core/server/src/streaming/segments/messages/messages_reader.rs
+++ b/core/server/src/streaming/segments/messages/messages_reader.rs
@@ -19,6 +19,7 @@
use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut};
use crate::streaming::utils::PooledBuffer;
use bytes::BytesMut;
+use compio::buf::{IntoInner, IoBuf};
use compio::fs::{File, OpenOptions};
use compio::io::AsyncReadAtExt;
use error_set::ErrContext;
@@ -178,8 +179,9 @@ impl MessagesReader {
) -> Result<PooledBuffer, std::io::Error> {
if use_pool {
let mut buf = PooledBuffer::with_capacity(len as usize);
- unsafe { buf.set_len(len as usize) };
- let (result, buf) = self.file.read_exact_at(buf, offset as u64).await.into();
+ let len = len as usize;
+ let (result, buf) = self.file.read_exact_at(buf.slice(..len), offset as u64).await.into();
+ let buf = buf.into_inner();
result?;
Ok(buf)
} else {
diff --git a/core/server/src/streaming/segments/reading_messages.rs b/core/server/src/streaming/segments/reading_messages.rs
index fa46c183..d72b1591 100644
--- a/core/server/src/streaming/segments/reading_messages.rs
+++ b/core/server/src/streaming/segments/reading_messages.rs
@@ -21,7 +21,7 @@ use crate::streaming::segments::segment::Segment;
use error_set::ErrContext;
use iggy_common::{IggyByteSize, IggyError};
use std::sync::atomic::Ordering;
-use tracing::trace;
+use tracing::{error, trace};
const COMPONENT: &str = "STREAMING_SEGMENT";
diff --git a/core/server/src/streaming/segments/segment.rs b/core/server/src/streaming/segments/segment.rs
index dfe7bc21..fdca6be6 100644
--- a/core/server/src/streaming/segments/segment.rs
+++ b/core/server/src/streaming/segments/segment.rs
@@ -28,6 +28,7 @@ use iggy_common::IggyByteSize;
use iggy_common::IggyError;
use iggy_common::IggyExpiry;
use iggy_common::IggyTimestamp;
+use tracing::error;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};