Feat/cross rate early exit

docs/getting-started/benchmark.md

@@ -178,6 +178,8 @@ guidellm run --profile kind=poisson,rate=16 --seed kind=static,value=42
 | `rate`            | Target rate(s) in requests per second   | `--profile kind=poisson,rate=10` or `--profile '{"kind":"poisson","rate":[10,20]}'` |
 | `max_concurrency` | Maximum concurrent requests to schedule | `--profile kind=poisson,rate=10,max_concurrency=32`                                 |
 
+When multiple rates are specified and a constraint with `stopping_scope="all"` is configured (e.g., over-saturation or error rate limits), triggering that constraint will skip all remaining rates in the list. For best results, specify rates in ascending order so that only higher (harder) rates are skipped. With non-ascending order, easier rates that follow a failed rate will also be skipped.
+
 Use `--seed kind=static,value=42` for reproducible Poisson scheduling.
 
 You can use the `--override` option to specify a list of rate values, to run a set of Poisson "strategies" (sub-benchmarks) with different rate values. For example, `--profile kind=poisson --override profile.rate 10,20,30` will run a Poisson strategy with 10 requests per second, a Poisson strategy with 20 requests per second, and a Poisson strategy with 30 requests per second.
@@ -188,7 +190,7 @@ The sweep profile applies a sequence of benchmark strategies to find the optimal
 
 1. It runs a `synchronous` strategy to measure the baseline rate,
 2. then runs a `throughput` strategy to determine peak throughput,
-3. and finally runs a series of asynchronous strategies with rates interpolated between the baseline and maximum throughput. (The number of interpolated strategies is `sweep_size` minus 2.) The asynchronous strategy type is determined by the `strategy_type` profile parameter. The default strategy type is `constant`.
+3. and finally runs a series of asynchronous strategies with rates interpolated between the baseline and maximum throughput. (The number of interpolated strategies is `sweep_size` minus 2.) The asynchronous strategy type is determined by the `strategy_type` profile parameter. The default strategy type is `constant`. During the async phase, if a constraint with `stopping_scope="all"` triggers at a given rate, remaining rates are skipped.
 
 For example, to run a sweep with 10 strategies, 10 seconds of rampup, and a strategy type of `poisson`:
 

src/guidellm/benchmark/profiles/asynchronous.py

@@ -107,17 +107,23 @@ 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)
+        If a previous rate was terminated by a constraint with
+        stopping_scope='all', remaining 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 escalation halted
         :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

@@ -93,15 +93,22 @@ 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)
+        If a previous stream count was terminated by a constraint with
+        stopping_scope='all', remaining 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 escalation halted
         """
-        _ = (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,32 @@ 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 constraint with stopping_scope="all".
+
+        Inspects the scheduler state's end_queuing_constraints for any constraint
+        whose stopping_scope is "all", indicating the system could not handle the
+        load and escalation to subsequent rates/streams should halt.
+
+        :param prev_benchmark: Benchmark instance
+        :return: True if escalation should stop, 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.stopping_scope == "all":
+                logger.debug(
+                    "Stopping rate escalation: constraint '{}' "
+                    "triggered (stopping_scope=all)",
+                    name,
+                )
+                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
@@ -130,6 +132,15 @@ def next_strategy(
                 )
             )[1:]  # don't rerun synchronous
 
+        # Stop escalation if a constraint with stopping_scope='all' triggered
+        # during the async phase. Throughput is excluded because it intentionally
+        # pushes beyond sustainable load. Synchronous never reaches here.
+        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
         )  # subtract synchronous and throughput

src/guidellm/scheduler/constraints/args.py

@@ -4,7 +4,7 @@
 
 from __future__ import annotations
 
-from typing import Annotated, ClassVar
+from typing import Annotated, ClassVar, Literal
 
 from annotated_types import Gt, Lt
 from pydantic import BeforeValidator, ConfigDict, Field
@@ -74,6 +74,15 @@ def __pydantic_schema_base_type__(cls) -> type[ConstraintArgs]:
         description="Constraint type discriminator for polymorphic serialization",
     )
 
+    stopping_scope: Literal["current", "all"] = Field(
+        default="current",
+        description=(
+            "Whether triggering this constraint should stop only the current "
+            "benchmark ('current') or also halt escalation to subsequent "
+            "rates/streams ('all')."
+        ),
+    )
+
     @property
     def constraint_key(self) -> str:
         """

