Repository navigation
Feat/http fleet - #16488
Closed
reasonsolo wants to merge 12 commits into
Closed
Feat/http fleet#16488reasonsolo wants to merge 12 commits into
reasonsolo wants to merge 12 commits into
Conversation
Signed-off-by: Lizhi Zhou <1432185+reasonsolo@users.noreply.github.com>
Signed-off-by: Lizhi Zhou <1432185+reasonsolo@users.noreply.github.com>
Signed-off-by: Lizhi Zhou <1432185+reasonsolo@users.noreply.github.com>
Signed-off-by: Lizhi Zhou <1432185+reasonsolo@users.noreply.github.com>
Signed-off-by: Lizhi Zhou <1432185+reasonsolo@users.noreply.github.com>
Signed-off-by: Lizhi Zhou <1432185+reasonsolo@users.noreply.github.com>
Signed-off-by: Lizhi Zhou <1432185+reasonsolo@users.noreply.github.com>
Signed-off-by: Lizhi Zhou <1432185+reasonsolo@users.noreply.github.com>
Signed-off-by: Lizhi Zhou <1432185+reasonsolo@users.noreply.github.com>
Signed-off-by: Lizhi Zhou <1432185+reasonsolo@users.noreply.github.com>
Signed-off-by: Lizhi Zhou <1432185+reasonsolo@users.noreply.github.com>
reasonsolo
requested review from
EmmaQiaoCh,
QiJune,
ZhanruiSunCh,
asfiyab-nvidia,
brnguyen2,
chuangz0,
fredricz-20070104,
laikhtewari,
nv-xtf and
yufeiwu-nv
July 16, 2026 09:25
Separate HTTP frontends from generator ownership while preserving shared serving state and reuse the fleet supervisor for disaggregated orchestration. Signed-off-by: Lizhi Zhou <1432185+reasonsolo@users.noreply.github.com>
reasonsolo
force-pushed
the
feat/http_fleet
branch
from
July 16, 2026 09:33
4d41d3f to
fb31828
Compare
Contributor
📝 WalkthroughWalkthroughDisaggregated serving now supports coordinator-backed routing, shared request IDs, multi-process HTTP fleets, generator IPC proxies, external coordinators, and expanded unit and integration coverage for routing, lifecycle, concurrency, and multi-orchestrator deployments. ChangesDisaggregated serving
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 2 | ❌ 3❌ Failed checks (2 warnings, 1 inconclusive)
✅ Passed checks (2 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 20
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
🟡 Minor comments (10)
tests/integration/defs/disaggregated/test_configs/disagg_config_multi_orchestrator.yaml-1-1 (1)
1-1: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winAdd the required NVIDIA copyright header.
This new YAML file starts directly with configuration content. As per coding guidelines, “Add the NVIDIA copyright header to all new files.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/integration/defs/disaggregated/test_configs/disagg_config_multi_orchestrator.yaml` at line 1, Add the standard NVIDIA copyright header at the beginning of the new YAML configuration file, before the hostname setting, using the repository’s established header format for YAML files.Source: Coding guidelines
tests/unittest/disaggregated/test_coordinator_worker.py-113-115 (1)
113-115: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winClose the HTTP server and join its thread during teardown.
shutdown()stopsserve_forever()but does not close the listening socket. Repeated tests can leak descriptors.Proposed teardown
def __exit__(self, *a): self._httpd.shutdown() + self._httpd.server_close() + self._thread.join(timeout=10)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unittest/disaggregated/test_coordinator_worker.py` around lines 113 - 115, Update the test server context manager’s __exit__ teardown to close the HTTP server after shutdown and join the serving thread before returning. Ensure the listening socket and thread are fully cleaned up so repeated tests do not leak resources.tests/unittest/disaggregated/test_coordinator_e2e.py-77-81 (1)
77-81: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winNarrow these exception handlers to the expected failures.
tests/unittest/disaggregated/test_coordinator_e2e.py#L77-L81andtests/unittest/disaggregated/test_coordinator_worker.py#L65-L69: catch theKeyErrorraised when a collector is already gone, instead of suppressing every exception.tests/unittest/disaggregated/test_coordinator_e2e.py#L245-L250andtests/unittest/disaggregated/test_coordinator_worker.py#L173-L178: catchaiohttp.ClientErrorandasyncio.TimeoutErroronly.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unittest/disaggregated/test_coordinator_e2e.py` around lines 77 - 81, Replace the broad exception handlers in tests/unittest/disaggregated/test_coordinator_e2e.py lines 77-81 and tests/unittest/disaggregated/test_coordinator_worker.py lines 65-69 with KeyError handling for already-unregistered collectors. In tests/unittest/disaggregated/test_coordinator_e2e.py lines 245-250 and tests/unittest/disaggregated/test_coordinator_worker.py lines 173-178, catch only aiohttp.ClientError and asyncio.TimeoutError, preserving the existing cleanup and request behavior.Sources: Coding guidelines, Linters/SAST tools
tests/unittest/serve/test_generator_ipc.py-243-250 (1)
243-250: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winReplace fixed sleeps with condition-based synchronization.
These tests assume IPC processing completes within exactly
0.1s, making them flaky on loaded CI runners. Poll the expected abort/queue state with a bounded timeout or expose an event from the fake generator.As per path instructions, test feedback must identify whether coverage is sufficient or needs concrete follow-up.
Also applies to: 267-275
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unittest/serve/test_generator_ipc.py` around lines 243 - 250, Replace fixed asyncio.sleep calls in test_client_disconnect_aborts_outstanding_generation and the corresponding test around the second occurrence with condition-based synchronization: wait for the expected abort or queue state using a bounded timeout, or expose an event from the fake generator. Preserve the existing assertions and ensure both disconnect scenarios are covered; if the state cannot be observed reliably, add the concrete event/state hook needed for follow-up.Source: Path instructions
tests/unittest/disaggregated/test_openai_disagg_service.py-433-439 (1)
433-439: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winAssert that the reserved generation server is preserved.
Coverage is insufficient:
assert_not_awaited()does not provesend_request()received"gen:9001". A regression passingserver=Nonecould trigger downstream rescheduling.gen_call = service._gen_client.send_request.call_args +assert gen_call.kwargs["server"] == "gen:9001" assert gen_call.kwargs["req_id"] == 101As per path instructions, test feedback must identify whether coverage is sufficient or needs concrete follow-up.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unittest/disaggregated/test_openai_disagg_service.py` around lines 433 - 439, Test coverage needs a concrete follow-up: extend the test around _send_disagg_request to assert that gen_client.send_request received the reserved generation server identifier "gen:9001" in its request arguments, while preserving the existing assertions that _gen_router.get_next_server was not awaited and the request IDs remain correct.Source: Path instructions
tests/unittest/disaggregated/test_disagg_utils.py-246-265 (1)
246-265: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winAdd negative lower-bound validation cases.
Coverage is insufficient for the
[0, limit)contract: regressions acceptingnode_id=-1,process_id=-1, or a negative worker environment value would pass.Proposed coverage
def test_get_global_disagg_request_id_range_validation(): + with pytest.raises(ValueError): + get_global_disagg_request_id(-1, 0) + with pytest.raises(ValueError): + get_global_disagg_request_id(0, -1) with pytest.raises(ValueError): get_global_disagg_request_id(256, 0) def test_worker_local_process_id_range_validation(monkeypatch): + monkeypatch.setenv("TRTLLM_DISAGG_WORKER_PROCESS_ID", "-1") + with pytest.raises(ValueError): + worker_local_process_id()As per path instructions, test feedback must identify whether coverage is sufficient or needs concrete follow-up.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unittest/disaggregated/test_disagg_utils.py` around lines 246 - 265, Add negative lower-bound assertions to test_get_global_disagg_request_id_range_validation for node_id=-1 and process_id=-1, expecting ValueError under the existing [0, limit) contract. In test_worker_local_process_id_range_validation, set TRTLLM_DISAGG_WORKER_PROCESS_ID to a negative value and assert worker_local_process_id() raises ValueError.Source: Path instructions
tests/unittest/disaggregated/test_router.py-103-108 (1)
103-108: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winVerify the retried release payload.
Coverage is insufficient because the test checks only call count. Decode both POST bodies and assert they retain
req_id=123,success=True, and the generation role.Proposed coverage
assert session.post.call_count == 2 +for call in session.post.call_args_list: + payload = msgpack.unpackb(call.kwargs["data"], raw=False) + assert payload == { + "role": "generation", + "req_id": 123, + "success": True, + } sleep.assert_awaited_once()As per path instructions, test feedback must identify whether coverage is sufficient or needs concrete follow-up.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unittest/disaggregated/test_router.py` around lines 103 - 108, Extend the test around router._finish_async to decode and inspect both session.post request bodies rather than checking only call count. Assert each payload preserves req_id=123, success=True, and the generation role, while retaining the existing sleep assertion.Source: Path instructions
tests/unittest/disaggregated/test_openai_disagg_server.py-58-64 (1)
58-64: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winVerify that the proxied response preserves
Content-Type.Coverage is insufficient because the coordinator returns
"application/json", but the test never checks that_proxy_cluster_storage_request()relays it.assert response.status_code == 200 assert response.body == b'{"result":true}' +assert response.headers["content-type"] == "application/json"As per path instructions, test feedback must identify whether coverage is sufficient or needs concrete follow-up.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unittest/disaggregated/test_openai_disagg_server.py` around lines 58 - 64, Test coverage is insufficient because it does not verify that the proxied response preserves the coordinator’s Content-Type. Update the test around _proxy_cluster_storage_request to have the mocked coordinator response include application/json and assert that response.headers contains the same Content-Type value, while retaining the existing status and body assertions.Source: Path instructions
docs/source/features/disagg-serving.md-101-103 (1)
101-103: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winRestore language identifiers on the changed code fences.
docs/source/features/disagg-serving.md#L101-L103: usetextfor the bit-layout block.docs/source/features/disagg-serving.md#L212-L221: usebashfor the curl example.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@docs/source/features/disagg-serving.md` around lines 101 - 103, Restore the language identifiers on both code fences in docs/source/features/disagg-serving.md: mark the bit-layout block at lines 101-103 as text and the curl example at lines 212-221 as bash.Source: Linters/SAST tools
docs/source/features/disagg-serving.md-97-109 (1)
97-109: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winSoften the uniqueness claim and fix the field mapping.
ctx_request_idis the generation routing key, and the snowflake’s uniqueness still depends on distinctnode_id/process_idassignments; rewrite this section so it doesn’t imply a single field or a blanket collision-free guarantee.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@docs/source/features/disagg-serving.md` around lines 97 - 109, Rewrite the disaggregated request-ID section to identify ctx_request_id as the generation routing key and accurately describe how it relates to DisaggregatedParams.disagg_request_id. Qualify the snowflake uniqueness claim as dependent on correctly distinct node_id/process_id assignments, and remove any blanket collision-free or unconditional global-uniqueness wording while preserving the documented bit layout and configuration sources.
🧹 Nitpick comments (4)
tests/unittest/disaggregated/test_disagg_utils.py (1)
221-265: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftAdd annotations to all newly introduced test functions and helpers.
tests/unittest/disaggregated/test_disagg_utils.py#L221-L265: annotateget_ids()and new tests, including-> None.tests/unittest/disaggregated/test_openai_disagg_server.py#L35-L50: annotate the async test andreceive().tests/unittest/disaggregated/test_openai_disagg_service.py#L412-L424: annotate the test and_ctx_response().tests/unittest/disaggregated/test_router.py#L80-L93: annotate the test and_response().tests/unittest/serve/test_generator_ipc.py#L70-L150: annotate new methods and helpers throughout the new file.tests/unittest/serve/test_generator_proxy.py#L42-L64: annotate both new tests and fixture parameters.As per coding guidelines, “Annotate every function” and use precise Python 3.10+ types.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unittest/disaggregated/test_disagg_utils.py` around lines 221 - 265, Annotate every newly introduced function, helper, test, async test, fixture, and method with precise Python 3.10+ types, including -> None where appropriate. Apply this to get_ids and the new tests in tests/unittest/disaggregated/test_disagg_utils.py (221-265), the async test and receive in tests/unittest/disaggregated/test_openai_disagg_server.py (35-50), the test and _ctx_response in tests/unittest/disaggregated/test_openai_disagg_service.py (412-424), the test and _response in tests/unittest/disaggregated/test_router.py (80-93), all new methods and helpers in tests/unittest/serve/test_generator_ipc.py (70-150), and both tests plus fixture parameters in tests/unittest/serve/test_generator_proxy.py (42-64).Source: Coding guidelines
tensorrt_llm/serve/generator_ipc.py (2)
717-729: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winDo not silently discard shutdown failures.
Unexpected failures from
promise.abort()or client notification are hidden, obscuring incomplete cancellation and cleanup. Catch expected disconnect/cancellation exceptions specifically and log unexpected failures.As per coding guidelines, catch the narrowest possible exceptions and avoid broad exception handling.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tensorrt_llm/serve/generator_ipc.py` around lines 717 - 729, Update the shutdown cleanup loops in the generator IPC server to catch only expected disconnect or cancellation exceptions from promise.abort() and _send(), while logging unexpected failures with relevant context. Remove the broad silent Exception handlers so incomplete cancellation or client notification errors are visible.Sources: Coding guidelines, Linters/SAST tools
328-329: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd precise return annotations to the new IPC interfaces.
tensorrt_llm/serve/generator_ipc.py#L328-L329: annotate the request-output await iterator.tensorrt_llm/serve/generator_ipc.py#L362-L363: annotate the unary-future await iterator.tensorrt_llm/serve/generator_proxy.py#L166-L170: annotate the dynamically returned async callable.tensorrt_llm/serve/generator_proxy.py#L278-L279: declareRemoteRequestOutput | RemoteUnaryFuture.As per coding guidelines, annotate every function and use precise callable types.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tensorrt_llm/serve/generator_ipc.py` around lines 328 - 329, Annotate the new IPC interfaces with precise return types: in tensorrt_llm/serve/generator_ipc.py lines 328-329, update the request-output await iterator __await__; at lines 362-363, update the unary-future await iterator __await__; in tensorrt_llm/serve/generator_proxy.py lines 166-170, type the dynamically returned async callable; and at lines 278-279, declare the return as RemoteRequestOutput | RemoteUnaryFuture.Source: Coding guidelines
tensorrt_llm/commands/serve.py (1)
1869-1871: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAnnotate the new fleet helper contracts.
Add concrete parameter and return types, including
-> Nonefor non-returning helpers. As per coding guidelines, “Annotate every function.”Also applies to: 1947-1949, 1969-1972, 2048-2048, 2111-2111
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tensorrt_llm/commands/serve.py` around lines 1869 - 1871, Annotate every newly added fleet helper, including _launch_disagg_fleet and the helpers at the referenced locations, with concrete parameter types and explicit return types. Use -> None for helpers that do not return a value, and ensure annotations reflect each function’s actual inputs and outputs without changing behavior.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tensorrt_llm/commands/serve.py`:
- Around line 1982-1996: Update the coordinator endpoint setup around
public_host, coord_port, and coord_url so the unauthenticated CoordinatorServer
never binds to the externally reachable public_host. Bind its TCP listener to
loopback when TCP is used, or use the existing Unix-domain socket exclusively
for internal coordinator traffic; keep the public endpoint only for the
separately required health interface and preserve the coordinator routing
behavior.
- Around line 1960-1964: Update the fleet supervision logic around fleet.wait()
in the serving command to raise a RuntimeError for any worker exit, including
return_code 0, unless shutdown was explicitly initiated. Apply the same behavior
to the corresponding worker-wait handling near the implicit coordinator path,
preserving normal shutdown without raising.
In `@tensorrt_llm/serve/coordinator_server.py`:
- Around line 58-62: Make both ASGI lifespan managers exception-safe by wrapping
their yield bodies in finally blocks. In
tensorrt_llm/serve/coordinator_server.py, update lifespan to always call
self._coordinator.stop(); in tensorrt_llm/serve/openai_disagg_server.py, use
nested finally blocks to always perform service teardown and drain metric tasks,
including when lifespan execution is cancelled or fails.
In `@tensorrt_llm/serve/disagg_coordinator.py`:
- Around line 214-249: Update CoordinatorServer.start and stop to track whether
_ctx_client and _gen_client were created internally when no clients are
injected. During stop, close only those service-owned clients using their client
shutdown API, while leaving externally supplied clients untouched; ensure
cleanup occurs alongside the existing cluster and monitoring shutdown.
- Around line 499-508: Update the readiness flow around is_ready so inference
requests do not perform serial /health and cluster_info coordinator calls on
every request. Add a short-lived cached readiness/topology result or background
synchronization, reuse it from both completion handlers, and ensure the cache
refreshes while preserving unavailable and not-ready outcomes.
- Around line 499-511: Replace broad exception handling with narrowly scoped
transport/timeout and MessagePack decode exceptions in disagg_coordinator.py at
lines 468-479, 499-511, and 530-539, preserving retry and health-response
behavior while allowing unrelated client or coordinator bugs to propagate. In
coordinator_server.py at lines 82-86 and 117-121, catch only malformed-body or
MessagePack decode errors when returning HTTP 400; leave unrelated failures
unhandled.
In `@tensorrt_llm/serve/generator_ipc.py`:
- Around line 178-217: The visual IPC payload handling in
_snapshot_visual_output and _restore_visual_output must use a private storage
directory and mode-0600 files, rather than predictable shared-directory
permissions and direct torch.save output. Create payload files with restrictive
permissions before writing, and add server-side lifecycle cleanup so orphaned
payloads are removed when clients disconnect or workers terminate, while
retaining restoration cleanup for successfully consumed files.
- Around line 625-649: Update _pump and _pump_unary so their
asyncio.CancelledError handlers send a terminal response envelope with status
"cancelled" before re-raising or completing cancellation. Add the corresponding
cancelled branch to RemoteUnaryFuture, matching the existing RemoteRequestOutput
handling, so unary callers resolve cancellation instead of remaining pending.
In `@tensorrt_llm/serve/http_fleet.py`:
- Line 294: Update the frontend setup around bind_reuseport_http_socket and the
subsequent worker-socket creation so that when frontend_config.port is 0, the
reserved port from reservation.getsockname()[1] is assigned back into
frontend_config before frontends bind. Preserve the configured port for nonzero
values and ensure all workers reuse the same reserved ephemeral port.
In `@tensorrt_llm/serve/openai_client.py`:
- Around line 163-175: Update the outer streaming generator around
_post_with_retry and _response_generator so it owns routing-reservation
finalization for the entire request lifetime, including cancellation or failures
before response headers are received. Move or broaden the existing cleanup
finally to cover POST setup and lazy iteration, and guard it so finalization
occurs exactly once without duplicate cleanup in _response_generator.
In `@tensorrt_llm/serve/openai_disagg_service.py`:
- Around line 49-50: Update the type hints in the relevant service definitions
to match their implementations: change client_factory to accept the three
arguments passed at its call site around the service request flow, and annotate
_check_conditional_disagg as returning tuple[str | None, bool] rather than bool.
Keep the runtime behavior unchanged.
In `@tensorrt_llm/serve/openai_server.py`:
- Line 392: Separate observability initialization from the
_owns_generator_lifecycle check in the worker setup around _is_visual_gen.
Ensure every worker initializes or proxies the energy monitor and iteration
buffer used by /energy_metrics and /metrics, while preserving generator
lifecycle ownership for generator-specific state and preventing non-owner
workers from draining shared generator statistics.
In `@tensorrt_llm/serve/openai_video_routes.py`:
- Around line 390-397: The terminal status-update paths around the output.video
failure handling and the additionally referenced completion paths must not use
separate get-then-upsert operations. Replace them with an atomic
update_if_present or compare-and-set operation that only updates an existing
job, or enforce an equivalent deletion tombstone, so concurrent delete_video
calls cannot be followed by background-task recreation.
In `@tensorrt_llm/serve/router_utils.py`:
- Around line 198-228: Initialize a synchronization lock in _init_block_hashing,
then use it to guard all reads, insertions, evictions, and move_to_end
operations on _tokenizers and _tok_prefix_cache in _get_tokenizer and
_encode_with_prefix_cache. Ensure tokenizer creation and cache initialization
cannot occur concurrently, while keeping tokenization outside the lock only if
the cache state is safely rechecked before storing the result.
- Around line 198-212: Update _get_tokenizer to load tokenizer artifacts only
from the configured local or allowlisted tokenizer directory, rather than
falling back to the request-provided model value. Remove trust_remote_code=True
from TransformersTokenizer.from_pretrained and preserve the existing
custom-tokenizer path.
In `@tensorrt_llm/serve/router.py`:
- Around line 1686-1724: The _finish_async method must reconcile local
coordinator state when all terminal /finish attempts fail. Ensure the associated
_req_routing_table or _coord_pending entry is either expired server-side after a
bounded interval or retained in a bounded retry queue and retried until
acknowledgement, so failed finishes cannot leave permanent phantom load.
- Around line 782-789: The routed-block flow around
_apply_routed_blocks_on_route and _finish must not retain speculative hashes
after failed or cancelled requests. Restore success-only block application, or
track speculative blocks per request with reference counts and commit successful
requests while rolling back unsuccessful ones; ensure every failure path removes
or otherwise releases its hashes so false cache hits cannot persist.
- Around line 930-935: Update the server-selection flow around the scores and
matches arrays so conditional disaggregation returns only the cache match for
the selected generation server, rather than aggregating every candidate’s match.
Track or reuse the selected server index and use it to retrieve the
corresponding entry from matches in both affected locations, while preserving
the existing candidate scoring behavior.
- Around line 865-875: Update KvCacheAwareRouter.get_next_server to preserve the
provided req_id and pass it into _route instead of discarding it. Update
finish_request to use req_id when provided, forwarding that identifier to
_finish so conditional disaggregation matches the reservation across different
request objects; retain the existing request-based fallback when no req_id is
supplied.
In `@tests/unittest/disaggregated/test_coordinator_e2e.py`:
- Around line 397-414: Update the drive function’s request orchestration to use
independent concurrent HTTP connections rather than sequential requests on one
persistent ClientSession, ensuring requests can reach multiple Uvicorn workers.
Expose a worker identifier in each response or response metadata, collect it
alongside the generated text, and assert that more than one distinct worker
handled the requests while preserving the GEN_TEXT assertions.
---
Minor comments:
In `@docs/source/features/disagg-serving.md`:
- Around line 101-103: Restore the language identifiers on both code fences in
docs/source/features/disagg-serving.md: mark the bit-layout block at lines
101-103 as text and the curl example at lines 212-221 as bash.
- Around line 97-109: Rewrite the disaggregated request-ID section to identify
ctx_request_id as the generation routing key and accurately describe how it
relates to DisaggregatedParams.disagg_request_id. Qualify the snowflake
uniqueness claim as dependent on correctly distinct node_id/process_id
assignments, and remove any blanket collision-free or unconditional
global-uniqueness wording while preserving the documented bit layout and
configuration sources.
In
`@tests/integration/defs/disaggregated/test_configs/disagg_config_multi_orchestrator.yaml`:
- Line 1: Add the standard NVIDIA copyright header at the beginning of the new
YAML configuration file, before the hostname setting, using the repository’s
established header format for YAML files.
In `@tests/unittest/disaggregated/test_coordinator_e2e.py`:
- Around line 77-81: Replace the broad exception handlers in
tests/unittest/disaggregated/test_coordinator_e2e.py lines 77-81 and
tests/unittest/disaggregated/test_coordinator_worker.py lines 65-69 with
KeyError handling for already-unregistered collectors. In
tests/unittest/disaggregated/test_coordinator_e2e.py lines 245-250 and
tests/unittest/disaggregated/test_coordinator_worker.py lines 173-178, catch
only aiohttp.ClientError and asyncio.TimeoutError, preserving the existing
cleanup and request behavior.
In `@tests/unittest/disaggregated/test_coordinator_worker.py`:
- Around line 113-115: Update the test server context manager’s __exit__
teardown to close the HTTP server after shutdown and join the serving thread
before returning. Ensure the listening socket and thread are fully cleaned up so
repeated tests do not leak resources.
In `@tests/unittest/disaggregated/test_disagg_utils.py`:
- Around line 246-265: Add negative lower-bound assertions to
test_get_global_disagg_request_id_range_validation for node_id=-1 and
process_id=-1, expecting ValueError under the existing [0, limit) contract. In
test_worker_local_process_id_range_validation, set
TRTLLM_DISAGG_WORKER_PROCESS_ID to a negative value and assert
worker_local_process_id() raises ValueError.
In `@tests/unittest/disaggregated/test_openai_disagg_server.py`:
- Around line 58-64: Test coverage is insufficient because it does not verify
that the proxied response preserves the coordinator’s Content-Type. Update the
test around _proxy_cluster_storage_request to have the mocked coordinator
response include application/json and assert that response.headers contains the
same Content-Type value, while retaining the existing status and body
assertions.
In `@tests/unittest/disaggregated/test_openai_disagg_service.py`:
- Around line 433-439: Test coverage needs a concrete follow-up: extend the test
around _send_disagg_request to assert that gen_client.send_request received the
reserved generation server identifier "gen:9001" in its request arguments, while
preserving the existing assertions that _gen_router.get_next_server was not
awaited and the request IDs remain correct.
In `@tests/unittest/disaggregated/test_router.py`:
- Around line 103-108: Extend the test around router._finish_async to decode and
inspect both session.post request bodies rather than checking only call count.
Assert each payload preserves req_id=123, success=True, and the generation role,
while retaining the existing sleep assertion.
In `@tests/unittest/serve/test_generator_ipc.py`:
- Around line 243-250: Replace fixed asyncio.sleep calls in
test_client_disconnect_aborts_outstanding_generation and the corresponding test
around the second occurrence with condition-based synchronization: wait for the
expected abort or queue state using a bounded timeout, or expose an event from
the fake generator. Preserve the existing assertions and ensure both disconnect
scenarios are covered; if the state cannot be observed reliably, add the
concrete event/state hook needed for follow-up.
---
Nitpick comments:
In `@tensorrt_llm/commands/serve.py`:
- Around line 1869-1871: Annotate every newly added fleet helper, including
_launch_disagg_fleet and the helpers at the referenced locations, with concrete
parameter types and explicit return types. Use -> None for helpers that do not
return a value, and ensure annotations reflect each function’s actual inputs and
outputs without changing behavior.
In `@tensorrt_llm/serve/generator_ipc.py`:
- Around line 717-729: Update the shutdown cleanup loops in the generator IPC
server to catch only expected disconnect or cancellation exceptions from
promise.abort() and _send(), while logging unexpected failures with relevant
context. Remove the broad silent Exception handlers so incomplete cancellation
or client notification errors are visible.
- Around line 328-329: Annotate the new IPC interfaces with precise return
types: in tensorrt_llm/serve/generator_ipc.py lines 328-329, update the
request-output await iterator __await__; at lines 362-363, update the
unary-future await iterator __await__; in tensorrt_llm/serve/generator_proxy.py
lines 166-170, type the dynamically returned async callable; and at lines
278-279, declare the return as RemoteRequestOutput | RemoteUnaryFuture.
In `@tests/unittest/disaggregated/test_disagg_utils.py`:
- Around line 221-265: Annotate every newly introduced function, helper, test,
async test, fixture, and method with precise Python 3.10+ types, including ->
None where appropriate. Apply this to get_ids and the new tests in
tests/unittest/disaggregated/test_disagg_utils.py (221-265), the async test and
receive in tests/unittest/disaggregated/test_openai_disagg_server.py (35-50),
the test and _ctx_response in
tests/unittest/disaggregated/test_openai_disagg_service.py (412-424), the test
and _response in tests/unittest/disaggregated/test_router.py (80-93), all new
methods and helpers in tests/unittest/serve/test_generator_ipc.py (70-150), and
both tests plus fixture parameters in
tests/unittest/serve/test_generator_proxy.py (42-64).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 0be5d43b-1a89-4852-bbf6-41da2fa20b9d
📒 Files selected for processing (33)
docs/source/features/disagg-serving.mdtensorrt_llm/commands/serve.pytensorrt_llm/llmapi/disagg_utils.pytensorrt_llm/serve/conversation_id.pytensorrt_llm/serve/coordinator_server.pytensorrt_llm/serve/disagg_coordinator.pytensorrt_llm/serve/generator_ipc.pytensorrt_llm/serve/generator_proxy.pytensorrt_llm/serve/http_fleet.pytensorrt_llm/serve/openai_client.pytensorrt_llm/serve/openai_disagg_server.pytensorrt_llm/serve/openai_disagg_service.pytensorrt_llm/serve/openai_server.pytensorrt_llm/serve/openai_video_routes.pytensorrt_llm/serve/perf_metrics.pytensorrt_llm/serve/responses_utils.pytensorrt_llm/serve/router.pytensorrt_llm/serve/router_utils.pytests/integration/defs/disaggregated/test_configs/disagg_config_multi_orchestrator.yamltests/integration/defs/disaggregated/test_disaggregated.pytests/integration/test_lists/test-db/l0_a10.ymltests/integration/test_lists/test-db/l0_h100.ymltests/unittest/api_stability/references/trtllm_serve_cli.yamltests/unittest/disaggregated/test_coordinator_e2e.pytests/unittest/disaggregated/test_coordinator_worker.pytests/unittest/disaggregated/test_disagg_openai_client.pytests/unittest/disaggregated/test_disagg_utils.pytests/unittest/disaggregated/test_openai_disagg_server.pytests/unittest/disaggregated/test_openai_disagg_service.pytests/unittest/disaggregated/test_router.pytests/unittest/serve/test_generator_ipc.pytests/unittest/serve/test_generator_proxy.pytests/unittest/serve/test_http_fleet.py
Comment on lines
+1960
to
+1964
| worker_id, process, return_code = fleet.wait() | ||
| if return_code != 0: | ||
| raise RuntimeError( | ||
| f"Disagg fleet worker {worker_id} (pid={process.pid}) " | ||
| f"exited with code {return_code}") |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Treat every unexpected fleet-worker exit as a supervisor failure.
An exit code of zero currently stops the external fleet or cancels the implicit coordinator without raising, allowing the command to terminate successfully after losing a required worker. Raise for any worker exit unless shutdown was explicitly initiated.
Also applies to: 2027-2031
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tensorrt_llm/commands/serve.py` around lines 1960 - 1964, Update the fleet
supervision logic around fleet.wait() in the serving command to raise a
RuntimeError for any worker exit, including return_code 0, unless shutdown was
explicitly initiated. Apply the same behavior to the corresponding worker-wait
handling near the implicit coordinator path, preserving normal shutdown without
raising.
Comment on lines
+1982
to
+1996
| public_host, public_port = disagg_cfg.hostname, disagg_cfg.port | ||
| coord_port = int( | ||
| os.environ.get(DisaggWorkerEnvs.TLLM_DISAGG_COORDINATOR_PORT, | ||
| public_port - 1)) | ||
| # The fleet is co-located with the implicit coordinator, so route its hot | ||
| # /select,/finish over a Unix domain socket (avoids the TCP loopback stack | ||
| # that dominated per-request latency). Opt-out via the UDS env = "0". | ||
| use_uds = os.environ.get(DisaggWorkerEnvs.TLLM_DISAGG_COORDINATOR_UDS, | ||
| "1") == "1" | ||
| coord_uds = ( | ||
| f"/tmp/trtllm_disagg_coord_{public_port}.sock" # nosec B108 | ||
| if use_uds else None) | ||
| # Fleet points at the UDS when enabled; the TCP port stays up for health. | ||
| coord_url = f"unix:{coord_uds}" if coord_uds else \ | ||
| f"http://{public_host}:{coord_port}" |
Contributor
There was a problem hiding this comment.
🔒 Security & Privacy | 🔴 Critical | ⚡ Quick win
Keep the unauthenticated coordinator off the public interface.
public_host may be a wildcard or externally reachable address, yet it is reused for the coordinator TCP server. CoordinatorServer exposes mutating /select, /finish, and optional cluster-storage operations without authentication. Bind this internal server to loopback/UDS, or add authentication and network isolation.
Also applies to: 2022-2024
🧰 Tools
🪛 ast-grep (0.44.1)
[info] 1991-1991: Do not hardcode temporary file or directory names
Context: f"/tmp/trtllm_disagg_coord_{public_port}.sock"
Note: [CWE-377] Insecure Temporary File.
(hardcoded-tmp-file)
[warning] 1995-1995: Do not make http calls without encryption
Context: f"http://{public_host}:{coord_port}"
Note: [CWE-319] Cleartext Transmission of Sensitive Information.
(requests-http)
🪛 Ruff (0.15.21)
[error] 1992-1992: Probable insecure usage of temporary file or directory: "/tmp/trtllm_disagg_coord_"
(S108)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tensorrt_llm/commands/serve.py` around lines 1982 - 1996, Update the
coordinator endpoint setup around public_host, coord_port, and coord_url so the
unauthenticated CoordinatorServer never binds to the externally reachable
public_host. Bind its TCP listener to loopback when TCP is used, or use the
existing Unix-domain socket exclusively for internal coordinator traffic; keep
the public endpoint only for the separately required health interface and
preserve the coordinator routing behavior.
Comment on lines
+58
to
+62
| @asynccontextmanager | ||
| async def lifespan(app: FastAPI): | ||
| await self._coordinator.start() | ||
| yield | ||
| await self._coordinator.stop() |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
ASGI lifespan cleanup is not exception-safe. Both lifespan managers place cleanup after an unprotected yield, so cancellation or lifespan failure bypasses shutdown.
tensorrt_llm/serve/coordinator_server.py#L58-L62: callDisaggCoordinatorService.stop()fromfinally.tensorrt_llm/serve/openai_disagg_server.py#L165-L175: call service teardown and drain metric tasks from nestedfinallyblocks.
📍 Affects 2 files
tensorrt_llm/serve/coordinator_server.py#L58-L62(this comment)tensorrt_llm/serve/openai_disagg_server.py#L165-L175
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tensorrt_llm/serve/coordinator_server.py` around lines 58 - 62, Make both
ASGI lifespan managers exception-safe by wrapping their yield bodies in finally
blocks. In tensorrt_llm/serve/coordinator_server.py, update lifespan to always
call self._coordinator.stop(); in tensorrt_llm/serve/openai_disagg_server.py,
use nested finally blocks to always perform service teardown and drain metric
tasks, including when lifespan execution is cancelled or fails.
Comment on lines
+214
to
+249
| async def start(self) -> None: | ||
| await self._ctx_router.prepare_servers() | ||
| await self._gen_router.prepare_servers() | ||
| if self._ctx_client is None or self._gen_client is None: | ||
| self._ctx_client = self._client_factory( | ||
| self._ctx_router, ServerRole.CONTEXT, self._config.max_retries | ||
| ) | ||
| self._gen_client = self._client_factory( | ||
| self._gen_router, ServerRole.GENERATION, self._config.max_retries | ||
| ) | ||
|
|
||
| if self._config.disagg_cluster_config and self._cluster_storage: | ||
| logger.info("Starting disagg cluster manager") | ||
| self._disagg_cluster_manager = DisaggClusterManager( | ||
| self._config.disagg_cluster_config, self._cluster_storage | ||
| ) | ||
| await self._disagg_cluster_manager.start() | ||
| await self._disagg_cluster_manager.watch_workers(on_event=self._on_worker_event) | ||
| logger.info("Disagg cluster manager started") | ||
| else: | ||
| if self._metadata_server and self._metadata_config: | ||
| logger.info("Starting server monitoring via metadata service") | ||
| await self._ctx_router.start_server_monitoring( | ||
| self._metadata_config.refresh_interval | ||
| ) | ||
| await self._gen_router.start_server_monitoring( | ||
| self._metadata_config.refresh_interval | ||
| ) | ||
| await self._wait_for_all_servers_ready() | ||
|
|
||
| async def stop(self) -> None: | ||
| if self._disagg_cluster_manager: | ||
| await self._disagg_cluster_manager.stop() | ||
| if self._metadata_server: | ||
| await self._ctx_router.stop_server_monitoring() | ||
| await self._gen_router.stop_server_monitoring() |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Close readiness clients owned by the coordinator service.
CoordinatorServer calls start() without injecting clients, so Lines 217-223 create two OpenAIHttpClient sessions. stop() never shuts them down, leaking sessions on shutdown or restart. Track client ownership and close only service-created clients.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tensorrt_llm/serve/disagg_coordinator.py` around lines 214 - 249, Update
CoordinatorServer.start and stop to track whether _ctx_client and _gen_client
were created internally when no clients are injected. During stop, close only
those service-owned clients using their client shutdown API, while leaving
externally supplied clients untouched; ensure cleanup occurs alongside the
existing cluster and monitoring shutdown.
Comment on lines
+499
to
+508
| async def is_ready(self) -> bool: | ||
| try: | ||
| async with self.session.get( | ||
| f"{self._remote_url}/health", timeout=self._request_timeout_s | ||
| ) as resp: | ||
| if resp.status != 200: | ||
| return False | ||
| info = await self.cluster_info() | ||
| await self._sync_stateless_routers(info) | ||
| return info.get("is_ready", False) |
Contributor
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
Avoid two coordinator round trips on every inference request.
Both completion handlers call coordinator.is_ready() per request, while this implementation serially fetches /health and /cluster_info. At fleet throughput this adds request latency and makes the coordinator a pre-routing bottleneck. Cache readiness/topology briefly or synchronize it in the background.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tensorrt_llm/serve/disagg_coordinator.py` around lines 499 - 508, Update the
readiness flow around is_ready so inference requests do not perform serial
/health and cluster_info coordinator calls on every request. Add a short-lived
cached readiness/topology result or background synchronization, reuse it from
both completion handlers, and ensure the cache refreshes while preserving
unavailable and not-ready outcomes.
Comment on lines
+782
to
789
| def _apply_routed_blocks_on_route(self, server: str, | ||
| block_hashes: list[list[BlockHash]], | ||
| hash_algo: str) -> None: | ||
| if not self._track_routed_blocks: | ||
| return | ||
| flat = [h for hl in block_hashes for h in hl] | ||
| self._pending_routed_blocks[id(request)] = (flat, hash_algo) | ||
|
|
||
| def _apply_routed_blocks_on_finish(self, request: OpenAIRequest, | ||
| server: Optional[str], | ||
| success: bool) -> None: | ||
| # Pop unconditionally to avoid leaks; apply only when eligible. | ||
| entry = self._pending_routed_blocks.pop(id(request), None) | ||
| if not (self._track_routed_blocks and success): | ||
| return | ||
| if entry is None: | ||
| return | ||
| if server is None or server not in self._server_state: | ||
| return | ||
| flat_block_hashes, hash_algo = entry | ||
| flat_block_hashes = [h for hashes in block_hashes for h in hashes] | ||
| self._server_state[server].add_blocks(flat_block_hashes, | ||
| hash_algo=hash_algo) |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Rollback or defer speculative routed blocks when requests fail.
Blocks are added before the worker stores them, while _finish discards success. A failed or cancelled request therefore leaves hashes that no worker removal event will clear, producing persistent false cache hits. Track speculative blocks per request with reference counts and commit/rollback them on completion, or restore success-only application.
Also applies to: 973-978, 1000-1006
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tensorrt_llm/serve/router.py` around lines 782 - 789, The routed-block flow
around _apply_routed_blocks_on_route and _finish must not retain speculative
hashes after failed or cancelled requests. Restore success-only block
application, or track speculative blocks per request with reference counts and
commit successful requests while rolling back unsuccessful ones; ensure every
failure path removes or otherwise releases its hashes so false cache hits cannot
persist.
Comment on lines
+865
to
+875
| async def get_next_server(self, | ||
| request: OpenAIRequest, | ||
| exclude_server: Optional[str] = None, | ||
| req_id: Optional[int] = None) -> tuple[str, dict]: | ||
| del req_id | ||
| # Standalone (in-process) entry point = routing_key(tokenize+hash) then | ||
| # the shared _route core -- the SAME core the coordinator path uses. | ||
| key = await asyncio.to_thread(self._routing_key_sync, request) | ||
| server, info, _handle = await self._route(key, | ||
| exclude_server=exclude_server, | ||
| request=request) |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm whether _get_gen_request returns a distinct request object.
ast-grep outline tensorrt_llm/serve/openai_disagg_service.py \
--match '_get_gen_request|_check_conditional_disagg' --view expanded
rg -n -C6 'def _get_gen_request|get_next_server\(.*req_id|finish_request\(.*req_id' \
tensorrt_llm/serve tests/unittest/disaggregatedRepository: NVIDIA/TensorRT-LLM
Length of output: 8298
Preserve req_id in the KV-cache router. openai_disagg_service.py passes gen_reservation_id into get_next_server() and finish_request(), but KvCacheAwareRouter.get_next_server() drops it and finish_request() keys state by id(request). Conditional disaggregation creates a new generation request object, so the reservation and finish can miss each other and leave router/load state stuck. Pass req_id through to _route() and use it in _finish() when provided.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tensorrt_llm/serve/router.py` around lines 865 - 875, Update
KvCacheAwareRouter.get_next_server to preserve the provided req_id and pass it
into _route instead of discarding it. Update finish_request to use req_id when
provided, forwarding that identifier to _finish so conditional disaggregation
matches the reservation across different request objects; retain the existing
request-based fallback when no req_id is supplied.
Comment on lines
+930
to
+935
| scores, matches = [], [] | ||
| for i, server in enumerate(servers): | ||
| algo, bh = _hashes(server) | ||
| matches.append(await server_states[server].matched_tokens(bh, algo)) | ||
| scores.append(matches[-1] / self._tokens_per_block - | ||
| self._load_weight * workloads[i]) |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Return the selected server’s cache match, not every candidate’s total.
matches contains one aggregate per candidate server, but conditional disaggregation sums the entire list as the selected generation server’s cache coverage. Matches on other servers can therefore suppress the context phase even when the chosen server lacks those blocks.
Proposed fix
- "matches": matches,
+ "matches": [matches[winner]],Also applies to: 979-985
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tensorrt_llm/serve/router.py` around lines 930 - 935, Update the
server-selection flow around the scores and matches arrays so conditional
disaggregation returns only the cache match for the selected generation server,
rather than aggregating every candidate’s match. Track or reuse the selected
server index and use it to retrieve the corresponding entry from matches in both
affected locations, while preserving the existing candidate scoring behavior.
Comment on lines
+1686
to
+1724
| async def _finish_async(self, req_id: int, success: bool): | ||
| _t0 = time.monotonic() | ||
| for attempt in range(1, COORDINATOR_FINISH_MAX_ATTEMPTS + 1): | ||
| try: | ||
| async with self.session.post( | ||
| f"{self._coordinator_url}/finish", | ||
| data=msgpack.packb( | ||
| { | ||
| "role": self._role, | ||
| "req_id": req_id, | ||
| "success": success | ||
| }, | ||
| use_bin_type=True), | ||
| headers=_MSGPACK_HEADERS, | ||
| timeout=self._request_timeout_s) as resp: | ||
| if resp.status != 200: | ||
| raw_body = await resp.read() | ||
| try: | ||
| body = msgpack.unpackb(raw_body, raw=False) | ||
| except Exception: # noqa: BLE001 | ||
| body = raw_body.decode(errors="replace") | ||
| error = body.get("error", body) if isinstance( | ||
| body, dict) else body | ||
| raise RuntimeError( | ||
| f"coordinator /finish returned {resp.status}: {error}" | ||
| ) | ||
| self._finish_lat.record(time.monotonic() - _t0) | ||
| return | ||
| except Exception as e: # noqa: BLE001 | ||
| if attempt == COORDINATOR_FINISH_MAX_ATTEMPTS: | ||
| logger.warning( | ||
| f"CoordinatorDelegatingRouter finish failed after " | ||
| f"{attempt} attempts: {e}") | ||
| return | ||
| logger.warning( | ||
| f"CoordinatorDelegatingRouter finish attempt {attempt} " | ||
| f"failed: {e}; retrying") | ||
| await asyncio.sleep(COORDINATOR_FINISH_RETRY_DELAY_S * | ||
| (2**(attempt - 1))) |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Reconcile coordinator state after terminal /finish failures.
After three failed attempts, this task returns while _req_routing_table or _coord_pending remains allocated indefinitely. A coordinator outage therefore leaves permanent phantom load and progressively skews stateful routing. Add server-side pending-entry expiry or retain failed finishes in a bounded retry queue until acknowledged.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tensorrt_llm/serve/router.py` around lines 1686 - 1724, The _finish_async
method must reconcile local coordinator state when all terminal /finish attempts
fail. Ensure the associated _req_routing_table or _coord_pending entry is either
expired server-side after a bounded interval or retained in a bounded retry
queue and retried until acknowledgement, so failed finishes cannot leave
permanent phantom load.
Comment on lines
+397
to
+414
| async def drive(): | ||
| # Fire several requests so the kernel spreads them across the | ||
| # 4 uvicorn workers. Every one must round-trip to GEN_TEXT. | ||
| async with aiohttp.ClientSession() as sess: | ||
| texts = [] | ||
| for i in range(8): | ||
| payload = {"model": "m", "prompt": f"hello-{i}", "max_tokens": 8} | ||
| headers = {"X-Session-ID": f"conv-{i}"} | ||
| async with sess.post( | ||
| f"{base}/v1/completions", json=payload, headers=headers, timeout=30 | ||
| ) as r: | ||
| assert r.status == 200, await r.text() | ||
| texts.append((await r.json())["choices"][0]["text"]) | ||
| logger.info(f"request {i} -> {texts[-1]!r}") | ||
| return texts | ||
|
|
||
| texts = asyncio.run(drive()) | ||
| assert all(t == GEN_TEXT for t in texts), texts |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Actually exercise more than one HTTP frontend worker.
These requests are sequential on one persistent ClientSession, so connection reuse can route every request through a single Uvicorn worker. Coverage in tests/unittest/disaggregated/test_coordinator_e2e.py is therefore insufficient for the four-worker claim. Send independent/concurrent connections and expose a worker identifier so the test can assert multiple workers handled requests.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/unittest/disaggregated/test_coordinator_e2e.py` around lines 397 - 414,
Update the drive function’s request orchestration to use independent concurrent
HTTP connections rather than sequential requests on one persistent
ClientSession, ensuring requests can reach multiple Uvicorn workers. Expose a
worker identifier in each response or response metadata, collect it alongside
the generated text, and assert that more than one distinct worker handled the
requests while preserving the GEN_TEXT assertions.
Source: Path instructions
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by CodeRabbit
--num_http_workersoptions for serving commands.Description
Test Coverage
PR Checklist
Please review the following before submitting your PR:
PR description clearly explains what and why. If using CodeRabbit's summary, please make sure it makes sense.
PR Follows TRT-LLM CODING GUIDELINES to the best of your knowledge.
Test cases are provided for new code paths (see test instructions)
If PR introduces API changes, an appropriate PR label is added - either
api-compatibleorapi-breaking. Forapi-breaking, includeBREAKINGin the PR title.Any new dependencies have been scanned for license and vulnerabilities
CODEOWNERS updated if ownership changes
Documentation updated as needed
Update tava architecture diagram if there is a significant design change in PR.
The reviewers assigned automatically/manually are appropriate for the PR.
Please check this after reviewing the above items as appropriate for this PR.
GitHub Bot Help
To see a list of available CI bot commands, please comment
/bot help.