Add cross-rate early exit for multi-rate benchmark profiles

When running multiple rates (constant, poisson, concurrent profiles) or
sweeping, stop escalating to higher rates if a failure constraint
(over-saturation, max errors, error rate) triggers at a lower rate.

- Sort rates/streams ascending in AsyncProfile and ConcurrentProfile
- Add _should_stop_escalating() on base Profile class using stop_all
  as the failure signal (vs stop_local for normal completions)
- Skip failure check after throughput phase in SweepProfile since
  over-saturation is expected at maximum load
- Log warning when rate order is changed by sorting
- Update CLI help and README with multi-rate documentation
- Add comprehensive unit tests for all profile types

Signed-off-by: Uri Shaket <ushaket@redhat.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Signed-off-by: Uri Shaket <ushaket@redhat.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: ushaket

README.md

@@ -180,7 +180,7 @@ guidellm run \
 **Key parameters:**
 
 - `--profile kind=<type>`: Defines the traffic pattern — `synchronous`, `concurrent`, `throughput`, `constant`, `poisson`, or `sweep`
-- `--profile kind=constant,rate=10`: For `constant`/`poisson`, set requests per second in the profile config; for `concurrent`, use `streams=`; for `throughput`, use `max_concurrency=`
+- `--profile kind=constant,rate=10`: For `constant`/`poisson`, set requests per second in the profile config; for `concurrent`, use `streams=`; for `throughput`, use `max_concurrency=`. For `constant`, `poisson`, and `concurrent`, multiple rates can be specified as a JSON list (e.g., `rate=[1,5,10]`). Rates are sorted ascending, and if a failure constraint (over-saturation, errors) triggers at a given rate, remaining higher rates are skipped.
 - `--constraint kind=max_duration,seconds=<seconds>` or `--constraint kind=max_requests,count=<count>`: Limit each strategy by time or request count
 
 ### Dataset Sources

src/guidellm/benchmark/profiles/asynchronous.py

@@ -9,6 +9,7 @@
 from pydantic import Field, PositiveFloat, PositiveInt, field_validator
 
 from guidellm.benchmark.schemas import ProfileArgs
+from guidellm.logger import logger
 from guidellm.scheduler import (
     AsyncConstantStrategy,
     AsyncPoissonStrategy,
@@ -56,7 +57,12 @@ def _coerce_rate_to_list(
         if not value:
             raise ValueError("rate requires at least one value")
         if isinstance(value, list | tuple):
-            return value
+            sorted_rates = sorted(value)
+            if list(sorted_rates) != list(value):
+                logger.warning(
+                    f"Rates reordered from {list(value)} to {sorted_rates} (ascending)"
+                )
+            return sorted_rates
         if isinstance(value, int | float):
             return [value]
         raise ValueError(
@@ -107,17 +113,24 @@ def next_strategy(
         """
         Generate async strategy for next configured rate.
 
-        :param prev_strategy: Previously completed strategy (unused)
-        :param prev_benchmark: Benchmark results from previous execution (unused)
+        Rates are sorted ascending, so if a previous rate was terminated by a
+        failure constraint (over-saturation, errors, etc.), all remaining higher
+        rates are skipped.
+
+        :param prev_strategy: Previously completed strategy
+        :param prev_benchmark: Benchmark results from previous execution
         :return: AsyncConstantStrategy or AsyncPoissonStrategy for next rate,
-            or None if all rates completed
+            or None if all rates completed or failure detected
         :raises ValueError: If strategy_type is neither 'constant' nor 'poisson'
         """
-        _ = (prev_strategy, prev_benchmark)  # unused
+        _ = prev_strategy
 
         if len(self.completed_strategies) >= len(self.args.rate):
             return None
 
+        if prev_benchmark is not None and self._should_stop_escalating(prev_benchmark):
+            return None
+
         current_rate = self.args.rate[len(self.completed_strategies)]
 
         if self._strategy_type == "constant":

src/guidellm/benchmark/profiles/concurrent.py

@@ -9,6 +9,7 @@
 from pydantic import Field, PositiveInt, field_validator
 
 from guidellm.benchmark.schemas import ProfileArgs
+from guidellm.logger import logger
 from guidellm.scheduler import (
     ConcurrentStrategy,
     ConstraintInitializer,
@@ -48,7 +49,13 @@ def _coerce_streams_to_list(cls, value: Any) -> Any:
         if not value:
             raise ValueError("streams requires at least one value")
         if isinstance(value, list | tuple):
-            return [int(stream) for stream in value]
+            streams = [int(stream) for stream in value]
+            sorted_streams = sorted(streams)
+            if sorted_streams != streams:
+                logger.warning(
+                    f"Streams reordered from {streams} to {sorted_streams} (ascending)"
+                )
+            return sorted_streams
         if isinstance(value, int | float):
             return [int(value)]
         raise ValueError(
@@ -93,15 +100,23 @@ def next_strategy(
         """
         Generate concurrent strategy for next stream count.
 
-        :param prev_strategy: Previously completed strategy (unused)
-        :param prev_benchmark: Benchmark results from previous execution (unused)
+        Stream counts are sorted ascending, so if a previous stream count was
+        terminated by a failure constraint (over-saturation, errors, etc.), all
+        remaining higher stream counts are skipped.
+
+        :param prev_strategy: Previously completed strategy
+        :param prev_benchmark: Benchmark results from previous execution
         :return: ConcurrentStrategy with next stream count, or None if complete
+            or failure detected
         """
-        _ = (prev_strategy, prev_benchmark)  # unused
+        _ = prev_strategy
 
         if len(self.completed_strategies) >= len(self.args.streams):
             return None
 
+        if prev_benchmark is not None and self._should_stop_escalating(prev_benchmark):
+            return None
+
         return ConcurrentStrategy(
             streams=self.args.streams[len(self.completed_strategies)],
             rampup_duration=self.args.rampup_duration,

src/guidellm/benchmark/profiles/profile.py

@@ -9,6 +9,7 @@
 from typing import Any
 
 from guidellm.benchmark.schemas import Benchmark, ProfileArgs
+from guidellm.logger import logger
 from guidellm.scheduler import (
     Constraint,
     ConstraintInitializer,
@@ -115,6 +116,33 @@ def strategy_types(self) -> list[str]:
         """
         return [strat.type_ for strat in self.completed_strategies]
 
+    @staticmethod
+    def _should_stop_escalating(prev_benchmark: Benchmark) -> bool:
+        """
+        Check if a benchmark was terminated by a failure constraint.
+
+        Inspects the scheduler state's end_queuing_constraints for any constraint
+        that used "stop_all" for request processing, which indicates the system
+        could not handle the load (over-saturation, excessive errors, etc.).
+        Constraints that use "stop_local" (max duration, max requests) are normal
+        completions and do not trigger escalation stops.
+
+        :param prev_benchmark: Benchmark instance with a scheduler_state attribute
+        :return: True if a failure constraint was triggered, False otherwise
+        """
+        scheduler_state = getattr(prev_benchmark, "scheduler_state", None)
+        if scheduler_state is None:
+            return False
+
+        for name, action in scheduler_state.end_queuing_constraints.items():
+            if action.request_processing == "stop_all":
+                logger.info(
+                    f"Stopping rate escalation: constraint '{name}' "
+                    f"triggered (request_processing=stop_all)"
+                )
+                return True
+        return False
+
     def strategies_generator(
         self,
     ) -> Generator[

src/guidellm/benchmark/profiles/sweep.py

@@ -97,7 +97,9 @@ def next_strategy(
         Generate next strategy in adaptive sweep sequence.
 
         Executes synchronous and throughput strategies first to measure baseline
-        rates, then generates interpolated rates for async strategies.
+        rates, then generates interpolated rates for async strategies. If a
+        failure constraint is triggered during the async phase, all remaining
+        higher rates are skipped.
 
         :param prev_strategy: Previously completed strategy instance
         :param prev_benchmark: Benchmark results from previous strategy execution
@@ -129,6 +131,18 @@ def next_strategy(
                     self.args.sweep_size - 1,
                 )
             )[1:]  # don't rerun synchronous
+            # After throughput, fall through to async rate logic below.
+            # Don't check escalation since throughput is designed to push
+            # beyond sustainable load (over-saturation is expected).
+
+        # Stop escalation if a failure constraint was triggered.
+        # The throughput guard above skips this via the != "throughput" check.
+        # Synchronous never reaches here (returns ThroughputStrategy above).
+        if (
+            prev_strategy.type_ != "throughput"
+            and self._should_stop_escalating(prev_benchmark)  # type: ignore[arg-type]
+        ):
+            return None
 
         next_index = (
             len(self.completed_strategies) - 1 - 1