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/15 21:54:52 UTC
(iggy) branch join_set_bench updated (fd55c0c3 -> 0f390b0e)
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a change to branch join_set_bench
in repository https://gitbox.apache.org/repos/asf/iggy.git
discard fd55c0c3 Cleanup
discard 89a3bdfa use join_set instead of select for bench actors
new 0f390b0e Improve benchmark, but parallelizing actors
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (fd55c0c3)
\
N -- N -- N refs/heads/join_set_bench (0f390b0e)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 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:
(iggy) 01/01: Improve benchmark, but parallelizing actors
Posted by gk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gkoszyk pushed a commit to branch join_set_bench
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 0f390b0e5f85e78378b6d6ce3b602802fab3c7cc
Author: numinex <g....@gmail.com>
AuthorDate: Sat Mar 15 22:54:42 2025 +0100
Improve benchmark, but parallelizing actors
---
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 ++++-------
12 files changed, 92 insertions(+), 71 deletions(-)
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());