diff --git a/tests/entrypoints/scale_out/derender/test_derender_stream.py b/tests/entrypoints/scale_out/derender/test_derender_stream.py index 4e325db32c..8510d63c4b 100644 --- a/tests/entrypoints/scale_out/derender/test_derender_stream.py +++ b/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 diff --git a/vllm/renderers/online_derenderer.py b/vllm/renderers/online_derenderer.py index 9dfa0c412f..6f7cd10758 100644 --- a/vllm/renderers/online_derenderer.py +++ b/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(