diff --git a/vllm/multimodal/processing/context.py b/vllm/multimodal/processing/context.py index ef9710374d81..bed66d0a4e9d 100644 --- a/vllm/multimodal/processing/context.py +++ b/vllm/multimodal/processing/context.py @@ -268,28 +268,6 @@ def call_hf_processor( try: output = hf_processor(**data, **allowed_kwargs) except Exception as exc: - # See https://github.com/huggingface/tokenizers/issues/537 - if ( - isinstance(exc, RuntimeError) - and exc - and exc.args[0] == "Already borrowed" - and num_tries < max_tries - ): - logger.warning( - "Failed to acquire tokenizer in current thread. " - "Retrying (%d/%d)...", - num_tries, - max_tries, - ) - time.sleep(0.5) - return self.call_hf_processor( - hf_processor, - data, - kwargs, - num_tries=num_tries + 1, - max_tries=max_tries, - ) - msg = ( f"Failed to apply {type(hf_processor).__name__} " f"on data={data} with kwargs={allowed_kwargs}" diff --git a/vllm/renderers/base.py b/vllm/renderers/base.py index 5824957c5b00..41d8c0075fb1 100644 --- a/vllm/renderers/base.py +++ b/vllm/renderers/base.py @@ -1,7 +1,6 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project import asyncio -import copy import time from abc import ABC, abstractmethod from collections.abc import Mapping, Sequence @@ -108,17 +107,10 @@ def __init__(self, config: "VllmConfig", tokenizer: _T | None) -> None: if mm_registry.supports_multimodal_inputs(config.model_config): mm_processor_cache = mm_registry.processor_cache_from_config(config) - # Deep-copy the tokenizer so the multimodal processor gets its - # own Rust tokenizer backend. Without this, concurrent access - # from AsyncMicrobatchTokenizer and call_hf_processor causes - # "RuntimeError: Already borrowed" from the Rust RefCell. - # See: https://github.com/huggingface/tokenizers/issues/537 - mm_tokenizer = copy.deepcopy(tokenizer) - with set_default_torch_num_threads(): self.mm_processor = mm_registry.create_processor( config.model_config, - tokenizer=mm_tokenizer, + tokenizer=self.tokenizer, cache=mm_processor_cache, ) @@ -130,11 +122,10 @@ def __init__(self, config: "VllmConfig", tokenizer: _T | None) -> None: # requests don't pollute the sender cache. ro_cache = mm_registry.processor_only_cache_from_config(config) if ro_cache is not None: - ro_tokenizer = copy.deepcopy(tokenizer) with set_default_torch_num_threads(): self._readonly_mm_processor = mm_registry.create_processor( config.model_config, - tokenizer=ro_tokenizer, + tokenizer=self.tokenizer, cache=ro_cache, ) diff --git a/vllm/renderers/hf.py b/vllm/renderers/hf.py index bff473fa6417..6425bc647a1c 100644 --- a/vllm/renderers/hf.py +++ b/vllm/renderers/hf.py @@ -2,6 +2,7 @@ # SPDX-FileCopyrightText: Copyright contributors to the vLLM project from __future__ import annotations +import copy import inspect import itertools import weakref @@ -42,7 +43,7 @@ apply_token_matches, find_mm_placeholders, ) -from vllm.tokenizers.hf import HfTokenizer +from vllm.tokenizers.hf import HfTokenizer, maybe_make_thread_pool from vllm.transformers_utils.chat_templates import get_chat_template_fallback_path from vllm.transformers_utils.processor import cached_get_processor from vllm.utils.async_utils import make_async @@ -785,6 +786,14 @@ def __init__( config: VllmConfig, tokenizer: HfTokenizer | None, ) -> None: + # Ensure the og tokenizer is never modified by maybe_make_thread_pool + tokenizer = copy.copy(tokenizer) + if ( + # Skip for mock configs and tokenizers + getattr(config.model_config, "enable_prompt_embeds", False) + and isinstance(tokenizer, HfTokenizer) + ): + _ensure_prompt_embeds_placeholder_token(tokenizer) super().__init__(config, tokenizer) self.use_unified_vision_chunk = getattr( @@ -795,6 +804,11 @@ def __init__( safe_apply_chat_template, executor=self._executor ) + if self.tokenizer is not None: + maybe_make_thread_pool( + self.tokenizer, config.model_config.renderer_num_workers + 1 + ) + def render_messages( self, messages: list[ChatCompletionMessageParam], diff --git a/vllm/tokenizers/__init__.py b/vllm/tokenizers/__init__.py index 2daba409881f..6531989a9f35 100644 --- a/vllm/tokenizers/__init__.py +++ b/vllm/tokenizers/__init__.py @@ -1,6 +1,7 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project +from .hf import maybe_make_thread_pool from .protocol import TokenizerLike from .registry import ( TokenizerRegistry, @@ -15,4 +16,5 @@ "cached_get_tokenizer", "get_tokenizer", "cached_tokenizer_from_config", + "maybe_make_thread_pool", ] diff --git a/vllm/tokenizers/hf.py b/vllm/tokenizers/hf.py index 85c812398529..03c71adb8d65 100644 --- a/vllm/tokenizers/hf.py +++ b/vllm/tokenizers/hf.py @@ -2,8 +2,9 @@ # SPDX-FileCopyrightText: Copyright contributors to the vLLM project import contextlib import copy +import queue from pathlib import Path -from typing import TypeAlias +from typing import TypeAlias, TypeVar from transformers import AutoTokenizer, PreTrainedTokenizer, PreTrainedTokenizerFast @@ -12,6 +13,92 @@ from .protocol import TokenizerLike HfTokenizer: TypeAlias = PreTrainedTokenizer | PreTrainedTokenizerFast +_T = TypeVar("_T", bound=TokenizerLike) + + +class ThreadSafeHFTokenizerMixin: + """Mixin class for thread-safe HF fast tokenizers.""" + + pass + + +def maybe_make_thread_pool(tokenizer: _T, copies: int = 1): + """ + If `tokenizer` is a `PreTrainedTokenizerFast`, modify the tokenizer + in-place to make the public interface thread-safe by routing calls + through a deep-copied tokenizer pool. + + Note that: + - Only ``TokenizerLike``'s public interface is thread-safe. + This doesn't include ``_tokenizer`` property nor any mutation + methods like ``add_special_tokens`` or ``add_tokens``. + - Adjacent method calls could happen on different deep copies. + """ + if not isinstance(tokenizer, PreTrainedTokenizerFast) or isinstance( + tokenizer, ThreadSafeHFTokenizerMixin + ): + return tokenizer + + og_tokenizer = copy.copy(tokenizer) + + tokenizer_pool: queue.Queue[PreTrainedTokenizerFast] = queue.Queue() + for _ in range(copies): + tokenizer_pool.put(copy.deepcopy(og_tokenizer)) + + @contextlib.contextmanager + def _borrow_from_pool(): + try: + tok = tokenizer_pool.get_nowait() + yield tok + except queue.Empty: + tok = copy.deepcopy(og_tokenizer) + yield tok + finally: + tokenizer_pool.put(tok) + + class TokenizerPool(tokenizer.__class__, ThreadSafeHFTokenizerMixin): # type: ignore + def apply_chat_template(self, *args, **kwargs): + with _borrow_from_pool() as tok: + return tok.apply_chat_template(*args, **kwargs) + + def batch_decode(self, *args, **kwargs): + with _borrow_from_pool() as tok: + return tok.batch_decode(*args, **kwargs) + + def batch_encode(self, *args, **kwargs): + with _borrow_from_pool() as tok: + return tok.batch_encode(*args, **kwargs) + + def convert_tokens_to_ids(self, *args, **kwargs): + with _borrow_from_pool() as tok: + return tok.convert_tokens_to_ids(*args, **kwargs) + + def convert_ids_to_tokens(self, *args, **kwargs): + with _borrow_from_pool() as tok: + return tok.convert_ids_to_tokens(*args, **kwargs) + + def convert_tokens_to_string(self, *args, **kwargs): + with _borrow_from_pool() as tok: + return tok.convert_tokens_to_string(*args, **kwargs) + + def decode(self, *args, **kwargs): + with _borrow_from_pool() as tok: + return tok.decode(*args, **kwargs) + + def encode(self, *args, **kwargs): + with _borrow_from_pool() as tok: + return tok.encode(*args, **kwargs) + + def __call__(self, *args, **kwargs): + with _borrow_from_pool() as tok: + return tok(*args, **kwargs) + + def __reduce__(self): + return maybe_make_thread_pool, (og_tokenizer, copies) + + TokenizerPool.__name__ = f"TokenizerPool{og_tokenizer.__class__.__name__}" + + tokenizer.__class__ = TokenizerPool def get_cached_tokenizer(tokenizer: HfTokenizer) -> HfTokenizer: