[Perf][Frontend] Offload streaming derender detokenization (#57528)

Signed-off-by: Shrey Gajjar <shreygajjar007@gmail.com>

tests/entrypoints/scale_out/derender/test_derender_stream.py

@@ -18,6 +18,8 @@ Tests are split into two layers:
 """
 
 import json
+import threading
+from concurrent.futures import ThreadPoolExecutor
 
 import pytest
 import pytest_asyncio
@@ -214,7 +216,7 @@ def tokenizer():
 
 
 @pytest.fixture(scope="module")
-def derenderer(tokenizer):
+def derenderer(tokenizer, request):
     """Construct a minimal OnlineDerenderer backed by a stub renderer."""
     from unittest.mock import MagicMock
 
@@ -223,6 +225,10 @@ def derenderer(tokenizer):
     renderer = MagicMock()
     renderer.get_tokenizer.return_value = tokenizer
 
+    executor = ThreadPoolExecutor(max_workers=1)
+    request.addfinalizer(executor.shutdown)
+    renderer._executor = executor
+
     model_config = MagicMock()
     model_config.hf_config.model_type = "llama"
     model_config.model = MODEL_NAME
@@ -244,13 +250,11 @@ def derenderer(tokenizer):
 def parsed_derenderer(tokenizer):
     """OnlineDerenderer with `_FakeParser` wired in as `self.parser`.
 
-    The parser configured path runs `_derender_chat_stream_parsed` via
-    `make_async(..., executor=renderer._executor)`, so unlike
-    `derenderer` above (whose streaming paths never touch the executor),
-    `renderer._executor` must be a real `ThreadPoolExecutor` here.
+    The parser-configured path runs `_derender_chat_stream_parsed` via
+    `make_async(..., executor=renderer._executor)`, so
+    `renderer._executor` must be a real `ThreadPoolExecutor`.
     `loop.run_in_executor` cannot submit work to a `MagicMock`.
     """
-    from concurrent.futures import ThreadPoolExecutor
     from unittest.mock import MagicMock
 
     from vllm.renderers.online_derenderer import OnlineDerenderer
@@ -390,6 +394,46 @@ class TestDetokenizeDelta:
         assert results[0] == self._one_shot(tokenizer, token_ids)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "stream_method",
+    ["derender_chat_stream", "derender_completion_stream"],
+)
+async def test_streaming_detokenization_runs_off_event_loop(
+    derenderer, monkeypatch, stream_method
+):
+    """Streaming detokenization runs on the renderer executor."""
+    import vllm.renderers.online_derenderer as online_derenderer_module
+
+    event_loop_thread_id = threading.get_ident()
+    executor_thread_id = derenderer.renderer._executor.submit(
+        threading.get_ident
+    ).result()
+
+    detokenize_thread_id = None
+    original_detokenize = online_derenderer_module.detokenize_incrementally
+
+    def record_detokenize_thread(*args, **kwargs):
+        nonlocal detokenize_thread_id
+        detokenize_thread_id = threading.get_ident()
+        return original_detokenize(*args, **kwargs)
+
+    monkeypatch.setattr(
+        online_derenderer_module,
+        "detokenize_incrementally",
+        record_detokenize_thread,
+    )
+
+    await getattr(derenderer, stream_method)(
+        model=MODEL_NAME,
+        generate_chunk=_make_stream_chunk([10]),
+    )
+
+    assert detokenize_thread_id is not None
+    assert detokenize_thread_id == executor_thread_id
+    assert detokenize_thread_id != event_loop_thread_id
+
+
 class TestDerenderCompletionStream:
     """derender_completion_stream: streaming output parity with one shot."""
 
@@ -875,7 +919,6 @@ def harmony_derenderer(harmony_tokenizer):
     with a real tokenizer and parser, so the replay path is exercised
     against Harmony's actual channel grammar rather than a stub.
     """
-    from concurrent.futures import ThreadPoolExecutor
     from unittest.mock import MagicMock
 
     from vllm.parser.harmony import HarmonyParser

vllm/renderers/online_derenderer.py

@@ -101,8 +101,10 @@ class OnlineDerenderer:
         self._derender_completion_async = make_async(
             self._derender_completion, executor=renderer._executor
         )
-        # Replay is O(n) per chunk (unlike the O(delta) detok-only paths),
-        # so it must not run on the event loop either.
+        self._detokenize_delta_async = make_async(
+            self._detokenize_delta, executor=renderer._executor
+        )
+        # Replay is O(n) per chunk, so it must not run on the event loop.
         self._derender_chat_stream_parsed_async = make_async(
             self._derender_chat_stream_parsed, executor=renderer._executor
         )
@@ -362,7 +364,7 @@ class OnlineDerenderer:
 
         for choice in generate_chunk.choices:
             delta_tids = choice.token_ids or []
-            new_text, updated_state = self._detokenize_delta(
+            new_text, updated_state = await self._detokenize_delta_async(
                 tokenizer, delta_tids, updated_state, skip_special_tokens=skip_special
             )
 
@@ -716,7 +718,7 @@ class OnlineDerenderer:
 
         for choice in generate_chunk.choices:
             delta_tids = choice.token_ids or []
-            new_text, updated_state = self._detokenize_delta(
+            new_text, updated_state = await self._detokenize_delta_async(
                 tokenizer, delta_tids, updated_state, skip_special_tokens=skip_special
             )
             stream_choices.append(