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/03/16 17:15:28 UTC
(iggy) branch master updated: feat(bench): improve benchmark by parallelizing actors (#1645)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 7fa78ce1 feat(bench): improve benchmark by parallelizing actors (#1645)
7fa78ce1 is described below
commit 7fa78ce1a86a76f5dd94fcaa42fbf50852fe8e29
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 04c37d4b..f6e29946 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 5be85509..37c97a89 100644
--- a/bench/Cargo.toml
+++ b/bench/Cargo.toml
@@ -17,7 +17,7 @@
[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
@@ -31,7 +31,6 @@ charming = "0.4.0"
chrono = "0.4.40"
clap = { version = "4.5.32", 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 ea4a68e5..f05e63d1 100644
--- a/bench/src/actors/consumer.rs
+++ b/bench/src/actors/consumer.rs
@@ -92,7 +92,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 3b2ba5f4..83d727c7 100644
--- a/bench/src/actors/producer.rs
+++ b/bench/src/actors/producer.rs
@@ -88,7 +88,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 6000a5a1..e906ec23 100644
--- a/bench/src/actors/producing_consumer.rs
+++ b/bench/src/actors/producing_consumer.rs
@@ -101,7 +101,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 9a704739..faa9af87 100644
--- a/bench/src/benchmarks/benchmark.rs
+++ b/bench/src/benchmarks/benchmark.rs
@@ -19,7 +19,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;
@@ -29,7 +28,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;
@@ -40,11 +40,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);
@@ -81,7 +76,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 f6d38488..b7531171 100644
--- a/bench/src/benchmarks/consumer_benchmark.rs
+++ b/bench/src/benchmarks/consumer_benchmark.rs
@@ -18,14 +18,17 @@
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::poll_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 {
@@ -44,14 +47,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();
@@ -94,12 +99,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 68b8c6f9..6e55f4a1 100644
--- a/bench/src/benchmarks/consumer_group_benchmark.rs
+++ b/bench/src/benchmarks/consumer_group_benchmark.rs
@@ -16,7 +16,7 @@
* under the License.
*/
-use super::benchmark::{BenchmarkFutures, Benchmarkable};
+use super::benchmark::Benchmarkable;
use crate::{
actors::consumer::Consumer,
args::common::IggyBenchArgs,
@@ -28,9 +28,12 @@ use iggy::{
client::ConsumerGroupClient, clients::client::IggyClient, error::IggyError,
messages::poll_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 {
@@ -87,7 +90,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)
@@ -99,11 +104,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);
@@ -128,14 +133,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 c0a0ef80..68782de0 100644
--- a/bench/src/benchmarks/producer_and_consumer_benchmark.rs
+++ b/bench/src/benchmarks/producer_and_consumer_benchmark.rs
@@ -19,14 +19,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::poll_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 {
@@ -45,7 +48,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();
@@ -56,8 +61,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(
@@ -77,8 +81,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;
@@ -103,14 +106,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 c3f958e9..a7b6cbfa 100644
--- a/bench/src/benchmarks/producer_and_consumer_group_benchmark.rs
+++ b/bench/src/benchmarks/producer_and_consumer_group_benchmark.rs
@@ -16,7 +16,7 @@
* under the License.
*/
-use super::benchmark::{BenchmarkFutures, Benchmarkable};
+use super::benchmark::Benchmarkable;
use crate::{
actors::{consumer::Consumer, producer::Producer},
args::common::IggyBenchArgs,
@@ -28,9 +28,12 @@ use iggy::{
client::ConsumerGroupClient, clients::client::IggyClient, error::IggyError,
messages::poll_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 ProducerAndConsumerGroupBenchmark {
@@ -87,7 +90,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)
@@ -116,9 +121,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);
@@ -140,8 +143,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);
@@ -168,15 +170,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 a87b7e58..4fde7cf2 100644
--- a/bench/src/benchmarks/producer_benchmark.rs
+++ b/bench/src/benchmarks/producer_benchmark.rs
@@ -18,12 +18,15 @@
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 {
@@ -42,7 +45,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);
@@ -53,7 +58,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();
@@ -76,11 +81,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 1f56918a..2cdf2bc0 100644
--- a/bench/src/benchmarks/producing_consumer_benchmark.rs
+++ b/bench/src/benchmarks/producing_consumer_benchmark.rs
@@ -18,14 +18,17 @@
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::poll_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 {
@@ -44,7 +47,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);
@@ -54,7 +59,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();
@@ -83,11 +89,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 42487832..bae9d5b1 100644
--- a/bench/src/benchmarks/producing_consumer_group_benchmark.rs
+++ b/bench/src/benchmarks/producing_consumer_group_benchmark.rs
@@ -18,7 +18,7 @@
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;
@@ -27,9 +27,11 @@ use iggy::clients::client::IggyClient;
use iggy::error::IggyError;
use iggy::messages::poll_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 {
@@ -86,7 +88,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)
@@ -105,11 +109,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();
@@ -140,11 +144,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 f52f4eb8..2548902d 100644
--- a/bench/src/runner.rs
+++ b/bench/src/runner.rs
@@ -23,7 +23,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;
@@ -60,18 +59,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());