You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2025/03/17 16:30:44 UTC
(iggy) branch zero-copy-no-batching-parallel-bench created (now 9da56188)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a change to branch zero-copy-no-batching-parallel-bench
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 9da56188 feat(bench): improve benchmark by parallelizing actors (#1645)
This branch includes the following new commits:
new 61b0b00b rename headers to user_headers
new 9da56188 feat(bench): improve benchmark by parallelizing actors (#1645)
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.
(iggy) 02/02: feat(bench): improve benchmark by parallelizing actors (#1645)
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch zero-copy-no-batching-parallel-bench
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 9da56188fa3fff082aa6db14f615e1f3bceb9a76
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Sun Mar 16 18:15:25 2025 +0100
feat(bench): improve benchmark by parallelizing actors (#1645)
This PR improves our benchmark suite, by parallelizing benchmark actors
using `JoinSet` rather than `select`
---
Cargo.lock | 3 +--
bench/Cargo.toml | 3 +--
bench/src/actors/consumer.rs | 2 +-
bench/src/actors/producer.rs | 2 +-
bench/src/actors/producing_consumer.rs | 2 +-
bench/src/benchmarks/benchmark.rs | 13 +++++-------
bench/src/benchmarks/consumer_benchmark.rs | 18 ++++++++++-------
bench/src/benchmarks/consumer_group_benchmark.rs | 19 +++++++++++-------
.../benchmarks/producer_and_consumer_benchmark.rs | 21 +++++++++++---------
.../producer_and_consumer_group_benchmark.rs | 23 +++++++++++-----------
bench/src/benchmarks/producer_benchmark.rs | 17 ++++++++++------
.../src/benchmarks/producing_consumer_benchmark.rs | 18 +++++++++++------
.../producing_consumer_group_benchmark.rs | 16 +++++++++------
bench/src/runner.rs | 12 ++++-------
14 files changed, 94 insertions(+), 75 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 6cee216b..78bbc4a3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -607,7 +607,7 @@ dependencies = [
[[package]]
name = "bench"
-version = "0.2.3"
+version = "0.2.4"
dependencies = [
"async-trait",
"atomic-time",
@@ -616,7 +616,6 @@ dependencies = [
"chrono",
"clap",
"figlet-rs",
- "futures",
"hostname",
"human-repr",
"iggy",
diff --git a/bench/Cargo.toml b/bench/Cargo.toml
index ff919c28..ea88d38e 100644
--- a/bench/Cargo.toml
+++ b/bench/Cargo.toml
@@ -1,6 +1,6 @@
[package]
name = "bench"
-version = "0.2.3"
+version = "0.2.4"
edition = "2021"
license = "Apache-2.0"
# Due to dependency to integration, which has a dependency to server, setting
@@ -14,7 +14,6 @@ charming = "0.4.0"
chrono = "0.4.39"
clap = { version = "4.5.30", features = ["derive"] }
figlet-rs = "0.1.5"
-futures = "0.3.31"
hostname = "0.4.0"
human-repr = "1.1.0"
iggy = { path = "../sdk" }
diff --git a/bench/src/actors/consumer.rs b/bench/src/actors/consumer.rs
index 0f5860fb..e919067a 100644
--- a/bench/src/actors/consumer.rs
+++ b/bench/src/actors/consumer.rs
@@ -75,7 +75,7 @@ impl Consumer {
}
}
- pub async fn run(&self) -> Result<BenchmarkIndividualMetrics, IggyError> {
+ pub async fn run(self) -> Result<BenchmarkIndividualMetrics, IggyError> {
let topic_id: u32 = 1;
let default_partition_id: u32 = 1;
let message_batches = self.message_batches as u64;
diff --git a/bench/src/actors/producer.rs b/bench/src/actors/producer.rs
index 4cca4713..affbd457 100644
--- a/bench/src/actors/producer.rs
+++ b/bench/src/actors/producer.rs
@@ -71,7 +71,7 @@ impl Producer {
}
}
- pub async fn run(&self) -> Result<BenchmarkIndividualMetrics, IggyError> {
+ pub async fn run(self) -> Result<BenchmarkIndividualMetrics, IggyError> {
let topic_id: u32 = 1;
let default_partition_id: u32 = 1;
let partitions_count = self.partitions_count;
diff --git a/bench/src/actors/producing_consumer.rs b/bench/src/actors/producing_consumer.rs
index e0cd55a8..0517c2a3 100644
--- a/bench/src/actors/producing_consumer.rs
+++ b/bench/src/actors/producing_consumer.rs
@@ -82,7 +82,7 @@ impl ProducingConsumer {
}
}
- pub async fn run(&self) -> Result<BenchmarkIndividualMetrics, IggyError> {
+ pub async fn run(self) -> Result<BenchmarkIndividualMetrics, IggyError> {
let topic_id: u32 = 1;
let default_partition_id: u32 = 1;
let message_batches = self.message_batches;
diff --git a/bench/src/benchmarks/benchmark.rs b/bench/src/benchmarks/benchmark.rs
index be87d3c8..cc2f8fd4 100644
--- a/bench/src/benchmarks/benchmark.rs
+++ b/bench/src/benchmarks/benchmark.rs
@@ -1,7 +1,6 @@
use crate::args::kind::BenchmarkKindCommand;
use crate::{args::common::IggyBenchArgs, utils::client_factory::create_client_factory};
use async_trait::async_trait;
-use futures::Future;
use iggy::client::{StreamClient, TopicClient};
use iggy::clients::client::IggyClient;
use iggy::compression::compression_algorithm::CompressionAlgorithm;
@@ -11,7 +10,8 @@ use iggy::utils::topic_size::MaxTopicSize;
use iggy_bench_report::benchmark_kind::BenchmarkKind;
use iggy_bench_report::individual_metrics::BenchmarkIndividualMetrics;
use integration::test_server::{login_root, ClientFactory};
-use std::{pin::Pin, sync::Arc};
+use std::sync::Arc;
+use tokio::task::JoinSet;
use tracing::info;
use super::consumer_benchmark::ConsumerBenchmark;
@@ -22,11 +22,6 @@ use super::producer_benchmark::ProducerBenchmark;
use super::producing_consumer_benchmark::EndToEndProducingConsumerBenchmark;
use super::producing_consumer_group_benchmark::EndToEndProducingConsumerGroupBenchmark;
-pub type BenchmarkFutures = Result<
- Vec<Pin<Box<dyn Future<Output = Result<BenchmarkIndividualMetrics, IggyError>> + Send>>>,
- IggyError,
->;
-
impl From<IggyBenchArgs> for Box<dyn Benchmarkable> {
fn from(args: IggyBenchArgs) -> Self {
let client_factory = create_client_factory(&args);
@@ -63,7 +58,9 @@ impl From<IggyBenchArgs> for Box<dyn Benchmarkable> {
#[async_trait]
pub trait Benchmarkable {
- async fn run(&mut self) -> BenchmarkFutures;
+ async fn run(
+ &mut self,
+ ) -> Result<JoinSet<Result<BenchmarkIndividualMetrics, IggyError>>, IggyError>;
fn kind(&self) -> BenchmarkKind;
fn args(&self) -> &IggyBenchArgs;
fn client_factory(&self) -> &Arc<dyn ClientFactory>;
diff --git a/bench/src/benchmarks/consumer_benchmark.rs b/bench/src/benchmarks/consumer_benchmark.rs
index 33c92521..c5fffdd6 100644
--- a/bench/src/benchmarks/consumer_benchmark.rs
+++ b/bench/src/benchmarks/consumer_benchmark.rs
@@ -1,13 +1,16 @@
use crate::actors::consumer::Consumer;
use crate::args::common::IggyBenchArgs;
-use crate::benchmarks::benchmark::{BenchmarkFutures, Benchmarkable};
+use crate::benchmarks::benchmark::Benchmarkable;
use crate::rate_limiter::RateLimiter;
use async_trait::async_trait;
+use iggy::error::IggyError;
use iggy::messages::PollingKind;
use iggy_bench_report::benchmark_kind::BenchmarkKind;
+use iggy_bench_report::individual_metrics::BenchmarkIndividualMetrics;
use integration::test_server::ClientFactory;
use std::sync::atomic::AtomicI64;
use std::sync::Arc;
+use tokio::task::JoinSet;
use tracing::info;
pub struct ConsumerBenchmark {
@@ -26,14 +29,16 @@ impl ConsumerBenchmark {
#[async_trait]
impl Benchmarkable for ConsumerBenchmark {
- async fn run(&mut self) -> BenchmarkFutures {
+ async fn run(
+ &mut self,
+ ) -> Result<JoinSet<Result<BenchmarkIndividualMetrics, IggyError>>, IggyError> {
self.check_streams().await?;
let consumers_count = self.args.consumers();
info!("Creating {} consumer(s)...", consumers_count);
let messages_per_batch = self.args.messages_per_batch();
let message_batches = self.args.message_batches();
- let mut futures: BenchmarkFutures = Ok(Vec::with_capacity(consumers_count as usize));
+ let mut set = JoinSet::new();
for consumer_id in 1..=consumers_count {
let args = self.args.clone();
let client_factory = self.client_factory.clone();
@@ -76,12 +81,11 @@ impl Benchmarkable for ConsumerBenchmark {
args.rate_limit()
.map(|rl| RateLimiter::new(rl.as_bytes_u64())),
);
-
- let future = Box::pin(async move { consumer.run().await });
- futures.as_mut().unwrap().push(future);
+ set.spawn(consumer.run());
}
+
info!("Created {} consumer(s).", consumers_count);
- futures
+ Ok(set)
}
fn kind(&self) -> BenchmarkKind {
diff --git a/bench/src/benchmarks/consumer_group_benchmark.rs b/bench/src/benchmarks/consumer_group_benchmark.rs
index 697fd9b6..42195f56 100644
--- a/bench/src/benchmarks/consumer_group_benchmark.rs
+++ b/bench/src/benchmarks/consumer_group_benchmark.rs
@@ -1,4 +1,4 @@
-use super::benchmark::{BenchmarkFutures, Benchmarkable};
+use super::benchmark::Benchmarkable;
use crate::{
actors::consumer::Consumer,
args::common::IggyBenchArgs,
@@ -10,9 +10,12 @@ use iggy::{
client::ConsumerGroupClient, clients::client::IggyClient, error::IggyError,
messages::PollingKind,
};
-use iggy_bench_report::benchmark_kind::BenchmarkKind;
+use iggy_bench_report::{
+ benchmark_kind::BenchmarkKind, individual_metrics::BenchmarkIndividualMetrics,
+};
use integration::test_server::{login_root, ClientFactory};
use std::sync::{atomic::AtomicI64, Arc};
+use tokio::task::JoinSet;
use tracing::{error, info};
pub struct ConsumerGroupBenchmark {
@@ -69,7 +72,9 @@ impl ConsumerGroupBenchmark {
#[async_trait]
impl Benchmarkable for ConsumerGroupBenchmark {
- async fn run(&mut self) -> BenchmarkFutures {
+ async fn run(
+ &mut self,
+ ) -> Result<JoinSet<Result<BenchmarkIndividualMetrics, IggyError>>, IggyError> {
self.check_streams().await?;
let consumer_groups_count = self.args.number_of_consumer_groups();
self.init_consumer_groups(consumer_groups_count)
@@ -81,11 +86,11 @@ impl Benchmarkable for ConsumerGroupBenchmark {
let consumers = self.args.consumers();
let messages_per_batch = self.args.messages_per_batch();
let warmup_time = self.args.warmup_time();
- let mut futures: BenchmarkFutures = Ok(Vec::with_capacity((consumers) as usize));
let polling_kind = PollingKind::Next;
let message_batches = self.args.message_batches();
let total_message_batches = Arc::new(AtomicI64::new((message_batches * consumers) as i64));
+ let mut set = JoinSet::new();
for consumer_id in 1..=consumers {
let consumer_group_id =
start_consumer_group_id + 1 + (consumer_id % consumer_groups_count);
@@ -110,14 +115,14 @@ impl Benchmarkable for ConsumerGroupBenchmark {
.rate_limit()
.map(|rl| RateLimiter::new(rl.as_bytes_u64())),
);
- let future = Box::pin(async move { consumer.run().await });
- futures.as_mut().unwrap().push(future);
+ set.spawn(consumer.run());
}
+
info!(
"Starting consumer group benchmark with {} messages",
self.total_messages()
);
- futures
+ Ok(set)
}
fn kind(&self) -> BenchmarkKind {
diff --git a/bench/src/benchmarks/producer_and_consumer_benchmark.rs b/bench/src/benchmarks/producer_and_consumer_benchmark.rs
index 1c4a3fd8..6bfe6bee 100644
--- a/bench/src/benchmarks/producer_and_consumer_benchmark.rs
+++ b/bench/src/benchmarks/producer_and_consumer_benchmark.rs
@@ -1,14 +1,17 @@
use crate::actors::consumer::Consumer;
use crate::actors::producer::Producer;
use crate::args::common::IggyBenchArgs;
-use crate::benchmarks::benchmark::{BenchmarkFutures, Benchmarkable};
+use crate::benchmarks::benchmark::Benchmarkable;
use crate::rate_limiter::RateLimiter;
use async_trait::async_trait;
+use iggy::error::IggyError;
use iggy::messages::PollingKind;
use iggy_bench_report::benchmark_kind::BenchmarkKind;
+use iggy_bench_report::individual_metrics::BenchmarkIndividualMetrics;
use integration::test_server::ClientFactory;
use std::sync::atomic::AtomicI64;
use std::sync::Arc;
+use tokio::task::JoinSet;
use tracing::info;
pub struct ProducerAndConsumerBenchmark {
@@ -27,7 +30,9 @@ impl ProducerAndConsumerBenchmark {
#[async_trait]
impl Benchmarkable for ProducerAndConsumerBenchmark {
- async fn run(&mut self) -> BenchmarkFutures {
+ async fn run(
+ &mut self,
+ ) -> Result<JoinSet<Result<BenchmarkIndividualMetrics, IggyError>>, IggyError> {
self.init_streams().await.expect("Failed to init streams!");
let start_stream_id = self.args.start_stream_id();
let producers = self.args.producers();
@@ -38,8 +43,7 @@ impl Benchmarkable for ProducerAndConsumerBenchmark {
let partitions_count = self.args.number_of_partitions();
let warmup_time = self.args.warmup_time();
- let mut futures: BenchmarkFutures =
- Ok(Vec::with_capacity((producers + consumers) as usize));
+ let mut set = JoinSet::new();
for producer_id in 1..=producers {
let stream_id = start_stream_id + producer_id;
let producer = Producer::new(
@@ -59,8 +63,7 @@ impl Benchmarkable for ProducerAndConsumerBenchmark {
.map(|rl| RateLimiter::new(rl.as_bytes_u64())),
false, // TODO: put timestamp into first message, it should be an argument to iggy-bench
);
- let future = Box::pin(async move { producer.run().await });
- futures.as_mut().unwrap().push(future);
+ set.spawn(producer.run());
}
let polling_kind = PollingKind::Offset;
@@ -85,14 +88,14 @@ impl Benchmarkable for ProducerAndConsumerBenchmark {
.rate_limit()
.map(|rl| RateLimiter::new(rl.as_bytes_u64())),
);
- let future = Box::pin(async move { consumer.run().await });
- futures.as_mut().unwrap().push(future);
+ set.spawn(consumer.run());
}
+
info!(
"Starting to send and poll {} messages",
self.total_messages()
);
- futures
+ Ok(set)
}
fn kind(&self) -> BenchmarkKind {
diff --git a/bench/src/benchmarks/producer_and_consumer_group_benchmark.rs b/bench/src/benchmarks/producer_and_consumer_group_benchmark.rs
index 00f46eca..7fee789b 100644
--- a/bench/src/benchmarks/producer_and_consumer_group_benchmark.rs
+++ b/bench/src/benchmarks/producer_and_consumer_group_benchmark.rs
@@ -1,4 +1,4 @@
-use super::benchmark::{BenchmarkFutures, Benchmarkable};
+use super::benchmark::Benchmarkable;
use crate::{
actors::{consumer::Consumer, producer::Producer},
args::common::IggyBenchArgs,
@@ -8,9 +8,12 @@ use crate::{
use async_trait::async_trait;
use iggy::messages::PollingKind;
use iggy::{client::ConsumerGroupClient, clients::client::IggyClient, error::IggyError};
-use iggy_bench_report::benchmark_kind::BenchmarkKind;
+use iggy_bench_report::{
+ benchmark_kind::BenchmarkKind, individual_metrics::BenchmarkIndividualMetrics,
+};
use integration::test_server::{login_root, ClientFactory};
use std::sync::{atomic::AtomicI64, Arc};
+use tokio::task::JoinSet;
use tracing::{error, info};
pub struct ProducerAndConsumerGroupBenchmark {
@@ -67,7 +70,9 @@ impl ProducerAndConsumerGroupBenchmark {
#[async_trait]
impl Benchmarkable for ProducerAndConsumerGroupBenchmark {
- async fn run(&mut self) -> BenchmarkFutures {
+ async fn run(
+ &mut self,
+ ) -> Result<JoinSet<Result<BenchmarkIndividualMetrics, IggyError>>, IggyError> {
self.init_streams().await.expect("Failed to init streams!");
let consumer_groups_count = self.args.number_of_consumer_groups();
self.init_consumer_groups(consumer_groups_count)
@@ -96,9 +101,7 @@ impl Benchmarkable for ProducerAndConsumerGroupBenchmark {
message_batches * producers,
);
- let mut futures: BenchmarkFutures =
- Ok(Vec::with_capacity((producers + consumers) as usize));
-
+ let mut set = JoinSet::new();
for producer_id in 1..=producers {
info!("Executing the benchmark on producer #{}...", producer_id);
let stream_id = self.args.start_stream_id() + 1 + (producer_id % streams_number);
@@ -120,8 +123,7 @@ impl Benchmarkable for ProducerAndConsumerGroupBenchmark {
.map(|rl| RateLimiter::new(rl.as_bytes_u64())),
false, // TODO: Put latency into payload of first message, it should be an argument to iggy-bench
);
- let future = Box::pin(async move { producer.run().await });
- futures.as_mut().unwrap().push(future);
+ set.spawn(producer.run());
}
info!("Created {} producer(s).", producers);
@@ -148,15 +150,14 @@ impl Benchmarkable for ProducerAndConsumerGroupBenchmark {
.rate_limit()
.map(|rl| RateLimiter::new(rl.as_bytes_u64())),
);
- let future = Box::pin(async move { consumer.run().await });
- futures.as_mut().unwrap().push(future);
+ set.spawn(consumer.run());
}
info!(
"Starting to send and poll {} messages",
self.total_messages()
);
- futures
+ Ok(set)
}
fn kind(&self) -> BenchmarkKind {
diff --git a/bench/src/benchmarks/producer_benchmark.rs b/bench/src/benchmarks/producer_benchmark.rs
index 0e9379e0..7453508b 100644
--- a/bench/src/benchmarks/producer_benchmark.rs
+++ b/bench/src/benchmarks/producer_benchmark.rs
@@ -1,11 +1,14 @@
use crate::actors::producer::Producer;
use crate::args::common::IggyBenchArgs;
-use crate::benchmarks::benchmark::{BenchmarkFutures, Benchmarkable};
+use crate::benchmarks::benchmark::Benchmarkable;
use crate::rate_limiter::RateLimiter;
use async_trait::async_trait;
+use iggy::error::IggyError;
use iggy_bench_report::benchmark_kind::BenchmarkKind;
+use iggy_bench_report::individual_metrics::BenchmarkIndividualMetrics;
use integration::test_server::ClientFactory;
use std::sync::Arc;
+use tokio::task::JoinSet;
use tracing::info;
pub struct ProducerBenchmark {
@@ -24,7 +27,9 @@ impl ProducerBenchmark {
#[async_trait]
impl Benchmarkable for ProducerBenchmark {
- async fn run(&mut self) -> BenchmarkFutures {
+ async fn run(
+ &mut self,
+ ) -> Result<JoinSet<Result<BenchmarkIndividualMetrics, IggyError>>, IggyError> {
self.init_streams().await.expect("Failed to init streams!");
let producers_count = self.args.producers();
info!("Creating {} producer(s)...", producers_count);
@@ -35,7 +40,7 @@ impl Benchmarkable for ProducerBenchmark {
let partitions_count = self.args.number_of_partitions();
let warmup_time = self.args.warmup_time();
- let mut futures: BenchmarkFutures = Ok(Vec::with_capacity(producers_count as usize));
+ let mut set = JoinSet::new();
for producer_id in 1..=producers_count {
let args = self.args.clone();
let client_factory = self.client_factory.clone();
@@ -58,11 +63,11 @@ impl Benchmarkable for ProducerBenchmark {
.map(|rl| RateLimiter::new(rl.as_bytes_u64())),
false, // TODO: Put timestamp in payload of first message, it should be an argument to iggy-bench
);
- let future = Box::pin(async move { producer.run().await });
- futures.as_mut().unwrap().push(future);
+ set.spawn(producer.run());
}
+
info!("Created {} producer(s).", producers_count);
- futures
+ Ok(set)
}
fn kind(&self) -> BenchmarkKind {
diff --git a/bench/src/benchmarks/producing_consumer_benchmark.rs b/bench/src/benchmarks/producing_consumer_benchmark.rs
index 8909636f..4692280b 100644
--- a/bench/src/benchmarks/producing_consumer_benchmark.rs
+++ b/bench/src/benchmarks/producing_consumer_benchmark.rs
@@ -1,13 +1,16 @@
use crate::actors::producing_consumer::ProducingConsumer;
use crate::args::common::IggyBenchArgs;
-use crate::benchmarks::benchmark::{BenchmarkFutures, Benchmarkable};
+use crate::benchmarks::benchmark::Benchmarkable;
use crate::rate_limiter::RateLimiter;
use async_trait::async_trait;
+use iggy::error::IggyError;
use iggy::messages::PollingKind;
use iggy_bench_report::benchmark_kind::BenchmarkKind;
+use iggy_bench_report::individual_metrics::BenchmarkIndividualMetrics;
use integration::test_server::ClientFactory;
use std::sync::atomic::AtomicI64;
use std::sync::Arc;
+use tokio::task::JoinSet;
use tracing::info;
pub struct EndToEndProducingConsumerBenchmark {
@@ -26,7 +29,9 @@ impl EndToEndProducingConsumerBenchmark {
#[async_trait]
impl Benchmarkable for EndToEndProducingConsumerBenchmark {
- async fn run(&mut self) -> BenchmarkFutures {
+ async fn run(
+ &mut self,
+ ) -> Result<JoinSet<Result<BenchmarkIndividualMetrics, IggyError>>, IggyError> {
self.init_streams().await.expect("Failed to init streams!");
let actors_count = self.args.producers();
info!("Creating {} producing consumer(s)...", actors_count);
@@ -36,7 +41,8 @@ impl Benchmarkable for EndToEndProducingConsumerBenchmark {
let message_size = self.args.message_size();
let warmup_time = self.args.warmup_time();
let polling_kind = PollingKind::Offset;
- let mut futures: BenchmarkFutures = Ok(Vec::with_capacity(actors_count as usize));
+
+ let mut set = JoinSet::new();
for actor_id in 1..=actors_count {
let args = self.args.clone();
let client_factory = self.client_factory.clone();
@@ -65,11 +71,11 @@ impl Benchmarkable for EndToEndProducingConsumerBenchmark {
polling_kind,
false,
);
- let future = Box::pin(async move { actor.run().await });
- futures.as_mut().unwrap().push(future);
+ set.spawn(actor.run());
}
+
info!("Created {} producing consumer(s).", actors_count);
- futures
+ Ok(set)
}
fn kind(&self) -> BenchmarkKind {
diff --git a/bench/src/benchmarks/producing_consumer_group_benchmark.rs b/bench/src/benchmarks/producing_consumer_group_benchmark.rs
index 81b8f52b..b5d57c7e 100644
--- a/bench/src/benchmarks/producing_consumer_group_benchmark.rs
+++ b/bench/src/benchmarks/producing_consumer_group_benchmark.rs
@@ -1,6 +1,6 @@
use crate::actors::producing_consumer::ProducingConsumer;
use crate::args::common::IggyBenchArgs;
-use crate::benchmarks::benchmark::{BenchmarkFutures, Benchmarkable};
+use crate::benchmarks::benchmark::Benchmarkable;
use crate::benchmarks::{CONSUMER_GROUP_BASE_ID, CONSUMER_GROUP_NAME_PREFIX};
use crate::rate_limiter::RateLimiter;
use async_trait::async_trait;
@@ -9,9 +9,11 @@ use iggy::clients::client::IggyClient;
use iggy::error::IggyError;
use iggy::messages::PollingKind;
use iggy_bench_report::benchmark_kind::BenchmarkKind;
+use iggy_bench_report::individual_metrics::BenchmarkIndividualMetrics;
use integration::test_server::{login_root, ClientFactory};
use std::sync::atomic::AtomicI64;
use std::sync::Arc;
+use tokio::task::JoinSet;
use tracing::{error, info};
pub struct EndToEndProducingConsumerGroupBenchmark {
@@ -68,7 +70,9 @@ impl EndToEndProducingConsumerGroupBenchmark {
#[async_trait]
impl Benchmarkable for EndToEndProducingConsumerGroupBenchmark {
- async fn run(&mut self) -> BenchmarkFutures {
+ async fn run(
+ &mut self,
+ ) -> Result<JoinSet<Result<BenchmarkIndividualMetrics, IggyError>>, IggyError> {
self.init_streams().await.expect("Failed to init streams!");
let consumer_groups_count = self.args.number_of_consumer_groups();
self.init_consumer_groups(consumer_groups_count)
@@ -87,11 +91,11 @@ impl Benchmarkable for EndToEndProducingConsumerGroupBenchmark {
let partitions_count = self.args.number_of_partitions();
let warmup_time = self.args.warmup_time();
let polling_kind = PollingKind::Next;
- let mut futures: BenchmarkFutures = Ok(Vec::with_capacity(actors_count as usize));
let consumer_groups_count = self.args.number_of_consumer_groups();
let total_message_batches =
Arc::new(AtomicI64::new((message_batches * actors_count) as i64));
+ let mut set = JoinSet::new();
for actor_id in 1..=actors_count {
let args = self.args.clone();
let client_factory = self.client_factory.clone();
@@ -122,11 +126,11 @@ impl Benchmarkable for EndToEndProducingConsumerGroupBenchmark {
polling_kind,
false,
);
- let future = Box::pin(async move { actor.run().await });
- futures.as_mut().unwrap().push(future);
+ set.spawn(actor.run());
}
+
info!("Created {actors_count} producing consumer(s) which would be part of {consumer_groups_count} consumer groups");
- futures
+ Ok(set)
}
fn kind(&self) -> BenchmarkKind {
diff --git a/bench/src/runner.rs b/bench/src/runner.rs
index 71cbfb68..60732720 100644
--- a/bench/src/runner.rs
+++ b/bench/src/runner.rs
@@ -5,7 +5,6 @@ use crate::plot::{plot_chart, ChartType};
use crate::utils::collect_server_logs_and_save_to_file;
use crate::utils::cpu_name::append_cpu_name_lowercase;
use crate::utils::server_starter::start_server_if_needed;
-use futures::future::select_all;
use iggy::error::IggyError;
use iggy_bench_report::hardware::BenchmarkHardware;
use iggy_bench_report::params::BenchmarkParams;
@@ -42,18 +41,15 @@ impl BenchmarkRunner {
let mut individual_metrics = Vec::new();
- while !join_handles.is_empty() {
- let (result, _index, remaining) = select_all(join_handles).await;
- join_handles = remaining;
-
- match result {
- Ok(r) => individual_metrics.push(r),
+ while let Some(individual_metric) = join_handles.join_next().await {
+ let individual_metric = individual_metric.expect("Failed to join actor!");
+ match individual_metric {
+ Ok(individual_metric) => individual_metrics.push(individual_metric),
Err(e) => return Err(e),
}
}
info!("All actors joined!");
-
let hardware =
BenchmarkHardware::get_system_info_with_identifier(benchmark.args().identifier());
let params = BenchmarkParams::from(benchmark.args());
(iggy) 01/02: rename headers to user_headers
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch zero-copy-no-batching-parallel-bench
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 61b0b00ba582a9b43be6465d0d7e164903a792aa
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Mon Mar 17 15:43:16 2025 +0100
rename headers to user_headers
---
bench/src/actors/consumer.rs | 3 +-
integration/tests/streaming/get_by_offset.rs | 5 +-
integration/tests/streaming/get_by_timestamp.rs | 4 +-
sdk/src/clients/producer.rs | 2 +-
sdk/src/models/messaging/message.rs | 17 +-
sdk/src/models/messaging/message_header.rs | 46 ++++
.../src/compat/index_rebuilding/index_rebuilder.rs | 56 ++--
server/src/http/messages.rs | 48 ++--
server/src/streaming/mod.rs | 2 +-
.../src/streaming/segments/indexes/indexes_mut.rs | 2 +-
.../streaming/segments/types/message_view_mut.rs | 1 -
.../src/streaming/segments/types/messages_batch.rs | 2 +-
server/src/streaming/systems/system.rs | 29 --
server/src/streaming/topics/messages.rs | 298 ++++++++++-----------
14 files changed, 257 insertions(+), 258 deletions(-)
diff --git a/bench/src/actors/consumer.rs b/bench/src/actors/consumer.rs
index 0c451009..0f5860fb 100644
--- a/bench/src/actors/consumer.rs
+++ b/bench/src/actors/consumer.rs
@@ -16,7 +16,6 @@ use iggy_bench_report::actor_kind::ActorKind;
use iggy_bench_report::benchmark_kind::BenchmarkKind;
use iggy_bench_report::individual_metrics::BenchmarkIndividualMetrics;
use integration::test_server::{login_root, ClientFactory};
-use std::hint::black_box;
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::Arc;
use std::time::Duration;
@@ -27,7 +26,7 @@ pub struct Consumer {
client_factory: Arc<dyn ClientFactory>,
benchmark_kind: BenchmarkKind,
consumer_id: u32,
- consumer_group_id: Option<u32>,
+ consumer_group_id: Option<u32>,
stream_id: u32,
messages_per_batch: u32,
message_batches: u32,
diff --git a/integration/tests/streaming/get_by_offset.rs b/integration/tests/streaming/get_by_offset.rs
index 6c41c6f9..474bfad2 100644
--- a/integration/tests/streaming/get_by_offset.rs
+++ b/integration/tests/streaming/get_by_offset.rs
@@ -242,7 +242,8 @@ async fn test_get_messages_by_offset(
let original_message = &all_messages[original_idx];
let original_msg_header = &original_message.header;
let original_headers = Some(
- HashMap::from_bytes(original_message.headers.as_ref().unwrap().to_bytes()).unwrap(),
+ HashMap::from_bytes(original_message.user_headers.as_ref().unwrap().to_bytes())
+ .unwrap(),
);
let original_payload = &original_message.payload;
@@ -257,7 +258,7 @@ async fn test_get_messages_by_offset(
msg.header.offset,
);
assert_eq!(
- &msg.headers, &original_headers,
+ &msg.user_headers, &original_headers,
"Headers mismatch at offset {}",
msg.header.offset,
);
diff --git a/integration/tests/streaming/get_by_timestamp.rs b/integration/tests/streaming/get_by_timestamp.rs
index 0e2afa51..49832f14 100644
--- a/integration/tests/streaming/get_by_timestamp.rs
+++ b/integration/tests/streaming/get_by_timestamp.rs
@@ -225,9 +225,9 @@ async fn test_get_messages_by_timestamp(
i, original_message.payload, loaded_message.payload
);
assert_eq!(
- loaded_message.headers, original_message.headers,
+ loaded_message.user_headers, original_message.user_headers,
"Headers mismatch at position {}: expected {:?}, got {:?}",
- i, original_message.headers, loaded_message.headers
+ i, original_message.user_headers, loaded_message.user_headers
);
assert!(
loaded_message.header.timestamp >= initial_timestamp.as_micros(),
diff --git a/sdk/src/clients/producer.rs b/sdk/src/clients/producer.rs
index e47e9a57..f66de103 100644
--- a/sdk/src/clients/producer.rs
+++ b/sdk/src/clients/producer.rs
@@ -4,8 +4,8 @@ use crate::diagnostic::DiagnosticEvent;
use crate::error::IggyError;
use crate::identifier::{IdKind, Identifier};
use crate::locking::{IggySharedMut, IggySharedMutFn};
-use crate::prelude::IggyMessage;
use crate::partitioner::Partitioner;
+use crate::prelude::IggyMessage;
use crate::prelude::Partitioning;
use crate::utils::crypto::EncryptorKind;
use crate::utils::duration::IggyDuration;
diff --git a/sdk/src/models/messaging/message.rs b/sdk/src/models/messaging/message.rs
index e85fb13e..45ea1a1f 100644
--- a/sdk/src/models/messaging/message.rs
+++ b/sdk/src/models/messaging/message.rs
@@ -7,6 +7,7 @@ use crate::utils::byte_size::IggyByteSize;
use crate::utils::sizeable::Sizeable;
use crate::utils::timestamp::IggyTimestamp;
use bytes::{BufMut, Bytes, BytesMut};
+use serde::{Deserialize, Serialize};
use serde_with::base64::Base64;
use serde_with::serde_as;
use std::collections::HashMap;
@@ -14,12 +15,12 @@ use std::str::FromStr;
/// The single message. It is exact format in which message is saved to / retrieved from the disk.
#[serde_as]
-#[derive(Default, Debug, serde::Serialize, serde::Deserialize, PartialEq)]
+#[derive(Default, Debug, Serialize, Deserialize, PartialEq)]
pub struct IggyMessage {
pub header: IggyMessageHeader,
#[serde_as(as = "Base64")]
pub payload: Bytes,
- pub headers: Option<HashMap<HeaderKey, HeaderValue>>,
+ pub user_headers: Option<HashMap<HeaderKey, HeaderValue>>,
}
impl IggyMessage {
@@ -82,7 +83,7 @@ impl IggyMessage {
messages.push(IggyMessage {
header,
payload,
- headers,
+ user_headers: headers,
});
}
@@ -125,7 +126,7 @@ impl std::fmt::Display for IggyMessage {
impl Sizeable for IggyMessage {
fn get_size_bytes(&self) -> IggyByteSize {
let payload_len = IggyByteSize::from(self.payload.len() as u64);
- let headers_len = get_headers_size_bytes(&self.headers);
+ let headers_len = get_headers_size_bytes(&self.user_headers);
let message_header_len = IggyByteSize::from(IGGY_MESSAGE_HEADER_SIZE as u64);
payload_len + headers_len + message_header_len
@@ -138,7 +139,7 @@ impl BytesSerializable for IggyMessage {
let message_header = self.header.to_bytes();
bytes.put_slice(&message_header);
bytes.put_slice(&self.payload);
- if let Some(headers) = &self.headers {
+ if let Some(headers) = &self.user_headers {
bytes.put_slice(&headers.to_bytes());
}
bytes.freeze()
@@ -170,7 +171,7 @@ impl BytesSerializable for IggyMessage {
Ok(IggyMessage {
header,
payload,
- headers,
+ user_headers: headers,
})
}
@@ -178,7 +179,7 @@ impl BytesSerializable for IggyMessage {
fn write_to_buffer(&self, buf: &mut BytesMut) {
buf.put_slice(&self.header.to_bytes());
buf.put_slice(&self.payload);
- if let Some(headers) = &self.headers {
+ if let Some(headers) = &self.user_headers {
buf.put_slice(&headers.to_bytes());
}
}
@@ -239,7 +240,7 @@ impl IggyMessageBuilder {
IggyMessage {
header: msg_header,
payload,
- headers: self.headers,
+ user_headers: self.headers,
}
}
}
diff --git a/sdk/src/models/messaging/message_header.rs b/sdk/src/models/messaging/message_header.rs
index 807bd371..ea636c03 100644
--- a/sdk/src/models/messaging/message_header.rs
+++ b/sdk/src/models/messaging/message_header.rs
@@ -34,6 +34,52 @@ impl Sizeable for IggyMessageHeader {
}
}
+impl IggyMessageHeader {
+ pub fn from_raw_bytes(bytes: &[u8]) -> Result<Self, IggyError> {
+ if bytes.len() != IGGY_MESSAGE_HEADER_SIZE as usize {
+ return Err(IggyError::InvalidCommand);
+ }
+
+ Ok(IggyMessageHeader {
+ checksum: u64::from_le_bytes(
+ bytes[IGGY_MESSAGE_CHECKSUM_OFFSET_RANGE]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ ),
+ id: u128::from_le_bytes(
+ bytes[IGGY_MESSAGE_ID_OFFSET_RANGE]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ ),
+ offset: u64::from_le_bytes(
+ bytes[IGGY_MESSAGE_OFFSET_OFFSET_RANGE]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ ),
+ timestamp: u64::from_le_bytes(
+ bytes[IGGY_MESSAGE_TIMESTAMP_OFFSET_RANGE]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ ),
+ origin_timestamp: u64::from_le_bytes(
+ bytes[IGGY_MESSAGE_ORIGIN_TIMESTAMP_OFFSET_RANGE]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ ),
+ headers_length: u32::from_le_bytes(
+ bytes[IGGY_MESSAGE_HEADERS_LENGTH_OFFSET_RANGE]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ ),
+ payload_length: u32::from_le_bytes(
+ bytes[IGGY_MESSAGE_PAYLOAD_LENGTH_OFFSET_RANGE]
+ .try_into()
+ .map_err(|_| IggyError::InvalidNumberEncoding)?,
+ ),
+ })
+ }
+}
+
impl BytesSerializable for IggyMessageHeader {
fn to_bytes(&self) -> Bytes {
let mut bytes = BytesMut::with_capacity(self.get_size_bytes().as_bytes_usize());
diff --git a/server/src/compat/index_rebuilding/index_rebuilder.rs b/server/src/compat/index_rebuilding/index_rebuilder.rs
index a3f3c777..a027f0bf 100644
--- a/server/src/compat/index_rebuilding/index_rebuilder.rs
+++ b/server/src/compat/index_rebuilding/index_rebuilder.rs
@@ -1,16 +1,9 @@
use crate::server_error::CompatError;
use crate::streaming::utils::file;
+use iggy::prelude::{IggyMessageHeader, IGGY_MESSAGE_HEADER_SIZE};
use std::io::SeekFrom;
use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader, BufWriter};
-// Same struct as RetainedMessageBatch, but without payload
-pub struct BatchHeader {
- base_offset: u64,
- last_offset_delta: u32,
- max_timestamp: u64,
- length: u32,
-}
-
pub struct IndexRebuilder {
pub log_path: String,
pub index_path: String,
@@ -26,61 +19,60 @@ impl IndexRebuilder {
}
}
- async fn read_batch_header(
+ async fn read_message_header(
reader: &mut BufReader<tokio::fs::File>,
- ) -> Result<BatchHeader, std::io::Error> {
- let base_offset = reader.read_u64_le().await?;
- let length = reader.read_u32_le().await?;
- let last_offset_delta = reader.read_u32_le().await?;
- let max_timestamp = reader.read_u64_le().await?;
-
- Ok(BatchHeader {
- base_offset,
- length,
- last_offset_delta,
- max_timestamp,
- })
+ ) -> Result<IggyMessageHeader, std::io::Error> {
+ let mut buf = [0u8; IGGY_MESSAGE_HEADER_SIZE as usize];
+ reader.read_exact(&mut buf).await?;
+ Ok(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>,
- header: &BatchHeader,
+ header: &IggyMessageHeader,
position: u32,
start_offset: u64,
) -> Result<(), CompatError> {
// Write offset (4 bytes) - base_offset + last_offset_delta - start_offset
- let offset = (header.base_offset + header.last_offset_delta as u64 - start_offset) as u32;
- writer.write_u32_le(offset).await?;
+ let offset = start_offset - header.offset;
+ debug_assert!(offset <= u32::MAX as u64);
+ writer.write_u32_le(offset as u32).await?;
// Write position (4 bytes)
writer.write_u32_le(position).await?;
// Write timestamp (8 bytes)
- writer.write_u64_le(header.max_timestamp).await?;
+ writer.write_u64_le(header.timestamp).await?;
Ok(())
}
pub async fn rebuild(&self) -> Result<(), CompatError> {
- //TODO: Fix me
- /*
let mut reader = BufReader::new(file::open(&self.log_path).await?);
let mut writer = BufWriter::new(file::overwrite(&self.index_path).await?);
let mut position = 0;
let mut next_position;
loop {
- match Self::read_batch_header(&mut reader).await {
+ match Self::read_message_header(&mut reader).await {
Ok(header) => {
// Calculate next position before writing current entry
- next_position = position + RETAINED_BATCH_HEADER_LEN as u32 + header.length;
+ next_position = position
+ + IGGY_MESSAGE_HEADER_SIZE as u32
+ + header.payload_length
+ + header.headers_length;
// Write index entry using current position
Self::write_index_entry(&mut writer, &header, position, self.start_offset)
.await?;
- // Skip batch messages
- reader.seek(SeekFrom::Current(header.length as i64)).await?;
+ // Skip message payload and headers
+ reader
+ .seek(SeekFrom::Current(
+ header.payload_length as i64 + header.headers_length as i64,
+ ))
+ .await?;
// Update position for next iteration
position = next_position;
@@ -92,7 +84,5 @@ impl IndexRebuilder {
writer.flush().await?;
Ok(())
- */
- todo!()
}
}
diff --git a/server/src/http/messages.rs b/server/src/http/messages.rs
index 4a30b1e4..7d1a3031 100644
--- a/server/src/http/messages.rs
+++ b/server/src/http/messages.rs
@@ -35,32 +35,30 @@ async fn poll_messages(
Extension(identity): Extension<Identity>,
Path((stream_id, topic_id)): Path<(String, String)>,
mut query: Query<PollMessages>,
-) -> Result<Json<()>, CustomError> {
- /*
- query.stream_id = Identifier::from_str_value(&stream_id)?;
- query.topic_id = Identifier::from_str_value(&topic_id)?;
- query.validate()?;
+) -> Result<Json<Vec<IggyMessage>>, CustomError> {
+ // query.stream_id = Identifier::from_str_value(&stream_id)?;
+ // query.topic_id = Identifier::from_str_value(&topic_id)?;
+ // query.validate()?;
- let consumer = Consumer::new(query.0.consumer.id);
- let system = state.system.read().await;
- let polled_messages = system
- .poll_messages(
- &Session::stateless(identity.user_id, identity.ip_address),
- &consumer,
- &query.0.stream_id,
- &query.0.topic_id,
- query.0.partition_id,
- PollingArgs::new(query.0.strategy, query.0.count, query.0.auto_commit),
- )
- .await
- .with_error_context(|error| {
- format!(
- "{COMPONENT} (error: {error}) - failed to poll messages, stream ID: {}, topic ID: {}, partition ID: {:?}",
- stream_id, topic_id, query.0.partition_id
- )
- })?;
- Ok(Json(polled_messages))
- */
+ // let consumer = Consumer::new(query.0.consumer.id);
+ // let system = state.system.read().await;
+ // let polled_messages = system
+ // .poll_messages(
+ // &Session::stateless(identity.user_id, identity.ip_address),
+ // &consumer,
+ // &query.0.stream_id,
+ // &query.0.topic_id,
+ // query.0.partition_id,
+ // PollingArgs::new(query.0.strategy, query.0.count, query.0.auto_commit),
+ // )
+ // .await
+ // .with_error_context(|error| {
+ // format!(
+ // "{COMPONENT} (error: {error}) - failed to poll messages, stream ID: {}, topic ID: {}, partition ID: {:?}",
+ // stream_id, topic_id, query.0.partition_id
+ // )
+ // })?;
+ // Ok(Json(polled_messages))
todo!()
}
diff --git a/server/src/streaming/mod.rs b/server/src/streaming/mod.rs
index d4b5b06a..00bb365d 100644
--- a/server/src/streaming/mod.rs
+++ b/server/src/streaming/mod.rs
@@ -1,4 +1,4 @@
-pub mod cache;
+// pub mod cache;
pub mod clients;
mod deduplication;
pub mod diagnostics;
diff --git a/server/src/streaming/segments/indexes/indexes_mut.rs b/server/src/streaming/segments/indexes/indexes_mut.rs
index b0817389..7f5a47f4 100644
--- a/server/src/streaming/segments/indexes/indexes_mut.rs
+++ b/server/src/streaming/segments/indexes/indexes_mut.rs
@@ -258,7 +258,7 @@ impl IggyIndexesMut {
);
0
};
-
+
// Ensure we don't have underflow when calculating messages count
let actual_messages_count = if end_position_in_array >= start_position_in_array {
end_position_in_array - start_position_in_array + 1
diff --git a/server/src/streaming/segments/types/message_view_mut.rs b/server/src/streaming/segments/types/message_view_mut.rs
index 9dcf53cd..3f14cc9b 100644
--- a/server/src/streaming/segments/types/message_view_mut.rs
+++ b/server/src/streaming/segments/types/message_view_mut.rs
@@ -41,7 +41,6 @@ impl<'a> IggyMessageViewMut<'a> {
/// Returns the size of the entire message (header + payload + user headers).
pub fn size(&self) -> usize {
- // TODO(hubcio): remove unwraps()
let hdr_view = self.msg_header();
(IGGY_MESSAGE_HEADER_SIZE + hdr_view.payload_length() + hdr_view.headers_length()) as usize
}
diff --git a/server/src/streaming/segments/types/messages_batch.rs b/server/src/streaming/segments/types/messages_batch.rs
index 05567cae..07e8c96a 100644
--- a/server/src/streaming/segments/types/messages_batch.rs
+++ b/server/src/streaming/segments/types/messages_batch.rs
@@ -198,7 +198,7 @@ impl IggyBatch {
messages.push(IggyMessage {
header,
payload,
- headers,
+ user_headers: headers,
});
}
}
diff --git a/server/src/streaming/systems/system.rs b/server/src/streaming/systems/system.rs
index 2a068f5f..6ec716ae 100644
--- a/server/src/streaming/systems/system.rs
+++ b/server/src/streaming/systems/system.rs
@@ -5,7 +5,6 @@ use crate::map_toggle_str;
use crate::state::file::FileState;
use crate::state::system::SystemState;
use crate::state::StateKind;
-use crate::streaming::cache::memory_tracker::CacheMemoryTracker;
use crate::streaming::clients::client_manager::ClientManager;
use crate::streaming::diagnostics::metrics::Metrics;
use crate::streaming::persistence::persister::*;
@@ -22,7 +21,6 @@ use iggy::error::IggyError;
use iggy::locking::IggySharedMut;
use iggy::locking::IggySharedMutFn;
use iggy::models::user_info::UserId;
-use iggy::utils::byte_size::IggyByteSize;
use iggy::utils::crypto::{Aes256GcmEncryptor, EncryptorKind};
use std::path::Path;
use std::sync::Arc;
@@ -76,10 +74,6 @@ pub struct System {
pub personal_access_token: PersonalAccessTokenConfig,
}
-/// For each cache eviction, we want to remove more than the size we need.
-/// This is done on purpose to avoid evicting messages on every write.
-const CACHE_OVER_EVICTION_FACTOR: u64 = 5;
-
impl System {
pub fn new(
config: Arc<SystemConfig>,
@@ -267,27 +261,4 @@ impl System {
Err(IggyError::Unauthenticated)
}
}
-
- pub async fn clean_cache(&self, size_to_clean: IggyByteSize) {
- //TODO: Fix me
- /*
- for stream in self.streams.values() {
- for topic in stream.get_topics() {
- for partition in topic.get_partitions().into_iter() {
- tokio::task::spawn(async move {
- let memory_tracker = CacheMemoryTracker::get_instance().unwrap();
- let mut partition_guard = partition.write().await;
- let cache = &mut partition_guard.cache.as_mut().unwrap();
- let size_to_remove = (cache.current_size().as_bytes_u64() as f64
- / memory_tracker.usage_bytes().as_bytes_u64() as f64
- * size_to_clean.as_bytes_u64() as f64)
- .ceil() as u64;
- cache.evict_by_size(size_to_remove * CACHE_OVER_EVICTION_FACTOR);
- });
- }
- }
- }
- */
- todo!()
- }
}
diff --git a/server/src/streaming/topics/messages.rs b/server/src/streaming/topics/messages.rs
index e9f038be..2ff01791 100644
--- a/server/src/streaming/topics/messages.rs
+++ b/server/src/streaming/topics/messages.rs
@@ -50,15 +50,13 @@ impl Topic {
let result = match strategy.kind {
PollingKind::Offset => partition.get_messages_by_offset(value, count).await,
PollingKind::Timestamp => {
- //TODO: Fix me
- /*
+
+
partition
.get_messages_by_timestamp(value.into(), count)
.await
.with_error_context(|error| format!("{COMPONENT} (error: {error}) - failed to get messages by timestamp: {value}, count: {count}"))
- */
- todo!()
- }
+ }
PollingKind::First => partition.get_first_messages(count).await,
PollingKind::Last => partition.get_last_messages(count).await,
PollingKind::Next => partition.get_next_messages(consumer, count).await,
@@ -83,11 +81,9 @@ impl Topic {
return Err(IggyError::TopicFull(self.topic_id, self.stream_id));
}
- /*
if messages.is_empty() {
return Ok(());
}
- */
let partition_id = match partitioning.kind {
PartitioningKind::Balanced => self.get_next_partition_id(),
@@ -195,8 +191,6 @@ impl Topic {
#[cfg(test)]
mod tests {
- //TODO: Fix me
- /*
use super::*;
use crate::configs::system::SystemConfig;
use crate::streaming::persistence::persister::FileWithSyncPersister;
@@ -204,152 +198,152 @@ mod tests {
use crate::streaming::storage::SystemStorage;
use bytes::Bytes;
use iggy::compression::compression_algorithm::CompressionAlgorithm;
+ use iggy::prelude::IggyMessage;
use iggy::utils::topic_size::MaxTopicSize;
use std::sync::atomic::AtomicU32;
use std::sync::atomic::AtomicU64;
use std::sync::Arc;
- #[tokio::test]
- async fn given_partition_id_key_messages_should_be_appended_only_to_the_chosen_partition() {
- let partition_id = 1;
- let partitioning = Partitioning::partition_id(partition_id);
- let partitions_count = 3;
- let messages_count: u32 = 1000;
- let topic = init_topic(partitions_count).await;
-
- for entity_id in 1..=messages_count {
- let messages = vec![Message::new(Some(entity_id as u128), Bytes::new(), None)];
- let batch_size = messages
- .iter()
- .map(|msg| msg.get_size_bytes())
- .sum::<IggyByteSize>();
- topic
- .append_messages(batch_size, partitioning.clone(), messages, None)
- .await
- .unwrap();
- }
-
- let partitions = topic.get_partitions();
- assert_eq!(partitions.len(), partitions_count as usize);
- for partition in partitions {
- let partition = partition.read().await;
- let messages = partition.cache.as_ref().unwrap().to_vec();
- if partition.partition_id == partition_id {
- assert_eq!(messages.len() as u32, messages_count);
- } else {
- assert_eq!(messages.len() as u32, 0);
- }
- }
- }
-
- #[tokio::test]
- async fn given_messages_key_key_messages_should_be_appended_to_the_calculated_partitions() {
- let partitions_count = 3;
- let messages_count = 1000;
- let topic = init_topic(partitions_count).await;
-
- for entity_id in 1..=messages_count {
- let partitioning = Partitioning::messages_key_u32(entity_id);
- let messages = vec![Message::new(Some(entity_id as u128), Bytes::new(), None)];
- let batch_size = messages
- .iter()
- .map(|msg| msg.get_size_bytes())
- .sum::<IggyByteSize>();
- topic
- .append_messages(batch_size, partitioning, messages, None)
- .await
- .unwrap();
- }
-
- let mut read_messages_count = 0;
- let partitions = topic.get_partitions();
- assert_eq!(partitions.len(), partitions_count as usize);
- for partition in partitions {
- let partition = partition.read().await;
- let messages = partition.cache.as_ref().unwrap().to_vec();
- read_messages_count += messages.len();
- assert!(messages.len() < messages_count as usize);
- }
-
- assert_eq!(read_messages_count, messages_count as usize);
- }
-
- #[tokio::test]
- async fn given_multiple_partitions_calculate_next_partition_id_should_return_next_partition_id_using_round_robin(
- ) {
- let partitions_count = 3;
- let messages_count = 1000;
- let topic = init_topic(partitions_count).await;
-
- let mut expected_partition_id = 0;
- for _ in 1..=messages_count {
- let partition_id = topic.get_next_partition_id();
- expected_partition_id += 1;
- if expected_partition_id > partitions_count {
- expected_partition_id = 1;
- }
-
- assert_eq!(partition_id, expected_partition_id);
- }
- }
-
- #[tokio::test]
- async fn given_multiple_partitions_calculate_partition_id_by_hash_should_return_next_partition_id(
- ) {
- let partitions_count = 3;
- let messages_count = 1000;
- let topic = init_topic(partitions_count).await;
-
- for entity_id in 1..=messages_count {
- let key = Partitioning::messages_key_u32(entity_id);
- let partition_id = topic.calculate_partition_id_by_messages_key_hash(&key.value);
- let entity_id_hash = hash::calculate_32(&key.value);
- let mut expected_partition_id = entity_id_hash % partitions_count;
- if expected_partition_id == 0 {
- expected_partition_id = partitions_count;
- }
-
- assert_eq!(partition_id, expected_partition_id);
- }
- }
-
- async fn init_topic(partitions_count: u32) -> Topic {
- let tempdir = tempfile::TempDir::new().unwrap();
- let config = Arc::new(SystemConfig {
- path: tempdir.path().to_str().unwrap().to_string(),
- ..Default::default()
- });
- let storage = Arc::new(SystemStorage::new(
- config.clone(),
- Arc::new(PersisterKind::FileWithSync(FileWithSyncPersister {})),
- ));
- let stream_id = 1;
- let id = 2;
- let name = "test";
- let compression_algorithm = CompressionAlgorithm::None;
- let size_of_parent_stream = Arc::new(AtomicU64::new(0));
- let messages_count_of_parent_stream = Arc::new(AtomicU64::new(0));
- let segments_count_of_parent_stream = Arc::new(AtomicU32::new(0));
-
- let topic = Topic::create(
- stream_id,
- id,
- name,
- partitions_count,
- config,
- storage,
- size_of_parent_stream,
- messages_count_of_parent_stream,
- segments_count_of_parent_stream,
- IggyExpiry::NeverExpire,
- compression_algorithm,
- MaxTopicSize::ServerDefault,
- 1,
- )
- .await
- .unwrap();
- topic.persist().await.unwrap();
- topic
- }
- */
+ // #[tokio::test]
+ // async fn given_partition_id_key_messages_should_be_appended_only_to_the_chosen_partition() {
+ // let partition_id = 1;
+ // let partitioning = Partitioning::partition_id(partition_id);
+ // let partitions_count = 3;
+ // let messages_count: u32 = 1000;
+ // let topic = init_topic(partitions_count).await;
+
+ // for entity_id in 1..=messages_count {
+ // let message = IggyMessage::builder()
+ // .id(entity_id as u128)
+ // .payload(Bytes::new())
+ // .build();
+ // let messages = IggyMessagesMut::from_messages(&[message], 1);
+ // topic
+ // .append_messages(&partitioning, messages, None)
+ // .await
+ // .unwrap();
+ // }
+
+ // let partitions = topic.get_partitions();
+ // assert_eq!(partitions.len(), partitions_count as usize);
+ // for partition in partitions {
+ // let partition = partition.read().await;
+ // let messages = partition.cache.as_ref().unwrap().to_vec();
+ // if partition.partition_id == partition_id {
+ // assert_eq!(messages.len() as u32, messages_count);
+ // } else {
+ // assert_eq!(messages.len() as u32, 0);
+ // }
+ // }
+ // }
+
+ // #[tokio::test]
+ // async fn given_messages_key_key_messages_should_be_appended_to_the_calculated_partitions() {
+ // let partitions_count = 3;
+ // let messages_count = 1000;
+ // let topic = init_topic(partitions_count).await;
+
+ // for entity_id in 1..=messages_count {
+ // let partitioning = Partitioning::messages_key_u32(entity_id);
+ // let message = IggyMessage::builder()
+ // .id(entity_id as u128)
+ // .payload(Bytes::new())
+ // .build();
+ // let messages = IggyMessagesMut::from_messages(&[message], 1);
+ // topic
+ // .append_messages(&partitioning, messages, None)
+ // .await
+ // .unwrap();
+ // }
+
+ // let mut read_messages_count = 0;
+ // let partitions = topic.get_partitions();
+ // assert_eq!(partitions.len(), partitions_count as usize);
+ // for partition in partitions {
+ // let partition = partition.read().await;
+ // let messages = partition.cache.as_ref().unwrap().to_vec();
+ // read_messages_count += messages.len();
+ // assert!(messages.len() < messages_count as usize);
+ // }
+
+ // assert_eq!(read_messages_count, messages_count as usize);
+ // }
+
+ // #[tokio::test]
+ // async fn given_multiple_partitions_calculate_next_partition_id_should_return_next_partition_id_using_round_robin(
+ // ) {
+ // let partitions_count = 3;
+ // let messages_count = 1000;
+ // let topic = init_topic(partitions_count).await;
+
+ // let mut expected_partition_id = 0;
+ // for _ in 1..=messages_count {
+ // let partition_id = topic.get_next_partition_id();
+ // expected_partition_id += 1;
+ // if expected_partition_id > partitions_count {
+ // expected_partition_id = 1;
+ // }
+
+ // assert_eq!(partition_id, expected_partition_id);
+ // }
+ // }
+
+ // #[tokio::test]
+ // async fn given_multiple_partitions_calculate_partition_id_by_hash_should_return_next_partition_id(
+ // ) {
+ // let partitions_count = 3;
+ // let messages_count = 1000;
+ // let topic = init_topic(partitions_count).await;
+
+ // for entity_id in 1..=messages_count {
+ // let key = Partitioning::messages_key_u32(entity_id);
+ // let partition_id = topic.calculate_partition_id_by_messages_key_hash(&key.value);
+ // let entity_id_hash = hash::calculate_32(&key.value);
+ // let mut expected_partition_id = entity_id_hash % partitions_count;
+ // if expected_partition_id == 0 {
+ // expected_partition_id = partitions_count;
+ // }
+
+ // assert_eq!(partition_id, expected_partition_id);
+ // }
+ // }
+
+ // async fn init_topic(partitions_count: u32) -> Topic {
+ // let tempdir = tempfile::TempDir::new().unwrap();
+ // let config = Arc::new(SystemConfig {
+ // path: tempdir.path().to_str().unwrap().to_string(),
+ // ..Default::default()
+ // });
+ // let storage = Arc::new(SystemStorage::new(
+ // config.clone(),
+ // Arc::new(PersisterKind::FileWithSync(FileWithSyncPersister {})),
+ // ));
+ // let stream_id = 1;
+ // let id = 2;
+ // let name = "test";
+ // let compression_algorithm = CompressionAlgorithm::None;
+ // let size_of_parent_stream = Arc::new(AtomicU64::new(0));
+ // let messages_count_of_parent_stream = Arc::new(AtomicU64::new(0));
+ // let segments_count_of_parent_stream = Arc::new(AtomicU32::new(0));
+
+ // let topic = Topic::create(
+ // stream_id,
+ // id,
+ // name,
+ // partitions_count,
+ // config,
+ // storage,
+ // size_of_parent_stream,
+ // messages_count_of_parent_stream,
+ // segments_count_of_parent_stream,
+ // IggyExpiry::NeverExpire,
+ // compression_algorithm,
+ // MaxTopicSize::ServerDefault,
+ // 1,
+ // )
+ // .await
+ // .unwrap();
+ // topic.persist().await.unwrap();
+ // topic
+ // }
}