From 6d6f42aaec066c9d4c044f3d2004b28b2ecf2e83 Mon Sep 17 00:00:00 2001 From: bofengluo Date: Wed, 26 Aug 2026 17:46:03 -0700 Subject: [PATCH 1/2] fix: fail fast on endpoint response stalls Add a no-progress deadline for in-flight endpoint requests. An accepted request that never produces a response chunk or final result would otherwise block the benchmark until its outer wall-time limit, because the phase drain waits on responses that never arrive. Disabled by default (`settings.timeouts.no_progress_timeout_s`, `null`). Rebased onto main and moved the setting into the Timeouts model added by #409, as requested in review. Review feedback addressed: - [P1] Do not retain every completed UUID. `completed_uuids` now only holds synthetic completions from `register_skipped`, so it no longer grows O(total requests) alongside `uuid_to_index`. - [P2] Keep the disabled path inert. Activity tracking is gated on the feature being enabled, and the receiver reuses the arrival timestamp it already takes rather than reading the clock a second time. Measured cost when disabled: +3.5 ns per response, +9.4 ns per request. - [P2] Replace the per-transition watchdog task with a single self-rearming `loop.call_later` TimerHandle. No task, no Event, no per-iteration `wait_for`. Under the production eager task factory this cuts arm+cancel from 2205 ns to 535 ns per request (4.1x). - [P2] Compute the remaining deadline from the last progress stamp instead of sleeping a full interval each iteration, which could delay detection by up to 2x the configured timeout. - Drop `cohort` from the docs and from the local variable in `issue()`. - Keep a single copy of the tuning guidance: the schema description is the short form, docs/config/DESIGN.md holds the detail. Also fixed while here: a receiver transport error could overwrite an earlier NoProgressError, masking the real diagnosis when a stalled endpoint also drops its connection. First error now wins. Validation: - `pytest tests/unit/config/test_schema.py tests/unit/commands/test_benchmark.py tests/unit/load_generator/test_async_session.py` (425 passed) - `pre-commit run --all-files` (mypy reports 3 pre-existing Darwin-only `os.sched_*affinity` errors in files this change does not touch) - `python scripts/regenerate_templates.py --check` - AGA disaggregated held-response canary, job 578946: failed as intended with `Endpoint made no response progress for 10.0s with 1 request(s) in flight`, 10.002 s after the phase started against a 10 s deadline. The 10 s value is fault-injection coverage, not the deployment recommendation. Not adopted: the suggestion to use one phase/session-lifetime watchdog driven by an activity event. The guard is still armed on the 0-to-1 in-flight transition and retired on drain, now with a TimerHandle rather than a task. This keeps the timer heap clear once a phase drains and measured 4.1x cheaper than the reviewed version; happy to revisit if a resident watchdog is preferred. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 14 + docs/CLI_DESIGN.md | 2 + docs/CLI_QUICK_REFERENCE.md | 1 + docs/config/DESIGN.md | 8 +- docs/load_generator/DESIGN.md | 7 +- .../commands/benchmark/execute.py | 1 + src/inference_endpoint/config/schema.py | 14 + .../templates/concurrency_template_full.yaml | 1 + .../templates/offline_template_full.yaml | 1 + .../templates/online_template_full.yaml | 1 + .../load_generator/session.py | 175 +++++++- tests/unit/commands/test_benchmark.py | 26 ++ tests/unit/config/test_schema.py | 15 + .../unit/load_generator/test_async_session.py | 420 +++++++++++++++++- 14 files changed, 652 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index 6dc13ad61..0979df68a 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,20 @@ Dataset Manager ──> Load Generator ──> Endpoint Client ──> External - **Online** (`poisson`): Fixed QPS with Poisson arrival distribution for latency profiling - **Concurrency**: Fixed concurrent request count +### Endpoint liveness + +An in-flight request can stall without ever returning a response. Set a no-progress deadline to fail the run instead of waiting out the overall wall-time limit: + +```yaml +settings: + timeouts: + no_progress_timeout_s: 30 +``` + +Or `--no-progress-timeout 30` for `benchmark offline` / `benchmark online`. + +The deadline is armed only while requests are in flight, and every streamed chunk or final response resets it, so long streaming generations are not failed. It is a client-side guard: it does not replace the engine's own request timeout. See [docs/config/DESIGN.md](docs/config/DESIGN.md#no-progress-deadline) for how to pick a value. + ### Performance Design The hot path is optimized for minimal overhead: diff --git a/docs/CLI_DESIGN.md b/docs/CLI_DESIGN.md index da4799a14..d378fe859 100644 --- a/docs/CLI_DESIGN.md +++ b/docs/CLI_DESIGN.md @@ -61,6 +61,8 @@ Both paths produce the **same subclass with the same defaults**. A YAML file wit 3. **Datasets injected after construction.** `--dataset` strings are parsed by a `BeforeValidator` on the `datasets` field, then merged via `config.with_updates(datasets=...)`. +Fields get their dotted path automatically; those needing a shorter operational spelling also declare a `cyclopts.Parameter(alias=...)`. For example `settings.timeouts.no_progress_timeout_s` accepts either `--timeouts.no-progress-timeout-s` or the alias `--no-progress-timeout`. + ### YAML path 1. **`from_yaml_file(path)`** loads YAML, resolves `${VAR}` env vars on parsed values, then passes the dict to a Pydantic `TypeAdapter` with `Discriminator`. diff --git a/docs/CLI_QUICK_REFERENCE.md b/docs/CLI_QUICK_REFERENCE.md index 0cad95822..c74e1d211 100644 --- a/docs/CLI_QUICK_REFERENCE.md +++ b/docs/CLI_QUICK_REFERENCE.md @@ -300,6 +300,7 @@ settings: timeouts: run_timeout_s: null # Whole-run watchdog; null = off performance_drain_timeout_s: null # null = unlimited + no_progress_timeout_s: 30 # Endpoint liveness deadline; null = off load_pattern: type: "max_throughput" target_qps: 10.0 diff --git a/docs/config/DESIGN.md b/docs/config/DESIGN.md index 795efbb66..e1862ed18 100644 --- a/docs/config/DESIGN.md +++ b/docs/config/DESIGN.md @@ -48,7 +48,7 @@ Key nested models: | Model | Purpose | | ---------------- | --------------------------------------------------- | | `LoadPattern` | Pattern type + parameters (target QPS, concurrency) | -| `RuntimeConfig` | Duration, sample count, RNG seeds | +| `RuntimeConfig` | Workload sizing, issue-duration cap, and RNG seeds | | `ClientSettings` | Worker count and HTTP client settings | | `EndpointConfig` | Endpoint URLs, API key | | `Dataset` | Dataset path, type (performance / accuracy) | @@ -71,6 +71,12 @@ Immutable snapshot of all parameters needed to execute a run. Once constructed, `RuntimeSettings` cannot be modified. All consumers receive the same instance. +### No-progress deadline + +`settings.timeouts.no_progress_timeout_s` is a client-side liveness deadline, `None` by default. Once a phase has an in-flight request, `BenchmarkSession` fails the run if no streamed chunk or final response arrives within the interval. It does not apply before the server accepts a request, and it does not replace an engine's own KV-transfer deadline. + +**Choosing a value.** Set it above the deployment's longest expected gap between responses — for a non-streaming endpoint, that is the full request latency. For TensorRT-LLM disaggregated serving, match the executor `hang_detection_timeout` (default 300 s), not `cache_transceiver_config.kv_transfer_timeout_ms`; start at `300` and raise it if the normal p99 response gap is longer. + ### `BenchmarkSuiteRuleset` (abstract base) Extension point for competition-specific constraints. diff --git a/docs/load_generator/DESIGN.md b/docs/load_generator/DESIGN.md index 563fba33c..a5bc7a1c7 100644 --- a/docs/load_generator/DESIGN.md +++ b/docs/load_generator/DESIGN.md @@ -35,6 +35,7 @@ each producing an independent report. BenchmarkSession.run(phases) | +-- STARTED + +-- [no-progress watchdog] armed only if no_progress_timeout_s is set (default null) +-- [warmup] strategy.execute() → drain_after=False (keep in-flight saturated) +-- [perf phase 1] START_PERFORMANCE_TRACKING → strategy.execute() → drain → STOP_PERFORMANCE_TRACKING +-- [warmup] strategy.execute() → drain_after=False (keep in-flight saturated) @@ -173,7 +174,7 @@ class BenchmarkSession: **`run(phases)`** lifecycle: 1. Publish `SessionEventType.STARTED` -2. Start receiver coroutine (`_receive_responses`) +2. Start receiver coroutine (`_receive_responses`). When the no-progress timeout is set, a single liveness timer is armed while work is in flight and fails the session if no streamed chunk or final result arrives within the interval. 3. For each phase: a. Create `SampleOrder` and `LoadStrategy` from phase settings b. Set `self._current_dataset` to phase dataset @@ -296,6 +297,8 @@ async def _receive_responses(self): self._strategy_task.cancel() break self._handle_response(resp) + if self._no_progress_timeout_ns is not None: + self._last_response_progress_ns = time.monotonic_ns() ``` Uses `recv()` exclusively — no `poll()` spin. The ZMQ fd is registered with @@ -303,6 +306,8 @@ the event loop, so `recv()` wakes exactly when a response is available with zero CPU overhead. Each `recv()` call yields to the event loop, ensuring strategy coroutines (call_at callbacks, semaphore waiters) are never starved. +The no-progress guard rides this same path to catch a client-visible worker or transport stall. It is not a per-request engine timeout: any chunk or final response counts as progress, and only a silent in-flight request fails the run. The receiver only stamps a timestamp, and only when the deadline is set — streaming frame rate is QPS x output length, so per-chunk work is paid per token. The armed `call_later` timer re-derives its own deadline from that stamp when it fires, and re-arms for the remainder if progress raced it. + For `ConcurrencyStrategy`, `_handle_response` calls `strategy.on_query_complete()` which releases the semaphore. Since `recv()` returns as soon as the fd is readable and `eager_task_factory` executes the woken semaphore waiter synchronously, there diff --git a/src/inference_endpoint/commands/benchmark/execute.py b/src/inference_endpoint/commands/benchmark/execute.py index a6ad406a5..f251a0773 100644 --- a/src/inference_endpoint/commands/benchmark/execute.py +++ b/src/inference_endpoint/commands/benchmark/execute.py @@ -871,6 +871,7 @@ async def _run_benchmark_async( loop=loop, on_sample_complete=on_sample_complete, session_id=session_id, + no_progress_timeout_s=ctx.config.settings.timeouts.no_progress_timeout_s, ) watchdog.bind_session(session) sigint.bind_session(session, abort_event) diff --git a/src/inference_endpoint/config/schema.py b/src/inference_endpoint/config/schema.py index 4b72244bd..f27f04188 100644 --- a/src/inference_endpoint/config/schema.py +++ b/src/inference_endpoint/config/schema.py @@ -801,6 +801,20 @@ class Timeouts(WithUpdatesMixin, BaseModel): gt=0, description="Whole-run watchdog seconds (None = off).", ) + no_progress_timeout_s: Annotated[ + float | None, + cyclopts.Parameter( + alias="--no-progress-timeout", + ), + ] = Field( + None, + gt=0, + description=( + "Endpoint liveness deadline: fail the run when work is in flight but " + "no response chunk or completion arrives for this many seconds " + "(None = off). See docs/config/DESIGN.md for choosing a value." + ), + ) interrupted_teardown_grace_s: float = Field( 30.0, gt=0, diff --git a/src/inference_endpoint/config/templates/concurrency_template_full.yaml b/src/inference_endpoint/config/templates/concurrency_template_full.yaml index 65e339663..6fff60e66 100644 --- a/src/inference_endpoint/config/templates/concurrency_template_full.yaml +++ b/src/inference_endpoint/config/templates/concurrency_template_full.yaml @@ -86,6 +86,7 @@ settings: worker_gc_mode: relaxed # Worker GC strategy | options: disabled, relaxed, system timeouts: # All global waits and deadlines (see config/schema.py) run_timeout_s: null # Whole-run watchdog seconds (None = off). + no_progress_timeout_s: null # Endpoint liveness deadline: fail the run when work is in flight but no response chunk or completion arrives for this many seconds (None = off). See docs/config/DESIGN.md for choosing a value. interrupted_teardown_grace_s: 30.0 # Abort drain grace before remaining services are killed. service_ready_timeout_s: 30.0 # Service startup wait in seconds. warmup_drain_timeout_s: 240.0 # Warmup drain seconds (None = unlimited). diff --git a/src/inference_endpoint/config/templates/offline_template_full.yaml b/src/inference_endpoint/config/templates/offline_template_full.yaml index b5e69ed73..aba0a3d83 100644 --- a/src/inference_endpoint/config/templates/offline_template_full.yaml +++ b/src/inference_endpoint/config/templates/offline_template_full.yaml @@ -86,6 +86,7 @@ settings: worker_gc_mode: relaxed # Worker GC strategy | options: disabled, relaxed, system timeouts: # All global waits and deadlines (see config/schema.py) run_timeout_s: null # Whole-run watchdog seconds (None = off). + no_progress_timeout_s: null # Endpoint liveness deadline: fail the run when work is in flight but no response chunk or completion arrives for this many seconds (None = off). See docs/config/DESIGN.md for choosing a value. interrupted_teardown_grace_s: 30.0 # Abort drain grace before remaining services are killed. service_ready_timeout_s: 30.0 # Service startup wait in seconds. warmup_drain_timeout_s: 240.0 # Warmup drain seconds (None = unlimited). diff --git a/src/inference_endpoint/config/templates/online_template_full.yaml b/src/inference_endpoint/config/templates/online_template_full.yaml index b723d1e9e..f94f619d5 100644 --- a/src/inference_endpoint/config/templates/online_template_full.yaml +++ b/src/inference_endpoint/config/templates/online_template_full.yaml @@ -87,6 +87,7 @@ settings: worker_gc_mode: relaxed # Worker GC strategy | options: disabled, relaxed, system timeouts: # All global waits and deadlines (see config/schema.py) run_timeout_s: null # Whole-run watchdog seconds (None = off). + no_progress_timeout_s: null # Endpoint liveness deadline: fail the run when work is in flight but no response chunk or completion arrives for this many seconds (None = off). See docs/config/DESIGN.md for choosing a value. interrupted_teardown_grace_s: 30.0 # Abort drain grace before remaining services are killed. service_ready_timeout_s: 30.0 # Service startup wait in seconds. warmup_drain_timeout_s: 240.0 # Warmup drain seconds (None = unlimited). diff --git a/src/inference_endpoint/load_generator/session.py b/src/inference_endpoint/load_generator/session.py index daa909409..71cbde487 100644 --- a/src/inference_endpoint/load_generator/session.py +++ b/src/inference_endpoint/load_generator/session.py @@ -46,6 +46,11 @@ logger = logging.getLogger(__name__) + +class NoProgressError(RuntimeError): + """An endpoint stopped returning response progress while work was in flight.""" + + # --------------------------------------------------------------------------- # Phase configuration # --------------------------------------------------------------------------- @@ -154,6 +159,7 @@ class PhaseIssuer: "_dataset", "_issuer", "_on_inflight_drained", + "_on_inflight_started", "_performance_tracking_stopped", "_prompt_warning_reasons", "_publisher", @@ -163,6 +169,7 @@ class PhaseIssuer: "uuid_to_conv_info", "completed_uuids", "inflight", + "inflight_started_ns", "issued_count", ) @@ -172,6 +179,7 @@ def __init__( issuer: SampleIssuer, publisher: EventPublisher, stop_check: Callable[[], bool], + on_inflight_started: Callable[[], None] | None = None, on_inflight_drained: Callable[[], None] | None = None, routing_headers: tuple[str, ...] = (), ): @@ -179,12 +187,15 @@ def __init__( self._issuer = issuer self._publisher = publisher self._stop_check = stop_check + self._on_inflight_started = on_inflight_started self._on_inflight_drained = on_inflight_drained or (lambda: None) self._routing_headers = routing_headers self.uuid_to_index: dict[str, int] = {} self.uuid_to_conv_info: dict[str, tuple[str, int | None]] = {} self.completed_uuids: set[str] = set() self.inflight: int = 0 + # Set on the 0 -> 1 transition and cleared when the phase drains. + self.inflight_started_ns: int | None = None self.issued_count: int = 0 self._performance_tracking_stopped = False self._prompt_warning_reasons: set[str] = set() @@ -197,8 +208,12 @@ def _warn_prompt_once(self, reason: str, message: str) -> None: logger.warning(message) def mark_inflight_complete(self) -> None: - self.inflight -= 1 if self.inflight <= 0: + logger.warning("Ignoring completion with no in-flight request") + return + self.inflight -= 1 + if self.inflight == 0: + self.inflight_started_ns = None self._on_inflight_drained() def stop_performance_tracking(self) -> None: @@ -312,7 +327,14 @@ def issue( ) ) self._issuer.issue(query) + starting_inflight = self.inflight == 0 + if starting_inflight and self._on_inflight_started is not None: + self.inflight_started_ns = time.monotonic_ns() self.inflight += 1 + # The callback arms the liveness guard, which reads this counter, so it + # must observe the incremented value rather than the prior idle state. + if starting_inflight and self._on_inflight_started is not None: + self._on_inflight_started() self.issued_count += 1 return query_id @@ -362,6 +384,7 @@ def __init__( loop: asyncio.AbstractEventLoop, on_sample_complete: Callable[[QueryResult], None] | None = None, session_id: str | None = None, + no_progress_timeout_s: float | None = None, ): self._issuer = issuer self._publisher = event_publisher @@ -379,6 +402,16 @@ def __init__( self._recv_task: asyncio.Task | None = None self._strategy_task: asyncio.Task | None = None self._drain_event = asyncio.Event() + self._fatal_error: RuntimeError | None = None + self._last_response_progress_ns: int | None = None + # Liveness guard: a single re-arming timer, not a task. It exists only + # while work is in flight, so the disabled and idle paths cost nothing. + self._no_progress_timeout_ns = ( + int(no_progress_timeout_s * 1_000_000_000) + if no_progress_timeout_s is not None + else None + ) + self._progress_timer: asyncio.TimerHandle | None = None def stop(self) -> None: """Signal early termination. Safe to call from signal handler. @@ -389,6 +422,7 @@ def stop(self) -> None: """ self._stop_requested = True self._drain_event.set() + self._disarm_no_progress() if self._strategy_task and not self._strategy_task.done(): self._strategy_task.cancel() @@ -431,6 +465,8 @@ async def run( try: for phase in phases: + if self._fatal_error is not None: + raise self._fatal_error if self._stop_requested: break if on_phase_start is not None: @@ -438,6 +474,8 @@ async def run( result = await self._run_phase(phase) if result is not None: phase_results.append(result) + if self._fatal_error is not None: + raise self._fatal_error finally: self._done = True if self._recv_task and not self._recv_task.done(): @@ -446,6 +484,7 @@ async def run( await self._recv_task except asyncio.CancelledError: pass + self._disarm_no_progress() if self._stop_requested: # Aborted run (Ctrl-C, transport closure, run watchdog): mark # it BEFORE the terminal ENDED so the aggregator's ENDED-driven @@ -468,6 +507,11 @@ async def _run_phase(self, phase: PhaseConfig) -> PhaseResult | None: # Per-phase stop flag is scoped to this phase; clear any cap left set by # a previous phase so it can't short-circuit this one. self._current_phase_stopped = False + # A new phase must not inherit a response timestamp from the previous + # phase: its first in-flight request gets a full liveness interval. + if self._no_progress_timeout_ns is not None: + self._last_response_progress_ns = None + self._disarm_no_progress() # Create per-phase state if phase.strategy is not None: @@ -485,7 +529,12 @@ async def _run_phase(self, phase: PhaseConfig) -> PhaseResult | None: issuer=self._issuer, publisher=self._publisher, stop_check=self._make_stop_check(phase.runtime_settings, phase_start), - on_inflight_drained=self._drain_event.set, + on_inflight_started=( + self._on_inflight_started + if self._no_progress_timeout_ns is not None + else None + ), + on_inflight_drained=self._on_inflight_drained, routing_headers=phase.routing_headers, ) @@ -566,27 +615,111 @@ async def _drain_inflight( async def _receive_responses(self) -> None: """Receive responses from the issuer. Runs as a concurrent task.""" - while not self._done: - resp = await self._issuer.recv() - if resp is None: - # Transport closed unexpectedly — trigger stop so strategy - # and drain don't hang waiting for responses that will never arrive. - logger.warning("Issuer recv() returned None — transport closed") - self._stop_requested = True - self._drain_event.set() # Unblock _drain_inflight - # Cancel the strategy task if it's blocked (e.g., ConcurrencyStrategy - # awaiting sem.acquire() that will never be released). - if self._strategy_task and not self._strategy_task.done(): - self._strategy_task.cancel() - break - self._handle_response(resp) - - def _handle_response(self, resp: QueryResult | StreamChunk) -> None: + try: + while not self._done: + resp = await self._issuer.recv() + if resp is None: + # Transport closed unexpectedly — trigger stop so strategy + # and drain don't hang waiting for responses that will never arrive. + logger.warning("Issuer recv() returned None — transport closed") + self._stop_requested = True + self._drain_event.set() # Unblock _drain_inflight + # Cancel a strategy blocked awaiting a semaphore that will + # never be released. + if self._strategy_task and not self._strategy_task.done(): + self._strategy_task.cancel() + break + # One clock read per response, shared by the event records and + # the liveness stamp. + now_ns = time.monotonic_ns() + self._handle_response(resp, now_ns) + if self._no_progress_timeout_ns is not None: + self._last_response_progress_ns = now_ns + except asyncio.CancelledError: + raise + except Exception as exc: + error = RuntimeError(f"Endpoint response receiver failed: {exc}") + error.__cause__ = exc + # First error wins: a stalled endpoint often drops its connection + # too, and the transport error that follows is a symptom of the + # diagnosis already recorded, not a better one. + if self._fatal_error is None: + self._fatal_error = error + logger.exception("%s", error) + self.stop() + + def _on_inflight_started(self) -> None: + """Arm the liveness guard for a 0-to-1 in-flight transition.""" + if self._no_progress_timeout_ns is None or self._progress_timer is not None: + return + self._progress_timer = self._loop.call_later( + self._no_progress_timeout_ns / 1e9, self._fire_no_progress + ) + + def _on_inflight_drained(self) -> None: + """Unblock drain waits and retire the liveness guard.""" + self._drain_event.set() + self._disarm_no_progress() + + def _disarm_no_progress(self) -> None: + if self._progress_timer is not None: + self._progress_timer.cancel() + self._progress_timer = None + + def _fire_no_progress(self) -> None: + """Fail the run when outstanding work makes no response progress. + + This is transport- and engine-agnostic: a response chunk or final result + is progress. It deliberately does not try to classify TensorRT-LLM or + vLLM errors, so it also catches the silent worker hang where no HTTP + response is ever produced. + """ + self._progress_timer = None + timeout_ns = self._no_progress_timeout_ns + assert timeout_ns is not None + if self._done or self._stop_requested: + return + phase_issuer = self._current_phase_issuer + if ( + phase_issuer is None + or phase_issuer.inflight <= 0 + or phase_issuer.inflight_started_ns is None + ): + return + last_progress_ns = max( + timestamp + for timestamp in ( + self._last_response_progress_ns, + phase_issuer.inflight_started_ns, + ) + if timestamp is not None + ) + idle_ns = time.monotonic_ns() - last_progress_ns + if idle_ns < timeout_ns: + # Progress raced the deadline: re-arm for the remainder rather than + # reporting a stall that did not happen. + self._progress_timer = self._loop.call_later( + (timeout_ns - idle_ns) / 1e9, self._fire_no_progress + ) + return + error = NoProgressError( + "Endpoint made no response progress for " + f"{timeout_ns / 1e9:.1f}s with {phase_issuer.inflight} request(s) in flight" + ) + logger.error("%s", error) + if self._fatal_error is None: + self._fatal_error = error + self.stop() + + def _handle_response(self, resp: QueryResult | StreamChunk, now_ns: int) -> None: """Route a response to the appropriate handler. Transport contract for streaming: the worker sends intermediate StreamChunk messages for timing events, then a final QueryResult with accumulated output for completion. + + `now_ns` is the caller's arrival timestamp, used for every event record + that would otherwise read the clock again. """ phase_issuer = self._current_phase_issuer @@ -622,7 +755,7 @@ def _handle_response(self, resp: QueryResult | StreamChunk) -> None: self._publisher.publish( EventRecord( event_type=ErrorEventType.GENERIC, - timestamp_ns=time.monotonic_ns(), + timestamp_ns=now_ns, sample_uuid=query_id, conversation_id=conv_id_str, turn=turn_num, @@ -637,7 +770,7 @@ def _handle_response(self, resp: QueryResult | StreamChunk) -> None: event_type=SampleEventType.COMPLETE, timestamp_ns=resp.completed_at if isinstance(resp.completed_at, int) - else time.monotonic_ns(), + else now_ns, sample_uuid=query_id, conversation_id=conv_id_str, turn=turn_num, @@ -678,7 +811,7 @@ def _handle_response(self, resp: QueryResult | StreamChunk) -> None: self._publisher.publish( EventRecord( event_type=event_type, - timestamp_ns=time.monotonic_ns(), + timestamp_ns=now_ns, sample_uuid=resp.id, conversation_id=conv_id_str, turn=turn_num, diff --git a/tests/unit/commands/test_benchmark.py b/tests/unit/commands/test_benchmark.py index a5ad0e6a4..13c8a6359 100644 --- a/tests/unit/commands/test_benchmark.py +++ b/tests/unit/commands/test_benchmark.py @@ -678,6 +678,32 @@ def test_use_legacy_loadgen_qps_metrics_default_and_disable(self): lp = bound.arguments["config"].settings.load_pattern assert lp.use_legacy_loadgen_qps_metrics is False + @pytest.mark.unit + @pytest.mark.parametrize( + ("flag", "value"), + [ + ("--no-progress-timeout", "15"), + ("--timeouts.no-progress-timeout-s", "20"), + ], + ) + def test_no_progress_timeout_cli_spellings(self, flag, value): + _, bound, _ = benchmark_app.parse_args( + [ + "offline", + "--endpoints", + "http://h:80", + "--model", + "m", + "--dataset", + "d.jsonl", + flag, + value, + ], + exit_on_error=False, + ) + config = bound.arguments["config"] + assert config.settings.timeouts.no_progress_timeout_s == float(value) + @pytest.mark.unit def test_warmup_salt_flag_default_and_negative(self): """warmup.salt defaults off; --warmup-salt enables it; --no-warmup-salt diff --git a/tests/unit/config/test_schema.py b/tests/unit/config/test_schema.py index 912e859e7..a21a51c1f 100644 --- a/tests/unit/config/test_schema.py +++ b/tests/unit/config/test_schema.py @@ -353,6 +353,21 @@ def test_online_max_throughput_rejected(self): settings={"load_pattern": {"type": "max_throughput"}}, ) + @pytest.mark.unit + def test_no_progress_timeout_requires_positive_value(self): + base = { + "type": TestType.OFFLINE, + "model_params": {"name": "M"}, + "endpoint_config": {"endpoints": ["http://x"]}, + "datasets": [{"path": "D"}], + } + config = BenchmarkConfig( + **base, settings={"timeouts": {"no_progress_timeout_s": 15}} + ) + assert config.settings.timeouts.no_progress_timeout_s == 15 + with pytest.raises(ValueError, match="greater than 0"): + BenchmarkConfig(**base, settings={"timeouts": {"no_progress_timeout_s": 0}}) + @pytest.mark.unit def test_submission_bad_benchmark_mode(self): with pytest.raises(ValueError, match="benchmark_mode"): diff --git a/tests/unit/load_generator/test_async_session.py b/tests/unit/load_generator/test_async_session.py index 6acbb774c..2af0fe088 100644 --- a/tests/unit/load_generator/test_async_session.py +++ b/tests/unit/load_generator/test_async_session.py @@ -19,6 +19,7 @@ import asyncio import random +import time from types import SimpleNamespace import pytest @@ -34,6 +35,7 @@ from inference_endpoint.dataset_manager.dataset import Dataset from inference_endpoint.load_generator.session import ( BenchmarkSession, + NoProgressError, PhaseConfig, PhaseIssuer, PhaseResult, @@ -163,6 +165,42 @@ def test_issue_builds_query_and_publishes(self): assert issued_events[0].conversation_id == "" assert issued_events[0].turn is None + def test_inflight_started_timestamp_tracks_outstanding_work(self): + phase_issuer = PhaseIssuer( + FakeDataset(2), + FakeIssuer(), + FakePublisher(), + lambda: False, + on_inflight_started=lambda: None, + ) + + phase_issuer.issue(0) + started_ns = phase_issuer.inflight_started_ns + assert started_ns is not None + + phase_issuer.issue(1) + assert phase_issuer.inflight_started_ns == started_ns + + phase_issuer.mark_inflight_complete() + assert phase_issuer.inflight_started_ns == started_ns + phase_issuer.mark_inflight_complete() + assert phase_issuer.inflight_started_ns is None + + def test_duplicate_completion_does_not_make_inflight_negative(self): + phase_issuer = PhaseIssuer( + FakeDataset(1), + FakeIssuer(), + FakePublisher(), + lambda: False, + on_inflight_started=lambda: None, + ) + phase_issuer.issue(0) + phase_issuer.mark_inflight_complete() + phase_issuer.mark_inflight_complete() + + assert phase_issuer.inflight == 0 + assert phase_issuer.inflight_started_ns is None + def test_issue_preserves_structured_chat_input_for_isl(self): class ChatDataset(FakeDataset): def load_sample(self, index: int) -> dict: @@ -405,6 +443,352 @@ def test_register_skipped_returns_none_when_stopped(self): @pytest.mark.unit class TestBenchmarkSession: + @pytest.mark.asyncio + async def test_no_progress_watchdog_waits_between_inflight_periods(self): + """The guard arms with in-flight work and disarms when the phase drains.""" + loop = asyncio.get_running_loop() + session = BenchmarkSession( + FakeIssuer(), FakePublisher(), loop, no_progress_timeout_s=60 + ) + phase_issuer = PhaseIssuer( + FakeDataset(1), + FakeIssuer(), + FakePublisher(), + lambda: False, + on_inflight_started=session._on_inflight_started, + on_inflight_drained=session._on_inflight_drained, + ) + session._current_phase_issuer = phase_issuer + + phase_issuer.issue(0) + assert session._progress_timer is not None + + phase_issuer.mark_inflight_complete() + assert session._progress_timer is None + assert session._fatal_error is None + + @pytest.mark.asyncio + async def test_no_progress_watchdog_waits_for_an_active_phase(self): + """No in-flight work means no armed timer, so an idle session never fires.""" + session = BenchmarkSession( + FakeIssuer(), + FakePublisher(), + asyncio.get_running_loop(), + no_progress_timeout_s=0.01, + ) + + assert session._progress_timer is None + await asyncio.sleep(0.05) + assert session._fatal_error is None + + @pytest.mark.asyncio + async def test_perf_cap_keeps_the_guard_watching_the_drain(self): + """The perf issue cap stops issuing, but in-flight work is still guarded. + + `stop_current_phase` deliberately leaves the drain running, so the + liveness guard must stay armed until the phase actually drains — + otherwise a stall during drain would go undetected. + """ + loop = asyncio.get_running_loop() + session = BenchmarkSession( + FakeIssuer(), FakePublisher(), loop, no_progress_timeout_s=60 + ) + phase_issuer = PhaseIssuer( + FakeDataset(1), + FakeIssuer(), + FakePublisher(), + lambda: False, + on_inflight_started=session._on_inflight_started, + on_inflight_drained=session._on_inflight_drained, + ) + phase_issuer.issue(0) + session._current_phase_issuer = phase_issuer + armed = session._progress_timer + assert armed is not None + + # A strategy must be in flight for stop_current_phase to do anything. + async def never() -> None: + await asyncio.sleep(3600) + + session._strategy_task = asyncio.create_task(never()) + session.stop_current_phase() + + assert session._current_phase_stopped is True + assert session._progress_timer is armed # still watching the drain + assert session._fatal_error is None + + phase_issuer.mark_inflight_complete() # drain finishes + assert session._progress_timer is None # now retired + + @pytest.mark.asyncio + async def test_no_progress_watchdog_stops_when_session_stops(self): + """A normal stop must not be reported as an endpoint stall.""" + loop = asyncio.get_running_loop() + session = BenchmarkSession( + FakeIssuer(), FakePublisher(), loop, no_progress_timeout_s=0.01 + ) + phase_issuer = PhaseIssuer( + FakeDataset(1), + FakeIssuer(), + FakePublisher(), + lambda: False, + on_inflight_started=session._on_inflight_started, + ) + phase_issuer.issue(0) + session._current_phase_issuer = phase_issuer + assert session._progress_timer is not None + + session.stop() + + assert session._progress_timer is None + await asyncio.sleep(0.05) + assert session._fatal_error is None + + @pytest.mark.asyncio + async def test_no_progress_watchdog_stops_when_phase_changes(self): + """A phase change leaves the session-lifetime guard idle, not failed.""" + loop = asyncio.get_running_loop() + session = BenchmarkSession( + FakeIssuer(), FakePublisher(), loop, no_progress_timeout_s=0.01 + ) + phase_issuer = PhaseIssuer( + FakeDataset(1), + FakeIssuer(), + FakePublisher(), + lambda: False, + on_inflight_started=session._on_inflight_started, + ) + phase_issuer.issue(0) + session._current_phase_issuer = phase_issuer + assert session._progress_timer is not None + + # Swap in a phase with nothing in flight, then let the armed timer fire. + session._current_phase_issuer = PhaseIssuer( + FakeDataset(1), FakeIssuer(), FakePublisher(), lambda: False + ) + session._fire_no_progress() + + assert session._fatal_error is None + assert session._progress_timer is None + + @pytest.mark.asyncio + async def test_no_progress_watchdog_uses_one_timer(self): + """Re-entering the armed state must not stack a second timer.""" + session = BenchmarkSession( + FakeIssuer(), + FakePublisher(), + asyncio.get_running_loop(), + no_progress_timeout_s=60, + ) + + session._on_inflight_started() + armed = session._progress_timer + assert armed is not None + + session._on_inflight_started() + + assert session._progress_timer is armed + session._disarm_no_progress() + + @pytest.mark.asyncio + async def test_response_activity_does_not_rearm_the_timer(self): + """The receiver stamps a timestamp per chunk; it must not re-arm the timer. + + Streaming frame rate is QPS x output length, so anything the receiver + does per chunk is paid per token. + """ + loop = asyncio.get_running_loop() + issuer = FakeIssuer() + issuer._auto_respond = False + session = BenchmarkSession( + issuer, FakePublisher(), loop, no_progress_timeout_s=60 + ) + phase_issuer = PhaseIssuer( + FakeDataset(1), + issuer, + FakePublisher(), + lambda: False, + on_inflight_started=session._on_inflight_started, + ) + query_id = phase_issuer.issue(0) + session._current_phase_issuer = phase_issuer + armed = session._progress_timer + assert armed is not None + before = session._last_response_progress_ns + + issuer.inject_response(StreamChunk(id=query_id, response_chunk="tok")) + issuer.shutdown() # None ends the receive loop + await session._receive_responses() + + assert session._last_response_progress_ns != before # stamp advanced + assert session._progress_timer is armed # same timer, not re-armed + session._disarm_no_progress() + + @pytest.mark.asyncio + async def test_run_retires_an_active_no_progress_watchdog(self): + """Session teardown leaves no guard task running, even on a long deadline.""" + loop = asyncio.get_running_loop() + issuer = FakeIssuer() + issuer._loop = loop + session = BenchmarkSession( + issuer, FakePublisher(), loop, no_progress_timeout_s=60 + ) + # The 60s deadline never elapses here, so the guard is retired by + # teardown rather than by its own timeout. + phases = [PhaseConfig("perf", _make_settings(n_samples=1), FakeDataset(1))] + + await asyncio.wait_for(session.run(phases), timeout=5.0) + + assert session._progress_timer is None + assert session._fatal_error is None + + @pytest.mark.asyncio + async def test_run_does_not_start_a_phase_after_a_fatal_watchdog_error(self): + """A pending fatal error prevents a later phase from issuing work.""" + session = BenchmarkSession( + FakeIssuer(), FakePublisher(), asyncio.get_running_loop() + ) + session._fatal_error = NoProgressError("endpoint stalled") + started_phases = [] + phase = PhaseConfig("perf", _make_settings(n_samples=1), FakeDataset(1)) + + with pytest.raises(NoProgressError, match="endpoint stalled"): + await session.run([phase], on_phase_start=started_phases.append) + + assert started_phases == [] + + @pytest.mark.asyncio + async def test_no_progress_fails_instead_of_hanging(self): + """A silent server hang must not wait for an unbounded drain.""" + loop = asyncio.get_running_loop() + previous_task_factory = loop.get_task_factory() + loop.set_task_factory(asyncio.eager_task_factory) + issuer = FakeIssuer() + issuer._loop = loop + issuer._auto_respond = False + session = BenchmarkSession( + issuer, FakePublisher(), loop, no_progress_timeout_s=0.02 + ) + phases = [PhaseConfig("perf", _make_settings(n_samples=1), FakeDataset(1))] + + try: + with pytest.raises(NoProgressError, match="no response progress"): + await asyncio.wait_for(session.run(phases), timeout=1.0) + finally: + loop.set_task_factory(previous_task_factory) + + @pytest.mark.asyncio + async def test_no_progress_guard_accepts_streaming_progress(self): + """Chunks refresh the guard; only a silent in-flight request is fatal.""" + loop = asyncio.get_running_loop() + issuer = FakeIssuer() + issuer._loop = loop + issuer._auto_respond = False + session = BenchmarkSession( + issuer, FakePublisher(), loop, no_progress_timeout_s=0.03 + ) + phases = [PhaseConfig("perf", _make_settings(n_samples=1), FakeDataset(1))] + + async def stream_response() -> None: + while not issuer._issued: + await asyncio.sleep(0.002) + query = issuer._issued[0] + for _ in range(3): + await asyncio.sleep(0.015) + issuer.inject_response(StreamChunk(id=query.id, response_chunk="x")) + issuer.inject_response(QueryResult(id=query.id, response_output="done")) + + injector = asyncio.create_task(stream_response()) + result = await asyncio.wait_for(session.run(phases), timeout=1.0) + await injector + assert result.perf_results[0].issued_count == 1 + + @pytest.mark.asyncio + async def test_no_progress_guard_rearms_after_an_idle_gap(self): + """A prior completion must not make the next request immediately stale.""" + + class GapStrategy: + async def execute(self, phase_issuer) -> None: + phase_issuer.issue(0) + await asyncio.sleep(0.04) + phase_issuer.issue(1) + + def on_query_complete(self, query_id: str) -> None: + pass + + loop = asyncio.get_running_loop() + issuer = FakeIssuer(response_delay=0.001) + issuer._loop = loop + session = BenchmarkSession( + issuer, FakePublisher(), loop, no_progress_timeout_s=0.02 + ) + phases = [ + PhaseConfig( + "perf", + _make_settings(n_samples=2), + FakeDataset(2), + strategy=GapStrategy(), + ) + ] + + result = await asyncio.wait_for(session.run(phases), timeout=1.0) + assert result.perf_results[0].issued_count == 2 + + @pytest.mark.asyncio + async def test_receiver_exception_fails_session_instead_of_hanging(self): + class FailingIssuer(FakeIssuer): + async def recv(self) -> QueryResult | StreamChunk | None: + raise OSError("connection reset") + + loop = asyncio.get_running_loop() + session = BenchmarkSession(FailingIssuer(), FakePublisher(), loop) + phases = [PhaseConfig("perf", _make_settings(n_samples=1), FakeDataset(1))] + + with pytest.raises(RuntimeError, match="Endpoint response receiver failed"): + await asyncio.wait_for(session.run(phases), timeout=1.0) + + @pytest.mark.asyncio + async def test_receiver_failure_does_not_mask_an_earlier_stall(self): + """A stalled endpoint that then drops its connection still reports the stall. + + Without first-error-wins, the transport error that follows a watchdog + firing overwrites the diagnosis the watchdog produced. + """ + loop = asyncio.get_running_loop() + session = BenchmarkSession( + FakeIssuer(), FakePublisher(), loop, no_progress_timeout_s=60 + ) + phase_issuer = PhaseIssuer( + FakeDataset(1), + FakeIssuer(), + FakePublisher(), + lambda: False, + on_inflight_started=session._on_inflight_started, + ) + phase_issuer.issue(0) + session._current_phase_issuer = phase_issuer + # The endpoint went silent long enough for the guard to fire. Both + # stamps must age: the deadline is measured from the later of them. + stalled_ns = time.monotonic_ns() - 120 * 1_000_000_000 + session._last_response_progress_ns = stalled_ns + phase_issuer.inflight_started_ns = stalled_ns + session._fire_no_progress() + assert isinstance(session._fatal_error, NoProgressError) + + # The dead connection now surfaces as a transport error in the receiver. + session._stop_requested = False + failing = FakeIssuer() + + async def boom() -> QueryResult | StreamChunk | None: + raise OSError("connection reset") + + failing.recv = boom # type: ignore[method-assign] + session._issuer = failing + await session._receive_responses() + + assert isinstance(session._fatal_error, NoProgressError) + assert "no response progress" in str(session._fatal_error) + @pytest.mark.asyncio async def test_single_perf_phase(self): loop = asyncio.get_running_loop() @@ -884,9 +1268,11 @@ async def test_handle_response_stamps_conversation_id_and_turn(self): # Streaming path: entry stays available for the terminal COMPLETE pop. phase_issuer.uuid_to_conv_info["q-stream"] = ("conv-s", 7) session._handle_response( - StreamChunk(id="q-stream", metadata={"first_chunk": True}) + StreamChunk(id="q-stream", metadata={"first_chunk": True}), 1_000 + ) + session._handle_response( + StreamChunk(id="q-stream", response_chunk="delta"), 1_001 ) - session._handle_response(StreamChunk(id="q-stream", response_chunk="delta")) assert ( publisher.events_of_type(SampleEventType.RECV_FIRST)[0].conversation_id, publisher.events_of_type(SampleEventType.RECV_FIRST)[0].turn, @@ -895,24 +1281,35 @@ async def test_handle_response_stamps_conversation_id_and_turn(self): publisher.events_of_type(SampleEventType.RECV_NON_FIRST)[0].conversation_id, publisher.events_of_type(SampleEventType.RECV_NON_FIRST)[0].turn, ) == ("conv-s", 7) + # Chunk events carry the caller's arrival stamp, not a fresh clock read. + assert ( + publisher.events_of_type(SampleEventType.RECV_FIRST)[0].timestamp_ns + == 1_000 + ) + assert ( + publisher.events_of_type(SampleEventType.RECV_NON_FIRST)[0].timestamp_ns + == 1_001 + ) assert "q-stream" in phase_issuer.uuid_to_conv_info # Success path: COMPLETE inherits conv info, entry is popped. phase_issuer.uuid_to_index["q-ok"] = 0 phase_issuer.uuid_to_conv_info["q-ok"] = ("conv-9", 5) phase_issuer.inflight = 1 - session._handle_response( - QueryResult( - id="q-ok", - response_output="ok", - metadata={"finish_reason": "stop", "worker_id": 3}, - completed_at=12345, - ) + ok_resp = QueryResult( + id="q-ok", + response_output="ok", + metadata={"finish_reason": "stop", "worker_id": 3}, ) + session._handle_response(ok_resp, 1_002) complete = publisher.events_of_type(SampleEventType.COMPLETE) assert [(e.conversation_id, e.turn) for e in complete] == [("conv-9", 5)] assert complete[0].finish_reason == "stop" assert complete[0].worker_id == 3 + # QueryResult stamps completed_at itself (force_setattr in __post_init__), + # and that stamp outranks the caller's arrival timestamp. + assert complete[0].timestamp_ns == ok_resp.completed_at + assert complete[0].timestamp_ns != 1_002 assert "q-ok" not in phase_issuer.uuid_to_conv_info assert "q-ok" not in phase_issuer.completed_uuids @@ -924,7 +1321,8 @@ async def test_handle_response_stamps_conversation_id_and_turn(self): QueryResult( id="q-err", error=ErrorData(error_type="boom", error_message="x"), - ) + ), + 1_003, ) error_events = [ e for e in publisher.events if isinstance(e.event_type, ErrorEventType) @@ -1287,7 +1685,7 @@ async def test_drops_late_response_after_timeout(self): id="q-late", error=ErrorData(error_type="late", error_message="late arrival"), ) - session._handle_response(late_resp) + session._handle_response(late_resp, 1_004) assert publisher.events == [] assert phase_issuer.inflight == 1 From 50368e041282120c3d4741915f5f9accc1a21d24 Mon Sep 17 00:00:00 2001 From: bofengluo Date: Thu, 27 Aug 2026 15:07:32 -0700 Subject: [PATCH 2/2] fix: finalize artifacts on endpoint stall; document liveness semantics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review feedback on the no-progress watchdog: - A NoProgressError no longer propagates out of _run_benchmark_async: the run drains the metrics pipeline, returns a BenchmarkResult carrying stall_error, finalizes the standard artifacts (result_summary.json, report.txt) with the report marked INTERRUPTED, and only then fails with a non-zero exit — matching the run-timeout and drain-timeout conventions. - Document that the deadline clock starts at local issuance (there is no server-acceptance ack) and that under the default client.stream_all_chunks=false only the first chunk and the final result reset it, so the value must exceed the longest single-request latency (>=300 s guidance). Co-Authored-By: Claude Fable 5 --- README.md | 8 +- docs/CLI_DESIGN.md | 2 +- docs/CLI_QUICK_REFERENCE.md | 2 +- docs/config/DESIGN.md | 8 +- docs/load_generator/DESIGN.md | 2 +- .../commands/benchmark/execute.py | 28 +++++ src/inference_endpoint/config/schema.py | 3 +- .../templates/concurrency_template_full.yaml | 2 +- .../templates/offline_template_full.yaml | 2 +- .../templates/online_template_full.yaml | 2 +- tests/unit/commands/test_benchmark.py | 112 +++++++++++++++++- 11 files changed, 151 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index 0979df68a..f02759c09 100644 --- a/README.md +++ b/README.md @@ -92,17 +92,15 @@ Dataset Manager ──> Load Generator ──> Endpoint Client ──> External ### Endpoint liveness -An in-flight request can stall without ever returning a response. Set a no-progress deadline to fail the run instead of waiting out the overall wall-time limit: +Set a no-progress deadline to fail the run when in-flight work stops responding. Use `>=300` seconds; raise it for long requests: ```yaml settings: timeouts: - no_progress_timeout_s: 30 + no_progress_timeout_s: 300 ``` -Or `--no-progress-timeout 30` for `benchmark offline` / `benchmark online`. - -The deadline is armed only while requests are in flight, and every streamed chunk or final response resets it, so long streaming generations are not failed. It is a client-side guard: it does not replace the engine's own request timeout. See [docs/config/DESIGN.md](docs/config/DESIGN.md#no-progress-deadline) for how to pick a value. +Or pass `--no-progress-timeout 300`. The guard is client-side; see [the configuration guide](docs/config/DESIGN.md#no-progress-deadline) for tuning. ### Performance Design diff --git a/docs/CLI_DESIGN.md b/docs/CLI_DESIGN.md index d378fe859..9af256e97 100644 --- a/docs/CLI_DESIGN.md +++ b/docs/CLI_DESIGN.md @@ -61,7 +61,7 @@ Both paths produce the **same subclass with the same defaults**. A YAML file wit 3. **Datasets injected after construction.** `--dataset` strings are parsed by a `BeforeValidator` on the `datasets` field, then merged via `config.with_updates(datasets=...)`. -Fields get their dotted path automatically; those needing a shorter operational spelling also declare a `cyclopts.Parameter(alias=...)`. For example `settings.timeouts.no_progress_timeout_s` accepts either `--timeouts.no-progress-timeout-s` or the alias `--no-progress-timeout`. +Fields get their dotted path automatically; those needing a shorter operational spelling also declare a `cyclopts.Parameter(alias=...)`. For example `settings.timeouts.no_progress_timeout_s` accepts either `--timeouts.no-progress-timeout-s` or the alias `--no-progress-timeout`; use `>=300` seconds when enabling it. ### YAML path diff --git a/docs/CLI_QUICK_REFERENCE.md b/docs/CLI_QUICK_REFERENCE.md index c74e1d211..900161931 100644 --- a/docs/CLI_QUICK_REFERENCE.md +++ b/docs/CLI_QUICK_REFERENCE.md @@ -300,7 +300,7 @@ settings: timeouts: run_timeout_s: null # Whole-run watchdog; null = off performance_drain_timeout_s: null # null = unlimited - no_progress_timeout_s: 30 # Endpoint liveness deadline; null = off + no_progress_timeout_s: 300 # Endpoint liveness deadline; use >=300 s, null = off load_pattern: type: "max_throughput" target_qps: 10.0 diff --git a/docs/config/DESIGN.md b/docs/config/DESIGN.md index e1862ed18..b48befd9c 100644 --- a/docs/config/DESIGN.md +++ b/docs/config/DESIGN.md @@ -73,9 +73,13 @@ Once constructed, `RuntimeSettings` cannot be modified. All consumers receive th ### No-progress deadline -`settings.timeouts.no_progress_timeout_s` is a client-side liveness deadline, `None` by default. Once a phase has an in-flight request, `BenchmarkSession` fails the run if no streamed chunk or final response arrives within the interval. It does not apply before the server accepts a request, and it does not replace an engine's own KV-transfer deadline. +`settings.timeouts.no_progress_timeout_s` is a client-side liveness deadline, `None` by default. Once a phase has an in-flight request, `BenchmarkSession` fails the run if no streamed chunk or final response arrives within the interval. The clock starts at local issuance — when the request is handed to the client transport. There is no server-acceptance acknowledgment in the client contract, so worker scheduling, connection acquisition, and local transport queueing all consume the deadline before the server even sees the request. It does not replace an engine's own KV-transfer deadline. -**Choosing a value.** Set it above the deployment's longest expected gap between responses — for a non-streaming endpoint, that is the full request latency. For TensorRT-LLM disaggregated serving, match the executor `hang_detection_timeout` (default 300 s), not `cache_transceiver_config.kv_transfer_timeout_ms`; start at `300` and raise it if the normal p99 response gap is longer. +**What counts as progress.** Only messages that reach the main process reset the clock. Under the default `client.stream_all_chunks: false`, workers forward a request's first streamed chunk and its final result — intermediate chunks never cross the IPC boundary. The largest gap a healthy streaming request can therefore produce is its full first-chunk-to-completion span, not a per-token gap. With `stream_all_chunks: true` every chunk resets the clock, at the cost of per-chunk IPC traffic. + +**Choosing a value.** Use `>=300` seconds and set it above the deployment's longest expected gap between main-process-visible events. Under the default `stream_all_chunks: false` — and for any non-streaming endpoint — that is the full latency of the longest single request, including pre-accept overhead; use `600` seconds or more for reasoning or long-output workloads. For TensorRT-LLM disaggregated serving, match the executor `hang_detection_timeout` (default 300 s), not `cache_transceiver_config.kv_transfer_timeout_ms`; raise the value if the normal p99 gap between visible events is longer. + +**On failure**, the run still drains the metrics pipeline and writes the standard artifacts (`performance/result_summary.json`, `report.txt`) with the report marked `INTERRUPTED`, then exits non-zero. ### `BenchmarkSuiteRuleset` (abstract base) diff --git a/docs/load_generator/DESIGN.md b/docs/load_generator/DESIGN.md index a5bc7a1c7..ab2ebc519 100644 --- a/docs/load_generator/DESIGN.md +++ b/docs/load_generator/DESIGN.md @@ -174,7 +174,7 @@ class BenchmarkSession: **`run(phases)`** lifecycle: 1. Publish `SessionEventType.STARTED` -2. Start receiver coroutine (`_receive_responses`). When the no-progress timeout is set, a single liveness timer is armed while work is in flight and fails the session if no streamed chunk or final result arrives within the interval. +2. Start receiver coroutine (`_receive_responses`). When the no-progress timeout is set, a single liveness timer is armed while work is in flight and fails the session if no streamed chunk or final result arrives within the interval. Configure `>=300` seconds; see `docs/config/DESIGN.md` for tuning. 3. For each phase: a. Create `SampleOrder` and `LoadStrategy` from phase settings b. Set `self._current_dataset` to phase dataset diff --git a/src/inference_endpoint/commands/benchmark/execute.py b/src/inference_endpoint/commands/benchmark/execute.py index f251a0773..5964be88a 100644 --- a/src/inference_endpoint/commands/benchmark/execute.py +++ b/src/inference_endpoint/commands/benchmark/execute.py @@ -101,6 +101,7 @@ from inference_endpoint.load_generator.conversation_manager import ConversationManager from inference_endpoint.load_generator.session import ( BenchmarkSession, + NoProgressError, PhaseConfig, PhaseType, SessionResult, @@ -170,6 +171,9 @@ class BenchmarkResult: profiling: dict[str, Any] | None = None run_timed_out: bool = False user_interrupted: bool = False + # NoProgressError message when the endpoint stalled; the run still drains + # and finalizes its artifacts, then run_benchmark fails with this message. + stall_error: str | None = None @dataclass @@ -884,6 +888,7 @@ async def _run_benchmark_async( ) _perf_cap_done = False session_completed_normally = False + stall_error: str | None = None def _on_perf_phase_timeout() -> None: if not _perf_cap_done: @@ -930,6 +935,23 @@ def _on_phase_start(phase: PhaseConfig) -> None: start_time_ns=0, end_time_ns=0, ) + elif isinstance(e, NoProgressError): + # Endpoint stall is fatal but the drained data is still + # worth keeping: fall through to the drain below so the + # interrupted report and standard artifacts are written, + # then run_benchmark fails with this message. + logger.error( + "Endpoint stalled — finalizing partial results " + "before failing the run: %s", + e, + ) + stall_error = str(e) + result = SessionResult( + session_id=session_id, + phase_results=[], + start_time_ns=0, + end_time_ns=0, + ) else: raise ExecutionError(f"Benchmark execution failed: {e}") from e finally: @@ -1011,6 +1033,7 @@ def _on_phase_start(phase: PhaseConfig) -> None: profiling=profiler.payload(), run_timed_out=watchdog.fired, user_interrupted=sigint.interrupted, + stall_error=stall_error, ) @@ -1158,6 +1181,7 @@ def finalize_benchmark(ctx: BenchmarkContext, bench: BenchmarkResult) -> None: aborted = ( bench.run_timed_out or bench.user_interrupted + or bench.stall_error is not None or (report is not None and report.state == "interrupted") ) if report is not None and aborted and report.state != "interrupted": @@ -1254,6 +1278,10 @@ def run_benchmark( f"Run timeout ({run_timeout_s}s) reached; run aborted and " "report marked INTERRUPTED" ) + if bench.stall_error is not None: + raise ExecutionError( + f"{bench.stall_error}; run aborted and report marked INTERRUPTED" + ) if bench.report is None: raise ExecutionError("Benchmark produced no usable metrics report") if not (report := bench.report).complete: diff --git a/src/inference_endpoint/config/schema.py b/src/inference_endpoint/config/schema.py index f27f04188..cb5df73d1 100644 --- a/src/inference_endpoint/config/schema.py +++ b/src/inference_endpoint/config/schema.py @@ -812,7 +812,8 @@ class Timeouts(WithUpdatesMixin, BaseModel): description=( "Endpoint liveness deadline: fail the run when work is in flight but " "no response chunk or completion arrives for this many seconds " - "(None = off). See docs/config/DESIGN.md for choosing a value." + "(None = off). Use >=300 seconds and size it above the longest " + "single-request latency. See docs/config/DESIGN.md." ), ) interrupted_teardown_grace_s: float = Field( diff --git a/src/inference_endpoint/config/templates/concurrency_template_full.yaml b/src/inference_endpoint/config/templates/concurrency_template_full.yaml index 6fff60e66..d11db5afb 100644 --- a/src/inference_endpoint/config/templates/concurrency_template_full.yaml +++ b/src/inference_endpoint/config/templates/concurrency_template_full.yaml @@ -86,7 +86,7 @@ settings: worker_gc_mode: relaxed # Worker GC strategy | options: disabled, relaxed, system timeouts: # All global waits and deadlines (see config/schema.py) run_timeout_s: null # Whole-run watchdog seconds (None = off). - no_progress_timeout_s: null # Endpoint liveness deadline: fail the run when work is in flight but no response chunk or completion arrives for this many seconds (None = off). See docs/config/DESIGN.md for choosing a value. + no_progress_timeout_s: null # Endpoint liveness deadline: fail the run when work is in flight but no response chunk or completion arrives for this many seconds (None = off). Use >=300 seconds and size it above the longest single-request latency. See docs/config/DESIGN.md. interrupted_teardown_grace_s: 30.0 # Abort drain grace before remaining services are killed. service_ready_timeout_s: 30.0 # Service startup wait in seconds. warmup_drain_timeout_s: 240.0 # Warmup drain seconds (None = unlimited). diff --git a/src/inference_endpoint/config/templates/offline_template_full.yaml b/src/inference_endpoint/config/templates/offline_template_full.yaml index aba0a3d83..83aaf646c 100644 --- a/src/inference_endpoint/config/templates/offline_template_full.yaml +++ b/src/inference_endpoint/config/templates/offline_template_full.yaml @@ -86,7 +86,7 @@ settings: worker_gc_mode: relaxed # Worker GC strategy | options: disabled, relaxed, system timeouts: # All global waits and deadlines (see config/schema.py) run_timeout_s: null # Whole-run watchdog seconds (None = off). - no_progress_timeout_s: null # Endpoint liveness deadline: fail the run when work is in flight but no response chunk or completion arrives for this many seconds (None = off). See docs/config/DESIGN.md for choosing a value. + no_progress_timeout_s: null # Endpoint liveness deadline: fail the run when work is in flight but no response chunk or completion arrives for this many seconds (None = off). Use >=300 seconds and size it above the longest single-request latency. See docs/config/DESIGN.md. interrupted_teardown_grace_s: 30.0 # Abort drain grace before remaining services are killed. service_ready_timeout_s: 30.0 # Service startup wait in seconds. warmup_drain_timeout_s: 240.0 # Warmup drain seconds (None = unlimited). diff --git a/src/inference_endpoint/config/templates/online_template_full.yaml b/src/inference_endpoint/config/templates/online_template_full.yaml index f94f619d5..e62fe37ad 100644 --- a/src/inference_endpoint/config/templates/online_template_full.yaml +++ b/src/inference_endpoint/config/templates/online_template_full.yaml @@ -87,7 +87,7 @@ settings: worker_gc_mode: relaxed # Worker GC strategy | options: disabled, relaxed, system timeouts: # All global waits and deadlines (see config/schema.py) run_timeout_s: null # Whole-run watchdog seconds (None = off). - no_progress_timeout_s: null # Endpoint liveness deadline: fail the run when work is in flight but no response chunk or completion arrives for this many seconds (None = off). See docs/config/DESIGN.md for choosing a value. + no_progress_timeout_s: null # Endpoint liveness deadline: fail the run when work is in flight but no response chunk or completion arrives for this many seconds (None = off). Use >=300 seconds and size it above the longest single-request latency. See docs/config/DESIGN.md. interrupted_teardown_grace_s: 30.0 # Abort drain grace before remaining services are killed. service_ready_timeout_s: 30.0 # Service startup wait in seconds. warmup_drain_timeout_s: 240.0 # Warmup drain seconds (None = unlimited). diff --git a/tests/unit/commands/test_benchmark.py b/tests/unit/commands/test_benchmark.py index 13c8a6359..30f8fd64e 100644 --- a/tests/unit/commands/test_benchmark.py +++ b/tests/unit/commands/test_benchmark.py @@ -102,6 +102,7 @@ ) from inference_endpoint.load_generator.sample_order import create_sample_order from inference_endpoint.load_generator.session import ( + NoProgressError, PhaseResult, PhaseType, SessionResult, @@ -1728,6 +1729,78 @@ async def _launch_ok(service_configs, *, timeout): with pytest.raises(expected_error, match=expected_match): await _run_benchmark_async(ctx, loop, sigint=SigintGovernor()) + @pytest.mark.unit + @pytest.mark.asyncio + async def test_no_progress_error_drains_and_returns_partial_result(self, tmp_path): + """A liveness failure must preserve the drained report for finalization.""" + config = OfflineConfig(**_OFFLINE_KWARGS, settings=OfflineSettings()) + ctx = self._make_ctx(config, tmp_path) + + async def _launch_ok(service_configs, *, timeout): + return None + + mock_zmq = MagicMock() + mock_zmq.socket_dir = str(tmp_path / "sockets") + mock_client = MagicMock() + mock_client.shutdown_async = AsyncMock() + mock_session = MagicMock() + mock_session.run = AsyncMock(side_effect=NoProgressError("endpoint stalled")) + report = MagicMock() + drain = AsyncMock(return_value=report) + loop = asyncio.get_event_loop() + + with ( + patch( + "inference_endpoint.commands.benchmark.pipeline.ManagedZMQContext" + ) as MockZMQ, + patch( + "inference_endpoint.commands.benchmark.pipeline.EventPublisherService" + ) as MockPub, + patch( + "inference_endpoint.commands.benchmark.pipeline.MetricsSnapshotSubscriber" + ) as MockSub, + patch( + "inference_endpoint.commands.benchmark.pipeline.ServiceLauncher" + ) as MockLauncher, + patch("inference_endpoint.commands.benchmark.execute.tqdm"), + patch( + "inference_endpoint.commands.benchmark.execute._create_issuer", + new=AsyncMock(return_value=(MagicMock(), mock_client)), + ), + patch( + "inference_endpoint.commands.benchmark.execute._build_agentic_strategy", + return_value=None, + ), + patch( + "inference_endpoint.commands.benchmark.execute.BenchmarkSession", + return_value=mock_session, + ), + patch( + "inference_endpoint.commands.benchmark.execute._build_phases", + return_value=[], + ), + patch( + "inference_endpoint.commands.benchmark.pipeline." + "MetricsPipeline.drain_and_build_report", + new=drain, + ), + patch.object(loop, "add_signal_handler"), + patch.object(loop, "remove_signal_handler"), + ): + MockZMQ.scoped.return_value.__enter__ = MagicMock(return_value=mock_zmq) + MockZMQ.scoped.return_value.__exit__ = MagicMock(return_value=False) + MockPub.return_value.socket_name = "test_pub" + MockSub.return_value.start = MagicMock() + MockLauncher.return_value.launch = _launch_ok + + bench = await _run_benchmark_async(ctx, loop, sigint=SigintGovernor()) + + assert bench.stall_error == "endpoint stalled" + assert bench.report is report + assert bench.session.phase_results == [] + drain.assert_awaited_once() + mock_client.shutdown_async.assert_awaited_once() + class TestAccuracyOnlyDatasetLoading: """`--accuracy-only` must skip the performance dataset even when the config @@ -2497,15 +2570,16 @@ def test_skip_endpoint_phase_scorer_reports_external_sample_count( @pytest.mark.unit @pytest.mark.parametrize( - ("abort_field", "snapshot_state"), + ("abort_field", "abort_value", "snapshot_state"), [ - ("run_timed_out", "complete"), - ("user_interrupted", "live"), - (None, "interrupted"), + ("run_timed_out", True, "complete"), + ("user_interrupted", True, "live"), + ("stall_error", "endpoint stalled", "complete"), + (None, None, "interrupted"), ], ) def test_aborted_run_never_scores_or_writes_complete_artifacts( - self, tmp_path, monkeypatch, abort_field, snapshot_state + self, tmp_path, monkeypatch, abort_field, abort_value, snapshot_state ): """Split-brain guard: an aborted run must never ship artifacts under any non-interrupted state. "complete": the aggregator finalized before @@ -2519,7 +2593,7 @@ def test_aborted_run_never_scores_or_writes_complete_artifacts( bench = _make_benchmark_result(tmp_path) bench.report = self._make_report(state=snapshot_state) if abort_field is not None: - setattr(bench, abort_field, True) + setattr(bench, abort_field, abort_value) score_accuracy = MagicMock() monkeypatch.setattr(execute_mod, "score_accuracy", score_accuracy) @@ -2589,6 +2663,32 @@ def _make_report(state: str) -> Report: ) +class TestRunBenchmarkStallOutcome: + """An endpoint stall must finalize artifacts before the run fails.""" + + @pytest.mark.unit + def test_stall_finalizes_then_raises(self, tmp_path, monkeypatch): + config = OfflineConfig(**_OFFLINE_KWARGS) + ctx = _make_benchmark_context(config, tmp_path) + bench = _make_benchmark_result(tmp_path) + bench.stall_error = "endpoint stalled" + finalize = MagicMock() + + monkeypatch.setattr(execute_mod, "setup_benchmark", MagicMock(return_value=ctx)) + monkeypatch.setattr( + execute_mod, "run_benchmark_async", MagicMock(return_value=bench) + ) + monkeypatch.setattr(execute_mod, "finalize_benchmark", finalize) + + with pytest.raises(ExecutionError, match="endpoint stalled"): + execute_mod.run_benchmark(config, TestMode.PERF) + + # The raise happens only after finalize wrote the standard artifacts; + # reaching pytest.raises without this call would mean the stall + # propagated before finalization (the pre-fix behavior). + finalize.assert_called_once_with(ctx, bench) + + class TestScorerMethodSync: """Ensure ScorerMethod enum stays in sync with the scorer registry."""