feat(bench): implement random message and batch size for benchmarking

This commit introduces the ability to specify random message sizes
and batch sizes in the benchmarking tool.

The changes include:

- Added support for specifying message size and batch size as either
  a single value or a range.
- Updated the benchmarking logic to handle random sizes, ensuring
  that the tool can generate messages and batches with varying sizes.
- Refactored the code to improve readability and maintainability,
  including the removal of unused code and the introduction of
  utility functions for common operations.
- Enhanced the rate limiting mechanism to support byte-based limits,
  ensuring that the benchmarking tool can accurately simulate
  different network conditions.

bench/Cargo.toml

@@ -1,20 +1,20 @@
 [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
 # mimalloc on server is also setting it on bench.
 
 [dependencies]
 async-trait = "0.1.86"
-atomic-time = "0.1.5"
-bytes = "1.10.0"
 charming = "0.4.0"
 chrono = "0.4.39"
 clap = { version = "4.5.30", features = ["derive"] }
 figlet-rs = "0.1.5"
+flume = "0.11.1"
 futures = "0.3.31"
+governor = "0.8.0"
 hostname = "0.4.0"
 human-repr = "1.1.0"
 iggy = { path = "../sdk" }

bench/report/Cargo.toml

@@ -1,17 +1,17 @@
 [package]
 name = "iggy-bench-report"
-version = "0.2.1"
+version = "0.2.2"
 edition = "2021"
 description = "Benchmark report and chart generation library for iggy-bench binary and iggy-benchmarks-dashboard web app"
 license = "Apache-2.0"
 
 [dependencies]
-byte-unit = "5.1.6"
 charming = "0.4.0"
 colored = "3.0.0"
 derive-new = "0.7.0"
 derive_more = { version = "2.0.1", features = ["full"] }
 human-repr = "1.1.0"
+rand = "0.9.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 sysinfo = "0.33.1"

bench/report/src/plotting/text/subtext.rs

@@ -1,13 +1,12 @@
 use crate::{
     benchmark_kind::BenchmarkKind, group_metrics::BenchmarkGroupMetrics,
-    group_metrics_kind::GroupMetricsKind, params::BenchmarkParams, report::BenchmarkReport,
+    group_metrics_kind::GroupMetricsKind, report::BenchmarkReport,
 };
-use byte_unit::{Byte, UnitType};
 use human_repr::HumanCount;
 
 impl BenchmarkReport {
     pub fn subtext(&self) -> String {
-        let params_text = self.params.format_params();
+        let params_text = self.format_params();
         let mut stats = Vec::new();
 
         // First add latency stats
@@ -61,6 +60,52 @@ impl BenchmarkReport {
             .collect::<Vec<String>>()
             .join("\n")
     }
+
+    pub fn format_params(&self) -> String {
+        let actors_info = self.params.format_actors_info();
+        let message_batches = self.params.message_batches;
+        let messages_per_batch = self.params.messages_per_batch;
+        let message_size = self.params.message_size;
+
+        let sent_msgs = self.total_messages_sent();
+        let polled_msgs = self.total_messages_received();
+        let total_bytes = self.total_bytes();
+
+        let mut user_data_print = String::new();
+
+        if sent_msgs > 0 && polled_msgs > 0 {
+            user_data_print.push_str(&format!(
+                "Sent {} Messages, Polled {} Messages, {} in Total",
+                sent_msgs.human_count_bare(),
+                polled_msgs.human_count_bare(),
+                total_bytes.human_count_bytes(),
+            ));
+        } else if polled_msgs > 0 {
+            user_data_print.push_str(&format!(
+                "Polled {} Messages, {} in Total",
+                polled_msgs.human_count_bare(),
+                total_bytes.human_count_bytes(),
+            ));
+        } else {
+            user_data_print.push_str(&format!(
+                "Sent {} Messages, {} in Total",
+                sent_msgs.human_count_bare(),
+                total_bytes.human_count_bytes(),
+            ));
+        }
+
+        let topics = "1 Topic per Stream".to_owned();
+        let partitions = if self.params.partitions == 0 {
+            "".to_owned()
+        } else {
+            format!("  •  {} Partitions per Topic", self.params.partitions)
+        };
+        let streams = format!("{} Streams", self.params.streams);
+
+        format!(
+            "{actors_info}  •  {streams}  •  {topics}{partitions}  •  {messages_per_batch} Msg/batch  •  {message_batches:?} Batches  •  {message_size} bytes/msg  •  {user_data_print}",
+        )
+    }
 }
 
 impl BenchmarkGroupMetrics {
@@ -89,50 +134,3 @@ impl BenchmarkGroupMetrics {
         )
     }
 }
-
-impl BenchmarkParams {
-    pub fn format_params(&self) -> String {
-        let actors_info = self.format_actors_info();
-        let message_batches = self.message_batches as u64;
-        let messages_per_batch = self.messages_per_batch as u64;
-        let message_size = self.message_size as u64;
-
-        let sent = message_batches * messages_per_batch * message_size * self.producers as u64;
-        let polled = message_batches * messages_per_batch * message_size * self.consumers as u64;
-
-        let mut user_data_print = String::new();
-
-        if sent > 0 && polled > 0 {
-            user_data_print.push_str(&format!(
-                "Sent {:.2}, Polled {:.2}",
-                Byte::from_u64(sent).get_appropriate_unit(UnitType::Decimal),
-                Byte::from_u64(polled).get_appropriate_unit(UnitType::Decimal)
-            ));
-        } else if polled > 0 {
-            user_data_print.push_str(&format!(
-                "Polled {:.2}",
-                Byte::from_u64(polled).get_appropriate_unit(UnitType::Decimal)
-            ));
-        } else {
-            user_data_print.push_str(&format!(
-                "Sent {:.2}",
-                Byte::from_u64(sent).get_appropriate_unit(UnitType::Decimal),
-            ));
-        }
-
-        let message_batches = message_batches.human_count_bare();
-        let messages_per_batch = messages_per_batch.human_count_bare();
-
-        let topics = "1 topic per stream".to_owned();
-        let partitions = if self.partitions == 0 {
-            "".to_owned()
-        } else {
-            format!("  •  {} partitions per topic", self.partitions)
-        };
-        let streams = format!("{} streams", self.streams);
-
-        format!(
-            "{actors_info}  •  {streams}  •  {topics}{partitions}  •  {messages_per_batch} msg/batch  •  {message_batches} batches  •  {message_size} bytes/msg  •  {user_data_print}",
-        )
-    }
-}