[Fix] tokenizer: Isolate and offload regular fast tokenization

python/sglang/srt/managers/tokenizer_manager.py

@@ -29,9 +29,12 @@
 import time
 from array import array
 from collections import deque
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import nullcontext
+from contextvars import copy_context
 from datetime import datetime
 from enum import Enum
+from functools import partial
 from http import HTTPStatus
 from typing import (
     Any,
@@ -54,6 +57,7 @@
 import zmq
 import zmq.asyncio
 from fastapi import BackgroundTasks
+from transformers import PreTrainedTokenizerFast
 
 from sglang.srt.beam_search.output import (
     build_beam_search_out,
@@ -589,6 +593,9 @@ def __init__(
         # Initialize tokenizer and multimodalprocessor
         self.init_tokenizer_and_processor()
 
+        # Give regular fast tokenization a backend the event loop never uses.
+        self.init_tokenizer_executor()
+
         # Init inter-process communication
         self.init_ipc_channels(port_args)
 
@@ -708,6 +715,40 @@ def _validate_cuda_vmm_feature_transport_support(self) -> None:
                 f"{model_class.__name__}"
             )
 
+    def init_tokenizer_executor(self):
+        self._tokenizer_executor = ThreadPoolExecutor(
+            max_workers=1, thread_name_prefix="sglang-tokenizer"
+        )
+        self._tokenizer_backend_source = None
+        self._tokenizer_backend = None
+        if isinstance(self.tokenizer, PreTrainedTokenizerFast):
+            # Copy once at startup: copying a real vocabulary on the request
+            # path can itself stall the loop. Other tokenizer implementations
+            # retain their synchronous path until they have an isolation contract.
+            self._tokenizer_backend_source = self.tokenizer.backend_tokenizer
+            self._tokenizer_backend = copy.deepcopy(self._tokenizer_backend_source)
+
+    def _regular_tokenizer_for_worker(self):
+        tokenizer = self.tokenizer
+        if (
+            not isinstance(tokenizer, PreTrainedTokenizerFast)
+            or tokenizer.backend_tokenizer is not self._tokenizer_backend_source
+        ):
+            # A replaced tokenizer must not use a stale backend or copy it
+            # during a request. Preserve the existing synchronous behavior.
+            return None
+        # Wrapper attributes (e.g. padding_side) may be configured after init.
+        # Pass this view explicitly; never swap the event loop's tokenizer.
+        worker_tokenizer = copy.copy(tokenizer)
+        worker_tokenizer._tokenizer = self._tokenizer_backend
+        return worker_tokenizer
+
+    async def run_in_tokenizer_executor(self, func, *args, **kwargs):
+        func_call = partial(copy_context().run, func, *args, **kwargs)
+        return await asyncio.get_running_loop().run_in_executor(
+            self._tokenizer_executor, func_call
+        )
+
     def init_ipc_channels(self, port_args: PortArgs):
         context = zmq.asyncio.Context(2)
         self.recv_from_detokenizer = get_zmq_socket(
@@ -1103,14 +1144,22 @@ async def _tokenize_texts(
         else:
             logger.debug(f"Using regular tokenizer for {len(tokenizer_input)} inputs")
 
-            if not is_cross_encoder and (not getattr(self.tokenizer, "is_fast", False)):
-                input_ids = [self.tokenizer.encode(t) for t in tokenizer_input]
-                token_type_ids = None
+            def tokenize_sync(tokenizer):
+                if not is_cross_encoder and (not getattr(tokenizer, "is_fast", False)):
+                    return [tokenizer.encode(t) for t in tokenizer_input], None
+
+                encoded = tokenizer(tokenizer_input, **tokenizer_kwargs)
+                return (
+                    encoded["input_ids"],
+                    encoded.get("token_type_ids") if is_cross_encoder else None,
+                )
+
+            worker_tokenizer = self._regular_tokenizer_for_worker()
+            if worker_tokenizer is None:
+                input_ids, token_type_ids = tokenize_sync(self.tokenizer)
             else:
-                encoded = self.tokenizer(tokenizer_input, **tokenizer_kwargs)
-                input_ids = encoded["input_ids"]
-                token_type_ids = (
-                    encoded.get("token_type_ids") if is_cross_encoder else None
+                input_ids, token_type_ids = await self.run_in_tokenizer_executor(
+                    tokenize_sync, worker_tokenizer
                 )
 
         # vLLM's OpenAI embeddings endpoint includes special tokens for

test/registered/unit/entrypoints/test_http_server_liveness.py

@@ -0,0 +1,121 @@
+"""ASGI integration coverage for HTTP liveness during regular tokenization."""
+
+import asyncio
+import threading
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import httpx
+from tokenizers import Tokenizer
+from tokenizers.models import WordLevel
+from transformers import PreTrainedTokenizerFast
+
+from sglang.test.test_utils import CustomTestCase, maybe_stub_sgl_kernel
+
+maybe_stub_sgl_kernel()
+
+from sglang.srt.entrypoints import http_server  # noqa: E402
+from sglang.srt.managers.tokenizer_manager import (  # noqa: E402
+    ServerStatus,
+    TokenizerManager,
+)
+from sglang.test.ci.ci_register import register_cpu_ci  # noqa: E402
+
+register_cpu_ci(est_time=5, suite="base-a-test-cpu")
+
+
+class _BlockingTokenizer(PreTrainedTokenizerFast):
+    def __init__(self, started, release):
+        backend = Tokenizer(WordLevel({"[UNK]": 0, "hello": 1}, unk_token="[UNK]"))
+        super().__init__(tokenizer_object=backend, unk_token="[UNK]")
+        self.started = started
+        self.release = release
+
+    def __call__(self, texts, **kwargs):
+        self.started.set()
+        if not self.release.wait(timeout=5):
+            raise TimeoutError("test did not release tokenization")
+        return super().__call__(texts, **kwargs)
+
+
+class _TokenizingManager(TokenizerManager):
+    """Exercise real tokenization without launching an inference engine."""
+
+    def __init__(self, started, release):
+        self.tokenizer = _BlockingTokenizer(started, release)
+        self.async_dynamic_batch_tokenizer = None
+        self.model_config = SimpleNamespace(is_embedding_gemma=False)
+        self.server_status = ServerStatus.Up
+        self.gracefully_exit = False
+        if hasattr(self, "init_tokenizer_executor"):
+            self.init_tokenizer_executor()
+
+    async def generate_request(self, obj, request):
+        input_ids, _ = await self._tokenize_texts(obj.text)
+        yield {"input_ids": input_ids}
+
+
+class TestHTTPServerLiveness(CustomTestCase):
+    def test_ping_responds_while_generate_tokenization_is_blocked(self):
+        tokenization_started = threading.Event()
+        ping_completed = threading.Event()
+        release_tokenization = threading.Event()
+        ping_was_starved = threading.Event()
+        tokenization_did_not_start = threading.Event()
+        manager = _TokenizingManager(tokenization_started, release_tokenization)
+
+        def release_after_ping():
+            if not tokenization_started.wait(timeout=2):
+                tokenization_did_not_start.set()
+                release_tokenization.set()
+                return
+            if not ping_completed.wait(timeout=1):
+                ping_was_starved.set()
+            release_tokenization.set()
+
+        watchdog = threading.Thread(target=release_after_ping)
+        watchdog.start()
+
+        async def call_generate_and_ping():
+            transport = httpx.ASGITransport(app=http_server.app)
+            async with httpx.AsyncClient(
+                transport=transport, base_url="http://test"
+            ) as client:
+                generate_task = asyncio.create_task(
+                    client.post("/generate", json={"text": "hello"})
+                )
+                # Queue liveness before tokenization gets its first loop turn.
+                ping_task = asyncio.create_task(client.get("/ping"))
+                ping_response = await ping_task
+                ping_completed.set()
+                generate_response = await generate_task
+                return ping_response, generate_response
+
+        try:
+            with patch.object(
+                http_server,
+                "_global_state",
+                http_server._GlobalState(manager, None, {}),
+            ):
+                ping_response, generate_response = asyncio.run(call_generate_and_ping())
+        finally:
+            release_tokenization.set()
+            watchdog.join(timeout=2)
+            executor = getattr(manager, "_tokenizer_executor", None)
+            if executor is not None:
+                executor.shutdown(wait=True)
+
+        self.assertEqual(ping_response.status_code, 200)
+        self.assertEqual(generate_response.status_code, 200)
+        self.assertEqual(generate_response.json(), {"input_ids": [1]})
+        self.assertFalse(tokenization_did_not_start.is_set())
+        self.assertFalse(watchdog.is_alive())
+        self.assertFalse(
+            ping_was_starved.is_set(),
+            "/ping was starved by synchronous regular tokenization",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()