A high-throughput and memory-efficient inference and serving engine for LLMs
[ASR] Optimize CPU preproc to get 2.5x RTFx via multi-threading (#44612)
Signed-off-by: Ekagra Ranjan <3116519+ekagra-ranjan@users.noreply.github.com> Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
Ekagra Ranjan committed
Jun 12, 2026 at 00:05 UTC
e0b9fb12902b0bed54d2f1b866a7ae00b30aa814
4 files changed
+105
-34
vllm/entrypoints/serve/utils/server_utils.py
+7
@@ -474,6 +474,13 @@ async def lifespan(app: FastAPI):
474
finally:
475
if task is not None:
476
task.cancel()
477
+ for attr_name in (
478
+ "openai_serving_transcription",
479
+ "openai_serving_translation",
480
+ ):
481
+ serving = getattr(app.state, attr_name, None)
482
+ if serving is not None and hasattr(serving, "shutdown"):
483
+ serving.shutdown()
484
finally:
485
# Ensure app state including engine ref is gc'd
486
del app.state
vllm/entrypoints/speech_to_text/base/serving.py
+61
-34
@@ -6,6 +6,7 @@ import math
6
import time
7
import zlib
8
from collections.abc import AsyncGenerator, Callable, Set
9
+from concurrent.futures import ThreadPoolExecutor
10
from functools import cached_property
11
from typing import Final, Literal, TypeAlias, TypeVar, cast
12
@@ -37,7 +38,7 @@ from vllm.renderers.inputs import DictPrompt, EncoderDecoderDictPrompt
38
from vllm.renderers.inputs.preprocess import parse_enc_dec_prompt, parse_model_prompt
39
from vllm.sampling_params import BeamSearchParams, SamplingParams
40
from vllm.tokenizers import get_tokenizer
40
-from vllm.utils.async_utils import merge_async_iterators
41
+from vllm.utils.async_utils import make_async_with_semaphore, merge_async_iterators
42
43
from ..transcription.protocol import (
44
TranscriptionResponse,
@@ -63,6 +64,7 @@ T = TypeVar("T", bound=SpeechToTextResponse)
64
V = TypeVar("V", bound=SpeechToTextResponseVerbose)
65
S = TypeVar("S", bound=SpeechToTextSegment)
66
67
+
68
ResponseType: TypeAlias = (
69
TranscriptionResponse
70
| TranslationResponse
@@ -131,6 +133,19 @@ class OpenAISpeechToText(OpenAIServing):
133
self.default_sampling_params,
134
)
135
136
+ # setup preprocess resources
137
+ # we keep separate thread pool for frontend preprocessing instead
138
+ # of reusing the one from Renderer which showed lower throughput
139
+ # https://github.com/vllm-project/vllm/pull/44612#issuecomment-4662757781
140
+ num_audio_preprocess_workers = envs.VLLM_MAX_AUDIO_PREPROCESS_WORKERS
141
+ self._preprocess_executor = ThreadPoolExecutor(
142
+ max_workers=num_audio_preprocess_workers,
143
+ thread_name_prefix="stt-preprocess",
144
+ )
145
+ self._decode_and_chunk_speech_async = make_async_with_semaphore(
146
+ self._decode_and_chunk_speech, executor=self._preprocess_executor
147
+ )
148
+
149
@cached_property
150
def model_cls(self) -> type[SupportsTranscription]:
151
from vllm.model_executor.model_loader import get_model_cls
@@ -138,6 +153,49 @@ class OpenAISpeechToText(OpenAIServing):
153
model_cls = get_model_cls(self.model_config)
154
return cast(type[SupportsTranscription], model_cls)
155
156
+ def shutdown(self) -> None:
157
+ self._preprocess_executor.shutdown(wait=False)
158
+
159
+ def _decode_and_chunk_speech(
160
+ self,
161
+ audio_data: bytes,
162
+ ) -> tuple[list[np.ndarray], float]:
163
+ # Decode audio bytes. For container formats (MP4, M4A, WebM) that
164
+ # soundfile cannot detect from a BytesIO stream, _load_audio_bytes
165
+ # transparently falls back to ffmpeg via an in-memory fd.
166
+ # NOTE resample to model SR here for efficiency. This is also a
167
+ # pre-requisite for chunking, as it assumes Whisper SR.
168
+ try:
169
+ with io.BytesIO(audio_data) as buf:
170
+ y, sr = load_audio(
171
+ buf,
172
+ sr=self.asr_config.sample_rate,
173
+ max_duration_s=self.max_audio_decode_duration_s,
174
+ )
175
+ except Exception as exc:
176
+ raise ValueError("Invalid or unsupported audio file.") from exc
177
+
178
+ duration = get_audio_duration(y=y, sr=sr)
179
+ do_split_audio = self.asr_config.allow_audio_chunking and (
180
+ self.asr_config.max_audio_clip_s is not None
181
+ and duration > self.asr_config.max_audio_clip_s
182
+ )
183
+
184
+ if not do_split_audio:
185
+ chunks = [y]
186
+ else:
187
+ assert self.asr_config.max_audio_clip_s is not None
188
+ assert self.asr_config.min_energy_split_window_size is not None
189
+ chunks = split_audio(
190
+ audio_data=y,
191
+ sample_rate=int(sr),
192
+ max_clip_duration_s=self.asr_config.max_audio_clip_s,
193
+ overlap_duration_s=self.asr_config.overlap_chunk_second,
194
+ min_energy_window_size=self.asr_config.min_energy_split_window_size,
195
+ )
196
+
197
+ return chunks, duration
198
+
199
async def _detect_language(
200
self,
201
audio_chunk: np.ndarray,
@@ -210,39 +268,8 @@ class OpenAISpeechToText(OpenAIServing):
268
value=len(audio_data) / 1024**2,
269
)
270
213
- # Decode audio bytes. For container formats (MP4, M4A, WebM) that
214
- # soundfile cannot detect from a BytesIO stream, _load_audio_bytes
215
- # transparently falls back to ffmpeg via an in-memory fd.
216
- # NOTE resample to model SR here for efficiency. This is also a
217
- # pre-requisite for chunking, as it assumes Whisper SR.
218
- try:
219
- with io.BytesIO(audio_data) as buf:
220
- y, sr = load_audio(
221
- buf,
222
- sr=self.asr_config.sample_rate,
223
- max_duration_s=self.max_audio_decode_duration_s,
224
- )
225
- except Exception as exc:
226
- raise ValueError("Invalid or unsupported audio file.") from exc
227
-
228
- duration = get_audio_duration(y=y, sr=sr)
229
- do_split_audio = self.asr_config.allow_audio_chunking and (
230
- self.asr_config.max_audio_clip_s is not None
231
- and duration > self.asr_config.max_audio_clip_s
232
- )
233
-
234
- if not do_split_audio:
235
- chunks = [y]
236
- else:
237
- assert self.asr_config.max_audio_clip_s is not None
238
- assert self.asr_config.min_energy_split_window_size is not None
239
- chunks = split_audio(
240
- audio_data=y,
241
- sample_rate=int(sr),
242
- max_clip_duration_s=self.asr_config.max_audio_clip_s,
243
- overlap_duration_s=self.asr_config.overlap_chunk_second,
244
- min_energy_window_size=self.asr_config.min_energy_split_window_size,
245
- )
271
+ # Run cpu intensive preprocess step in a separate thread pool executor.
272
+ chunks, duration = await self._decode_and_chunk_speech_async(audio_data)
273
274
if request.language is None and getattr(
275
self.model_cls, "supports_explicit_language_detection", False
vllm/envs.py
+11
@@ -78,6 +78,7 @@ if TYPE_CHECKING:
78
VLLM_MEDIA_LOADING_THREAD_COUNT: int = 8
79
VLLM_MAX_AUDIO_CLIP_FILESIZE_MB: int = 25
80
VLLM_MAX_AUDIO_DECODE_DURATION_S: int = 600
81
+ VLLM_MAX_AUDIO_PREPROCESS_WORKERS: int = max(1, min(os.cpu_count() or 1, 2))
82
VLLM_VIDEO_LOADER_BACKEND: str = "opencv"
83
VLLM_MEDIA_CONNECTOR: str = "http"
84
VLLM_MM_HASHER_ALGORITHM: str = "blake3"
@@ -928,6 +929,15 @@ environment_variables: dict[str, Callable[[], Any]] = {
929
"VLLM_MAX_AUDIO_DECODE_DURATION_S": lambda: int(
930
os.getenv("VLLM_MAX_AUDIO_DECODE_DURATION_S", "600")
931
),
932
+ # Maximum number of worker threads used for STT preprocessing. The default
933
+ # intentionally caps at 2 because that performed best in profiling.
934
+ # https://github.com/vllm-project/vllm/pull/44612#issuecomment-4662757781
935
+ "VLLM_MAX_AUDIO_PREPROCESS_WORKERS": lambda: int(
936
+ os.getenv(
937
+ "VLLM_MAX_AUDIO_PREPROCESS_WORKERS",
938
+ str(max(1, min(os.cpu_count() or 1, 2))),
939
+ )
940
+ ),
941
# Backend for Video IO — selects the frame-sampling algorithm.
942
# - "opencv": uniform sampling.
943
# - "opencv_dynamic": duration-aware dynamic sampling.
@@ -1997,6 +2007,7 @@ def compile_factors() -> dict[str, object]:
2007
"VLLM_MEDIA_LOADING_THREAD_COUNT",
2008
"VLLM_MAX_AUDIO_CLIP_FILESIZE_MB",
2009
"VLLM_MAX_AUDIO_DECODE_DURATION_S",
2010
+ "VLLM_MAX_AUDIO_PREPROCESS_WORKERS",
2011
"VLLM_VIDEO_LOADER_BACKEND",
2012
"VLLM_MEDIA_CONNECTOR",
2013
"VLLM_OBJECT_STORAGE_SHM_BUFFER_NAME",
vllm/utils/async_utils.py
+26
@@ -248,6 +248,32 @@ def make_async(
248
return _async_wrapper
249
250
251
+def make_async_with_semaphore(
252
+ func: Callable[P, T],
253
+ executor: ThreadPoolExecutor,
254
+) -> Callable[P, Awaitable[T]]:
255
+ """
256
+ Take a blocking function, and run it on in an executor thread.
257
+
258
+ This function prevents the blocking function from blocking the
259
+ asyncio event loop.
260
+ The code in this function needs to be thread safe.
261
+
262
+ The function is wrapped in a semaphore to limit the number of
263
+ concurrent executions making it easier to cancel tasks before they start.
264
+ """
265
+
266
+ semaphore = asyncio.Semaphore(executor._max_workers)
267
+
268
+ async def _async_wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
269
+ loop = asyncio.get_event_loop()
270
+ p_func = partial(func, *args, **kwargs)
271
+ async with semaphore:
272
+ return await loop.run_in_executor(executor, p_func)
273
+
274
+ return _async_wrapper
275
+
276
+
277
def run_in_loop(loop: AbstractEventLoop, function: Callable, *args):
278
if in_loop(loop):
279
function(*args)