From 1c38c003427c03b8de11a8ac75691801b446f2ee Mon Sep 17 00:00:00 2001 From: kimseunghyun Date: Wed, 14 Oct 2026 15:45:56 +0900 Subject: [PATCH 1/7] fix(session): stamp COMPLETE with the arrival time of its chunks In streaming runs a sample's COMPLETE could be earlier than its own RECV_FIRST, which gave a negative TPOT. The ZMQ transport decodes a whole drain before the session handles any message, so QueryResult.completed_at (set at decode) predates the handling-time stamp of chunks from the same drain. COMPLETE now uses the same per-response monotonic_ns() read as the RECV_FIRST/RECV_NON_FIRST events. Closes #533 Co-Authored-By: Claude Opus 5.5 --- docs/load_generator/DESIGN.md | 20 ++++--- .../load_generator/session.py | 8 ++- .../unit/load_generator/test_async_session.py | 58 +++++++++++++++++-- 3 files changed, 71 insertions(+), 15 deletions(-) diff --git a/docs/load_generator/DESIGN.md b/docs/load_generator/DESIGN.md index ac9327a7b..5b6d19322 100644 --- a/docs/load_generator/DESIGN.md +++ b/docs/load_generator/DESIGN.md @@ -352,12 +352,16 @@ poll-based approach. until the batch threshold is reached or `flush()` is called. The timestamp itself is still accurate (captured before buffering), but the EventRecord reaches subscribers with batching latency. -- COMPLETE: `QueryResult.completed_at` is set via - `force_setattr(monotonic_ns())` in `__post_init__`, regenerated on - deserialization. Both ISSUED and COMPLETE timestamps share the same ZMQ - transit bias. TTFT (`RECV_FIRST - ISSUED`) is still sensitive to this overhead - since it spans the full ZMQ round-trip. TPOT avoids cross-process clock skew - by computing time deltas between consecutive chunks within the same process. +- RECV_FIRST, RECV_NON_FIRST and COMPLETE: one `monotonic_ns()` read per + response in `_receive_responses`, when the session handles it. All three use + this same clock point, so a response's COMPLETE is never earlier than its own + chunks. (`QueryResult.completed_at` is set at decode; the transport decodes a + whole drain before the session handles any message, so a result decoded in the + same drain as its chunks would precede them and give a negative TPOT.) Both + ISSUED and COMPLETE timestamps share the same ZMQ transit bias. TTFT + (`RECV_FIRST - ISSUED`) is still sensitive to this overhead since it spans the + full ZMQ round-trip. TPOT avoids cross-process clock skew by computing time + deltas between consecutive chunks within the same process. ### LoadStrategy (Protocol) @@ -635,7 +639,7 @@ sequenceDiagram alt Non-streaming W-->>I: ZMQ PUSH (QueryResult) I-->>B: recv() - B->>E: publish(COMPLETE, uuid, completed_at) + B->>E: publish(COMPLETE, uuid, timestamp_ns) else Streaming W-->>I: ZMQ PUSH (StreamChunk first_chunk) I-->>B: recv() @@ -647,7 +651,7 @@ sequenceDiagram end W-->>I: ZMQ PUSH (QueryResult — final accumulated output) I-->>B: recv() - B->>E: publish(COMPLETE, uuid, completed_at) + B->>E: publish(COMPLETE, uuid, timestamp_ns) end E->>M: ZMQ PUB (EventRecord) diff --git a/src/inference_endpoint/load_generator/session.py b/src/inference_endpoint/load_generator/session.py index 896646f81..5d8e16cf0 100644 --- a/src/inference_endpoint/load_generator/session.py +++ b/src/inference_endpoint/load_generator/session.py @@ -793,9 +793,11 @@ def _handle_response(self, resp: QueryResult | StreamChunk, now_ns: int) -> None self._publisher.publish( EventRecord( event_type=SampleEventType.COMPLETE, - timestamp_ns=resp.completed_at - if isinstance(resp.completed_at, int) - else now_ns, + # The same arrival stamp as RECV_FIRST/RECV_NON_FIRST. + # resp.completed_at is set at decode, so a result decoded + # in the same transport drain as its chunks would precede + # them (negative TPOT). + timestamp_ns=now_ns, sample_uuid=query_id, conversation_id=conv_id_str, turn=turn_num, diff --git a/tests/unit/load_generator/test_async_session.py b/tests/unit/load_generator/test_async_session.py index 99cc04ced..ec88720bd 100644 --- a/tests/unit/load_generator/test_async_session.py +++ b/tests/unit/load_generator/test_async_session.py @@ -1200,6 +1200,57 @@ async def inject_streaming_responses(): recv_first = publisher.events_of_type(SampleEventType.RECV_FIRST) assert len(recv_first) == 2 + @pytest.mark.asyncio + async def test_streaming_complete_not_before_chunks_of_one_drain(self): + """COMPLETE is never stamped before the chunks of its own response. + + One transport drain decodes [chunk, chunk, result] before the session + handles any of them, so QueryResult.completed_at is older than the + chunks' handling time. Regression test for mlcommons/endpoints#533. + """ + loop = asyncio.get_running_loop() + publisher = FakePublisher() + + issuer = FakeIssuer() + issuer._loop = loop + issuer._auto_respond = False + + session = BenchmarkSession(issuer, publisher, loop) + settings = _make_settings( + load_pattern=LoadPattern( + type=LoadPatternType.CONCURRENCY, target_concurrency=1 + ), + n_samples=1, + ) + phases = [PhaseConfig("perf", settings, FakeDataset(1))] + + async def inject_one_drain(): + while not issuer._issued: + await asyncio.sleep(0.005) + query_id = issuer._issued[0].id + # Decoded first, as the transport decodes a whole drain at once. + result = QueryResult(id=query_id, response_output="out") + await asyncio.sleep(0.01) + issuer.inject_response( + StreamChunk(id=query_id, metadata={"first_chunk": True}) + ) + issuer.inject_response(StreamChunk(id=query_id, response_chunk="more")) + issuer.inject_response(result) + + injector = asyncio.create_task(inject_one_drain()) + await asyncio.wait_for(session.run(phases), timeout=5.0) + await injector + + recv_first = publisher.events_of_type(SampleEventType.RECV_FIRST) + recv_non_first = publisher.events_of_type(SampleEventType.RECV_NON_FIRST) + complete = publisher.events_of_type(SampleEventType.COMPLETE) + assert len(recv_first) == len(recv_non_first) == len(complete) == 1 + assert ( + recv_first[0].timestamp_ns + <= recv_non_first[0].timestamp_ns + <= complete[0].timestamp_ns + ) + @pytest.mark.asyncio async def test_concurrency_strategy_transport_close_no_deadlock(self): """ConcurrencyStrategy must not deadlock when transport closes mid-phase.""" @@ -1371,10 +1422,9 @@ async def test_handle_response_stamps_conversation_id_and_turn(self): 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 + # COMPLETE carries the caller's arrival stamp, like the chunk events, + # not the decode-time QueryResult.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 From 75f229dcf50e24e1dba17a93a49e412e47b3cbe4 Mon Sep 17 00:00:00 2001 From: kimseunghyun Date: Wed, 14 Oct 2026 16:05:12 +0900 Subject: [PATCH 2/7] docs(load_generator): update the second timestamp section for COMPLETE The "Timestamp Fidelity" section and its _handle_response example still described COMPLETE as stamped with QueryResult.completed_at. Both now match the implementation: one arrival stamp per response, shared by the chunk events and COMPLETE. Co-Authored-By: Claude Opus 5.5 --- docs/load_generator/DESIGN.md | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/docs/load_generator/DESIGN.md b/docs/load_generator/DESIGN.md index 5b6d19322..ee40a51ab 100644 --- a/docs/load_generator/DESIGN.md +++ b/docs/load_generator/DESIGN.md @@ -947,9 +947,11 @@ shorter strings, no parsing overhead. Update `Query.default_factory` to match. - **ISSUED**: `monotonic_ns()` taken in `PhaseIssuer.issue()` immediately before `issuer.issue(query)`. ZMQ push is sync/non-blocking — timestamp is honest. -- **COMPLETE**: `QueryResult.completed_at` set in `__post_init__` on - deserialization in the main process. Measures main-process receipt time, not - worker-side completion. +- **RECV_FIRST / RECV_NON_FIRST / COMPLETE**: one `monotonic_ns()` read per + response in `_receive_responses`, when the session handles it. Measures + main-process handling time, not worker-side completion. All three use the same + clock point, so a response's COMPLETE is never earlier than its own chunks. + `QueryResult.completed_at` is set at decode and does not give this guarantee. - **TTFT**: `RECV_FIRST - ISSUED` includes full round-trip ZMQ overhead (outbound to worker + return to main). This adds 20-100μs of systematic bias. Acceptable for most benchmarks; document as a known measurement overhead. @@ -963,7 +965,7 @@ After a warmup phase (no drain), in-flight responses arrive during the perf phase. The receiver must distinguish stale vs current-phase completions: ```python -def _handle_response(self, resp: QueryResult) -> None: +def _handle_response(self, resp: QueryResult, now_ns: int) -> None: query_id = resp.id phase_issuer = self._current_phase_issuer # ERROR is published before COMPLETE (the aggregator depends on this order @@ -975,7 +977,7 @@ def _handle_response(self, resp: QueryResult) -> None: if self._current_phase_type != PhaseType.WARMUP: self._publisher.publish(EventRecord( event_type=SampleEventType.COMPLETE, - timestamp_ns=resp.completed_at, + timestamp_ns=now_ns, # the caller's per-response arrival stamp sample_uuid=query_id, )) # Only route to the current phase's strategy if this is a current-phase query. From ce2902acdf71993d0ec0b96aa1982ecb8e8cc1d1 Mon Sep 17 00:00:00 2001 From: kimseunghyun Date: Wed, 14 Oct 2026 17:05:21 +0900 Subject: [PATCH 3/7] docs(load_generator): show the shared now_ns in the receiver example The _receive_responses example called _handle_response(resp) and read the clock again for the liveness stamp. It now reads monotonic_ns() once per response and passes it to _handle_response, as the implementation does. Co-Authored-By: Claude Opus 5.5 --- docs/load_generator/DESIGN.md | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/docs/load_generator/DESIGN.md b/docs/load_generator/DESIGN.md index ee40a51ab..453e62b77 100644 --- a/docs/load_generator/DESIGN.md +++ b/docs/load_generator/DESIGN.md @@ -309,9 +309,12 @@ async def _receive_responses(self): if self._strategy_task and not self._strategy_task.done(): self._strategy_task.cancel() break - self._handle_response(resp) + # One clock read per response, shared by the event records + # (RECV_FIRST/RECV_NON_FIRST/COMPLETE) and the liveness stamp. + now_ns = time.monotonic_ns() + self._handle_response(resp, now_ns) if self._endpoint_response_idle_timeout_ns is not None: - self._last_response_progress_ns = time.monotonic_ns() + self._last_response_progress_ns = now_ns ``` Uses `recv()` exclusively — no `poll()` spin. The ZMQ fd is registered with the @@ -335,7 +338,7 @@ returns as soon as the fd is readable and `eager_task_factory` executes the woken semaphore waiter synchronously, there is no added latency compared to a poll-based approach. -**`_handle_response(resp)`**: +**`_handle_response(resp, now_ns)`**: - `QueryResult`: publish COMPLETE event, decrement `_inflight`, call `on_sample_complete`, call `strategy.on_query_complete(query_id)` if strategy From 9809875282220371641d4a97187d7723f3130010 Mon Sep 17 00:00:00 2001 From: kimseunghyun Date: Mon, 2 Nov 2026 11:44:41 +0900 Subject: [PATCH 4/7] fix(session): keep completed_at for COMPLETE unless its chunks came later Review feedback: stamping COMPLETE at handling time also moved COMPLETE for non-streaming responses, so client backlog entered COMPLETE - ISSUED latency and OSL/s. COMPLETE now keeps QueryResult.completed_at, and only a response whose own chunks were handled later (same transport drain) takes its last chunk's stamp. The session keeps that stamp per query and pops it at the result. Co-Authored-By: Claude Opus 5.5 Signed-off-by: kimseunghyun --- docs/load_generator/DESIGN.md | 48 +++++++++++-------- .../load_generator/session.py | 33 +++++++++++-- .../unit/load_generator/test_async_session.py | 36 ++++++++++++-- 3 files changed, 89 insertions(+), 28 deletions(-) diff --git a/docs/load_generator/DESIGN.md b/docs/load_generator/DESIGN.md index 453e62b77..86728cd2a 100644 --- a/docs/load_generator/DESIGN.md +++ b/docs/load_generator/DESIGN.md @@ -309,8 +309,8 @@ async def _receive_responses(self): 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 - # (RECV_FIRST/RECV_NON_FIRST/COMPLETE) and the liveness stamp. + # One clock read per response, shared by the chunk event records + # (RECV_FIRST/RECV_NON_FIRST), the COMPLETE floor and the liveness stamp. now_ns = time.monotonic_ns() self._handle_response(resp, now_ns) if self._endpoint_response_idle_timeout_ns is not None: @@ -355,16 +355,19 @@ poll-based approach. until the batch threshold is reached or `flush()` is called. The timestamp itself is still accurate (captured before buffering), but the EventRecord reaches subscribers with batching latency. -- RECV_FIRST, RECV_NON_FIRST and COMPLETE: one `monotonic_ns()` read per - response in `_receive_responses`, when the session handles it. All three use - this same clock point, so a response's COMPLETE is never earlier than its own - chunks. (`QueryResult.completed_at` is set at decode; the transport decodes a - whole drain before the session handles any message, so a result decoded in the - same drain as its chunks would precede them and give a negative TPOT.) Both - ISSUED and COMPLETE timestamps share the same ZMQ transit bias. TTFT - (`RECV_FIRST - ISSUED`) is still sensitive to this overhead since it spans the - full ZMQ round-trip. TPOT avoids cross-process clock skew by computing time - deltas between consecutive chunks within the same process. +- RECV_FIRST and RECV_NON_FIRST: one `monotonic_ns()` read per response in + `_receive_responses`, when the session handles it. +- COMPLETE: `QueryResult.completed_at` is set via + `force_setattr(monotonic_ns())` in `__post_init__`, regenerated on + deserialization. The transport decodes a whole drain before the session + handles any message, so a result decoded in the same drain as its chunks is + older than their RECV stamps (negative TPOT). In that case only, COMPLETE + takes the handling stamp of the response's last chunk, so a response's + COMPLETE is never earlier than its own chunks. A response without chunks keeps + `completed_at`. Both ISSUED and COMPLETE timestamps share the same ZMQ transit + bias. TTFT (`RECV_FIRST - ISSUED`) is still sensitive to this overhead since + it spans the full ZMQ round-trip. TPOT avoids cross-process clock skew by + computing time deltas between consecutive chunks within the same process. ### LoadStrategy (Protocol) @@ -642,7 +645,7 @@ sequenceDiagram alt Non-streaming W-->>I: ZMQ PUSH (QueryResult) I-->>B: recv() - B->>E: publish(COMPLETE, uuid, timestamp_ns) + B->>E: publish(COMPLETE, uuid, completed_at) else Streaming W-->>I: ZMQ PUSH (StreamChunk first_chunk) I-->>B: recv() @@ -654,7 +657,7 @@ sequenceDiagram end W-->>I: ZMQ PUSH (QueryResult — final accumulated output) I-->>B: recv() - B->>E: publish(COMPLETE, uuid, timestamp_ns) + B->>E: publish(COMPLETE, uuid, max(completed_at, last chunk stamp)) end E->>M: ZMQ PUB (EventRecord) @@ -950,11 +953,14 @@ shorter strings, no parsing overhead. Update `Query.default_factory` to match. - **ISSUED**: `monotonic_ns()` taken in `PhaseIssuer.issue()` immediately before `issuer.issue(query)`. ZMQ push is sync/non-blocking — timestamp is honest. -- **RECV_FIRST / RECV_NON_FIRST / COMPLETE**: one `monotonic_ns()` read per - response in `_receive_responses`, when the session handles it. Measures - main-process handling time, not worker-side completion. All three use the same - clock point, so a response's COMPLETE is never earlier than its own chunks. - `QueryResult.completed_at` is set at decode and does not give this guarantee. +- **COMPLETE**: `QueryResult.completed_at` set in `__post_init__` on + deserialization in the main process. Measures main-process receipt time, not + worker-side completion. If a chunk of the same response was handled later + (same transport drain), COMPLETE takes that chunk's stamp instead, so it is + never earlier than its own chunks. +- **RECV_FIRST / RECV_NON_FIRST**: one `monotonic_ns()` read per response in + `_receive_responses`, when the session handles it. Measures main-process + handling time. - **TTFT**: `RECV_FIRST - ISSUED` includes full round-trip ZMQ overhead (outbound to worker + return to main). This adds 20-100μs of systematic bias. Acceptable for most benchmarks; document as a known measurement overhead. @@ -970,6 +976,7 @@ phase. The receiver must distinguish stale vs current-phase completions: ```python def _handle_response(self, resp: QueryResult, now_ns: int) -> None: query_id = resp.id + last_chunk_ns = self._last_chunk_ns.pop(query_id, None) # set by its chunks phase_issuer = self._current_phase_issuer # ERROR is published before COMPLETE (the aggregator depends on this order # for tracked-failure counting). @@ -980,7 +987,8 @@ def _handle_response(self, resp: QueryResult, now_ns: int) -> None: if self._current_phase_type != PhaseType.WARMUP: self._publisher.publish(EventRecord( event_type=SampleEventType.COMPLETE, - timestamp_ns=now_ns, # the caller's per-response arrival stamp + # completed_at, never earlier than this response's chunks + timestamp_ns=self._complete_timestamp_ns(resp, now_ns, last_chunk_ns), sample_uuid=query_id, )) # Only route to the current phase's strategy if this is a current-phase query. diff --git a/src/inference_endpoint/load_generator/session.py b/src/inference_endpoint/load_generator/session.py index 5d8e16cf0..74cb02e5a 100644 --- a/src/inference_endpoint/load_generator/session.py +++ b/src/inference_endpoint/load_generator/session.py @@ -402,6 +402,9 @@ def __init__( self._drain_event = asyncio.Event() self._fatal_error: RuntimeError | None = None self._last_response_progress_ns: int | None = None + # Handling stamp of each streaming response's latest chunk, so that its + # COMPLETE is never earlier than its own chunks. Popped at the result. + self._last_chunk_ns: dict[str, int] = {} # 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._endpoint_response_idle_timeout_ns = ( @@ -750,6 +753,7 @@ def _handle_response(self, resp: QueryResult | StreamChunk, now_ns: int) -> None if isinstance(resp, QueryResult): query_id = resp.id + last_chunk_ns = self._last_chunk_ns.pop(query_id, None) # Drop late responses for queries already synthetically terminated # (e.g. by AgenticInferenceStrategy._handle_timeout). Without this gate, # a real response arriving after timeout double-publishes ERROR/COMPLETE @@ -793,11 +797,9 @@ def _handle_response(self, resp: QueryResult | StreamChunk, now_ns: int) -> None self._publisher.publish( EventRecord( event_type=SampleEventType.COMPLETE, - # The same arrival stamp as RECV_FIRST/RECV_NON_FIRST. - # resp.completed_at is set at decode, so a result decoded - # in the same transport drain as its chunks would precede - # them (negative TPOT). - timestamp_ns=now_ns, + timestamp_ns=self._complete_timestamp_ns( + resp, now_ns, last_chunk_ns + ), sample_uuid=query_id, conversation_id=conv_id_str, turn=turn_num, @@ -835,6 +837,7 @@ def _handle_response(self, resp: QueryResult | StreamChunk, now_ns: int) -> None conv_id_str, turn_num = phase_issuer.uuid_to_conv_info.get( resp.id, ("", None) ) + self._last_chunk_ns[resp.id] = now_ns self._publisher.publish( EventRecord( event_type=event_type, @@ -845,6 +848,26 @@ def _handle_response(self, resp: QueryResult | StreamChunk, now_ns: int) -> None ) ) + @staticmethod + def _complete_timestamp_ns( + resp: QueryResult, now_ns: int, last_chunk_ns: int | None + ) -> int: + """COMPLETE stamp: the decode time, but never before the response's chunks. + + `resp.completed_at` is set when the transport decodes the result. The + transport decodes a whole drain before the session handles any message, + so a result decoded in the same drain as its chunks is older than the + chunks' handling stamps (RECV_FIRST/RECV_NON_FIRST), which would give a + negative TPOT. Only that case moves COMPLETE, to the last chunk's stamp; + a response without chunks keeps its decode time. + """ + completed_ns = ( + resp.completed_at if isinstance(resp.completed_at, int) else now_ns + ) + if last_chunk_ns is None: + return completed_ns + return max(completed_ns, last_chunk_ns) + def _make_stop_check( self, settings: RuntimeSettings, phase_start_ns: int ) -> Callable[[], bool]: diff --git a/tests/unit/load_generator/test_async_session.py b/tests/unit/load_generator/test_async_session.py index ec88720bd..4a27002a6 100644 --- a/tests/unit/load_generator/test_async_session.py +++ b/tests/unit/load_generator/test_async_session.py @@ -1369,6 +1369,37 @@ async def inject_error(): f"complete at idx {complete_idx}" ) + @pytest.mark.asyncio + async def test_complete_stamp_is_decode_time_floored_by_own_chunks(self): + """COMPLETE keeps QueryResult.completed_at unless one of the response's + own chunks was handled later; then it takes that chunk's stamp.""" + loop = asyncio.get_running_loop() + publisher = FakePublisher() + session = BenchmarkSession(FakeIssuer(), publisher, loop) + + late = QueryResult(id="q-late", response_output="out") + session._handle_response( + StreamChunk(id="q-late", metadata={"first_chunk": True}), + late.completed_at + 500, + ) + session._handle_response(late, late.completed_at + 900) + early = QueryResult(id="q-early", response_output="out") + session._handle_response( + StreamChunk(id="q-early", metadata={"first_chunk": True}), + early.completed_at - 500, + ) + session._handle_response(early, early.completed_at + 900) + + complete = { + e.sample_uuid: e.timestamp_ns + for e in publisher.events_of_type(SampleEventType.COMPLETE) + } + assert complete == { + "q-late": late.completed_at + 500, + "q-early": early.completed_at, + } + assert session._last_chunk_ns == {} + @pytest.mark.asyncio async def test_handle_response_stamps_conversation_id_and_turn(self): """All event types inherit (conv_id, turn) seeded at issue time; streaming @@ -1422,9 +1453,8 @@ async def test_handle_response_stamps_conversation_id_and_turn(self): 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 - # COMPLETE carries the caller's arrival stamp, like the chunk events, - # not the decode-time QueryResult.completed_at. - assert complete[0].timestamp_ns == 1_002 + # Without chunks, COMPLETE keeps the decode-time QueryResult.completed_at. + assert complete[0].timestamp_ns == ok_resp.completed_at assert "q-ok" not in phase_issuer.uuid_to_conv_info assert "q-ok" not in phase_issuer.completed_uuids From b1852db87f551fd171a2db8d23bf2c946a05362f Mon Sep 17 00:00:00 2001 From: kimseunghyun Date: Mon, 2 Nov 2026 17:48:59 +0900 Subject: [PATCH 5/7] fix(session): keep chunk stamps only for active queries; qualify COMPLETE bias Chunks of a previous phase or of a terminated query may never get a QueryResult, so their stamps would stay in _last_chunk_ns for the whole session. Keep a stamp only while the query is active in the current phase (any query when no phase is set). The design doc now says that COMPLETE includes the session's handling backlog when it is floored to the last chunk. Co-Authored-By: Claude Opus 5.5 Signed-off-by: kimseunghyun --- docs/load_generator/DESIGN.md | 11 +++++---- .../load_generator/session.py | 6 ++++- .../unit/load_generator/test_async_session.py | 23 +++++++++++++++++++ 3 files changed, 35 insertions(+), 5 deletions(-) diff --git a/docs/load_generator/DESIGN.md b/docs/load_generator/DESIGN.md index 86728cd2a..51f5644bb 100644 --- a/docs/load_generator/DESIGN.md +++ b/docs/load_generator/DESIGN.md @@ -364,8 +364,9 @@ poll-based approach. older than their RECV stamps (negative TPOT). In that case only, COMPLETE takes the handling stamp of the response's last chunk, so a response's COMPLETE is never earlier than its own chunks. A response without chunks keeps - `completed_at`. Both ISSUED and COMPLETE timestamps share the same ZMQ transit - bias. TTFT (`RECV_FIRST - ISSUED`) is still sensitive to this overhead since + `completed_at`. When COMPLETE keeps `completed_at`, ISSUED and COMPLETE share + the same ZMQ transit bias. When it takes the last chunk's stamp, COMPLETE also + includes the session's handling backlog for that drain. TTFT (`RECV_FIRST - ISSUED`) is still sensitive to this overhead since it spans the full ZMQ round-trip. TPOT avoids cross-process clock skew by computing time deltas between consecutive chunks within the same process. @@ -965,8 +966,10 @@ shorter strings, no parsing overhead. Update `Query.default_factory` to match. (outbound to worker + return to main). This adds 20-100μs of systematic bias. Acceptable for most benchmarks; document as a known measurement overhead. - **Latency (COMPLETE - ISSUED)**: Both timestamps taken on the main process - side. ZMQ transit bias is symmetric and cancels. This is the most accurate - measurement. + side. When COMPLETE keeps `completed_at`, the ZMQ transit bias is symmetric + and cancels; this is the most accurate measurement. When COMPLETE is floored + to the response's last chunk, it also includes the session's handling + backlog for that drain. ### Stale Completions After Saturation diff --git a/src/inference_endpoint/load_generator/session.py b/src/inference_endpoint/load_generator/session.py index 74cb02e5a..fa37235f7 100644 --- a/src/inference_endpoint/load_generator/session.py +++ b/src/inference_endpoint/load_generator/session.py @@ -837,7 +837,11 @@ def _handle_response(self, resp: QueryResult | StreamChunk, now_ns: int) -> None conv_id_str, turn_num = phase_issuer.uuid_to_conv_info.get( resp.id, ("", None) ) - self._last_chunk_ns[resp.id] = now_ns + if phase_issuer is None or ( + resp.id in phase_issuer.uuid_to_conv_info + and resp.id not in phase_issuer.completed_uuids + ): + self._last_chunk_ns[resp.id] = now_ns self._publisher.publish( EventRecord( event_type=event_type, diff --git a/tests/unit/load_generator/test_async_session.py b/tests/unit/load_generator/test_async_session.py index 4a27002a6..273fc6185 100644 --- a/tests/unit/load_generator/test_async_session.py +++ b/tests/unit/load_generator/test_async_session.py @@ -1400,6 +1400,29 @@ async def test_complete_stamp_is_decode_time_floored_by_own_chunks(self): } assert session._last_chunk_ns == {} + @pytest.mark.asyncio + async def test_chunk_stamp_kept_only_for_active_queries(self): + """Chunks of stale (previous-phase) or terminated queries may never get + a QueryResult, so their stamps are not kept.""" + loop = asyncio.get_running_loop() + issuer = FakeIssuer() + issuer._auto_respond = False + publisher = FakePublisher() + session = BenchmarkSession(issuer, publisher, loop) + phase_issuer = PhaseIssuer(FakeDataset(2), issuer, publisher, lambda: False) + session._current_phase_issuer = phase_issuer + active = phase_issuer.issue(0) + terminated = phase_issuer.issue(1) + phase_issuer.completed_uuids.add(terminated) + + for query_id in (active, terminated, "q-stale"): + session._handle_response( + StreamChunk(id=query_id, metadata={"first_chunk": True}), 100 + ) + + assert session._last_chunk_ns == {active: 100} + assert len(publisher.events_of_type(SampleEventType.RECV_FIRST)) == 3 + @pytest.mark.asyncio async def test_handle_response_stamps_conversation_id_and_turn(self): """All event types inherit (conv_id, turn) seeded at issue time; streaming From 6287873dd4a7a0bb7060a74c1691472d360bfc07 Mon Sep 17 00:00:00 2001 From: kimseunghyun Date: Mon, 2 Nov 2026 17:50:47 +0900 Subject: [PATCH 6/7] docs(design): prettier wrap Co-Authored-By: Claude Opus 5.5 Signed-off-by: kimseunghyun --- docs/load_generator/DESIGN.md | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/docs/load_generator/DESIGN.md b/docs/load_generator/DESIGN.md index 51f5644bb..a3c999170 100644 --- a/docs/load_generator/DESIGN.md +++ b/docs/load_generator/DESIGN.md @@ -366,9 +366,10 @@ poll-based approach. COMPLETE is never earlier than its own chunks. A response without chunks keeps `completed_at`. When COMPLETE keeps `completed_at`, ISSUED and COMPLETE share the same ZMQ transit bias. When it takes the last chunk's stamp, COMPLETE also - includes the session's handling backlog for that drain. TTFT (`RECV_FIRST - ISSUED`) is still sensitive to this overhead since - it spans the full ZMQ round-trip. TPOT avoids cross-process clock skew by - computing time deltas between consecutive chunks within the same process. + includes the session's handling backlog for that drain. TTFT + (`RECV_FIRST - ISSUED`) is still sensitive to this overhead since it spans the + full ZMQ round-trip. TPOT avoids cross-process clock skew by computing time + deltas between consecutive chunks within the same process. ### LoadStrategy (Protocol) @@ -968,8 +969,8 @@ shorter strings, no parsing overhead. Update `Query.default_factory` to match. - **Latency (COMPLETE - ISSUED)**: Both timestamps taken on the main process side. When COMPLETE keeps `completed_at`, the ZMQ transit bias is symmetric and cancels; this is the most accurate measurement. When COMPLETE is floored - to the response's last chunk, it also includes the session's handling - backlog for that drain. + to the response's last chunk, it also includes the session's handling backlog + for that drain. ### Stale Completions After Saturation From be1e54b90bdb28947f7cd64b784f53db225a1302 Mon Sep 17 00:00:00 2001 From: kimseunghyun Date: Tue, 3 Nov 2026 12:49:12 +0900 Subject: [PATCH 7/7] fix(session): drop chunk stamps of queries terminated without a result A turn timeout can terminate a streaming query after some of its chunks were stamped; it publishes a synthetic failure and no QueryResult follows, so nothing popped the stamp. Drop such a stamp when a later chunk of the terminated query arrives, drop the stamps of the phase's terminated queries at phase end, and clear all stamps at session teardown. Co-Authored-By: Claude Opus 5.5 Signed-off-by: kimseunghyun --- .../load_generator/session.py | 12 +++- .../unit/load_generator/test_async_session.py | 62 +++++++++++++++++++ 2 files changed, 73 insertions(+), 1 deletion(-) diff --git a/src/inference_endpoint/load_generator/session.py b/src/inference_endpoint/load_generator/session.py index fa37235f7..bffc2492a 100644 --- a/src/inference_endpoint/load_generator/session.py +++ b/src/inference_endpoint/load_generator/session.py @@ -403,7 +403,8 @@ def __init__( self._fatal_error: RuntimeError | None = None self._last_response_progress_ns: int | None = None # Handling stamp of each streaming response's latest chunk, so that its - # COMPLETE is never earlier than its own chunks. Popped at the result. + # COMPLETE is never earlier than its own chunks. Popped at the result; + # stamps of synthetically terminated queries are dropped at phase end. self._last_chunk_ns: dict[str, int] = {} # 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. @@ -486,6 +487,7 @@ async def run( except asyncio.CancelledError: pass self._disarm_endpoint_response_idle_timeout() + self._last_chunk_ns.clear() if self._stop_requested: # Aborted run (Ctrl-C, transport closure, run watchdog): mark # it BEFORE the terminal ENDED so the aggregator's ENDED-driven @@ -581,6 +583,12 @@ async def _run_phase(self, phase: PhaseConfig) -> PhaseResult | None: if phase.drain_after: await self._drain_inflight(phase_issuer, phase.drain_timeout) + # A query terminated without a QueryResult (e.g. a turn timeout after + # some chunks) leaves a stamp that nothing else would pop. + for query_id in [ + q for q in self._last_chunk_ns if q in phase_issuer.completed_uuids + ]: + del self._last_chunk_ns[query_id] if phase.phase_type == PhaseType.PERFORMANCE: phase_issuer.stop_performance_tracking() @@ -842,6 +850,8 @@ def _handle_response(self, resp: QueryResult | StreamChunk, now_ns: int) -> None and resp.id not in phase_issuer.completed_uuids ): self._last_chunk_ns[resp.id] = now_ns + else: + self._last_chunk_ns.pop(resp.id, None) self._publisher.publish( EventRecord( event_type=event_type, diff --git a/tests/unit/load_generator/test_async_session.py b/tests/unit/load_generator/test_async_session.py index 273fc6185..87443a4ee 100644 --- a/tests/unit/load_generator/test_async_session.py +++ b/tests/unit/load_generator/test_async_session.py @@ -1423,6 +1423,68 @@ async def test_chunk_stamp_kept_only_for_active_queries(self): assert session._last_chunk_ns == {active: 100} assert len(publisher.events_of_type(SampleEventType.RECV_FIRST)) == 3 + @pytest.mark.asyncio + async def test_late_chunk_drops_stamp_of_terminated_query(self): + """A chunk that arrives after a synthetic termination removes the stamp + recorded before it, since no QueryResult will pop it.""" + loop = asyncio.get_running_loop() + issuer = FakeIssuer() + issuer._auto_respond = False + publisher = FakePublisher() + session = BenchmarkSession(issuer, publisher, loop) + phase_issuer = PhaseIssuer(FakeDataset(1), issuer, publisher, lambda: False) + session._current_phase_issuer = phase_issuer + query_id = phase_issuer.issue(0) + + session._handle_response( + StreamChunk(id=query_id, metadata={"first_chunk": True}), 100 + ) + phase_issuer.completed_uuids.add(query_id) + phase_issuer.uuid_to_conv_info.pop(query_id) + session._handle_response(StreamChunk(id=query_id, response_chunk="x"), 200) + + assert session._last_chunk_ns == {} + + @pytest.mark.asyncio + async def test_phase_end_drops_chunk_stamps_of_terminated_queries(self): + """A turn timeout after some chunks publishes a synthetic failure and no + QueryResult follows, so the phase end drops that query's stamp.""" + loop = asyncio.get_running_loop() + issuer = FakeIssuer() + issuer._auto_respond = False + issuer._loop = loop + session = BenchmarkSession(issuer, FakePublisher(), loop) + + class TimeoutAfterChunkStrategy: + async def execute(self, phase_issuer) -> None: + query_id = phase_issuer.issue(0) + issuer.inject_response( + StreamChunk(id=query_id, metadata={"first_chunk": True}) + ) + while query_id not in session._last_chunk_ns: + await asyncio.sleep(0) + # The state changes of AgenticInferenceStrategy._handle_timeout. + phase_issuer.mark_inflight_complete() + phase_issuer.completed_uuids.add(query_id) + phase_issuer.uuid_to_conv_info.pop(query_id) + + def on_query_complete(self, query_id: str) -> None: + pass + + phase = PhaseConfig( + "perf", + _make_settings(n_samples=1), + FakeDataset(1), + strategy=TimeoutAfterChunkStrategy(), + ) + session._recv_task = asyncio.create_task(session._receive_responses()) + try: + await asyncio.wait_for(session._run_phase(phase), timeout=1.0) + assert session._last_chunk_ns == {} + finally: + issuer.shutdown() + await session._recv_task + @pytest.mark.asyncio async def test_handle_response_stamps_conversation_id_and_turn(self): """All event types inherit (conv_id, turn) seeded at issue time; streaming