src/guidellm/scheduler/constraints/error.py

@@ -159,6 +159,7 @@ def __call__(
         return SchedulerUpdateAction(
             request_queuing="stop" if errors_exceeded else "continue",
             request_processing="stop_all" if errors_exceeded else "continue",
+            stopping_scope=self.args.stopping_scope,
             metadata={
                 "max_errors": max_errors,
                 "errors_exceeded": errors_exceeded,
@@ -238,6 +239,7 @@ def __call__(
         return SchedulerUpdateAction(
             request_queuing="stop" if exceeded else "continue",
             request_processing="stop_all" if exceeded else "continue",
+            stopping_scope=self.args.stopping_scope,
             metadata={
                 "max_error_rate": max_error_rate,
                 "window_size": self.args.window,
@@ -316,6 +318,7 @@ def __call__(
         return SchedulerUpdateAction(
             request_queuing="stop" if exceeded else "continue",
             request_processing="stop_all" if exceeded else "continue",
+            stopping_scope=self.args.stopping_scope,
             metadata={
                 "max_error_rate": max_error_rate,
                 "min_processed": self.args.minimum,

src/guidellm/scheduler/constraints/request.py

@@ -134,6 +134,7 @@ def __call__(
         return SchedulerUpdateAction(
             request_queuing="stop" if create_exceeded else "continue",
             request_processing="stop_local" if processed_exceeded else "continue",
+            stopping_scope=self.args.stopping_scope,
             metadata={
                 "max_requests": max_num,
                 "create_exceeded": create_exceeded,
@@ -206,6 +207,7 @@ def __call__(
         return SchedulerUpdateAction(
             request_queuing="stop" if duration_exceeded else "continue",
             request_processing="stop_local" if duration_exceeded else "continue",
+            stopping_scope=self.args.stopping_scope,
             metadata={
                 "max_duration": max_duration,
                 "elapsed_time": elapsed,

src/guidellm/scheduler/constraints/saturation.py

@@ -362,6 +362,7 @@ def __init__(
         confidence: float = 0.95,
         eps: float = 1e-12,
         mode: Literal["enforce", "monitor"] = "enforce",
+        stopping_scope: Literal["current", "all"] = "current",
     ) -> None:  # noqa: PLR0913
         """
         Initialize the over-saturation constraint.
@@ -390,6 +391,9 @@ def __init__(
             (default: 1e-12)
         :param mode: Whether to stop when over-saturation is detected, or only monitor
             (default: "enforce")
+        :param stopping_scope: Whether to halt only the current benchmark
+            or also prevent escalation to subsequent rates/streams.
+            (default: "current")
         """
         self.minimum_duration = minimum_duration
         self.minimum_ttft = minimum_ttft
@@ -400,6 +404,7 @@ def __init__(
         self.confidence = confidence
         self.eps = eps
         self.mode = mode
+        self.stopping_scope = stopping_scope
         self.reset()
 
     @property
@@ -617,6 +622,7 @@ def __call__(
         return SchedulerUpdateAction(
             request_queuing="stop" if should_stop else "continue",
             request_processing="stop_all" if should_stop else "continue",
+            stopping_scope=self.stopping_scope,
             metadata={
                 "ttft_slope": ttft_slope,
                 "ttft_slope_moe": ttft_slope_moe,
@@ -670,4 +676,5 @@ def create_constraint(self, **_kwargs) -> Constraint:
             minimum_window_size=self.args.minimum_window_size,
             confidence=self.args.confidence,
             mode=self.args.mode,
+            stopping_scope=self.args.stopping_scope,
         )

src/guidellm/scheduler/schemas.py

@@ -288,6 +288,14 @@ class SchedulerUpdateAction(StandardBaseModel):
         default="continue",
         description="Action to take for request processing operations",
     )
+    stopping_scope: Literal["current", "all"] = Field(
+        default="current",
+        description=(
+            "Whether this constraint's stop signal should halt only the current "
+            "benchmark ('current') or also prevent escalation to subsequent "
+            "rates/streams ('all')."
+        ),
+    )
     metadata: dict[str, Any] = Field(
         default_factory=dict,
         description="Additional context and data for the scheduler action",