feat(server): add deferred PollMessages wait timeout

core/bench/src/actors/consumer/client/interface.rs

@@ -27,6 +27,7 @@ pub struct BenchmarkConsumerConfig {
     pub stream_id: String,
     pub messages_per_batch: BenchmarkNumericParameter,
     pub warmup_time: IggyDuration,
+    pub poll_wait_timeout: IggyDuration,
     pub polling_kind: PollingKind,
     pub origin_timestamp_latency_calculation: bool,
     pub pretty: bool,

core/bench/src/actors/consumer/client/low_level.rs

@@ -64,14 +64,15 @@ impl ConsumerClient for LowLevelConsumerClient {
 
         let before_poll = Instant::now();
         let polled = client
-            .poll_messages(
+            .poll_messages_with_timeout(
                 &self.stream_id,
                 &self.topic_id,
                 self.partition_id,
                 consumer,
                 &self.polling_strategy,
                 messages_to_receive,
                 self.auto_commit,
+                self.config.poll_wait_timeout.get_duration(),
             )
             .await;
 

core/bench/src/actors/consumer/typed_benchmark_consumer.rs

@@ -51,6 +51,7 @@ impl TypedBenchmarkConsumer {
         messages_per_batch: BenchmarkNumericParameter,
         finish_condition: Arc<BenchmarkFinishCondition>,
         warmup_time: IggyDuration,
+        poll_wait_timeout: IggyDuration,
         sampling_time: IggyDuration,
         moving_average_window: u32,
         polling_kind: PollingKind,
@@ -64,6 +65,7 @@ impl TypedBenchmarkConsumer {
             stream_id,
             messages_per_batch,
             warmup_time,
+            poll_wait_timeout,
             polling_kind,
             origin_timestamp_latency_calculation,
             pretty,

core/bench/src/actors/producing_consumer/typed_benchmark_producing_consumer.rs

@@ -58,6 +58,7 @@ impl TypedBenchmarkProducingConsumer {
         send_finish_condition: Arc<BenchmarkFinishCondition>,
         poll_finish_condition: Arc<BenchmarkFinishCondition>,
         warmup_time: IggyDuration,
+        poll_wait_timeout: IggyDuration,
         sampling_time: IggyDuration,
         moving_average_window: u32,
         limit_bytes_per_second: Option<IggyByteSize>,
@@ -81,6 +82,7 @@ impl TypedBenchmarkProducingConsumer {
             stream_id,
             messages_per_batch,
             warmup_time,
+            poll_wait_timeout,
             polling_kind,
             origin_timestamp_latency_calculation,
             pretty,

core/bench/src/args/common.rs

@@ -21,7 +21,8 @@ use super::props::{BenchmarkKindProps, BenchmarkTransportProps};
 use super::{
     defaults::{
         DEFAULT_MESSAGE_BATCHES, DEFAULT_MESSAGE_SIZE, DEFAULT_MESSAGES_PER_BATCH,
-        DEFAULT_MOVING_AVERAGE_WINDOW, DEFAULT_SAMPLING_TIME, DEFAULT_WARMUP_TIME,
+        DEFAULT_MOVING_AVERAGE_WINDOW, DEFAULT_POLL_WAIT_TIMEOUT, DEFAULT_SAMPLING_TIME,
+        DEFAULT_WARMUP_TIME,
     },
     transport::BenchmarkTransportCommand,
 };
@@ -71,6 +72,10 @@ pub struct IggyBenchArgs {
     #[arg(long, short = 'w', default_value_t = IggyDuration::from_str(DEFAULT_WARMUP_TIME).unwrap())]
     pub warmup_time: IggyDuration,
 
+    /// Poll wait timeout, e.g. "10ms", "1s". Use "0s" to disable deferred polling.
+    #[arg(long, default_value_t = IggyDuration::from_str(DEFAULT_POLL_WAIT_TIMEOUT).unwrap(), value_parser = IggyDuration::from_str)]
+    pub poll_wait_timeout: IggyDuration,
+
     /// Sampling time for metrics collection. It is also used as bucket size for time series calculations.
     #[arg(long, short = 't', default_value_t = IggyDuration::from_str(DEFAULT_SAMPLING_TIME).unwrap(), value_parser = IggyDuration::from_str)]
     pub sampling_time: IggyDuration,
@@ -163,6 +168,38 @@ impl IggyBenchArgs {
                 .exit();
         }
 
+        if self.high_level_api && !self.poll_wait_timeout.is_zero() {
+            Self::command()
+                .error(
+                    ErrorKind::ArgumentConflict,
+                    "--poll-wait-timeout is only supported by low-level benchmark consumers",
+                )
+                .exit();
+        }
+
+        if matches!(
+            self.kind(),
+            BenchmarkKind::PinnedProducer | BenchmarkKind::BalancedProducer
+        ) && !self.poll_wait_timeout.is_zero()
+        {
+            Self::command()
+                .error(
+                    ErrorKind::ArgumentConflict,
+                    "--poll-wait-timeout requires a benchmark with consumers",
+                )
+                .exit();
+        }
+
+        if matches!(self.transport(), TransportProtocol::Http) && !self.poll_wait_timeout.is_zero()
+        {
+            Self::command()
+                .error(
+                    ErrorKind::ArgumentConflict,
+                    "--poll-wait-timeout is not supported by HTTP transport",
+                )
+                .exit();
+        }
+
         self.benchmark_kind.inner().validate();
     }
 
@@ -218,6 +255,10 @@ impl IggyBenchArgs {
         self.warmup_time
     }
 
+    pub const fn poll_wait_timeout(&self) -> IggyDuration {
+        self.poll_wait_timeout
+    }
+
     pub const fn sampling_time(&self) -> IggyDuration {
         self.sampling_time
     }
@@ -389,6 +430,10 @@ impl IggyBenchArgs {
             transport.to_string(),
         ];
 
+        if !self.poll_wait_timeout().is_zero() {
+            parts.push(format!("poll_wait_{}", self.poll_wait_timeout()));
+        }
+
         if let Some(remark) = &self.remark() {
             parts.push(remark.clone());
         }
@@ -447,3 +492,30 @@ impl IggyBenchArgs {
         name
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn poll_wait_timeout_defaults_to_zero() {
+        let args = IggyBenchArgs::try_parse_from(["iggy-bench", "pinned-consumer", "tcp"])
+            .expect("args should parse");
+
+        assert!(args.poll_wait_timeout().is_zero());
+    }
+
+    #[test]
+    fn poll_wait_timeout_arg_parses() {
+        let args = IggyBenchArgs::try_parse_from([
+            "iggy-bench",
+            "--poll-wait-timeout",
+            "10ms",
+            "pinned-consumer",
+            "tcp",
+        ])
+        .expect("args should parse");
+
+        assert_eq!(args.poll_wait_timeout().as_micros(), 10_000);
+    }
+}

core/bench/src/args/defaults.rs

@@ -46,6 +46,7 @@ pub const DEFAULT_NUMBER_OF_CONSUMER_GROUPS: NonZeroU32 = u32!(1);
 pub const DEFAULT_NUMBER_OF_PRODUCERS: NonZeroU32 = u32!(8);
 
 pub const DEFAULT_WARMUP_TIME: &str = "0s";
+pub const DEFAULT_POLL_WAIT_TIMEOUT: &str = "0s";
 
 pub const DEFAULT_SAMPLING_TIME: &str = "10ms";
 pub const DEFAULT_MOVING_AVERAGE_WINDOW: u32 = 20;

core/bench/src/benchmarks/benchmark.rs

@@ -193,7 +193,12 @@ pub trait Benchmarkable: Send {
             .rate_limit()
             .map(|rl| format!(" global rate limit: {rl}/s"))
             .unwrap_or_default();
+        let poll_wait_timeout = if self.args().poll_wait_timeout().is_zero() {
+            String::new()
+        } else {
+            format!(" poll wait timeout: {},", self.args().poll_wait_timeout())
+        };
 
-        format!("{message_size}{messages_per_batch}{data}{rate_limit}")
+        format!("{message_size}{messages_per_batch}{data}{poll_wait_timeout}{rate_limit}")
     }
 }

core/bench/src/benchmarks/common.rs

@@ -165,6 +165,7 @@ pub fn build_consumer_futures(
     let consumers = args.consumers();
     let actors = args.producers() + args.consumers();
     let warmup_time = args.warmup_time();
+    let poll_wait_timeout = args.poll_wait_timeout();
     let messages_per_batch = args.messages_per_batch();
     let sampling_time = args.sampling_time();
     let moving_average_window = args.moving_average_window();
@@ -233,6 +234,7 @@ pub fn build_consumer_futures(
                     messages_per_batch,
                     finish_condition,
                     warmup_time,
+                    poll_wait_timeout,
                     sampling_time,
                     moving_average_window,
                     polling_kind,
@@ -255,6 +257,7 @@ pub fn build_producing_consumers_futures(
     let streams = args.streams();
     let partitions = args.number_of_partitions();
     let warmup_time = args.warmup_time();
+    let poll_wait_timeout = args.poll_wait_timeout();
     let messages_per_batch = args.messages_per_batch();
     let message_size = args.message_size();
     let polling_kind = PollingKind::Offset;
@@ -296,6 +299,7 @@ pub fn build_producing_consumers_futures(
                     send_finish_condition.clone(),
                     poll_finish_condition.clone(),
                     warmup_time,
+                    poll_wait_timeout,
                     args_clone.sampling_time(),
                     args_clone.moving_average_window(),
                     rate_limit,
@@ -320,6 +324,7 @@ pub fn build_producing_consumer_groups_futures(
     let partitions = args.number_of_partitions();
     let cg_count = args.number_of_consumer_groups();
     let warmup_time = args.warmup_time();
+    let poll_wait_timeout = args.poll_wait_timeout();
     let messages_per_batch = args.messages_per_batch();
     let message_size = args.message_size();
     let start_consumer_group_id = CONSUMER_GROUP_BASE_ID;
@@ -399,6 +404,7 @@ pub fn build_producing_consumer_groups_futures(
                     send_finish_condition,
                     poll_finish_condition,
                     warmup_time,
+                    poll_wait_timeout,
                     args_clone.sampling_time(),
                     args_clone.moving_average_window(),
                     rate_limit,

core/bench/src/utils/mod.rs

@@ -21,7 +21,7 @@ use bench_report::{
     transport::BenchmarkTransport,
 };
 use iggy::prelude::*;
-use std::{fs, path::Path};
+use std::{fmt::Write, fs, path::Path};
 use tracing::{error, info};
 
 use crate::args::{
@@ -31,8 +31,8 @@ use crate::args::{
         DEFAULT_HTTP_SERVER_ADDRESS, DEFAULT_MESSAGE_BATCHES, DEFAULT_MESSAGE_SIZE,
         DEFAULT_MESSAGES_PER_BATCH, DEFAULT_NUMBER_OF_CONSUMER_GROUPS, DEFAULT_NUMBER_OF_CONSUMERS,
         DEFAULT_NUMBER_OF_PRODUCERS, DEFAULT_PINNED_NUMBER_OF_PARTITIONS,
-        DEFAULT_PINNED_NUMBER_OF_STREAMS, DEFAULT_QUIC_SERVER_ADDRESS, DEFAULT_TCP_SERVER_ADDRESS,
-        DEFAULT_TOTAL_MESSAGES_SIZE, DEFAULT_WARMUP_TIME,
+        DEFAULT_PINNED_NUMBER_OF_STREAMS, DEFAULT_POLL_WAIT_TIMEOUT, DEFAULT_QUIC_SERVER_ADDRESS,
+        DEFAULT_TCP_SERVER_ADDRESS, DEFAULT_TOTAL_MESSAGES_SIZE, DEFAULT_WARMUP_TIME,
     },
 };
 
@@ -140,7 +140,10 @@ pub fn params_from_args_and_metrics(
         consumer_groups.to_string(),
     ];
 
-    let params_identifier = params_identifier.join("_");
+    let mut params_identifier = params_identifier.join("_");
+    if !args.poll_wait_timeout().is_zero() {
+        let _ = write!(params_identifier, "_poll_wait_{}", args.poll_wait_timeout());
+    }
 
     BenchmarkParams {
         benchmark_kind,
@@ -227,6 +230,13 @@ fn add_basic_arguments(parts: &mut Vec<String>, args: &IggyBenchArgs) {
     if args.warmup_time().to_string() != DEFAULT_WARMUP_TIME {
         parts.push(format!("--warmup-time \'{}\'", args.warmup_time()));
     }
+
+    if args.poll_wait_timeout().to_string() != DEFAULT_POLL_WAIT_TIMEOUT {
+        parts.push(format!(
+            "--poll-wait-timeout \'{}\'",
+            args.poll_wait_timeout()
+        ));
+    }
 }
 
 fn add_benchmark_kind_arguments(parts: &mut Vec<String>, args: &IggyBenchArgs) {