[None][perf] use native protobuf and streamline gRPC result delivery

docker/Dockerfile.multi

@@ -28,6 +28,8 @@ FROM ${BASE_IMAGE}:${CUDA_COMPAT_TAG} AS cuda_compat
 
 FROM ${BASE_IMAGE}:${BASE_TAG} AS base
 
+ENV PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=upb
+
 # Add NVIDIA EULA and AI Terms labels
 LABEL com.nvidia.eula="https://www.nvidia.com/en-us/agreements/enterprise-software/nvidia-software-license-agreement/"
 LABEL com.nvidia.ai-terms="https://www.nvidia.com/en-us/agreements/enterprise-software/product-specific-terms-for-ai-products/"
@@ -193,6 +195,8 @@ RUN --mount=type=cache,target=/root/.cache/pip --mount=type=cache,target=${CCACH
 
 FROM ${DEVEL_IMAGE} AS release
 
+ENV PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=upb
+
 WORKDIR /app/tensorrt_llm
 ARG GITHUB_MIRROR=""
 RUN --mount=type=cache,target=/root/.cache/pip --mount=type=bind,from=wheel,source=/src/tensorrt_llm/build,target=/tmp/wheel \

docs/source/installation/containers.md

@@ -10,6 +10,17 @@ TensorRT LLM uses a [multi-stage Dockerfile](https://github.com/NVIDIA/TensorRT-
 | **`wheel`** | Intermediate build stage. Extends `devel`, copies the source tree, and compiles the TensorRT LLM wheel. Not published as a standalone image. | -- |
 | **`release`** | Runtime image. Extends `devel`, installs the pre-built wheel from the `wheel` stage, including the OpenEngine gRPC runtime. Ready to use with no further compilation. | [`release`](https://catalog.ngc.nvidia.com/orgs/nvidia/teams/tensorrt-llm/containers/release) |
 
+## Protobuf runtime for gRPC
+
+Native protobuf (`upb`) is recommended for SMG and OpenEngine gRPC throughput.
+The `devel` and `release` images set `PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=upb`.
+For older or custom images, set this variable before starting Python, for example
+with `docker run -e PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=upb ...`.
+The installed protobuf wheel must include the native extension to use `upb`.
+Both gRPC launchers warn when Python protobuf is active and continue startup.
+Native `cpp` is also accepted without a warning. Changing the variable after
+protobuf has been imported does not change the active implementation.
+
 ## Pre-built Images on NGC
 
 The `devel` and `release` images are published to NGC and can be pulled directly:

jenkins/current_image_tags.properties

@@ -13,8 +13,8 @@
 #     images are adopted from PostMerge pipelines, the abbreviated commit hash is used instead.
 IMAGE_NAME=artifactory.nvidia.com/sw-tensorrt-llm-docker-local/tensorrt-llm
 
-LLM_DOCKER_IMAGE=artifactory.nvidia.com/sw-tensorrt-llm-docker-local/tensorrt-llm:pytorch-26.09-py3-x86_64-ubuntu24.04-skip-tritondevel-202610150933-20252
-LLM_SBSA_DOCKER_IMAGE=artifactory.nvidia.com/sw-tensorrt-llm-docker-local/tensorrt-llm:pytorch-26.09-py3-sbsa-ubuntu24.04-skip-tritondevel-202610150933-20252
-LLM_ROCKYLINUX8_PY310_DOCKER_IMAGE=artifactory.nvidia.com/sw-tensorrt-llm-docker-local/tensorrt-llm:cuda-13.4.1-devel-rocky8-x86_64-rocky8-py310-skip-tritondevel-202610150933-20252
-LLM_ROCKYLINUX8_PY312_DOCKER_IMAGE=artifactory.nvidia.com/sw-tensorrt-llm-docker-local/tensorrt-llm:cuda-13.4.1-devel-rocky8-x86_64-rocky8-py312-skip-tritondevel-202610150933-20252
-LLM_SBSA_WHEEL_DOCKER_IMAGE=artifactory.nvidia.com/sw-tensorrt-llm-docker-local/tensorrt-llm:cuda-13.4.1-devel-ubuntu24.04-sbsa-ubuntu24.04-py312-skip-tritondevel-202610150933-20252
+LLM_DOCKER_IMAGE=artifactory.nvidia.com/sw-tensorrt-llm-docker-local/tensorrt-llm-staging/tritondevel:x86_64-tritondevel-torch_skip-3b1f4b8-github-pr-19760-64027
+LLM_SBSA_DOCKER_IMAGE=artifactory.nvidia.com/sw-tensorrt-llm-docker-local/tensorrt-llm-staging/tritondevel:sbsa-tritondevel-torch_skip-3b1f4b8-github-pr-19760-64027
+LLM_ROCKYLINUX8_PY310_DOCKER_IMAGE=artifactory.nvidia.com/sw-tensorrt-llm-docker-local/tensorrt-llm-staging/tritondevel:x86_64-rockylinux8-torch_skip-py310-3b1f4b8-github-pr-19760-64027
+LLM_ROCKYLINUX8_PY312_DOCKER_IMAGE=artifactory.nvidia.com/sw-tensorrt-llm-docker-local/tensorrt-llm-staging/tritondevel:x86_64-rockylinux8-torch_skip-py312-3b1f4b8-github-pr-19760-64027
+LLM_SBSA_WHEEL_DOCKER_IMAGE=artifactory.nvidia.com/sw-tensorrt-llm-docker-local/tensorrt-llm-staging/tritondevel:sbsa-ubuntu24-torch_skip-py312-3b1f4b8-github-pr-19760-64027

tensorrt_llm/grpc/_protobuf.py

@@ -0,0 +1,25 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Protobuf performance guidance shared by the gRPC launchers."""
+
+from tensorrt_llm.logger import logger
+
+
+def _warn_if_python_protobuf() -> None:
+    """Warn for Python protobuf if its implementation can be detected."""
+    try:
+        from google.protobuf.internal import api_implementation
+
+        implementation = api_implementation.Type()
+    except (ImportError, AttributeError):
+        return
+
+    if implementation == "python":
+        logger.warning(
+            "TensorRT-LLM gRPC is using Python protobuf, which can reduce throughput. Set "
+            "PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=upb before starting Python "
+            "and install a supported protobuf wheel with the upb extension. "
+            "Changing the environment after importing protobuf does not change "
+            "the active implementation."
+        )

tensorrt_llm/grpc/openengine/README.md

@@ -7,6 +7,13 @@
 
 OpenEngine support ships in the base TensorRT-LLM package. `pip install "tensorrt_llm[openengine]"` remains available as a compatibility spelling but installs nothing extra.
 
+Native protobuf (`upb`) is recommended for gRPC throughput. TensorRT-LLM images select it
+by default. In older or custom environments, set
+`PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=upb` before starting Python and use a
+protobuf wheel that includes the native extension. Both gRPC launchers warn when
+Python protobuf is active and continue startup. Native `cpp` is also accepted
+without a warning.
+
 Then select OpenEngine when starting the gRPC server:
 
 ```bash

tensorrt_llm/grpc/openengine/server.py

@@ -20,6 +20,7 @@
 import uvloop
 
 from tensorrt_llm import LLM as PyTorchLLM
+from tensorrt_llm.grpc._protobuf import _warn_if_python_protobuf
 from tensorrt_llm.logger import logger
 from tensorrt_llm.serve._frontend_processes import (
     FrontendStartupCancelled,
@@ -210,6 +211,7 @@ def launch_server(
         enable_load_metrics: Enable scheduler-owned routing-load snapshots, independently of KV events.
         subagent_affinity_auth_key: Shared key for authenticated parent affinity hints.
     """
+    _warn_if_python_protobuf()
 
     async def serve() -> None:
         logger.info("Initializing TensorRT-LLM OpenEngine server...")

tensorrt_llm/grpc/smg/server.py

@@ -24,6 +24,7 @@
 import uvloop
 
 from tensorrt_llm import LLM as PyTorchLLM
+from tensorrt_llm.grpc._protobuf import _warn_if_python_protobuf
 from tensorrt_llm.logger import logger
 
 from .bindings import trtllm_service_pb2, trtllm_service_pb2_grpc
@@ -48,6 +49,7 @@ def launch_smg_server(
         served_model_name: Model name returned by discovery RPCs. Defaults to
             the model path.
     """
+    _warn_if_python_protobuf()
     try:
         from grpc_reflection.v1alpha import reflection
     except ModuleNotFoundError as e:

tensorrt_llm/grpc/smg/servicer.py

@@ -421,7 +421,7 @@ def _chunk_responses(
             index = completion.index
             # Use cumulative token_ids and compute delta ourselves
             # because token_ids_diff doesn't clear between iterations for n>1
-            all_tokens = list(completion.token_ids) if completion.token_ids else []
+            all_tokens = completion.token_ids or []
             sent_count = sent_token_counts.get(index, 0)
             delta_tokens = all_tokens[sent_count:]
 
@@ -436,7 +436,7 @@ def _chunk_responses(
                 token_ids=delta_tokens,
                 sequence_index=completion.index,
                 prompt_tokens=len(prompt_token_ids),
-                completion_tokens=len(completion.token_ids) if completion.token_ids else 0,
+                completion_tokens=len(all_tokens),
                 cached_tokens=cached_tokens,
             )
 
@@ -496,7 +496,7 @@ def _complete_responses(
 
         # Process all outputs (for n>1 support)
         for completion in gen_result.outputs:
-            output_tokens = list(completion.token_ids) if completion.token_ids else []
+            output_tokens = completion.token_ids or []
 
             complete = trtllm_service_pb2.GenerateComplete(
                 output_token_ids=output_tokens,

tensorrt_llm/llmapi/utils.py

@@ -484,24 +484,25 @@ def __init__(self,
         self._aq = queue
         self._loop = loop or asyncio.get_event_loop()
 
-    async def _notify(self):
-        self._aq.notify()
-
     def put(self, item) -> None:
         self._aq.put_nowait(item)
 
-        if self._loop.is_running():
-            asyncio.run_coroutine_threadsafe(self._notify(), self._loop)
-        else:
+        if not self._loop.is_running():
             raise AsyncQueue.EventLoopShutdownError()
+        try:
+            self._loop.call_soon_threadsafe(self._aq.notify)
+        except RuntimeError as error:
+            if self._loop.is_closed():
+                raise AsyncQueue.EventLoopShutdownError() from error
+            raise
 
     def put_nowait(self, item) -> None:
         """ Put item without notify the event. """
         self._aq.put_nowait(item)
 
-    # Notify many queues in one coroutine, to cut down context switch overhead.
+    # Notify many queues in one callback, to cut down context switch overhead.
     @staticmethod
-    async def _notify_many(queues: Iterable["_SyncQueue"]):
+    def _notify_many(queues: Iterable["_SyncQueue"]):
         for queue in queues:
             queue._aq.notify()
 
@@ -510,11 +511,15 @@ def notify_many(loop: asyncio.AbstractEventLoop,
                     queues: List["_SyncQueue"]) -> None:
         """ Notify the events in the loop. """
 
-        if loop.is_running():
-            asyncio.run_coroutine_threadsafe(
-                _SyncQueue._notify_many(frozenset(queues)), loop)
-        else:
+        if not loop.is_running():
             raise AsyncQueue.EventLoopShutdownError()
+        try:
+            loop.call_soon_threadsafe(_SyncQueue._notify_many,
+                                      frozenset(queues))
+        except RuntimeError as error:
+            if loop.is_closed():
+                raise AsyncQueue.EventLoopShutdownError() from error
+            raise
 
     @property
     def loop(self) -> asyncio.AbstractEventLoop:

tests/integration/test_lists/test-db/l0_cpu.yml

@@ -179,6 +179,13 @@ l0_cpu:
   - unittest/grpc/openengine/test_node.py
   - unittest/grpc/openengine/test_openengine.py
   - unittest/grpc/openengine/test_servicer.py
+  - unittest/grpc/smg/test_smg.py::test_streaming_deltas_preserve_sequences_logprobs_and_final_tokens
+  - unittest/grpc/smg/test_smg.py::test_streaming_empty_token_history
+  - unittest/grpc/test_grpc_optional.py::test_protobuf_warning_in_fresh_process
+  - unittest/grpc/test_grpc_optional.py::test_cpp_protobuf_does_not_warn
+  - unittest/grpc/test_grpc_optional.py::test_protobuf_warning_skips_unavailable_api
+  - unittest/grpc/test_grpc_optional.py::test_protobuf_warning_does_not_block_model_startup
+  - unittest/grpc/test_grpc_optional.py::test_smg_server_startup_failure_cleans_up
   - unittest/inputs
   - unittest/kv_cache_manager_v2_tests
   - unittest/llmapi/apps/test_anthropic_adapter.py

tests/unittest/disaggregated/test_remoteDictionary.py

@@ -1,3 +1,6 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
 import os
 import signal
 import socket
@@ -48,9 +51,6 @@ def stop_etcd_server(process):
 class TestEtcdDictionary(unittest.TestCase):
 
     def setUp(self):
-        # Set the protocol buffers implementation to python
-        os.environ["PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION"] = "python"
-
         # Start etcd server
         self.etcd_process = start_etcd_server()
 
@@ -76,9 +76,6 @@ def tearDown(self):
         # Stop etcd server
         stop_etcd_server(self.etcd_process)
 
-        # Unset the protocol buffers implementation
-        del os.environ["PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION"]
-
     def _cleanup_test_keys(self):
         # Helper method to remove test keys
         test_keys = [

tests/unittest/grpc/smg/test_smg.py

@@ -17,6 +17,7 @@
 import asyncio
 import io
 import os
+from types import SimpleNamespace
 
 import pytest
 import torch
@@ -54,6 +55,52 @@
 pytestmark = pytest.mark.threadleak(enabled=False)
 
 
+@pytest.mark.cpu_only
+def test_streaming_deltas_preserve_sequences_logprobs_and_final_tokens() -> None:
+    servicer = TrtllmServiceServicer(None)
+    first = SimpleNamespace(index=0, token_ids=[10], logprobs=[-0.5], finish_reason=None)
+    second = SimpleNamespace(index=1, token_ids=[20, 21], logprobs=[-1.0, -1.5], finish_reason=None)
+    result = SimpleNamespace(outputs=[first, second], cached_tokens=2)
+    sent: dict[int, int] = {}
+    initial = servicer._chunk_responses("request", result, [1, 2, 3], sent)
+    initial_bytes = [response.SerializeToString() for response in initial]
+    assert [list(response.chunk.token_ids) for response in initial] == [[10], [20, 21]]
+    assert sent == {0: 1, 1: 2}
+
+    first.token_ids.append(11)
+    first.logprobs.append(-0.25)
+    delta = servicer._chunk_responses("request", result, [1, 2, 3], sent)
+    assert len(delta) == 1
+    assert delta[0].chunk.sequence_index == 0
+    assert list(delta[0].chunk.token_ids) == [11]
+    assert delta[0].chunk.completion_tokens == 2
+    assert delta[0].chunk.prompt_tokens == 3
+    assert delta[0].chunk.cached_tokens == 2
+    assert [(entry.token_id, entry.logprob) for entry in delta[0].chunk.logprobs] == [(11, -0.25)]
+    assert [response.SerializeToString() for response in initial] == initial_bytes
+    assert servicer._chunk_responses("request", result, [1, 2, 3], sent) == []
+
+    first.finish_reason = second.finish_reason = "length"
+    final = servicer._complete_responses("request", result, [1, 2, 3])
+    assert [list(response.complete.output_token_ids) for response in final] == [[10, 11], [20, 21]]
+    assert [response.complete.sequence_index for response in final] == [0, 1]
+    assert all(response.complete.finish_reason == "length" for response in final)
+    final_bytes = [response.SerializeToString() for response in final]
+    first.token_ids.append(12)
+    second.token_ids.clear()
+    assert [response.SerializeToString() for response in final] == final_bytes
+
+
+@pytest.mark.cpu_only
+@pytest.mark.parametrize("token_ids", [None, []])
+def test_streaming_empty_token_history(token_ids: list[int] | None) -> None:
+    servicer = TrtllmServiceServicer(None)
+    result = SimpleNamespace(outputs=[SimpleNamespace(index=0, token_ids=token_ids)])
+    sent: dict[int, int] = {}
+    assert servicer._chunk_responses("request", result, [], sent) == []
+    assert sent == {}
+
+
 @pytest.mark.cpu_only
 class TestSamplingParamsConversion:
     """Tests for proto to SamplingParams conversion."""