[None][feat] configure input processor workers

docs/source/commands/trtllm-serve/trtllm-serve.rst

@@ -39,6 +39,24 @@ The following abbreviated command syntax shows the commonly used arguments to st
 
 For the full syntax and argument descriptions, refer to :ref:`syntax`.
 
+Preprocessing Workers
+~~~~~~~~~~~~~~~~~~~~~
+
+Set ``--num_input_processor_workers N`` or use the same setting in ``--config``:
+
+.. code-block:: yaml
+
+   num_input_processor_workers: 8
+
+The default remains eight input-processing threads per frontend for chat and
+completion requests, including text-only models. Zero runs inline on the request
+event loop. Qwen3-VL uses a separate HF processor and tokenizer per thread.
+Additional workers increase host memory use. Offline ``LLM`` calls remain synchronous.
+
+An explicit flag overrides the YAML value, and ``--set
+num_input_processor_workers=N`` has final precedence. Media loading
+uses its separate ``--num_media_load_workers`` pool.
+
 .. _tool-parser-selection:
 
 Tool-call Parser Selection

tensorrt_llm/_torch/models/modeling_qwen3vl.py

@@ -5,6 +5,7 @@
 import dataclasses
 import math
 import re
+import threading
 from functools import lru_cache
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Tuple, Union
 from typing import Mapping as TypingMapping
