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/06/28 12:05:10 UTC
(iggy) branch io_uring_tpc updated (df3a05ba -> 2a22241a)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch io_uring_tpc
in repository https://gitbox.apache.org/repos/asf/iggy.git
from df3a05ba server compiles
new 639a9a88 fix sdk
new 2a22241a fix segment writing/reading
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/common/src/types/confirmation/mod.rs | 1 -
core/integration/tests/archiver/disk.rs | 6 +
core/integration/tests/state/mod.rs | 7 +-
.../tests/streaming/common/test_setup.rs | 5 +-
core/integration/tests/streaming/get_by_offset.rs | 2 +-
.../tests/streaming/get_by_timestamp.rs | 2 +-
core/integration/tests/streaming/messages.rs | 2 +-
core/integration/tests/streaming/mod.rs | 2 +-
core/integration/tests/streaming/partition.rs | 2 +-
core/integration/tests/streaming/segment.rs | 160 +--------------
.../tests/streaming/{system.rs => shard.rs} | 5 +-
core/integration/tests/streaming/snapshot.rs | 5 +-
core/integration/tests/streaming/stream.rs | 2 +-
core/integration/tests/streaming/topic.rs | 2 +-
core/integration/tests/streaming/topic_messages.rs | 8 +-
core/sdk/src/clients/client.rs | 10 +-
core/sdk/src/clients/consumer.rs | 10 +-
core/sdk/src/clients/consumer_builder.rs | 6 +-
core/sdk/src/clients/producer.rs | 6 +-
core/sdk/src/clients/producer_builder.rs | 6 +-
core/sdk/src/http/http_client.rs | 6 +-
core/server/Cargo.toml | 2 +-
core/server/src/archiver/disk.rs | 5 +-
core/server/src/archiver/mod.rs | 8 +-
core/server/src/archiver/s3.rs | 4 +
.../handlers/messages/send_messages_handler.rs | 1 -
core/server/src/bootstrap.rs | 1 -
.../src/compat/index_rebuilding/index_rebuilder.rs | 43 +++--
core/server/src/io/file.rs | 122 ++++++++++++
core/server/src/io/mod.rs | 3 +
core/server/src/io/reader.rs | 44 +++++
core/server/src/io/writer.rs | 57 ++++++
core/server/src/lib.rs | 1 +
core/server/src/shard/system/messages.rs | 3 +-
core/server/src/shard/system/storage.rs | 14 +-
core/server/src/state/file.rs | 25 ++-
core/server/src/state/mod.rs | 8 +-
core/server/src/streaming/partitions/messages.rs | 29 ++-
core/server/src/streaming/partitions/storage.rs | 44 +++--
core/server/src/streaming/persistence/persister.rs | 58 +++---
core/server/src/streaming/persistence/task.rs | 24 +--
.../streaming/segments/messages/messages_reader.rs | 30 ++-
.../streaming/segments/messages/messages_writer.rs | 94 +++------
core/server/src/streaming/segments/messages/mod.rs | 58 ++++--
.../streaming/segments/messages/persister_task.rs | 215 ---------------------
core/server/src/streaming/segments/segment.rs | 7 +-
.../src/streaming/segments/writing_messages.rs | 13 +-
core/server/src/streaming/topics/messages.rs | 10 +-
core/server/src/streaming/topics/persistence.rs | 2 +-
core/server/src/streaming/utils/file.rs | 12 +-
50 files changed, 538 insertions(+), 654 deletions(-)
rename core/integration/tests/streaming/{system.rs => shard.rs} (98%)
create mode 100644 core/server/src/io/file.rs
create mode 100644 core/server/src/io/mod.rs
create mode 100644 core/server/src/io/reader.rs
create mode 100644 core/server/src/io/writer.rs
delete mode 100644 core/server/src/streaming/segments/messages/persister_task.rs
(iggy) 02/02: fix segment writing/reading
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch io_uring_tpc
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 2a22241af50ebbafe8c39891262ccd93ef6051eb
Author: numinex <g....@gmail.com>
AuthorDate: Sat Jun 28 14:05:05 2025 +0200
fix segment writing/reading
---
core/common/src/types/confirmation/mod.rs | 1 -
core/integration/tests/archiver/disk.rs | 6 +
core/integration/tests/state/mod.rs | 7 +-
.../tests/streaming/common/test_setup.rs | 5 +-
core/integration/tests/streaming/get_by_offset.rs | 2 +-
.../tests/streaming/get_by_timestamp.rs | 2 +-
core/integration/tests/streaming/messages.rs | 2 +-
core/integration/tests/streaming/mod.rs | 2 +-
core/integration/tests/streaming/partition.rs | 2 +-
core/integration/tests/streaming/segment.rs | 160 +--------------
.../tests/streaming/{system.rs => shard.rs} | 5 +-
core/integration/tests/streaming/snapshot.rs | 5 +-
core/integration/tests/streaming/stream.rs | 2 +-
core/integration/tests/streaming/topic.rs | 2 +-
core/integration/tests/streaming/topic_messages.rs | 8 +-
core/server/Cargo.toml | 2 +-
core/server/src/archiver/disk.rs | 5 +-
core/server/src/archiver/mod.rs | 8 +-
core/server/src/archiver/s3.rs | 4 +
.../handlers/messages/send_messages_handler.rs | 1 -
core/server/src/bootstrap.rs | 1 -
.../src/compat/index_rebuilding/index_rebuilder.rs | 43 +++--
core/server/src/io/file.rs | 122 ++++++++++++
core/server/src/io/mod.rs | 3 +
core/server/src/io/reader.rs | 44 +++++
core/server/src/io/writer.rs | 57 ++++++
core/server/src/lib.rs | 1 +
core/server/src/shard/system/messages.rs | 3 +-
core/server/src/shard/system/storage.rs | 14 +-
core/server/src/state/file.rs | 25 ++-
core/server/src/state/mod.rs | 8 +-
core/server/src/streaming/partitions/messages.rs | 29 ++-
core/server/src/streaming/partitions/storage.rs | 44 +++--
core/server/src/streaming/persistence/persister.rs | 58 +++---
core/server/src/streaming/persistence/task.rs | 24 +--
.../streaming/segments/messages/messages_reader.rs | 30 ++-
.../streaming/segments/messages/messages_writer.rs | 94 +++------
core/server/src/streaming/segments/messages/mod.rs | 58 ++++--
.../streaming/segments/messages/persister_task.rs | 215 ---------------------
core/server/src/streaming/segments/segment.rs | 7 +-
.../src/streaming/segments/writing_messages.rs | 13 +-
core/server/src/streaming/topics/messages.rs | 10 +-
core/server/src/streaming/topics/persistence.rs | 2 +-
core/server/src/streaming/utils/file.rs | 12 +-
44 files changed, 516 insertions(+), 632 deletions(-)
diff --git a/core/common/src/types/confirmation/mod.rs b/core/common/src/types/confirmation/mod.rs
index 5289a627..bc26cffa 100644
--- a/core/common/src/types/confirmation/mod.rs
+++ b/core/common/src/types/confirmation/mod.rs
@@ -24,7 +24,6 @@ use strum::{Display, EnumString};
pub enum Confirmation {
#[default]
Wait,
- NoWait,
}
#[cfg(test)]
diff --git a/core/integration/tests/archiver/disk.rs b/core/integration/tests/archiver/disk.rs
index f676d23f..00901190 100644
--- a/core/integration/tests/archiver/disk.rs
+++ b/core/integration/tests/archiver/disk.rs
@@ -110,11 +110,16 @@ async fn should_fail_when_file_to_archive_does_not_exist() {
}
async fn create_file(path: &str, content: &str) {
+ // TODO: Fixme
+ /*
let mut file = file::overwrite(path).await.unwrap();
file.write_all(content.as_bytes()).await.unwrap();
+ */
}
async fn assert_archived_file(file_to_archive_path: &str, archived_file_path: &str, content: &str) {
+ // TODO: Fixme
+ /*
assert!(Path::new(&file_to_archive_path).exists());
assert!(Path::new(&archived_file_path).exists());
let archived_file = file::open(archived_file_path).await;
@@ -126,4 +131,5 @@ async fn assert_archived_file(file_to_archive_path: &str, archived_file_path: &s
.await
.unwrap();
assert_eq!(content, archived_file_content);
+ */
}
diff --git a/core/integration/tests/state/mod.rs b/core/integration/tests/state/mod.rs
index d6729d2e..e125a395 100644
--- a/core/integration/tests/state/mod.rs
+++ b/core/integration/tests/state/mod.rs
@@ -50,11 +50,8 @@ impl StateSetup {
let version = SemanticVersion::from_str("1.2.3").unwrap();
let persister = PersisterKind::FileWithSync(FileWithSyncPersister {});
- let encryptor = encryption_key.map(|key| {
- Arc::new(EncryptorKind::Aes256Gcm(
- Aes256GcmEncryptor::new(key).unwrap(),
- ))
- });
+ let encryptor = encryption_key
+ .map(|key| EncryptorKind::Aes256Gcm(Aes256GcmEncryptor::new(key).unwrap()));
let state = FileState::new(
&messages_file_path,
&version,
diff --git a/core/integration/tests/streaming/common/test_setup.rs b/core/integration/tests/streaming/common/test_setup.rs
index c5971762..19363249 100644
--- a/core/integration/tests/streaming/common/test_setup.rs
+++ b/core/integration/tests/streaming/common/test_setup.rs
@@ -20,13 +20,14 @@ 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 {
pub config: Arc<SystemConfig>,
- pub storage: Arc<SystemStorage>,
+ pub storage: Rc<SystemStorage>,
}
impl TestSetup {
@@ -42,7 +43,7 @@ impl TestSetup {
let config = Arc::new(config);
fs::create_dir(config.get_system_path()).await.unwrap();
let persister = PersisterKind::FileWithSync(FileWithSyncPersister {});
- let storage = Arc::new(SystemStorage::new(config.clone(), Arc::new(persister)));
+ let storage = Rc::new(SystemStorage::new(config.clone(), Arc::new(persister)));
MemoryPool::init_pool(config.clone());
TestSetup { config, storage }
}
diff --git a/core/integration/tests/streaming/get_by_offset.rs b/core/integration/tests/streaming/get_by_offset.rs
index 68690c4f..df9f8f40 100644
--- a/core/integration/tests/streaming/get_by_offset.rs
+++ b/core/integration/tests/streaming/get_by_offset.rs
@@ -195,7 +195,7 @@ async fn test_get_messages_by_offset(
let batch = IggyMessagesBatchMut::from_messages(messages_slice_to_append, messages_size);
assert_eq!(batch.count(), batch_len);
- partition.append_messages(batch, None).await.unwrap();
+ partition.append_messages(batch).await.unwrap();
batch_offsets.push(partition.current_offset);
current_pos += batch_len as usize;
diff --git a/core/integration/tests/streaming/get_by_timestamp.rs b/core/integration/tests/streaming/get_by_timestamp.rs
index ae0ab88b..d0cc2e5b 100644
--- a/core/integration/tests/streaming/get_by_timestamp.rs
+++ b/core/integration/tests/streaming/get_by_timestamp.rs
@@ -203,7 +203,7 @@ async fn test_get_messages_by_timestamp(
let batch = IggyMessagesBatchMut::from_messages(messages_slice_to_append, messages_size);
assert_eq!(batch.count(), batch_len);
partition
- .append_messages(batch, Some(Confirmation::Wait))
+ .append_messages(batch)
.await
.unwrap();
diff --git a/core/integration/tests/streaming/messages.rs b/core/integration/tests/streaming/messages.rs
index e3529ab7..6ab6ca5b 100644
--- a/core/integration/tests/streaming/messages.rs
+++ b/core/integration/tests/streaming/messages.rs
@@ -104,7 +104,7 @@ async fn should_persist_messages_and_then_load_them_from_disk() {
.map(|msg| msg.get_size_bytes().as_bytes_u32())
.sum::<u32>();
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition.append_messages(batch, None).await.unwrap();
+ partition.append_messages(batch).await.unwrap();
assert_eq!(
partition.unsaved_messages_count, 0,
"Expected unsaved messages count to be 0, but got {}",
diff --git a/core/integration/tests/streaming/mod.rs b/core/integration/tests/streaming/mod.rs
index b5df4c0a..f51b875b 100644
--- a/core/integration/tests/streaming/mod.rs
+++ b/core/integration/tests/streaming/mod.rs
@@ -26,9 +26,9 @@ mod get_by_timestamp;
mod messages;
mod partition;
mod segment;
+mod shard;
mod snapshot;
mod stream;
-mod system;
mod topic;
mod topic_messages;
diff --git a/core/integration/tests/streaming/partition.rs b/core/integration/tests/streaming/partition.rs
index 5a7e5866..4fff0d04 100644
--- a/core/integration/tests/streaming/partition.rs
+++ b/core/integration/tests/streaming/partition.rs
@@ -196,7 +196,7 @@ async fn should_purge_existing_partition_on_disk() {
.map(|msg| msg.get_size_bytes().as_bytes_u32())
.sum();
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition.append_messages(batch, None).await.unwrap();
+ partition.append_messages(batch).await.unwrap();
let loaded_messages = partition.get_messages_by_offset(0, 100).await.unwrap();
assert_eq!(loaded_messages.count(), messages_count);
partition.purge().await.unwrap();
diff --git a/core/integration/tests/streaming/segment.rs b/core/integration/tests/streaming/segment.rs
index 403418d8..c5ec065e 100644
--- a/core/integration/tests/streaming/segment.rs
+++ b/core/integration/tests/streaming/segment.rs
@@ -25,7 +25,6 @@ use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig};
use server::configs::system::{PartitionConfig, SegmentConfig, SystemConfig};
use server::streaming::segments::*;
use server::streaming::session::Session;
-use server::streaming::systems::system::System;
use std::fs::DirEntry;
use std::net::{Ipv4Addr, SocketAddr};
use std::str::FromStr;
@@ -192,7 +191,7 @@ async fn should_persist_and_load_segment_with_messages() {
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
segment.append_batch(0, batch, None).await.unwrap();
- segment.persist_messages(None).await.unwrap();
+ segment.persist_messages().await.unwrap();
let mut loaded_segment = Segment::create(
stream_id,
topic_id,
@@ -216,158 +215,6 @@ async fn should_persist_and_load_segment_with_messages() {
assert_eq!(messages.count(), messages_count);
}
-#[tokio::test]
-async fn should_persist_and_load_segment_with_messages_with_nowait_confirmation() {
- let setup = TestSetup::init_with_config(SystemConfig {
- segment: SegmentConfig {
- server_confirmation: Confirmation::NoWait,
- ..Default::default()
- },
- ..Default::default()
- })
- .await;
- let stream_id = 1;
- let topic_id = 2;
- let partition_id = 3;
- let start_offset = 0;
- let mut segment = Segment::create(
- stream_id,
- topic_id,
- partition_id,
- start_offset,
- setup.config.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- true,
- );
-
- setup
- .create_partition_directory(stream_id, topic_id, partition_id)
- .await;
- segment.persist().await.unwrap();
- assert_persisted_segment(
- &setup
- .config
- .get_partition_path(stream_id, topic_id, partition_id),
- start_offset,
- )
- .await;
- let messages_count = 10;
- let mut messages = Vec::new();
- let mut messages_size = 0;
- for i in 0..messages_count {
- let message = IggyMessage::builder()
- .id(i as u128)
- .payload(Bytes::from("test"))
- .build()
- .expect("Failed to create message");
- messages_size += message.get_size_bytes().as_bytes_u32();
- messages.push(message);
- }
- let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- segment.append_batch(0, batch, None).await.unwrap();
- segment
- .persist_messages(Some(Confirmation::NoWait))
- .await
- .unwrap();
- sleep(Duration::from_millis(200)).await;
- let mut loaded_segment = Segment::create(
- stream_id,
- topic_id,
- partition_id,
- start_offset,
- setup.config.clone(),
- IggyExpiry::NeverExpire,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- false,
- );
- loaded_segment.load_from_disk().await.unwrap();
- let messages = loaded_segment
- .get_messages_by_offset(0, messages_count)
- .await
- .unwrap();
- assert_eq!(messages.count(), messages_count);
-}
-
-#[tokio::test]
-async fn given_all_expired_messages_segment_should_be_expired() {
- let config = SystemConfig {
- partition: PartitionConfig {
- enforce_fsync: true,
- ..Default::default()
- },
- segment: SegmentConfig {
- size: IggyByteSize::from_str("10B").unwrap(), // small size to force expiration
- ..Default::default()
- },
- ..Default::default()
- };
- let setup = TestSetup::init_with_config(config).await;
- let stream_id = 1;
- let topic_id = 2;
- let partition_id = 3;
- let start_offset = 0;
- let message_expiry_us = 100000;
- let message_expiry = message_expiry_us.into();
- let mut segment = Segment::create(
- stream_id,
- topic_id,
- partition_id,
- start_offset,
- setup.config.clone(),
- message_expiry,
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- Arc::new(AtomicU64::new(0)),
- true,
- );
-
- setup
- .create_partition_directory(stream_id, topic_id, partition_id)
- .await;
- segment.persist().await.unwrap();
- assert_persisted_segment(
- &setup
- .config
- .get_partition_path(stream_id, topic_id, partition_id),
- start_offset,
- )
- .await;
- let messages_count = 10;
- let mut messages = Vec::new();
- let mut messages_size = 0;
- for i in 0..messages_count {
- let message = IggyMessage::builder()
- .id(i as u128)
- .payload(Bytes::from("test"))
- .build()
- .expect("Failed to create message");
- messages_size += message.get_size_bytes().as_bytes_u32();
- messages.push(message);
- }
- let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- segment.append_batch(0, batch, None).await.unwrap();
- segment.persist_messages(None).await.unwrap();
- let not_expired_ts = IggyTimestamp::now();
- let expired_ts = not_expired_ts + IggyDuration::from(message_expiry_us + 1);
-
- assert!(segment.is_expired(expired_ts).await);
- assert!(!segment.is_expired(not_expired_ts).await);
-}
-
#[tokio::test]
async fn given_at_least_one_not_expired_message_segment_should_not_be_expired() {
let config = SystemConfig {
@@ -444,7 +291,7 @@ async fn given_at_least_one_not_expired_message_segment_should_not_be_expired()
let second_message_expired_ts =
IggyTimestamp::now() + IggyDuration::from(message_expiry_us * 2);
- segment.persist_messages(None).await.unwrap();
+ segment.persist_messages().await.unwrap();
assert!(
!segment.is_expired(nothing_expired_ts).await,
@@ -460,6 +307,8 @@ async fn given_at_least_one_not_expired_message_segment_should_not_be_expired()
);
}
+/*
+//TODO: Fixme use shard instead of system
#[tokio::test]
async fn should_delete_persisted_segments() -> Result<(), Box<dyn std::error::Error>> {
let config = SystemConfig {
@@ -648,6 +497,7 @@ 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/integration/tests/streaming/system.rs b/core/integration/tests/streaming/shard.rs
similarity index 98%
rename from core/integration/tests/streaming/system.rs
rename to core/integration/tests/streaming/shard.rs
index ce950e84..454a5841 100644
--- a/core/integration/tests/streaming/system.rs
+++ b/core/integration/tests/streaming/shard.rs
@@ -20,10 +20,11 @@ use crate::streaming::common::test_setup::TestSetup;
use iggy::prelude::Identifier;
use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig};
use server::streaming::session::Session;
-use server::streaming::systems::system::System;
use std::net::{Ipv4Addr, SocketAddr};
use tokio::fs;
+//TODO: Fix me use shard instead of system
+/*
#[tokio::test]
async fn should_initialize_system_and_base_directories() {
let setup = TestSetup::init().await;
@@ -127,3 +128,5 @@ async fn assert_persisted_stream(streams_path: &str, stream_id: u32) {
let stream_metadata = fs::metadata(stream_path).await.unwrap();
assert!(stream_metadata.is_dir());
}
+
+*/
diff --git a/core/integration/tests/streaming/snapshot.rs b/core/integration/tests/streaming/snapshot.rs
index 52b9419a..ae576c6a 100644
--- a/core/integration/tests/streaming/snapshot.rs
+++ b/core/integration/tests/streaming/snapshot.rs
@@ -20,11 +20,12 @@ use crate::streaming::common::test_setup::TestSetup;
use iggy::prelude::{SnapshotCompression, SystemSnapshotType};
use server::configs::server::{DataMaintenanceConfig, PersonalAccessTokenConfig};
use server::streaming::session::Session;
-use server::streaming::systems::system::System;
use std::io::{Cursor, Read};
use std::net::{Ipv4Addr, SocketAddr};
use zip::ZipArchive;
+//TODO: Fix me use shard instead of system
+/*
#[tokio::test]
async fn should_create_snapshot_file() {
let setup = TestSetup::init().await;
@@ -55,3 +56,5 @@ async fn should_create_snapshot_file() {
test_file.read_to_string(&mut test_content).unwrap();
assert_eq!(test_content, "test\n");
}
+
+*/
diff --git a/core/integration/tests/streaming/stream.rs b/core/integration/tests/streaming/stream.rs
index 417c00d4..5b340750 100644
--- a/core/integration/tests/streaming/stream.rs
+++ b/core/integration/tests/streaming/stream.rs
@@ -146,7 +146,7 @@ async fn should_purge_existing_stream_on_disk() {
.get_topic(&Identifier::numeric(topic_id).unwrap())
.unwrap();
topic
- .append_messages(&Partitioning::partition_id(1), batch, None)
+ .append_messages(&Partitioning::partition_id(1), batch)
.await
.unwrap();
let (_, loaded_messages) = topic
diff --git a/core/integration/tests/streaming/topic.rs b/core/integration/tests/streaming/topic.rs
index 3ae2c5f0..f75ffe2b 100644
--- a/core/integration/tests/streaming/topic.rs
+++ b/core/integration/tests/streaming/topic.rs
@@ -226,7 +226,7 @@ async fn should_purge_existing_topic_on_disk() {
.sum::<u32>();
let batch = IggyMessagesBatchMut::from_messages(&messages, batch_size);
topic
- .append_messages(&Partitioning::partition_id(1), batch, None)
+ .append_messages(&Partitioning::partition_id(1), batch)
.await
.unwrap();
let (_, loaded_messages) = topic
diff --git a/core/integration/tests/streaming/topic_messages.rs b/core/integration/tests/streaming/topic_messages.rs
index f716ef9d..f21263c4 100644
--- a/core/integration/tests/streaming/topic_messages.rs
+++ b/core/integration/tests/streaming/topic_messages.rs
@@ -63,7 +63,7 @@ async fn assert_polling_messages() {
.sum::<IggyByteSize>();
let batch = IggyMessagesBatchMut::from_messages(&messages, batch_size.as_bytes_u32());
topic
- .append_messages(&partitioning, batch, None)
+ .append_messages(&partitioning, batch)
.await
.unwrap();
@@ -110,7 +110,7 @@ async fn given_key_none_messages_should_be_appended_to_the_next_partition_using_
batch_size.as_bytes_u32(),
);
topic
- .append_messages(&partitioning, messages, None)
+ .append_messages(&partitioning, messages)
.await
.unwrap();
}
@@ -139,7 +139,7 @@ async fn given_key_partition_id_messages_should_be_appended_to_the_chosen_partit
batch_size.as_bytes_u32(),
);
topic
- .append_messages(&partitioning, messages, None)
+ .append_messages(&partitioning, messages)
.await
.unwrap();
}
@@ -172,7 +172,7 @@ async fn given_key_messages_key_messages_should_be_appended_to_the_calculated_pa
batch_size.as_bytes_u32(),
);
topic
- .append_messages(&partitioning, messages, None)
+ .append_messages(&partitioning, messages)
.await
.unwrap();
}
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index 1246e049..59a10c29 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -66,7 +66,7 @@ lending-iterator = "0.1.7"
hash32 = "1.0.0"
mimalloc = { workspace = true, optional = true }
moka = { version = "0.12.10", features = ["future"] }
-monoio = { version = "0.2.4", features = ["mkdirat", "unlinkat"] }
+monoio = { version = "0.2.4", features = ["mkdirat", "unlinkat", "renameat"] }
monoio-native-tls = "0.4.0"
nix = { version = "0.30", features = ["fs"] }
once_cell = "1.21.3"
diff --git a/core/server/src/archiver/disk.rs b/core/server/src/archiver/disk.rs
index c0affe7f..0eafa20d 100644
--- a/core/server/src/archiver/disk.rs
+++ b/core/server/src/archiver/disk.rs
@@ -70,7 +70,8 @@ impl Archiver for DiskArchiver {
files: &[&str],
base_directory: Option<String>,
) -> Result<(), ArchiverError> {
- debug!("Archiving files on disk: {:?}", files);
+ //TODO: Fixme figure this out, we can't use tokio methods there.
+ /* debug!("Archiving files on disk: {:?}", files);
for file in files {
debug!("Archiving file: {file}");
let source = Path::new(file);
@@ -93,7 +94,7 @@ impl Archiver for DiskArchiver {
})?;
debug!("Archived file: {file} at: {destination_path}");
}
-
+ */
Ok(())
}
}
diff --git a/core/server/src/archiver/mod.rs b/core/server/src/archiver/mod.rs
index 1ecf48b7..6d61dcad 100644
--- a/core/server/src/archiver/mod.rs
+++ b/core/server/src/archiver/mod.rs
@@ -53,18 +53,18 @@ impl FromStr for ArchiverKindType {
}
}
-pub trait Archiver: Send {
- fn init(&self) -> impl Future<Output = Result<(), ArchiverError>> + Send;
+pub trait Archiver {
+ fn init(&self) -> impl Future<Output = Result<(), ArchiverError>>;
fn is_archived(
&self,
file: &str,
base_directory: Option<String>,
- ) -> impl Future<Output = Result<bool, ArchiverError>> + Send;
+ ) -> impl Future<Output = Result<bool, ArchiverError>>;
fn archive(
&self,
files: &[&str],
base_directory: Option<String>,
- ) -> impl Future<Output = Result<(), ArchiverError>> + Send;
+ ) -> impl Future<Output = Result<(), ArchiverError>>;
}
#[derive(Debug)]
diff --git a/core/server/src/archiver/s3.rs b/core/server/src/archiver/s3.rs
index 036945d9..bbc9cd3b 100644
--- a/core/server/src/archiver/s3.rs
+++ b/core/server/src/archiver/s3.rs
@@ -157,6 +157,9 @@ impl Archiver for S3Archiver {
let base_directory = base_directory.as_deref().unwrap_or_default();
let destination = Path::new(&base_directory).join(path);
let destination_path = destination.to_str().unwrap_or_default().to_owned();
+ // TODO: Fixme figure this out.
+ // The `put_object_stream` method requires `AsyncRead` trait from tokio as its reader.
+ /*
let response = self
.bucket
.put_object_stream(&mut file, destination_path)
@@ -188,6 +191,7 @@ impl Archiver for S3Archiver {
return Err(ArchiverError::CannotArchiveFile {
file_path: (*path).to_string(),
});
+ */
}
Ok(())
}
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 5603dfe9..d9a4770f 100644
--- a/core/server/src/binary/handlers/messages/send_messages_handler.rs
+++ b/core/server/src/binary/handlers/messages/send_messages_handler.rs
@@ -114,7 +114,6 @@ impl ServerCommandHandler for SendMessages {
&self.topic_id,
&self.partitioning,
batch,
- None,
)
.await?;
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index 4f356a9b..ebf8946e 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -122,7 +122,6 @@ where
pub fn create_shard_executor() -> Runtime<TimeDriver<monoio::IoUringDriver>> {
// TODO: Figure out what else we could tweak there
// We for sure want to disable the userspace interrupts on new cq entry (set_coop_taskrun)
- // let urb = io_uring::IoUring::builder();
// TODO: Shall we make the size of ring be configureable ?
let builder = monoio::RuntimeBuilder::<monoio::IoUringDriver>::new()
//.uring_builder(urb.setup_coop_taskrun()) // broken shit.
diff --git a/core/server/src/compat/index_rebuilding/index_rebuilder.rs b/core/server/src/compat/index_rebuilding/index_rebuilder.rs
index 542620ee..1eebdacd 100644
--- a/core/server/src/compat/index_rebuilding/index_rebuilder.rs
+++ b/core/server/src/compat/index_rebuilding/index_rebuilder.rs
@@ -16,11 +16,13 @@
* under the License.
*/
-use crate::server_error::CompatError;
+use crate::io::file::IggyFile;
+use crate::io::writer::IggyWriter;
use crate::streaming::utils::file;
+use crate::{io::reader::IggyReader, server_error::CompatError};
use iggy_common::{IGGY_MESSAGE_HEADER_SIZE, IggyMessageHeader};
-use std::io::SeekFrom;
-use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader, BufWriter};
+use monoio::io::{AsyncReadRent, AsyncReadRentExt, AsyncWriteRent, AsyncWriteRentExt};
+use std::io::{Seek, SeekFrom};
pub struct IndexRebuilder {
pub messages_file_path: String,
@@ -38,16 +40,17 @@ impl IndexRebuilder {
}
async fn read_message_header(
- reader: &mut BufReader<tokio::fs::File>,
+ reader: &mut IggyReader<IggyFile>,
) -> Result<IggyMessageHeader, std::io::Error> {
- let mut buf = [0u8; IGGY_MESSAGE_HEADER_SIZE];
- reader.read_exact(&mut buf).await?;
- IggyMessageHeader::from_raw_bytes(&buf)
+ let buf = [0u8; IGGY_MESSAGE_HEADER_SIZE];
+ let (result, buf) = reader.read_exact(Box::new(buf)).await;
+ result?;
+ IggyMessageHeader::from_raw_bytes(&*buf)
.map_err(|_| std::io::Error::from(std::io::ErrorKind::InvalidData))
}
async fn write_index_entry(
- writer: &mut BufWriter<tokio::fs::File>,
+ writer: &mut IggyWriter<IggyFile>,
header: &IggyMessageHeader,
position: usize,
start_offset: u64,
@@ -55,20 +58,26 @@ impl IndexRebuilder {
// Write offset (4 bytes) - base_offset + last_offset_delta - start_offset
let offset = start_offset - header.offset;
debug_assert!(offset <= u32::MAX as u64);
- writer.write_u32_le(offset as u32).await?;
+ let (result, _) = writer.write_all(Box::new(offset.to_le_bytes())).await;
+ result?;
// Write position (4 bytes)
- writer.write_u32_le(position as u32).await?;
+ let (result, _) = writer.write_all(Box::new(position.to_le_bytes())).await;
+ result?;
// Write timestamp (8 bytes)
- writer.write_u64_le(header.timestamp).await?;
+ let (result, _) = writer
+ .write_all(Box::new(header.timestamp.to_le_bytes()))
+ .await;
+ result?;
Ok(())
}
pub async fn rebuild(&self) -> Result<(), CompatError> {
- let mut reader = BufReader::new(file::open(&self.messages_file_path).await?);
- let mut writer = BufWriter::new(file::overwrite(&self.index_path).await?);
+ let mut reader =
+ IggyReader::new(IggyFile::new(file::open(&self.messages_file_path).await?));
+ let mut writer = IggyWriter::new(IggyFile::new(file::overwrite(&self.index_path).await?));
let mut position = 0;
let mut next_position;
@@ -84,11 +93,9 @@ impl IndexRebuilder {
.await?;
// Skip message payload and headers
- reader
- .seek(SeekFrom::Current(
- header.payload_length as i64 + header.user_headers_length as i64,
- ))
- .await?;
+ reader.seek(SeekFrom::Current(
+ header.payload_length as i64 + header.user_headers_length as i64,
+ ))?;
// Update position for next iteration
position = next_position;
diff --git a/core/server/src/io/file.rs b/core/server/src/io/file.rs
new file mode 100644
index 00000000..61e51023
--- /dev/null
+++ b/core/server/src/io/file.rs
@@ -0,0 +1,122 @@
+use std::io::SeekFrom;
+
+use monoio::{
+ BufResult,
+ buf::{IoBuf, IoBufMut, IoVecBuf, IoVecBufMut, IoVecWrapper},
+ fs::{File, Metadata},
+ io::{AsyncReadRent, AsyncWriteRent},
+};
+
+/// Wrapper around `monoio::fs::File` to provide a consistent API for reading and writing files
+/// in an asynchronous context. This struct maintains the current position in the file and provides
+/// methods to read and write data at specific positions.
+#[derive(Debug)]
+pub struct IggyFile {
+ file: File,
+ position: u64,
+}
+
+impl From<File> for IggyFile {
+ fn from(file: File) -> Self {
+ Self { file, position: 0 }
+ }
+}
+
+impl std::io::Seek for IggyFile {
+ /// This method doesn't do bound checking aswell as, the `End` variant is not supported.
+ fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
+ self.position = match pos {
+ SeekFrom::Start(n) => n as i64,
+ SeekFrom::End(_) => unreachable!("End variant is not supported in IggyFile"),
+ SeekFrom::Current(n) => self.position as i64 + n,
+ } as u64;
+ Ok(self.position)
+ }
+}
+
+impl IggyFile {
+ pub fn new(file: File) -> Self {
+ Self { file, position: 0 }
+ }
+
+ pub async fn metadata(&self) -> std::io::Result<Metadata> {
+ self.file.metadata().await
+ }
+
+ pub async fn write_at<T: IoBuf>(&self, buf: T, pos: usize) -> BufResult<usize, T> {
+ self.file.write_at(buf, pos as u64).await
+ }
+
+ pub async fn write_all_at<T: IoBuf>(&self, buf: T, pos: u64) -> BufResult<(), T> {
+ self.file.write_all_at(buf, pos).await
+ }
+
+ pub async fn read_exact_at<T: IoBufMut>(&self, buf: T, pos: u64) -> BufResult<(), T> {
+ self.file.read_exact_at(buf, pos).await
+ }
+
+ pub async fn read_at<T: IoBufMut>(&self, buf: T, pos: usize) -> BufResult<usize, T> {
+ self.file.read_at(buf, pos as u64).await
+ }
+
+ pub async fn sync_all(&self) -> std::io::Result<()> {
+ self.file.sync_all().await
+ }
+}
+
+impl AsyncReadRent for IggyFile {
+ /// Reads `exactly` `buf.len()` bytes into the buffer.
+ async fn read<T: IoBufMut>(&mut self, buf: T) -> BufResult<usize, T> {
+ let (res, buf) = self.file.read_at(buf, self.position).await;
+ let n = match res {
+ Ok(n) => n,
+ Err(e) => return (Err(e), buf),
+ };
+ self.position += n as u64;
+ (Ok(n), buf)
+ }
+
+ //TODO(numinex) - maybe implement this ?
+ fn readv<T: IoVecBufMut>(&mut self, buf: T) -> impl Future<Output = BufResult<usize, T>> {
+ async move { (Ok(0), buf) }
+ }
+}
+
+impl AsyncWriteRent for IggyFile {
+ /// Writes entire buffer to the file at the current position.
+ async fn write<B: IoBuf>(&mut self, buf: B) -> BufResult<usize, B> {
+ let (res, buf) = self.file.write_at(buf, self.position).await;
+ let n = match res {
+ Ok(n) => n,
+ Err(e) => return (Err(e), buf),
+ };
+ self.position += n as u64;
+ (Ok(n), buf)
+ }
+
+ async fn writev<T: IoVecBuf>(&mut self, buf_vec: T) -> BufResult<usize, T> {
+ let slice = match IoVecWrapper::new(buf_vec) {
+ Ok(slice) => slice,
+ Err(buf) => return (Ok(0), buf),
+ };
+
+ let (result, slice) = self.write(slice).await;
+ (result, slice.into_inner())
+ }
+
+ async fn flush(&mut self) -> std::io::Result<()> {
+ self.file.sync_all().await
+ }
+
+ //TODO(numinex) - How to implement this ?
+ async fn shutdown(&mut self) -> std::io::Result<()> {
+ panic!("shutdown is not supported for IggyFile");
+ /*
+ self.file.sync_all().await;
+ unsafe {
+ let file = *self.file;
+ file.close().await
+ }
+ */
+ }
+}
diff --git a/core/server/src/io/mod.rs b/core/server/src/io/mod.rs
new file mode 100644
index 00000000..ea2bfab5
--- /dev/null
+++ b/core/server/src/io/mod.rs
@@ -0,0 +1,3 @@
+pub mod file;
+pub mod reader;
+pub mod writer;
diff --git a/core/server/src/io/reader.rs b/core/server/src/io/reader.rs
new file mode 100644
index 00000000..63443214
--- /dev/null
+++ b/core/server/src/io/reader.rs
@@ -0,0 +1,44 @@
+use std::io::{Seek, SeekFrom};
+
+use monoio::{
+ BufResult,
+ buf::{IoBufMut, IoVecBufMut},
+ io::{AsyncReadRent, BufReader},
+};
+
+pub struct IggyReader<R: AsyncReadRent + Seek> {
+ inner: BufReader<R>,
+}
+
+impl<R: AsyncReadRent + Seek> IggyReader<R> {
+ pub fn new(reader: R) -> Self {
+ Self {
+ inner: BufReader::new(reader),
+ }
+ }
+
+ pub fn with_capacity(capacity: usize, reader: R) -> Self {
+ Self {
+ inner: BufReader::with_capacity(capacity, reader),
+ }
+ }
+}
+
+impl<R: AsyncReadRent + Seek> Seek for IggyReader<R> {
+ fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
+ self.inner.get_mut().seek(pos)
+ }
+}
+
+impl<R> AsyncReadRent for IggyReader<R>
+where
+ R: AsyncReadRent + Seek,
+{
+ fn read<T: IoBufMut>(&mut self, buf: T) -> impl Future<Output = BufResult<usize, T>> {
+ self.inner.read(buf)
+ }
+
+ fn readv<T: IoVecBufMut>(&mut self, buf: T) -> impl Future<Output = BufResult<usize, T>> {
+ self.inner.readv(buf)
+ }
+}
diff --git a/core/server/src/io/writer.rs b/core/server/src/io/writer.rs
new file mode 100644
index 00000000..e249e405
--- /dev/null
+++ b/core/server/src/io/writer.rs
@@ -0,0 +1,57 @@
+use std::io::{Seek, SeekFrom};
+
+use monoio::{
+ BufResult,
+ io::{AsyncWriteRent, BufWriter},
+};
+
+pub struct IggyWriter<W: AsyncWriteRent + Seek> {
+ inner: BufWriter<W>,
+}
+
+impl<W: AsyncWriteRent + Seek> IggyWriter<W> {
+ pub fn new(writer: W) -> Self {
+ Self {
+ inner: BufWriter::new(writer),
+ }
+ }
+
+ pub fn with_capacity(capacity: usize, writer: W) -> Self {
+ Self {
+ inner: BufWriter::with_capacity(capacity, writer),
+ }
+ }
+}
+
+impl<W: AsyncWriteRent + Seek> Seek for IggyWriter<W> {
+ fn seek(&mut self, pos: SeekFrom) -> std::io::Result<u64> {
+ self.inner.get_mut().seek(pos)
+ }
+}
+
+impl<W> AsyncWriteRent for IggyWriter<W>
+where
+ W: AsyncWriteRent + Seek,
+{
+ fn write<T: monoio::buf::IoBuf>(
+ &mut self,
+ buf: T,
+ ) -> impl Future<Output = BufResult<usize, T>> {
+ self.inner.write(buf)
+ }
+
+ fn writev<T: monoio::buf::IoVecBuf>(
+ &mut self,
+ buf_vec: T,
+ ) -> impl Future<Output = BufResult<usize, T>> {
+ self.inner.writev(buf_vec)
+ }
+
+ fn flush(&mut self) -> impl Future<Output = std::io::Result<()>> {
+ self.inner.flush()
+ }
+
+ fn shutdown(&mut self) -> impl Future<Output = std::io::Result<()>> {
+ self.inner.shutdown()
+ }
+}
diff --git a/core/server/src/lib.rs b/core/server/src/lib.rs
index f618d3e0..68be8afd 100644
--- a/core/server/src/lib.rs
+++ b/core/server/src/lib.rs
@@ -36,6 +36,7 @@ pub mod channels;
pub(crate) mod compat;
pub mod configs;
pub mod http;
+pub mod io;
pub mod log;
pub mod quic;
pub mod server_error;
diff --git a/core/server/src/shard/system/messages.rs b/core/server/src/shard/system/messages.rs
index 7668566a..9a23a9bb 100644
--- a/core/server/src/shard/system/messages.rs
+++ b/core/server/src/shard/system/messages.rs
@@ -105,7 +105,6 @@ impl IggyShard {
topic_id: &Identifier,
partitioning: &Partitioning,
messages: IggyMessagesBatchMut,
- confirmation: Option<Confirmation>,
) -> Result<(), IggyError> {
self.ensure_authenticated(session)?;
let stream = self.get_stream(stream_id).with_error_context(|error| {
@@ -133,7 +132,7 @@ impl IggyShard {
};
topic
- .append_messages(partitioning, messages, confirmation)
+ .append_messages(partitioning, messages)
.await?;
self.metrics.increment_messages(messages_count as u64);
diff --git a/core/server/src/shard/system/storage.rs b/core/server/src/shard/system/storage.rs
index cc413583..3d895d4c 100644
--- a/core/server/src/shard/system/storage.rs
+++ b/core/server/src/shard/system/storage.rs
@@ -17,6 +17,7 @@
*/
use super::COMPONENT;
+use crate::io::file::IggyFile;
use crate::shard::system::info::SystemInfo;
use crate::streaming::persistence::persister::PersisterKind;
use crate::streaming::storage::SystemInfoStorage;
@@ -25,6 +26,7 @@ use crate::streaming::utils::file;
use anyhow::Context;
use error_set::ErrContext;
use iggy_common::IggyError;
+use monoio::io::AsyncReadRentExt;
use std::sync::Arc;
use tokio::io::AsyncReadExt;
use tracing::info;
@@ -48,7 +50,7 @@ impl SystemInfoStorage for FileSystemInfoStorage {
return Err(IggyError::ResourceNotFound(self.path.to_owned()));
}
- let mut file = file.unwrap();
+ let file = file.unwrap();
let file_size = file
.metadata()
.await
@@ -60,13 +62,15 @@ impl SystemInfoStorage for FileSystemInfoStorage {
})
.map_err(|_| IggyError::CannotReadFileMetadata)?
.len() as usize;
+
+ let mut file = IggyFile::new(file);
let mut buffer = PooledBuffer::with_capacity(file_size);
buffer.put_bytes(0, file_size);
- file.read_exact(&mut buffer)
- .await
+ let (result, buffer) = file.read_exact(buffer).await;
+ result
.with_error_context(|error| {
format!(
- "{COMPONENT} (error: {error}) - failed to read file content from path: {}",
+ "{COMPONENT} Failed to read system info from file at path: {} (error: {error})",
self.path
)
})
@@ -83,7 +87,7 @@ impl SystemInfoStorage for FileSystemInfoStorage {
.with_context(|| "Failed to serialize system info")
.map_err(|_| IggyError::CannotSerializeResource)?;
self.persister
- .overwrite(&self.path, &data)
+ .overwrite(&self.path, data)
.await
.with_error_context(|error| {
format!(
diff --git a/core/server/src/state/file.rs b/core/server/src/state/file.rs
index 41117c3a..b845d877 100644
--- a/core/server/src/state/file.rs
+++ b/core/server/src/state/file.rs
@@ -16,6 +16,8 @@
* under the License.
*/
+use crate::io::file::IggyFile;
+use crate::io::reader::IggyReader;
use crate::state::command::EntryCommand;
use crate::state::{COMPONENT, State, StateEntry};
use crate::streaming::persistence::persister::PersisterKind;
@@ -28,11 +30,11 @@ use iggy_common::EncryptorKind;
use iggy_common::IggyByteSize;
use iggy_common::IggyError;
use iggy_common::IggyTimestamp;
+use monoio::io::AsyncReadRentExt;
use std::fmt::Debug;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
-use tokio::io::{AsyncReadExt, BufReader};
use tracing::{debug, error, info};
pub const BUF_READER_CAPACITY_BYTES: usize = 512 * 1000;
@@ -115,6 +117,7 @@ impl State for FileState {
)
})
.map_err(|_| IggyError::CannotReadFile)?;
+ let file = IggyFile::new(file);
let file_size = file
.metadata()
.await
@@ -137,7 +140,7 @@ impl State for FileState {
);
let mut entries = Vec::new();
let mut total_size: u64 = 0;
- let mut reader = BufReader::with_capacity(BUF_READER_CAPACITY_BYTES, file);
+ let mut reader = IggyReader::with_capacity(BUF_READER_CAPACITY_BYTES, file);
let mut current_index = 0;
let mut entries_count = 0;
loop {
@@ -215,9 +218,10 @@ impl State for FileState {
total_size += 4;
let mut context = BytesMut::with_capacity(context_length);
context.put_bytes(0, context_length);
- reader
- .read_exact(&mut context)
- .await
+ let (result, context) = reader.read_exact(context).await;
+
+ result
+ .with_error_context(|error| format!("{FILE_STATE_PARSE_ERROR} code. {error}"))
.map_err(|_| IggyError::CannotReadFile)?;
let context = context.freeze();
total_size += context_length as u64;
@@ -238,9 +242,9 @@ impl State for FileState {
total_size += 4;
let mut command = BytesMut::with_capacity(command_length);
command.put_bytes(0, command_length);
- reader
- .read_exact(&mut command)
- .await
+ let (result, command) = reader.read_exact(command).await;
+ result
+ .with_error_context(|error| format!("{FILE_STATE_PARSE_ERROR} command. {error}"))
.map_err(|_| IggyError::CannotReadFile)?;
total_size += command_length as u64;
let command_payload;
@@ -353,15 +357,16 @@ impl State for FileState {
command,
);
let bytes = entry.to_bytes();
+ let len = bytes.len();
self.entries_count.fetch_add(1, Ordering::SeqCst);
self.persister
- .append(&self.path, &bytes)
+ .append(&self.path, bytes)
.await
.with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - failed to append state entry data to file, path: {}, data size: {}",
self.path,
- bytes.len()
+ len
)
})?;
debug!("Applied state entry: {entry}");
diff --git a/core/server/src/state/mod.rs b/core/server/src/state/mod.rs
index 656fc852..6ea022a6 100644
--- a/core/server/src/state/mod.rs
+++ b/core/server/src/state/mod.rs
@@ -40,14 +40,14 @@ pub enum StateKind {
}
#[cfg_attr(test, automock)]
-pub trait State: Send {
- fn init(&self) -> impl Future<Output = Result<Vec<StateEntry>, IggyError>> + Send;
- fn load_entries(&self) -> impl Future<Output = Result<Vec<StateEntry>, IggyError>> + Send;
+pub trait State {
+ fn init(&self) -> impl Future<Output = Result<Vec<StateEntry>, IggyError>>;
+ fn load_entries(&self) -> impl Future<Output = Result<Vec<StateEntry>, IggyError>>;
fn apply(
&self,
user_id: u32,
command: &EntryCommand,
- ) -> impl Future<Output = Result<(), IggyError>> + Send;
+ ) -> impl Future<Output = Result<(), IggyError>>;
}
impl StateKind {
diff --git a/core/server/src/streaming/partitions/messages.rs b/core/server/src/streaming/partitions/messages.rs
index b136aeea..cc961370 100644
--- a/core/server/src/streaming/partitions/messages.rs
+++ b/core/server/src/streaming/partitions/messages.rs
@@ -221,7 +221,6 @@ impl Partition {
pub async fn append_messages(
&mut self,
batch: IggyMessagesBatchMut,
- confirmation: Option<Confirmation>,
) -> Result<(), IggyError> {
if batch.count() == 0 {
return Ok(());
@@ -316,7 +315,7 @@ impl Partition {
}
);
- last_segment.persist_messages(confirmation).await.with_error_context(|error| {
+ last_segment.persist_messages().await.with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - failed to persist messages, partition id: {}, start offset: {}",
self.partition_id, last_segment.start_offset()
@@ -346,7 +345,7 @@ impl Partition {
self.partition_id
);
- last_segment.persist_messages(None).await.with_error_context(|error| {
+ last_segment.persist_messages().await.with_error_context(|error| {
format!(
"{COMPONENT} (error: {error}) - failed to persist messages, partition id: {}, start offset: {}",
self.partition_id, last_segment.start_offset()
@@ -384,7 +383,7 @@ mod tests {
.sum();
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition.append_messages(batch, None).await.unwrap();
+ partition.append_messages(batch).await.unwrap();
let loaded_messages = partition
.get_messages_by_offset(0, messages_count)
@@ -406,7 +405,7 @@ mod tests {
assert_eq!(batch.count(), messages_count);
let unique_messages_count = 3;
- partition.append_messages(batch, None).await.unwrap();
+ partition.append_messages(batch).await.unwrap();
let loaded_messages = partition
.get_messages_by_offset(0, messages_count)
@@ -433,7 +432,7 @@ mod tests {
.sum();
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition.append_messages(batch, None).await.unwrap();
+ partition.append_messages(batch).await.unwrap();
let loaded_messages = partition.get_messages_by_offset(0, 10).await.unwrap();
@@ -464,7 +463,7 @@ mod tests {
.sum();
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition.append_messages(batch, None).await.unwrap();
+ partition.append_messages(batch).await.unwrap();
let loaded_messages = partition.get_messages_by_offset(0, 10).await.unwrap();
@@ -495,7 +494,7 @@ mod tests {
.sum();
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition.append_messages(batch, None).await.unwrap();
+ partition.append_messages(batch).await.unwrap();
let loaded_messages = partition.get_messages_by_offset(0, 10).await.unwrap();
@@ -528,7 +527,7 @@ mod tests {
.sum();
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition.append_messages(batch, None).await.unwrap();
+ partition.append_messages(batch).await.unwrap();
let loaded_messages = partition.get_messages_by_offset(0, 10).await.unwrap();
@@ -566,7 +565,7 @@ mod tests {
.sum();
let initial_batch = IggyMessagesBatchMut::from_messages(&initial_messages, initial_size);
partition
- .append_messages(initial_batch, None)
+ .append_messages(initial_batch)
.await
.unwrap();
@@ -584,7 +583,7 @@ mod tests {
let duplicate_batch =
IggyMessagesBatchMut::from_messages(&duplicate_messages, duplicate_size);
partition
- .append_messages(duplicate_batch, None)
+ .append_messages(duplicate_batch)
.await
.unwrap();
@@ -614,7 +613,7 @@ mod tests {
.sum();
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition.append_messages(batch, None).await.unwrap();
+ partition.append_messages(batch).await.unwrap();
let loaded_messages = partition.get_messages_by_offset(0, 10).await.unwrap();
@@ -642,7 +641,7 @@ mod tests {
.map(|m| m.get_size_bytes().as_bytes_u32())
.sum();
let batch1 = IggyMessagesBatchMut::from_messages(&batch1, batch1_size);
- partition.append_messages(batch1, None).await.unwrap();
+ partition.append_messages(batch1).await.unwrap();
// Second batch with mix of new and duplicate messages
let batch2 = vec![
@@ -656,7 +655,7 @@ mod tests {
.map(|m| m.get_size_bytes().as_bytes_u32())
.sum();
let batch2 = IggyMessagesBatchMut::from_messages(&batch2, batch2_size);
- partition.append_messages(batch2, None).await.unwrap();
+ partition.append_messages(batch2).await.unwrap();
let loaded_messages = partition.get_messages_by_offset(0, 10).await.unwrap();
@@ -695,7 +694,7 @@ mod tests {
.sum();
let batch = IggyMessagesBatchMut::from_messages(&messages, messages_size);
- partition.append_messages(batch, None).await.unwrap();
+ partition.append_messages(batch).await.unwrap();
let loaded_messages = partition.get_messages_by_offset(0, 10).await.unwrap();
diff --git a/core/server/src/streaming/partitions/storage.rs b/core/server/src/streaming/partitions/storage.rs
index ba3c2981..954025d1 100644
--- a/core/server/src/streaming/partitions/storage.rs
+++ b/core/server/src/streaming/partitions/storage.rs
@@ -18,6 +18,7 @@
use crate::compat::index_rebuilding::index_rebuilder::IndexRebuilder;
use crate::configs::cache_indexes::CacheIndexesConfig;
+use crate::io::file::IggyFile;
use crate::state::system::PartitionState;
use crate::streaming::partitions::COMPONENT;
use crate::streaming::partitions::partition::{ConsumerOffset, Partition};
@@ -28,12 +29,12 @@ use crate::streaming::utils::file;
use error_set::ErrContext;
use iggy_common::ConsumerKind;
use iggy_common::IggyError;
+use monoio::fs;
+use monoio::fs::create_dir_all;
+use monoio::io::AsyncReadRentExt;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::Ordering;
-use tokio::fs;
-use tokio::fs::create_dir_all;
-use tokio::io::AsyncReadExt;
use tracing::{error, info, trace, warn};
#[derive(Debug)]
@@ -61,27 +62,24 @@ impl PartitionStorage for FilePartitionStorage {
partition.partition_path
);
partition.created_at = state.created_at;
- let dir_entries = fs::read_dir(&partition.partition_path).await;
- if fs::read_dir(&partition.partition_path)
- .await
+ // TODO: Replace this with the dir walk impl, that is mentined
+ // in the main function.
+ let mut dir_entries = std::fs::read_dir(&partition.partition_path)
.with_error_context(|error| format!(
"{COMPONENT} (error: {error}) - failed to read partition with ID: {} for stream with ID: {} and topic with ID: {} and path: {}.",
partition.partition_id, partition.stream_id, partition.topic_id, partition.partition_path,
- )).is_err()
- {
- return Err(IggyError::CannotReadPartitions);
- }
-
- let mut dir_entries = dir_entries.unwrap();
+ ))
+ .map_err(|_| IggyError::CannotReadPartitions)?;
let mut log_files = Vec::new();
- while let Some(dir_entry) = dir_entries.next_entry().await.unwrap_or(None) {
+ while let Some(dir_entry) = dir_entries.next() {
+ let dir_entry = dir_entry.unwrap();
let path = dir_entry.path();
let extension = path.extension();
if extension.is_none() || extension.unwrap() != LOG_EXTENSION {
continue;
}
- let metadata = dir_entry.metadata().await.unwrap();
+ let metadata = dir_entry.metadata().unwrap();
if metadata.is_dir() {
continue;
}
@@ -368,7 +366,7 @@ impl PartitionStorage for FilePartitionStorage {
));
}
- if fs::remove_dir_all(&partition.partition_path).await.is_err() {
+ if std::fs::remove_dir_all(&partition.partition_path).is_err() {
error!(
"Cannot delete partition directory: {} for partition with ID: {} for topic with ID: {} for stream with ID: {}.",
partition.partition_path,
@@ -391,7 +389,7 @@ impl PartitionStorage for FilePartitionStorage {
async fn save_consumer_offset(&self, offset: u64, path: &str) -> Result<(), IggyError> {
self.persister
- .overwrite(path, &offset.to_le_bytes())
+ .overwrite(path, Box::new(offset.to_le_bytes()))
.await
.with_error_context(|error| format!(
"{COMPONENT} (error: {error}) - failed to overwrite consumer offset with value: {offset}, path: {path}",
@@ -406,15 +404,17 @@ impl PartitionStorage for FilePartitionStorage {
path: &str,
) -> Result<Vec<ConsumerOffset>, IggyError> {
trace!("Loading consumer offsets from path: {path}...");
- let dir_entries = fs::read_dir(&path).await;
+ //TODO: replace with async once its there
+ let dir_entries = std::fs::read_dir(&path);
if dir_entries.is_err() {
return Err(IggyError::CannotReadConsumerOffsets(path.to_owned()));
}
let mut consumer_offsets = Vec::new();
let mut dir_entries = dir_entries.unwrap();
- while let Some(dir_entry) = dir_entries.next_entry().await.unwrap_or(None) {
- let metadata = dir_entry.metadata().await;
+ while let Some(dir_entry) = dir_entries.next() {
+ let dir_entry = dir_entry.unwrap();
+ let metadata = dir_entry.metadata();
if metadata.is_err() {
break;
}
@@ -439,7 +439,7 @@ impl PartitionStorage for FilePartitionStorage {
let path = Arc::new(path.unwrap().to_string());
let consumer_id = consumer_id.unwrap();
- let mut file = file::open(&path)
+ let file = file::open(&path)
.await
.with_error_context(|error| {
format!(
@@ -447,6 +447,7 @@ impl PartitionStorage for FilePartitionStorage {
)
})
.map_err(|_| IggyError::CannotReadFile)?;
+ let mut file = IggyFile::new(file);
let offset = file
.read_u64_le()
.await
@@ -473,7 +474,8 @@ impl PartitionStorage for FilePartitionStorage {
return Ok(());
}
- if fs::remove_dir_all(path).await.is_err() {
+ //TODO: replace with async once its there
+ if std::fs::remove_dir_all(path).is_err() {
error!("Cannot delete consumer offsets directory: {}.", path);
return Err(IggyError::CannotDeleteConsumerOffsetsDirectory(
path.to_owned(),
diff --git a/core/server/src/streaming/persistence/persister.rs b/core/server/src/streaming/persistence/persister.rs
index 75282e04..6ba5c87b 100644
--- a/core/server/src/streaming/persistence/persister.rs
+++ b/core/server/src/streaming/persistence/persister.rs
@@ -20,10 +20,10 @@ use crate::streaming::persistence::COMPONENT;
use crate::streaming::utils::file;
use error_set::ErrContext;
use iggy_common::IggyError;
+use monoio::buf::IoBuf;
use std::fmt::Debug;
use std::future::Future;
use tokio::fs;
-use tokio::io::AsyncWriteExt;
#[cfg(test)]
use mockall::automock;
@@ -37,7 +37,7 @@ pub enum PersisterKind {
}
impl PersisterKind {
- pub async fn append(&self, path: &str, bytes: &[u8]) -> Result<(), IggyError> {
+ pub async fn append<B: IoBuf>(&self, path: &str, bytes: B) -> Result<(), IggyError> {
match self {
PersisterKind::File(p) => p.append(path, bytes).await,
PersisterKind::FileWithSync(p) => p.append(path, bytes).await,
@@ -46,7 +46,7 @@ impl PersisterKind {
}
}
- pub async fn overwrite(&self, path: &str, bytes: &[u8]) -> Result<(), IggyError> {
+ pub async fn overwrite<B: IoBuf>(&self, path: &str, bytes: B) -> Result<(), IggyError> {
match self {
PersisterKind::File(p) => p.overwrite(path, bytes).await,
PersisterKind::FileWithSync(p) => p.overwrite(path, bytes).await,
@@ -66,18 +66,15 @@ impl PersisterKind {
}
#[cfg_attr(test, automock)]
-pub trait Persister: Send {
- fn append(
+pub trait Persister {
+ fn append<B: IoBuf>(&self, path: &str, bytes: B)
+ -> impl Future<Output = Result<(), IggyError>>;
+ fn overwrite<B: IoBuf>(
&self,
path: &str,
- bytes: &[u8],
- ) -> impl Future<Output = Result<(), IggyError>> + Send;
- fn overwrite(
- &self,
- path: &str,
- bytes: &[u8],
- ) -> impl Future<Output = Result<(), IggyError>> + Send;
- fn delete(&self, path: &str) -> impl Future<Output = Result<(), IggyError>> + Send;
+ bytes: B,
+ ) -> impl Future<Output = Result<(), IggyError>>;
+ fn delete(&self, path: &str) -> impl Future<Output = Result<(), IggyError>>;
}
#[derive(Debug)]
@@ -87,15 +84,16 @@ pub struct FilePersister;
pub struct FileWithSyncPersister;
impl Persister for FilePersister {
- async fn append(&self, path: &str, bytes: &[u8]) -> Result<(), IggyError> {
- let mut file = file::append(path)
+ async fn append<B: IoBuf>(&self, path: &str, bytes: B) -> Result<(), IggyError> {
+ let file = file::append(path)
.await
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to append to file: {path}")
})
.map_err(|_| IggyError::CannotAppendToFile)?;
- file.write_all(bytes)
+ file.write_all_at(bytes, 0)
.await
+ .0
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to write data to file: {path}")
})
@@ -103,15 +101,17 @@ impl Persister for FilePersister {
Ok(())
}
- async fn overwrite(&self, path: &str, bytes: &[u8]) -> Result<(), IggyError> {
- let mut file = file::overwrite(path)
+ async fn overwrite<B: IoBuf>(&self, path: &str, bytes: B) -> Result<(), IggyError> {
+ let file = file::overwrite(path)
.await
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to overwrite file: {path}")
})
.map_err(|_| IggyError::CannotOverwriteFile)?;
- file.write_all(bytes)
+ let position = 0;
+ file.write_all_at(bytes, position)
.await
+ .0
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to write data to file: {path}")
})
@@ -131,15 +131,21 @@ impl Persister for FilePersister {
}
impl Persister for FileWithSyncPersister {
- async fn append(&self, path: &str, bytes: &[u8]) -> Result<(), IggyError> {
- let mut file = file::append(path)
+ async fn append<B: IoBuf>(&self, path: &str, bytes: B) -> Result<(), IggyError> {
+ let file = file::append(path)
.await
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to append to file: {path}")
})
.map_err(|_| IggyError::CannotAppendToFile)?;
- file.write_all(bytes)
+ let position = file
+ .metadata()
+ .await
+ .map_err(|_| IggyError::CannotReadFileMetadata)?
+ .len();
+ file.write_all_at(bytes, position)
.await
+ .0
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to write data to file: {path}")
})
@@ -155,15 +161,17 @@ impl Persister for FileWithSyncPersister {
Ok(())
}
- async fn overwrite(&self, path: &str, bytes: &[u8]) -> Result<(), IggyError> {
- let mut file = file::overwrite(path)
+ async fn overwrite<B: IoBuf>(&self, path: &str, bytes: B) -> Result<(), IggyError> {
+ let file = file::overwrite(path)
.await
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to overwrite file: {path}")
})
.map_err(|_| IggyError::CannotOverwriteFile)?;
- file.write_all(bytes)
+ let position = 0;
+ file.write_all_at(bytes, position)
.await
+ .0
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to write data to file: {path}")
})
diff --git a/core/server/src/streaming/persistence/task.rs b/core/server/src/streaming/persistence/task.rs
index 70510ed5..d1c716aa 100644
--- a/core/server/src/streaming/persistence/task.rs
+++ b/core/server/src/streaming/persistence/task.rs
@@ -21,18 +21,26 @@ use bytes::Bytes;
use error_set::ErrContext;
use flume::{Receiver, Sender, unbounded};
use iggy_common::IggyError;
+use monoio::task;
use std::{sync::Arc, time::Duration};
-use tokio::task;
use tracing::error;
use super::persister::PersisterKind;
-#[derive(Debug)]
pub struct LogPersisterTask {
_sender: Option<Sender<Bytes>>,
_task_handle: Option<task::JoinHandle<()>>,
}
+impl std::fmt::Debug for LogPersisterTask {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("LogPersisterTask")
+ .field("_sender", &self._sender.is_some())
+ .field("_task_handle", &self._task_handle.is_some())
+ .finish()
+ }
+}
+
impl LogPersisterTask {
pub fn new(
path: String,
@@ -42,7 +50,7 @@ impl LogPersisterTask {
) -> Self {
let (sender, receiver): (Sender<Bytes>, Receiver<Bytes>) = unbounded();
- let task_handle = task::spawn(async move {
+ let task_handle = monoio::spawn(async move {
loop {
match receiver.recv_async().await {
Ok(data) => {
@@ -82,7 +90,7 @@ impl LogPersisterTask {
let mut retries = 0;
while retries < max_retries {
- match persister.append(path, &data).await {
+ match persister.append(path, data.clone()).await {
Ok(_) => return Ok(()),
Err(e) => {
error!(
@@ -121,13 +129,7 @@ impl Drop for LogPersisterTask {
self._sender.take();
if let Some(handle) = self._task_handle.take() {
- tokio::spawn(async move {
- if let Err(error) = handle.await {
- error!(
- "{COMPONENT} (error: {error}) - error while shutting down task in Drop.",
- );
- }
- });
+ monoio::spawn(async move { handle.await });
}
}
}
diff --git a/core/server/src/streaming/segments/messages/messages_reader.rs b/core/server/src/streaming/segments/messages/messages_reader.rs
index b7e4f434..0f413650 100644
--- a/core/server/src/streaming/segments/messages/messages_reader.rs
+++ b/core/server/src/streaming/segments/messages/messages_reader.rs
@@ -16,12 +16,13 @@
* under the License.
*/
+use crate::io::file::{IggyFile};
use crate::streaming::segments::{IggyIndexesMut, IggyMessagesBatchMut};
-use crate::streaming::utils::PooledBuffer;
+use crate::streaming::utils::{file, PooledBuffer};
use bytes::BytesMut;
use error_set::ErrContext;
use iggy_common::IggyError;
-use std::{fs::File as StdFile, os::unix::prelude::FileExt};
+use monoio::fs::File;
use std::{
io::ErrorKind,
sync::{
@@ -29,15 +30,13 @@ use std::{
atomic::{AtomicU64, Ordering},
},
};
-use tokio::fs::OpenOptions;
-use tokio::task::spawn_blocking;
use tracing::{error, trace};
/// A dedicated struct for reading from the messages file.
#[derive(Debug)]
pub struct MessagesReader {
file_path: String,
- file: Arc<StdFile>,
+ file: IggyFile,
messages_size_bytes: Arc<AtomicU64>,
}
@@ -47,10 +46,7 @@ impl MessagesReader {
file_path: &str,
messages_size_bytes: Arc<AtomicU64>,
) -> Result<Self, IggyError> {
- let file = OpenOptions::new()
- .read(true)
- .open(file_path)
- .await
+ let file = file::open_std(file_path)
.with_error_context(|error| {
format!("Failed to open messages file: {file_path}, error: {error}")
})
@@ -71,6 +67,7 @@ impl MessagesReader {
)
});
}
+ let file = File::from_std(file).unwrap();
trace!(
"Opened messages file for reading: {file_path}, size: {}",
@@ -79,7 +76,7 @@ impl MessagesReader {
Ok(Self {
file_path: file_path.to_string(),
- file: Arc::new(file.into_std().await),
+ file: file.into(),
messages_size_bytes,
})
}
@@ -177,20 +174,19 @@ impl MessagesReader {
len: u32,
use_pool: bool,
) -> Result<PooledBuffer, std::io::Error> {
- let file = self.file.clone();
- spawn_blocking(move || {
- if use_pool {
+ if use_pool {
let mut buf = PooledBuffer::with_capacity(len as usize);
unsafe { buf.set_len(len as usize) };
- file.read_exact_at(&mut buf, offset as u64)?;
+ let (result, buf) = self.file.read_exact_at(buf, offset as u64).await;
+ result?;
Ok(buf)
+
} else {
let mut buf = BytesMut::with_capacity(len as usize);
unsafe { buf.set_len(len as usize) };
- file.read_exact_at(&mut buf, offset as u64)?;
+ let (result, buf) = self.file.read_exact_at(buf, offset as u64).await;
+ result?;
Ok(PooledBuffer::from_existing(buf))
}
- })
- .await?
}
}
diff --git a/core/server/src/streaming/segments/messages/messages_writer.rs b/core/server/src/streaming/segments/messages/messages_writer.rs
index 01a661ac..0429817f 100644
--- a/core/server/src/streaming/segments/messages/messages_writer.rs
+++ b/core/server/src/streaming/segments/messages/messages_writer.rs
@@ -16,15 +16,17 @@
* under the License.
*/
-use super::PersisterTask;
-use crate::streaming::segments::{IggyMessagesBatchSet, messages::write_batch};
+use crate::{
+ io::file::IggyFile,
+ streaming::segments::{IggyMessagesBatchSet, messages::write_batch},
+};
use error_set::ErrContext;
use iggy_common::{Confirmation, IggyByteSize, IggyError};
+use monoio::fs::{File, OpenOptions};
use std::sync::{
Arc,
atomic::{AtomicU64, Ordering},
};
-use tokio::fs::{File, OpenOptions};
use tracing::{error, trace};
/// A dedicated struct for writing to the messages file.
@@ -32,9 +34,8 @@ use tracing::{error, trace};
pub struct MessagesWriter {
file_path: String,
/// Holds the file for synchronous writes; when asynchronous persistence is enabled, this will be None.
- file: Option<File>,
+ file: Option<IggyFile>,
/// When set, asynchronous writes are handled by this persister task.
- persister_task: Option<PersisterTask>,
messages_size_bytes: Arc<AtomicU64>,
fsync: bool,
}
@@ -49,7 +50,6 @@ impl MessagesWriter {
file_path: &str,
messages_size_bytes: Arc<AtomicU64>,
fsync: bool,
- server_confirmation: Confirmation,
file_exists: bool,
) -> Result<Self, IggyError> {
let file = OpenOptions::new()
@@ -82,23 +82,10 @@ impl MessagesWriter {
messages_size_bytes.load(Ordering::Acquire)
);
- let (file, persister_task) = match server_confirmation {
- Confirmation::NoWait => {
- let persister = PersisterTask::new(
- file,
- file_path.to_string(),
- fsync,
- messages_size_bytes.clone(),
- );
- (None, Some(persister))
- }
- Confirmation::Wait => (Some(file), None),
- };
-
+ let file = Some(IggyFile::new(file));
Ok(Self {
file_path: file_path.to_string(),
file,
- persister_task,
messages_size_bytes,
fsync,
})
@@ -108,7 +95,6 @@ impl MessagesWriter {
pub async fn save_batch_set(
&mut self,
batch_set: IggyMessagesBatchSet,
- confirmation: Confirmation,
) -> Result<IggyByteSize, IggyError> {
let messages_size = batch_set.size();
let messages_count = batch_set.count();
@@ -117,46 +103,32 @@ impl MessagesWriter {
"Saving batch set of size {messages_size} bytes ({containers_count} containers, {messages_count} messages) to messages file: {}",
self.file_path
);
- match confirmation {
- Confirmation::Wait => {
- if let Some(ref mut file) = self.file {
- write_batch(file, &self.file_path, batch_set)
- .await
- .with_error_context(|error| {
- format!(
- "Failed to write batch to messages file: {}. {error}",
- self.file_path
- )
- })
- .map_err(|_| IggyError::CannotWriteToFile)?;
- } else {
- error!("File handle is not available for synchronous write.");
- return Err(IggyError::CannotWriteToFile);
- }
-
- if self.fsync {
- let _ = self.fsync().await;
- }
-
- self.messages_size_bytes
- .fetch_add(messages_size as u64, Ordering::Release);
- trace!(
- "Written batch set of size {messages_size} bytes ({containers_count} containers, {messages_count} messages) to disk messages file: {}",
- self.file_path
- );
- }
- Confirmation::NoWait => {
- if let Some(task) = &self.persister_task {
- task.persist(batch_set).await;
- } else {
- panic!(
- "Confirmation::NoWait is used, but MessagesPersisterTask is not set for messages file: {}",
+ if let Some(ref mut file) = self.file {
+ write_batch(file, &self.file_path, batch_set)
+ .await
+ .with_error_context(|error| {
+ format!(
+ "Failed to write batch to messages file: {}. {error}",
self.file_path
- );
- }
- }
+ )
+ })
+ .map_err(|_| IggyError::CannotWriteToFile)?;
+ } else {
+ error!("File handle is not available for synchronous write.");
+ return Err(IggyError::CannotWriteToFile);
}
+ if self.fsync {
+ let _ = self.fsync().await;
+ }
+
+ self.messages_size_bytes
+ .fetch_add(messages_size as u64, Ordering::Release);
+ trace!(
+ "Written batch set of size {messages_size} bytes ({containers_count} containers, {messages_count} messages) to disk messages file: {}",
+ self.file_path
+ );
+
Ok(IggyByteSize::from(messages_size as u64))
}
@@ -172,10 +144,4 @@ impl MessagesWriter {
Ok(())
}
-
- pub async fn shutdown_persister_task(self) {
- if let Some(task) = self.persister_task {
- task.shutdown().await;
- }
- }
}
diff --git a/core/server/src/streaming/segments/messages/mod.rs b/core/server/src/streaming/segments/messages/mod.rs
index 7f9ef61e..512c3894 100644
--- a/core/server/src/streaming/segments/messages/mod.rs
+++ b/core/server/src/streaming/segments/messages/mod.rs
@@ -18,42 +18,72 @@
mod messages_reader;
mod messages_writer;
-mod persister_task;
+
+use crate::{io::file::IggyFile, to_iovec};
use super::IggyMessagesBatchSet;
use error_set::ErrContext;
use iggy_common::IggyError;
-use std::io::IoSlice;
-use tokio::{fs::File, io::AsyncWriteExt};
+use monoio::io::AsyncWriteRent;
+use std::{io::IoSlice, mem::take};
pub use messages_reader::MessagesReader;
pub use messages_writer::MessagesWriter;
-pub use persister_task::PersisterTask;
+
+use nix::libc::iovec;
/// Vectored write a batches of messages to file
async fn write_batch(
- file: &mut File,
+ file: &mut IggyFile,
file_path: &str,
batches: IggyMessagesBatchSet,
) -> Result<usize, IggyError> {
- let mut slices: Vec<IoSlice> = batches.iter().map(|b| IoSlice::new(b)).collect();
-
- let slices = &mut slices.as_mut_slice();
+ let mut slices = batches.iter().map(|b| to_iovec(&b)).collect::<Vec<iovec>>();
let mut total_written = 0;
- while !slices.is_empty() {
- let bytes_written = file
- .write_vectored(slices)
- .await
+ loop {
+ let (result, vomited) = file.writev(slices).await;
+ slices = vomited;
+ let bytes_written = result
.with_error_context(|error| {
format!("Failed to write messages to file: {file_path}, error: {error}",)
})
.map_err(|_| IggyError::CannotWriteToFile)?;
total_written += bytes_written;
-
- IoSlice::advance_slices(slices, bytes_written);
+ advance_slices(&mut slices.as_mut_slice(), bytes_written);
+ if slices.is_empty() {
+ break;
+ }
}
Ok(total_written)
}
+
+fn advance_slices(mut bufs: &mut [iovec], n: usize) {
+ // Number of buffers to remove.
+ let mut remove = 0;
+ // Remaining length before reaching n. This prevents overflow
+ // that could happen if the length of slices in `bufs` were instead
+ // accumulated. Those slice may be aliased and, if they are large
+ // enough, their added length may overflow a `usize`.
+ let mut left = n;
+ for buf in bufs.iter() {
+ if let Some(remainder) = left.checked_sub(buf.iov_len as _) {
+ left = remainder;
+ remove += 1;
+ } else {
+ break;
+ }
+ }
+
+ bufs = &mut bufs[remove..];
+ if bufs.is_empty() {
+ assert!(left == 0, "advancing io slices beyond their length");
+ } else {
+ unsafe {
+ bufs[0].iov_len -= n;
+ bufs[0].iov_base = bufs[0].iov_base.add(n);
+ }
+ }
+}
diff --git a/core/server/src/streaming/segments/messages/persister_task.rs b/core/server/src/streaming/segments/messages/persister_task.rs
deleted file mode 100644
index 74056293..00000000
--- a/core/server/src/streaming/segments/messages/persister_task.rs
+++ /dev/null
@@ -1,215 +0,0 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
- * 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
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-use crate::streaming::segments::IggyMessagesBatchSet;
-use error_set::ErrContext;
-use flume::{Receiver, unbounded};
-use std::{
- sync::{
- Arc,
- atomic::{AtomicU64, Ordering},
- },
- time::Duration,
-};
-use tokio::{fs::File, select, time::sleep};
-use tracing::{error, trace, warn};
-
-use super::write_batch;
-
-#[derive(Debug)]
-/// A command to the persister task.
-enum PersisterTaskCommand {
- WriteRequest(IggyMessagesBatchSet),
- Shutdown,
-}
-
-/// A background task that writes data asynchronously.
-#[derive(Debug)]
-pub struct PersisterTask {
- sender: flume::Sender<PersisterTaskCommand>,
- file_path: String, // used only for logging
- _handle: tokio::task::JoinHandle<()>,
-}
-
-impl PersisterTask {
- /// Creates a new persister task that takes ownership of `file`.
- pub fn new(file: File, file_path: String, fsync: bool, log_file_size: Arc<AtomicU64>) -> Self {
- let (sender, receiver) = unbounded();
- let log_file_size = log_file_size.clone();
- let file_path_clone = file_path.clone();
- let handle = tokio::spawn(async move {
- Self::run(file, file_path_clone, receiver, fsync, log_file_size).await;
- });
- Self {
- sender,
- file_path,
- _handle: handle,
- }
- }
-
- /// Sends the batch bytes to the persister task (fire-and-forget).
- pub async fn persist(&self, messages: IggyMessagesBatchSet) {
- if let Err(e) = self
- .sender
- .send_async(PersisterTaskCommand::WriteRequest(messages))
- .await
- {
- error!(
- "Failed to send write request to LogPersisterTask for file {}: {:?}",
- self.file_path, e
- );
- }
- }
-
- /// Sends the shutdown command to the persister task and waits for a response.
- pub async fn shutdown(self) {
- let start_time = tokio::time::Instant::now();
-
- if let Err(e) = self.sender.send_async(PersisterTaskCommand::Shutdown).await {
- error!(
- "Failed to send shutdown command to LogPersisterTask for file {}: {:?}",
- self.file_path, e
- );
- return;
- }
-
- let mut handle_future = self._handle;
-
- select! {
- result = &mut handle_future => {
- match result {
- Ok(_) => {
- let elapsed = start_time.elapsed();
- trace!(
- "PersisterTask shutdown complete for file {} in {:.2}s",
- self.file_path,
- elapsed.as_secs_f64()
- );
- }
- Err(e) => {
- error!(
- "Error during joining PersisterTask for file {}: {:?}",
- self.file_path, e
- );
- }
- }
- return;
- }
- _ = sleep(Duration::from_secs(1)) => {
- warn!(
- "PersisterTask for file {} is still shutting down after 1s",
- self.file_path
- );
- }
- }
-
- select! {
- result = &mut handle_future => {
- match result {
- Ok(_) => {
- let elapsed = start_time.elapsed();
- trace!(
- "PersisterTask shutdown complete for file {} in {:.2}s",
- self.file_path,
- elapsed.as_secs_f64()
- );
- }
- Err(e) => {
- error!(
- "Error during joining PersisterTask for file {}: {:?}",
- self.file_path, e
- );
- }
- }
- return;
- }
- _ = sleep(Duration::from_secs(4)) => {
- warn!(
- "PersisterTask for file {} is still shutting down after 5s",
- self.file_path
- );
- }
- }
-
- match handle_future.await {
- Ok(_) => {
- let elapsed = start_time.elapsed();
- warn!(
- "PersisterTask shutdown complete for file {} in {:.2}s",
- self.file_path,
- elapsed.as_secs_f64()
- );
- }
- Err(e) => {
- error!(
- "Error during joining PersisterTask for file {}: {:?}",
- self.file_path, e
- );
- }
- }
- }
-
- /// The background task loop. Processes write requests until the channel is closed.
- async fn run(
- mut file: File,
- file_path: String,
- receiver: Receiver<PersisterTaskCommand>,
- fsync: bool,
- log_file_size: Arc<AtomicU64>,
- ) {
- while let Ok(request) = receiver.recv_async().await {
- match request {
- PersisterTaskCommand::WriteRequest(messages) => {
- match write_batch(&mut file, &file_path, messages).await {
- Ok(bytes_written) => {
- if fsync {
- file.sync_all()
- .await
- .with_error_context(|error| {
- format!(
- "Failed to fsync messages file: {file_path}. {error}"
- )
- })
- .expect("Failed to fsync messages file");
- }
-
- log_file_size.fetch_add(bytes_written as u64, Ordering::Acquire);
- }
- Err(e) => {
- error!(
- "Failed to persist data in LogPersisterTask for file {file_path}: {:?}",
- e
- )
- }
- }
- }
- PersisterTaskCommand::Shutdown => {
- trace!("LogPersisterTask for file {file_path} received shutdown command");
- if let Err(e) = file.sync_all().await {
- error!(
- "Failed to sync_all() in LogPersisterTask for file {file_path}: {:?}",
- e
- );
- }
- break;
- }
- }
- }
- trace!("PersisterTask for file {file_path} has finished processing requests");
- }
-}
diff --git a/core/server/src/streaming/segments/segment.rs b/core/server/src/streaming/segments/segment.rs
index 60886df6..92dd254f 100644
--- a/core/server/src/streaming/segments/segment.rs
+++ b/core/server/src/streaming/segments/segment.rs
@@ -224,13 +224,10 @@ impl Segment {
let log_fsync = self.config.partition.enforce_fsync;
let index_fsync = self.config.partition.enforce_fsync;
- let server_confirmation = self.config.segment.server_confirmation;
-
let messages_writer = MessagesWriter::new(
&self.messages_path,
self.messages_size.clone(),
log_fsync,
- server_confirmation,
file_exists,
)
.await?;
@@ -304,9 +301,9 @@ impl Segment {
pub async fn shutdown_writing(&mut self) {
if let Some(log_writer) = self.messages_writer.take() {
- tokio::spawn(async move {
+ //TODO: Fixme not sure whether we should spawn a task here.
+ monoio::spawn(async move {
let _ = log_writer.fsync().await;
- log_writer.shutdown_persister_task().await;
});
} else {
warn!(
diff --git a/core/server/src/streaming/segments/writing_messages.rs b/core/server/src/streaming/segments/writing_messages.rs
index 607fd261..9c385b02 100644
--- a/core/server/src/streaming/segments/writing_messages.rs
+++ b/core/server/src/streaming/segments/writing_messages.rs
@@ -23,7 +23,6 @@ use crate::{
streaming::deduplication::message_deduplicator::MessageDeduplicator,
};
use error_set::ErrContext;
-use iggy_common::Confirmation;
use iggy_common::IggyError;
use std::sync::atomic::Ordering;
use tracing::{info, trace};
@@ -66,10 +65,7 @@ impl Segment {
Ok(())
}
- pub async fn persist_messages(
- &mut self,
- confirmation: Option<Confirmation>,
- ) -> Result<usize, IggyError> {
+ pub async fn persist_messages(&mut self) -> Result<usize, IggyError> {
if self.accumulator.is_empty() {
return Ok(0);
}
@@ -83,11 +79,6 @@ impl Segment {
let accumulator = std::mem::take(&mut self.accumulator);
let batches = accumulator.into_batch_set();
- let confirmation = match confirmation {
- Some(val) => val,
- None => self.config.segment.server_confirmation,
- };
-
let batch_size = batches.size();
let batch_count = batches.count();
@@ -97,7 +88,7 @@ impl Segment {
.messages_writer
.as_mut()
.expect("Messages writer not initialized")
- .save_batch_set(batches, confirmation)
+ .save_batch_set(batches)
.await
.with_error_context(|error| {
format!(
diff --git a/core/server/src/streaming/topics/messages.rs b/core/server/src/streaming/topics/messages.rs
index c01687a6..acbe8c28 100644
--- a/core/server/src/streaming/topics/messages.rs
+++ b/core/server/src/streaming/topics/messages.rs
@@ -80,7 +80,6 @@ impl Topic {
&self,
partitioning: &Partitioning,
messages: IggyMessagesBatchMut,
- confirmation: Option<Confirmation>,
) -> Result<(), IggyError> {
if !self.has_partitions() {
return Err(IggyError::NoPartitions(self.topic_id, self.stream_id));
@@ -108,7 +107,7 @@ impl Topic {
}
};
- self.append_messages_to_partition(messages, partition_id, confirmation)
+ self.append_messages_to_partition(messages, partition_id)
.await
}
@@ -134,7 +133,6 @@ impl Topic {
&self,
messages: IggyMessagesBatchMut,
partition_id: u32,
- confirmation: Option<Confirmation>,
) -> Result<(), IggyError> {
let partition = self.partitions.get(&partition_id);
partition
@@ -145,7 +143,7 @@ impl Topic {
))?
.write()
.await
- .append_messages(messages, confirmation)
+ .append_messages(messages)
.await
.with_error_context(|error| {
format!("{COMPONENT} (error: {error}) - failed to append messages")
@@ -230,7 +228,7 @@ mod tests {
.expect("Failed to create message with valid payload and headers");
let messages = IggyMessagesBatchMut::from_messages(&[message], 1);
topic
- .append_messages(&partitioning, messages, None)
+ .append_messages(&partitioning, messages)
.await
.unwrap();
}
@@ -263,7 +261,7 @@ mod tests {
.expect("Failed to create message with valid payload and headers");
let messages = IggyMessagesBatchMut::from_messages(&[message], 1);
topic
- .append_messages(&partitioning, messages, None)
+ .append_messages(&partitioning, messages)
.await
.unwrap();
}
diff --git a/core/server/src/streaming/topics/persistence.rs b/core/server/src/streaming/topics/persistence.rs
index a423c67d..f75e36ec 100644
--- a/core/server/src/streaming/topics/persistence.rs
+++ b/core/server/src/streaming/topics/persistence.rs
@@ -54,7 +54,7 @@ impl Topic {
let mut partition = partition.write().await;
let partition_id = partition.partition_id;
for segment in partition.get_segments_mut() {
- saved_messages_number += segment.persist_messages(None).await.with_error_context(|error| format!("{COMPONENT} (error: {error}) - failed to persist messages in segment, partition ID: {partition_id}"))?;
+ saved_messages_number += segment.persist_messages().await.with_error_context(|error| format!("{COMPONENT} (error: {error}) - failed to persist messages in segment, partition ID: {partition_id}"))?;
}
}
diff --git a/core/server/src/streaming/utils/file.rs b/core/server/src/streaming/utils/file.rs
index af26454d..7a5bb524 100644
--- a/core/server/src/streaming/utils/file.rs
+++ b/core/server/src/streaming/utils/file.rs
@@ -16,8 +16,12 @@
* under the License.
*/
+use monoio::fs::{File, OpenOptions, remove_file};
use std::path::Path;
-use tokio::fs::{File, OpenOptions, remove_file};
+
+pub fn open_std(path: &str) -> Result<std::fs::File, std::io::Error> {
+ std::fs::OpenOptions::new().read(true).open(path)
+}
pub async fn open(path: &str) -> Result<File, std::io::Error> {
OpenOptions::new().read(true).open(path).await
@@ -35,14 +39,16 @@ pub async fn overwrite(path: &str) -> Result<File, std::io::Error> {
.open(path)
.await
}
+
pub async fn remove(path: &str) -> Result<(), std::io::Error> {
remove_file(path).await
}
pub async fn rename(old_path: &str, new_path: &str) -> Result<(), std::io::Error> {
- tokio::fs::rename(Path::new(old_path), Path::new(new_path)).await
+ monoio::fs::rename(Path::new(old_path), Path::new(new_path)).await
}
pub async fn exists(path: &str) -> Result<bool, std::io::Error> {
- tokio::fs::try_exists(path).await
+ //TODO: Does monoio support that ?
+ std::fs::exists(path)
}
(iggy) 01/02: fix sdk
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch io_uring_tpc
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 639a9a889758513dd41a5348519e176b2b8df38d
Author: numinex <g....@gmail.com>
AuthorDate: Fri Jun 27 16:50:29 2025 +0200
fix sdk
---
core/sdk/src/clients/client.rs | 10 +++++-----
core/sdk/src/clients/consumer.rs | 10 +++++-----
core/sdk/src/clients/consumer_builder.rs | 6 +++---
core/sdk/src/clients/producer.rs | 6 +++---
core/sdk/src/clients/producer_builder.rs | 6 +++---
core/sdk/src/http/http_client.rs | 6 +++---
6 files changed, 22 insertions(+), 22 deletions(-)
diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs
index 4896e629..a60892ba 100644
--- a/core/sdk/src/clients/client.rs
+++ b/core/sdk/src/clients/client.rs
@@ -17,7 +17,7 @@
*/
use crate::clients::client_builder::IggyClientBuilder;
-use iggy_common::locking::{IggySharedMut, IggySharedMutFn};
+use iggy_common::locking::{IggyRwLock, IggySharedMutFn};
use crate::http::http_client::HttpClient;
use crate::prelude::EncryptorKind;
@@ -45,7 +45,7 @@ use tracing::{debug, error, info};
#[derive(Debug)]
#[allow(dead_code)]
pub struct IggyClient {
- pub(crate) client: IggySharedMut<Box<dyn Client>>,
+ pub(crate) client: IggyRwLock<Box<dyn Client>>,
partitioner: Option<Arc<dyn Partitioner>>,
pub(crate) encryptor: Option<Arc<EncryptorKind>>,
}
@@ -71,7 +71,7 @@ impl IggyClient {
/// Creates a new `IggyClient` with the provided client implementation for the specific transport.
pub fn new(client: Box<dyn Client>) -> Self {
- let client = IggySharedMut::new(client);
+ let client = IggyRwLock::new(client);
IggyClient {
client,
partitioner: None,
@@ -106,7 +106,7 @@ impl IggyClient {
info!("Client-side encryption is enabled.");
}
- let client = IggySharedMut::new(client);
+ let client = IggyRwLock::new(client);
IggyClient {
client,
partitioner,
@@ -115,7 +115,7 @@ impl IggyClient {
}
/// Returns the underlying client implementation for the specific transport.
- pub fn client(&self) -> IggySharedMut<Box<dyn Client>> {
+ pub fn client(&self) -> IggyRwLock<Box<dyn Client>> {
self.client.clone()
}
diff --git a/core/sdk/src/clients/consumer.rs b/core/sdk/src/clients/consumer.rs
index d0eeaf3a..99dd5197 100644
--- a/core/sdk/src/clients/consumer.rs
+++ b/core/sdk/src/clients/consumer.rs
@@ -21,7 +21,7 @@ use dashmap::DashMap;
use futures::Stream;
use futures_util::{FutureExt, StreamExt};
use iggy_binary_protocol::Client;
-use iggy_common::locking::{IggySharedMut, IggySharedMutFn};
+use iggy_common::locking::{IggyRwLock, IggySharedMutFn};
use iggy_common::{
Consumer, ConsumerKind, DiagnosticEvent, EncryptorKind, IdKind, Identifier, IggyDuration,
IggyError, IggyMessage, IggyTimestamp, PolledMessages, PollingKind, PollingStrategy,
@@ -93,7 +93,7 @@ unsafe impl Sync for IggyConsumer {}
pub struct IggyConsumer {
initialized: bool,
can_poll: Arc<AtomicBool>,
- client: IggySharedMut<Box<dyn Client>>,
+ client: IggyRwLock<Box<dyn Client>>,
consumer_name: String,
consumer: Arc<Consumer>,
is_consumer_group: bool,
@@ -129,7 +129,7 @@ pub struct IggyConsumer {
impl IggyConsumer {
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
- client: IggySharedMut<Box<dyn Client>>,
+ client: IggyRwLock<Box<dyn Client>>,
consumer_name: String,
consumer: Consumer,
stream_id: Identifier,
@@ -405,7 +405,7 @@ impl IggyConsumer {
#[allow(clippy::too_many_arguments)]
async fn store_consumer_offset(
- client: &IggySharedMut<Box<dyn Client>>,
+ client: &IggyRwLock<Box<dyn Client>>,
consumer: &Consumer,
stream_id: &Identifier,
topic_id: &Identifier,
@@ -781,7 +781,7 @@ impl IggyConsumer {
}
async fn initialize_consumer_group(
- client: IggySharedMut<Box<dyn Client>>,
+ client: IggyRwLock<Box<dyn Client>>,
create_consumer_group_if_not_exists: bool,
stream_id: Arc<Identifier>,
topic_id: Arc<Identifier>,
diff --git a/core/sdk/src/clients/consumer_builder.rs b/core/sdk/src/clients/consumer_builder.rs
index 908a367c..425a765c 100644
--- a/core/sdk/src/clients/consumer_builder.rs
+++ b/core/sdk/src/clients/consumer_builder.rs
@@ -18,13 +18,13 @@
use crate::prelude::{AutoCommit, AutoCommitWhen, IggyConsumer};
use iggy_binary_protocol::Client;
-use iggy_common::locking::IggySharedMut;
+use iggy_common::locking::IggyRwLock;
use iggy_common::{Consumer, EncryptorKind, Identifier, IggyDuration, PollingStrategy};
use std::sync::Arc;
#[derive(Debug)]
pub struct IggyConsumerBuilder {
- client: IggySharedMut<Box<dyn Client>>,
+ client: IggyRwLock<Box<dyn Client>>,
consumer_name: String,
consumer: Consumer,
stream: Identifier,
@@ -46,7 +46,7 @@ pub struct IggyConsumerBuilder {
impl IggyConsumerBuilder {
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
- client: IggySharedMut<Box<dyn Client>>,
+ client: IggyRwLock<Box<dyn Client>>,
consumer_name: String,
consumer: Consumer,
stream_id: Identifier,
diff --git a/core/sdk/src/clients/producer.rs b/core/sdk/src/clients/producer.rs
index 2c76b4ad..69d0042a 100644
--- a/core/sdk/src/clients/producer.rs
+++ b/core/sdk/src/clients/producer.rs
@@ -23,7 +23,7 @@ use crate::clients::producer_dispatcher::ProducerDispatcher;
use bytes::Bytes;
use futures_util::StreamExt;
use iggy_binary_protocol::Client;
-use iggy_common::locking::{IggySharedMut, IggySharedMutFn};
+use iggy_common::locking::{IggyRwLock, IggySharedMutFn};
use iggy_common::{
CompressionAlgorithm, DiagnosticEvent, EncryptorKind, IdKind, Identifier, IggyDuration,
IggyError, IggyExpiry, IggyMessage, IggyTimestamp, MaxTopicSize, Partitioner, Partitioning,
@@ -52,7 +52,7 @@ pub trait ProducerCoreBackend: Send + Sync + 'static {
pub struct ProducerCore {
initialized: AtomicBool,
can_send: Arc<AtomicBool>,
- client: Arc<IggySharedMut<Box<dyn Client>>>,
+ client: Arc<IggyRwLock<Box<dyn Client>>>,
stream_id: Arc<Identifier>,
stream_name: String,
topic_id: Arc<Identifier>,
@@ -431,7 +431,7 @@ pub struct IggyProducer {
impl IggyProducer {
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
- client: IggySharedMut<Box<dyn Client>>,
+ client: IggyRwLock<Box<dyn Client>>,
stream: Identifier,
stream_name: String,
topic: Identifier,
diff --git a/core/sdk/src/clients/producer_builder.rs b/core/sdk/src/clients/producer_builder.rs
index fcd0c4a0..15abef42 100644
--- a/core/sdk/src/clients/producer_builder.rs
+++ b/core/sdk/src/clients/producer_builder.rs
@@ -18,7 +18,7 @@
use crate::clients::producer_config::{BackgroundConfig, DirectConfig};
use crate::prelude::IggyProducer;
use iggy_binary_protocol::Client;
-use iggy_common::locking::IggySharedMut;
+use iggy_common::locking::IggyRwLock;
use iggy_common::{
EncryptorKind, Identifier, IggyDuration, IggyExpiry, MaxTopicSize, Partitioner, Partitioning,
};
@@ -36,7 +36,7 @@ impl Default for SendMode {
}
pub struct IggyProducerBuilder {
- client: IggySharedMut<Box<dyn Client>>,
+ client: IggyRwLock<Box<dyn Client>>,
stream: Identifier,
stream_name: String,
topic: Identifier,
@@ -58,7 +58,7 @@ pub struct IggyProducerBuilder {
impl IggyProducerBuilder {
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
- client: IggySharedMut<Box<dyn Client>>,
+ client: IggyRwLock<Box<dyn Client>>,
stream: Identifier,
stream_name: String,
topic: Identifier,
diff --git a/core/sdk/src/http/http_client.rs b/core/sdk/src/http/http_client.rs
index be619dad..6c6eb066 100644
--- a/core/sdk/src/http/http_client.rs
+++ b/core/sdk/src/http/http_client.rs
@@ -20,7 +20,7 @@ use crate::http::http_transport::HttpTransport;
use crate::prelude::{Client, HttpClientConfig, IggyDuration, IggyError};
use async_broadcast::{Receiver, Sender, broadcast};
use async_trait::async_trait;
-use iggy_common::locking::{IggySharedMut, IggySharedMutFn};
+use iggy_common::locking::{IggyRwLock, IggySharedMutFn};
use iggy_common::{
ConnectionString, ConnectionStringUtils, DiagnosticEvent, HttpConnectionStringOptions,
IdentityInfo, TransportProtocol,
@@ -51,7 +51,7 @@ pub struct HttpClient {
pub api_url: Url,
pub(crate) heartbeat_interval: IggyDuration,
client: ClientWithMiddleware,
- access_token: IggySharedMut<String>,
+ access_token: IggyRwLock<String>,
events: (Sender<DiagnosticEvent>, Receiver<DiagnosticEvent>),
}
@@ -280,7 +280,7 @@ impl HttpClient {
api_url,
client,
heartbeat_interval: IggyDuration::from_str("5s").unwrap(),
- access_token: IggySharedMut::new("".to_string()),
+ access_token: IggyRwLock::new("".to_string()),
events: broadcast(1000),
})
}