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 19:42:28 UTC
(iggy) branch join_set_bench created (now 89a3bdfa)
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
at 89a3bdfa use join_set instead of select for bench actors
This branch includes the following new commits:
new 89a3bdfa use join_set instead of select for bench actors
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.
(iggy) 01/01: use join_set instead of select for bench 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 89a3bdfaabe3eb0ae0185e3a60ddae46f6a66d72
Author: numinex <g....@gmail.com>
AuthorDate: Sat Mar 15 20:42:08 2025 +0100
use join_set instead of select for bench 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 | 3 ++-
bench/src/benchmarks/consumer_benchmark.rs | 12 +++++++-----
bench/src/benchmarks/consumer_group_benchmark.rs | 12 ++++++------
bench/src/benchmarks/producer_and_consumer_benchmark.rs | 16 ++++++++--------
.../benchmarks/producer_and_consumer_group_benchmark.rs | 16 +++++++---------
bench/src/benchmarks/producer_benchmark.rs | 13 ++++++++-----
bench/src/benchmarks/producing_consumer_benchmark.rs | 12 +++++++-----
.../src/benchmarks/producing_consumer_group_benchmark.rs | 11 ++++++-----
bench/src/runner.rs | 14 ++++++--------
12 files changed, 60 insertions(+), 55 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..f6d0cfbb 100644
--- a/bench/src/benchmarks/benchmark.rs
+++ b/bench/src/benchmarks/benchmark.rs
@@ -29,6 +29,7 @@ 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 tokio::task::JoinSet;
use std::{pin::Pin, sync::Arc};
use tracing::info;
@@ -81,7 +82,7 @@ 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..caf10edc 100644
--- a/bench/src/benchmarks/consumer_benchmark.rs
+++ b/bench/src/benchmarks/consumer_benchmark.rs
@@ -21,9 +21,12 @@ use crate::args::common::IggyBenchArgs;
use crate::benchmarks::benchmark::{BenchmarkFutures, 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 tokio::task::JoinSet;
use std::sync::atomic::AtomicI64;
use std::sync::Arc;
use tracing::info;
@@ -44,14 +47,14 @@ 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();
@@ -95,11 +98,10 @@ impl Benchmarkable for ConsumerBenchmark {
.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..4e526a7f 100644
--- a/bench/src/benchmarks/consumer_group_benchmark.rs
+++ b/bench/src/benchmarks/consumer_group_benchmark.rs
@@ -28,8 +28,9 @@ 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 tokio::task::JoinSet;
use std::sync::{atomic::AtomicI64, Arc};
use tracing::{error, info};
@@ -87,7 +88,7 @@ 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 +100,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 +129,13 @@ 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..5ae5b357 100644
--- a/bench/src/benchmarks/producer_and_consumer_benchmark.rs
+++ b/bench/src/benchmarks/producer_and_consumer_benchmark.rs
@@ -22,9 +22,12 @@ use crate::args::common::IggyBenchArgs;
use crate::benchmarks::benchmark::{BenchmarkFutures, 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 tokio::task::JoinSet;
use std::sync::atomic::AtomicI64;
use std::sync::Arc;
use tracing::info;
@@ -45,7 +48,7 @@ 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 +59,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 +79,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 +104,13 @@ 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..aa508308 100644
--- a/bench/src/benchmarks/producer_and_consumer_group_benchmark.rs
+++ b/bench/src/benchmarks/producer_and_consumer_group_benchmark.rs
@@ -28,8 +28,9 @@ 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 tokio::task::JoinSet;
use std::sync::{atomic::AtomicI64, Arc};
use tracing::{error, info};
@@ -87,7 +88,7 @@ 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 +117,8 @@ 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 +140,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 +167,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..2964fa25 100644
--- a/bench/src/benchmarks/producer_benchmark.rs
+++ b/bench/src/benchmarks/producer_benchmark.rs
@@ -21,8 +21,11 @@ use crate::args::common::IggyBenchArgs;
use crate::benchmarks::benchmark::{BenchmarkFutures, 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 tokio::task::JoinSet;
use std::sync::Arc;
use tracing::info;
@@ -42,7 +45,7 @@ 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 +56,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 +79,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..1e73d913 100644
--- a/bench/src/benchmarks/producing_consumer_benchmark.rs
+++ b/bench/src/benchmarks/producing_consumer_benchmark.rs
@@ -21,9 +21,12 @@ use crate::args::common::IggyBenchArgs;
use crate::benchmarks::benchmark::{BenchmarkFutures, 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 tokio::task::JoinSet;
use std::sync::atomic::AtomicI64;
use std::sync::Arc;
use tracing::info;
@@ -44,7 +47,7 @@ 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 +57,7 @@ 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 +86,10 @@ 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..beab3092 100644
--- a/bench/src/benchmarks/producing_consumer_group_benchmark.rs
+++ b/bench/src/benchmarks/producing_consumer_group_benchmark.rs
@@ -27,7 +27,9 @@ 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 tokio::task::JoinSet;
use std::sync::atomic::AtomicI64;
use std::sync::Arc;
use tracing::{error, info};
@@ -86,7 +88,7 @@ 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 +107,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 +142,10 @@ 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..97858bed 100644
--- a/bench/src/runner.rs
+++ b/bench/src/runner.rs
@@ -26,6 +26,7 @@ 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::individual_metrics;
use iggy_bench_report::params::BenchmarkParams;
use integration::test_server::TestServer;
use std::path::Path;
@@ -60,18 +61,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),
- Err(e) => return Err(e),
+ 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());