diff --git a/benches/README.md b/benches/README.md index dbb368eb..2cec41ac 100644 --- a/benches/README.md +++ b/benches/README.md @@ -208,6 +208,17 @@ slice's backends disagree the charts name them ("corpus differs by backend: an SQS row changes — a rate is a rate — but a shorter window is a noisier estimate, which is why the size is on the chart and not just in the script. +The FIFO cell deviates too, and separately, because the drain deviation never +reaches it: `consume_fifo` holds no barrier and takes no drain, so it publishes +the tier's 5 000 messages per shard and consumes them through the sequenced +path. LocalStack serves that path at a few messages per second, which makes +each of the three FIFO cells (one per payload) a ten-hour cell. SQS runs +**100 per FIFO worker** instead (one worker per shard on this backend), set +as `SQS_FIFO_MESSAGES` in `scripts/bench.sh` and passed as `--fifo-messages`, +which replaces the tier's per-consumer count in the same unit; every other +backend runs the tier's count. The row records the corpus it ran in +`messages`. + Kafka is the reference backend for the batched-consume flow. Its batch and parallel scenarios declare one partition per consumer so every group member gets work. Its `publish_single` row is bounded by librdkafka's default 5 ms diff --git a/examples/common/stress_test.rs b/examples/common/stress_test.rs index 08d1dc39..905ce376 100644 --- a/examples/common/stress_test.rs +++ b/examples/common/stress_test.rs @@ -271,6 +271,18 @@ pub struct Cli { #[arg(long, value_parser = clap::value_parser!(u64).range(1..))] pub drain_messages: Option, + /// Corpus per FIFO worker for the `consume-fifo` cell, in place of the + /// tier's per-consumer count and in the same unit: the cell's corpus is + /// this value times the workers the backend reports for FIFO (one per + /// shard on most backends, one in total on Kafka), exactly as the tier's + /// count is. FIFO holds no barrier and takes no drain, so this is the + /// only knob that sizes it; every other flow keeps the tier's count. It + /// exists for a backend whose FIFO path is slow enough that the tier's + /// corpus stops being a cell and becomes a day (SQS on LocalStack). The + /// row records the corpus it ran in `messages`. + #[arg(long, value_parser = clap::value_parser!(u64).range(1..))] + pub fifo_messages: Option, + /// Cap on a drain corpus in bytes (`corpus × payload_bytes`), default /// 2 GiB. A cell whose cap admits no message at its payload is refused /// rather than clamped. Only meaningful with `--drain-messages`; refused @@ -1317,6 +1329,10 @@ fn build_scenarios( // width instead of holding it constant. let tier_messages = match flow { Flow::Broadcast => per_consumer, + Flow::ConsumeFifo => cli + .fifo_messages + .unwrap_or(per_consumer) + .saturating_mul(consumers as u64), _ => per_consumer.saturating_mul(consumers as u64), }; // Only the flow with a batch to size carries the knobs; @@ -7272,6 +7288,85 @@ mod tests { assert_eq!(scenarios[0].messages, 5_000); } + #[test] + fn the_fifo_corpus_override_is_per_worker_like_the_tier() { + // Kafka's wrapper reports one FIFO worker, and the tier already sizes + // its corpus for that one worker (see the test above). The override + // replaces the tier's per-consumer count in the same unit, so it + // must not silently multiply by the shard count instead. + let scenarios = build_scenarios( + &cli_args(&[ + "--tier", + "moderate", + "--handler", + "zero", + "--flow", + "consume-fifo", + "--consumers", + "8", + "--fifo-messages", + "100", + ]), + Flow::ConsumerGroup, + 1, + None, + ); + assert_eq!(scenarios.len(), 1); + assert_eq!(scenarios[0].consumers, 1); + assert_eq!(scenarios[0].messages, 100); + } + + #[test] + fn the_fifo_corpus_can_be_deviated_per_worker() { + // SQS on LocalStack drains a FIFO shard at a few messages per second, + // so the tier's 5,000 per shard is a ten-hour cell three times over. + // The knob sizes the FIFO corpus per worker (one per shard here) and + // leaves every other flow on the tier's count, the same shape as the + // drain deviation. + let fifo = build_scenarios_cg(&cli_args(&[ + "--tier", + "moderate", + "--handler", + "zero", + "--flow", + "consume-fifo", + "--consumers", + "8", + "--fifo-messages", + "100", + ])); + assert_eq!(fifo.len(), 1); + assert_eq!(fifo[0].messages, 100 * SEQ_SHARDS as u64); + + // Compared against the same build without the knob rather than a + // literal: the parallel corpus is the tier's count raised to the + // framework floor, and this test is about the knob, not the floor. + let parallel_args = [ + "--tier", + "moderate", + "--handler", + "zero", + "--flow", + "consume-parallel", + "--consumers", + "8", + ]; + let baseline: Vec = build_scenarios_cg(&cli_args(¶llel_args)) + .iter() + .map(|s| s.messages) + .collect(); + let mut with_knob = parallel_args.to_vec(); + with_knob.extend(["--fifo-messages", "100"]); + let deviated: Vec = build_scenarios_cg(&cli_args(&with_knob)) + .iter() + .map(|s| s.messages) + .collect(); + assert_eq!( + deviated, baseline, + "--fifo-messages must not resize any other flow" + ); + } + #[test] fn fifo_scenarios_never_ask_for_concurrent_processing() { // FIFO concurrency comes from shards, not from concurrent dispatch diff --git a/scripts/bench.sh b/scripts/bench.sh index 8cabb8e7..0036af62 100755 --- a/scripts/bench.sh +++ b/scripts/bench.sh @@ -68,6 +68,15 @@ MATRIX=( # This is a per-backend corpus, not a second matrix: every other knob is # shared, so the flows and the axes stay comparable. SQS_DRAIN_MESSAGES=60000 +# The FIFO cell has no drain and no barrier, so the corpus deviation above +# never reaches it: it publishes the tier's 5 000 messages per shard and +# consumes them through the sequenced path, which LocalStack serves at a few +# messages per second. That is a ten-hour cell three times over (one per +# payload) for a number that measures LocalStack. 100 per FIFO worker (one +# per shard on this backend), the tier's own unit, keeps the cell a few +# minutes long and, as with the drain, the row records the corpus it ran +# (`messages`). +SQS_FIFO_MESSAGES=100 usage() { sed -n '2,7p' "$0" | sed 's/^# \{0,1\}//' @@ -133,6 +142,9 @@ if [ "$target" = sqs ]; then done [ "$deviated" = 1 ] \ || die "the matrix has no --drain-messages for the sqs corpus deviation to replace" + # Appended rather than substituted: the matrix carries no --fifo-messages, + # because every other backend runs the tier's FIFO corpus. + MATRIX+=(--fifo-messages "$SQS_FIFO_MESSAGES") fi if [ "$fresh" = 1 ] && [ -f "$RESULTS_FILE" ]; then @@ -147,7 +159,7 @@ log="$LOG_DIR/$target-$(date -u +%Y%m%dT%H%M%SZ).log" echo "backend: $target ($example, --features $features)" echo "matrix: ${MATRIX[*]} ${extra[*]:-}" if [ "$target" = sqs ]; then - echo "deviation: drain corpus $SQS_DRAIN_MESSAGES, not the pinned 6000000 — see the comment in this script" + echo "deviation: drain corpus $SQS_DRAIN_MESSAGES, not the pinned 6000000, and FIFO corpus $SQS_FIFO_MESSAGES per FIFO worker — see the comments in this script" fi echo "results: $RESULTS_FILE" echo "log: $log"