@@ -258,6 +259,35 @@ def __init__(
         # Qwen3-VL keeps ``torch_dtype`` only on ``text_config`` under transformers 5.x.
         self._dtype = self.config.text_config.dtype
 
+    def configure_preprocessing(self, num_workers: int) -> None:
+        if num_workers > 1:
+            if getattr(self, "_preprocess_thread_state", None) is None:
+                self._preprocess_thread_state = threading.local()
+                self._preprocess_copy_lock = threading.Lock()
+                self._get_preprocessing_state()
+        else:
+            self._preprocess_thread_state = None
+
+    def _get_preprocessing_state(self) -> Optional[threading.local]:
+        state = getattr(self, "_preprocess_thread_state", None)
+        if state is not None and not hasattr(state, "processor"):
+            # HF processors and fast tokenizers can change padding/truncation
+            # state during a call. Each preprocessing thread owns its copies.
+            with self._preprocess_copy_lock:
+                state.processor = copy.deepcopy(self._processor)
+                state.tokenizer = copy.deepcopy(self._tokenizer)
+        return state
+
+    @property
+    def processor(self):
+        state = self._get_preprocessing_state()
+        return self._processor if state is None else state.processor
+
+    @property
+    def tokenizer(self):
+        state = self._get_preprocessing_state()
+        return self._tokenizer if state is None else state.tokenizer
+
     @classmethod
     def _build_temporal_block(
         cls,
@@ -457,7 +487,7 @@ def get_num_tokens_per_video(
         # wrong-formula fallback that would break chunked prefill, so we must
         # run the full processor here.
         do_rescale = not (video and isinstance(video[0], torch.Tensor))
-        processed = self._processor(
+        processed = self.processor(
             text=["<|vision_start|><|video_pad|><|vision_end|>"],
             videos=[video],
             padding=True,

tensorrt_llm/_torch/shared_tensor/shared_tensor.py

@@ -156,6 +156,8 @@ class SharedTensorContainer:
 
     """
 
+    _cpu_storage_lock = threading.Lock()
+
     def __init__(self, method_key: int, tensor_handle: Dict[str, Any]):
         """Initialize the SharedTensorContainer.
 
@@ -452,14 +454,16 @@ def dump_to_dict(self) -> Dict[str, Any]:
             tensor_dict = SharedTensorContainer.cuda_handle_to_dict(
                 self.tensor_handle)
         elif self.method_key == _SharedTensorRebuildMethodRegistry.REBUILD_CPU:
-            sharing_strategy = get_sharing_strategy()
-            # Here we use file_system sharing strategy to make it serializable between two non-python independent processes
-            set_sharing_strategy("file_system")
-            storage = self.tensor_handle[1]
-            meta_data = self.tensor_handle[2]
-            storage_handle = reduce_storage(storage)
-            # restore the original sharing strategy
-            set_sharing_strategy(sharing_strategy)
+            # The sharing strategy also affects preprocessing threads.
+            with self._cpu_storage_lock:
+                sharing_strategy = get_sharing_strategy()
+                try:
+                    set_sharing_strategy("file_system")
+                    storage = self.tensor_handle[1]
+                    meta_data = self.tensor_handle[2]
+                    storage_handle = reduce_storage(storage)
+                finally:
+                    set_sharing_strategy(sharing_strategy)
             # exclude the first element which is the type of the storage
             storage_metadata = storage_handle[-1][1:]
             tensor_dict = SharedTensorContainer.cpu_handle_to_dict(

tensorrt_llm/commands/serve.py

@@ -276,6 +276,7 @@ def get_llm_args(
         enable_chunked_prefill: bool = False,
         enable_attention_dp: bool = False,
         video_pruning_rate: Optional[float] = None,
+        num_input_processor_workers: Optional[int] = None,
         telemetry: bool = True,
         agent_percentage: float = 0.0,
         agent_types: Optional[str] = None,
@@ -363,6 +364,8 @@ def get_llm_args(
         build_cli_config("multimodal_config",
                          video_pruning_rate=video_pruning_rate)
         if video_pruning_rate is not None else None,
+        "num_input_processor_workers":
+        num_input_processor_workers,
         "telemetry_config":
         _telemetry_config.TelemetryConfig(
             disabled=not telemetry,
@@ -530,7 +533,7 @@ def launch_server(
         multimodal_server_config: Optional[MultimodalServerConfig] = None,
         served_model_name: Optional[str] = None,
         allow_request_chat_template: bool = False,
-        num_input_processor_workers: int = 8,
+        num_input_processor_workers: Optional[int] = None,
         num_media_load_workers: int = 8,
         multi_frontend_enabled: bool = True,
         internal_disagg_auth_key: Optional[str] = None,
@@ -544,6 +547,10 @@ def launch_server(
         os.environ[
             DisaggLauncherEnvs.TLLM_DISAGG_ROLE] = server_role.name.lower()
 
+    if num_input_processor_workers is not None:
+        llm_args = update_llm_args_with_extra_dict(
+            llm_args,
+            {"num_input_processor_workers": num_input_processor_workers})
     backend = llm_args["backend"]
     model = served_model_name or llm_args["model"]
 
@@ -642,7 +649,6 @@ def launch_server(
                     multimodal_server_config=multimodal_server_config,
                     chat_template=chat_template,
                     allow_request_chat_template=allow_request_chat_template,
-                    input_processor_workers=num_input_processor_workers,
                     media_load_workers=num_media_load_workers,
                     internal_disagg_auth_key=internal_disagg_auth_key,
                     enable_runtime_control_endpoints=
@@ -1047,11 +1053,11 @@ def launch_visual_gen_server(
     "SO_REUSEPORT (classic IPC executor path only).",
     status="prototype")
 @stability_option("--num_input_processor_workers",
-                  type=click.IntRange(min=1),
-                  default=8,
-                  help="Size of the dedicated thread pool that runs the HF "
-                  "input processor (multimodal preprocess) on the chat "
-                  "and completion endpoints.",
+                  type=click.IntRange(min=0),
+                  default=None,
+                  help="Input-processing threads per frontend for chat and "
+                  "completion requests (default: 8). 0 runs inline. Sets "
+                  "num_input_processor_workers.",
                   status="prototype")
 @stability_option("--num_media_load_workers",
                   type=click.IntRange(min=1),
@@ -1284,7 +1290,7 @@ def serve(
     kv_cache_dtype: str,
     num_postprocess_workers: int,
     num_serve_frontends: int,
-    num_input_processor_workers: int,
+    num_input_processor_workers: Optional[int],
     num_media_load_workers: int,
     trust_remote_code: bool,
     revision: Optional[str],
@@ -1456,6 +1462,7 @@ def _serve_llm():
             enable_attention_dp=enable_attention_dp,
             enable_locality_domains=enable_locality_domains,
             video_pruning_rate=video_pruning_rate,
+            num_input_processor_workers=num_input_processor_workers,
             telemetry=telemetry,
             agent_percentage=agent_percentage,
             agent_types=agent_types,
@@ -1617,7 +1624,6 @@ def _serve_llm():
                 multimodal_server_config,
                 served_model_name=served_model_name,
                 allow_request_chat_template=allow_request_chat_template,
-                num_input_processor_workers=num_input_processor_workers,
                 num_media_load_workers=num_media_load_workers,
                 internal_disagg_auth_key=internal_disagg_auth_key,
                 report_addr=report_addr,

tensorrt_llm/inputs/registry.py

@@ -244,6 +244,9 @@ def __init__(self,
         self._trust_remote_code = trust_remote_code
         self._multimodal_hashing_supported: Optional[bool] = None
 
+    def configure_preprocessing(self, num_workers: int) -> None:
+        """Prepare processor state for the serving frontend's worker count."""
+
     def _init_tokenization_cache(self, enable_tokenization_cache: bool,
                                  tokenizer: PreTrainedTokenizerBase) -> None:
         """Build the cache on `tokenizer` if `enable_tokenization_cache`.

tensorrt_llm/llmapi/llm.py

@@ -2120,6 +2120,9 @@ def _build_model(self):
             trust_remote_code=self.args.trust_remote_code,
             enable_tokenization_cache=self.args.enable_tokenization_cache,
             **input_processor_kwargs)
+        if isinstance(self.input_processor, BaseMultimodalInputProcessor):
+            self.input_processor.configure_preprocessing(
+                self.args.num_input_processor_workers)
         self._tokenizer = self.input_processor.tokenizer
 
         # Resolve encode_only mode (opt-in only)

tensorrt_llm/llmapi/llm_args.py

@@ -6473,6 +6473,16 @@ def _check_removed_sampler_type(value: Any) -> None:
 
 class TorchLlmArgs(BaseLlmArgs):
     # PyTorch backend specific configurations
+    num_input_processor_workers: NonNegativeInt = Field(
+        default=8,
+        description=(
+            "Number of input-processing threads per serving frontend for "
+            "chat and completion requests, including text-only models. "
+            "0 runs inline on the request event loop. Offline LLM calls "
+            "remain synchronous."),
+        status="prototype",
+    )
+
     generation_config: Literal["auto", "trtllm"] = Field(
         default="trtllm",
         description=