[None][perf] Bypass static transfer admission for async Python Scheduler V2

tensorrt_llm/_torch/disaggregation/transceiver.py

@@ -74,6 +74,10 @@ def _find_consensus_request_ids(request_ids_all_ranks, sync_size):
 
 
 class KvCacheTransceiverV2(KvCacheTransceiver):
+    @property
+    def consumes_transfer_buffer(self) -> bool:
+        return False
+
     def __init__(
         self,
         mapping: Mapping,

tensorrt_llm/_torch/pyexecutor/kv_cache_transceiver.py

@@ -232,6 +232,11 @@ def create_kv_cache_transceiver(
 
 class KvCacheTransceiver(ABC):
 
+    @property
+    def consumes_transfer_buffer(self) -> bool:
+        """Return whether this runtime consumes the C++ CacheTransBuffer budget."""
+        return True
+
     @abstractmethod
     def respond_and_send_async(self, req: LlmRequest):
         raise NotImplementedError

tensorrt_llm/_torch/pyexecutor/py_executor.py

@@ -893,6 +893,16 @@ def on_detected():
                                    None)
         self._disagg_transfer_admission_controller = DisaggTransferAdmissionController(
             max_tokens_in_buffer, tokens_per_block)
+        if (self.global_rank == 0
+                and self._disagg_transfer_admission_controller.enabled()
+                and self._is_disagg_transfer_window_bypass_eligible()):
+            logger.warning_once(
+                f"[PyExecutor] Bypassing the executor transfer window "
+                f"configured by max_tokens_in_buffer={max_tokens_in_buffer} "
+                "for asynchronous Python generation with KV cache manager "
+                "V2 and pp_size=1; "
+                "scheduler KV cache capacity admission remains active.",
+                key="disagg_transfer_window_bypass")
         self.is_benchmark_disagg = (self.benchmark_req_queues_size > 0
                                     and self.kv_cache_transceiver is not None)
         # True while the benchmark disagg fill phase is in progress (waiting
@@ -2618,7 +2628,7 @@ def _executor_loop_pp(self):
                         local_disagg_candidates = getattr(
                             local_scheduler_output,
                             "fitting_disagg_gen_init_requests", [])
-                        self._revert_deferred_disagg_gen_init_alloc(
+                        self._reconcile_disagg_gen_init_allocations(
                             local_disagg_candidates,
                             fitting_disagg_gen_init_requests)
 
@@ -3492,6 +3502,27 @@ def _get_disagg_transfer_admission_controller(
             getattr(kv_cache_manager, "tokens_per_block", None),
         )
 
+    def _is_disagg_transfer_window_bypass_eligible(self) -> bool:
+        """Return whether this runtime may bypass an enabled transfer window."""
+        transceiver = getattr(self, "kv_cache_transceiver", None)
+        return (transceiver is not None
+                and transceiver.consumes_transfer_buffer is False
+                and self._uses_async_disagg_gen_transfer()
+                and self.dist.pp_size == 1 and self._uses_kv_manager_v2())
+
+    def _disagg_transfer_window_is_active(self) -> bool:
+        """Return whether the executor-level transfer window is active.
+
+        ``max_tokens_in_buffer`` describes the C++ transceiver's physical
+        buffer. The asynchronous Python transceiver does not consume that
+        buffer. With KV cache manager V2 and PP1, its generation requests
+        remain constrained by inline scheduler KV admission without a second
+        executor-level budget. Other configurations retain the window.
+        """
+        return (getattr(self, "kv_cache_transceiver", None) is not None
+                and self._get_disagg_transfer_admission_controller().enabled()
+                and not self._is_disagg_transfer_window_bypass_eligible())
+
     @staticmethod
     def _is_disagg_gen_only_no_context_benchmark() -> bool:
         """Return whether ``gen_only_no_context`` skips KV transfer."""
@@ -3519,8 +3550,8 @@ def _apply_disagg_transfer_admission(
             return fitting_disagg_gen_init_requests, False
 
         controller = self._get_disagg_transfer_admission_controller()
-        if not (getattr(self, "kv_cache_transceiver", None)
-                and controller.enabled() and fitting_disagg_gen_init_requests):
+        if not (self._disagg_transfer_window_is_active()
+                and fitting_disagg_gen_init_requests):
             return fitting_disagg_gen_init_requests, False
 
         admission_result = controller.select(self.active_requests,
@@ -3534,29 +3565,36 @@ def _apply_disagg_transfer_admission(
                          f"{admission_result.admitted_transfer_blocks}, "
                          f"budget={controller.max_transfer_blocks}")
 
-        self._revert_deferred_disagg_gen_init_alloc(
+        self._reconcile_disagg_gen_init_allocations(
             fitting_disagg_gen_init_requests,
             admission_result.admitted_requests)
 
         return (admission_result.admitted_requests,
                 admission_result.is_blocked_by_active_transfers())
 
-    def _revert_deferred_disagg_gen_init_alloc(
+    def _reconcile_disagg_gen_init_allocations(
             self, candidates: List[LlmRequest],
-            admitted_requests: List[LlmRequest]) -> None:
+            selected_requests: List[LlmRequest]) -> None:
+        """Revert Scheduler V2 allocations absent from a selected request set.
+
+        Scheduler V2 allocates KV while evaluating generation-init requests.
+        This reconciliation is required both after transfer-window admission
+        and when a PP follower's local candidates differ from the canonical
+        schedule propagated by rank 0.
+        """
         if not (self._uses_kv_manager_v2() and candidates):
             return
 
-        admitted_request_ids = {
+        selected_request_ids = {
             request.py_request_id
-            for request in admitted_requests
+            for request in selected_requests
         }
-        deferred_requests = [
+        unselected_requests = [
             request for request in candidates
-            if request.py_request_id not in admitted_request_ids
+            if request.py_request_id not in selected_request_ids
         ]
-        if deferred_requests:
-            self._revert_ctx_alloc(deferred_requests)
+        if unselected_requests:
+            self._revert_ctx_alloc(unselected_requests)
 
     @staticmethod
     def _dist_size(dist, name: str) -> int:

tests/unittest/_torch/executor/test_disagg_inflight_cancel_gate.py

@@ -644,7 +644,13 @@ def test_flag_unset_preserves_legacy_backend_env_precedence(monkeypatch):
         ("UCX", "CPP", "UCX", False),
     ],
 )
-def test_cpp_capability_is_config_scoped(monkeypatch, backend, runtime, nixl_backend, expected):
+def test_cpp_capability_is_config_scoped(
+    monkeypatch: pytest.MonkeyPatch,
+    backend: str,
+    runtime: str | None,
+    nixl_backend: str | None,
+    expected: bool,
+) -> None:
     if nixl_backend is not None:
         monkeypatch.setenv(transceiver_module._NIXL_KVCACHE_BACKEND_ENV, nixl_backend)
     config = CacheTransceiverConfig(backend=backend, transceiver_runtime=runtime)
@@ -668,15 +674,17 @@ def test_cpp_capability_is_config_scoped(monkeypatch, backend, runtime, nixl_bac
 
     transceiver = BindKvCacheTransceiver(Mock(), dist, kv_cache_manager, Mock(), config)
 
+    assert transceiver.consumes_transfer_buffer
     assert transceiver.supports_inflight_request_cancellation() is expected
     constructor.assert_called_once()
 
 
-def test_python_transceiver_capability_defaults_to_unsupported():
+def test_python_transceiver_capability_defaults_to_unsupported() -> None:
     from tensorrt_llm._torch.disaggregation.transceiver import KvCacheTransceiverV2
 
     transceiver = object.__new__(KvCacheTransceiverV2)
 
+    assert not transceiver.consumes_transfer_buffer
     assert not transceiver.supports_inflight_request_cancellation()
     assert not transceiver.has_poisoned_transfer_buffer()