[VLM] remove AsyncMMDataProcessor wrapper

[yhyang201]

Motivation

Revert #12066

Remove AsyncMMDataProcessor as it provides no real value and has fundamental design flaws.

All multimodal processors inherit from BaseMultiModalProcessor, which defines process_mm_data_async as @abstractmethod. This means:

1. Every processor already implements process_mm_data_async, so AsyncMMDataProcessor always takes the "async" path. The sync fallback (ThreadPoolExecutor + run_in_executor) is dead code — no processor ever triggers it. Moreover, the processors are not thread-safe (HuggingFace processors and tokenizers carry mutable state), making this fallback path not just dead but also incorrect by design.

2. The remaining features provide limited value for most processors:

Feature	Reality
Async non-blocking	Most processors are "fake async" — synchronous CPU work inside async def with no real await points (e.g. gemma3, pixtral), blocking the event loop regardless. Only a few (e.g. llava) use genuine await loop.run_in_executor().
Sync fallback (run_in_executor)	Dead code — all processors implement process_mm_data_async, no processor takes this path
Semaphore concurrency limit	Only effective for processors with real await points; for the majority that block synchronously, the event loop is never yielded so the semaphore has no opportunity to gate concurrency
Timeout (wait_for)	Only effective at await points; for processors that block synchronously, asyncio.wait_for cannot interrupt the blocking work

The two call sites in TokenizerManager now call mm_processor.process_mm_data_async() directly. The associated server args (--mm-max-concurrent-calls, --mm-per-request-timeout) are also removed.

Modifications

Accuracy Tests

Speed Tests and Profiling

Checklist

• Format your code according to the Format code with pre-commit.
• Add unit tests according to the Run and add unit tests.
• Update documentation according to Write documentations.
• Provide accuracy and speed benchmark results according to Test the accuracy and Benchmark the speed.
• Follow the SGLang code style guidance.

Review and Merge Process

1. Ping Merge Oncalls to start the process. See the PR Merge Process.
2. Get approvals from CODEOWNERS and other reviewers.
3. Trigger CI tests with comments or contact authorized users to do so.
   • Common commands include /tag-and-rerun-ci, /tag-run-ci-label, /rerun-failed-ci
4. After green CI and required approvals, ask Merge Oncalls or people with Write permission to merge the PR.

[gemini-code-assist]

Code Review

This pull request removes the AsyncMMDataProcessor utility and its corresponding unit tests, opting instead to call process_mm_data_async directly within the TokenizerManager. This change simplifies the multi-modal data processing pipeline and results in the removal of the mm_max_concurrent_calls and mm_per_request_timeout configuration options from the server arguments and CLI. I have no feedback to provide.

[yhyang201]

/tag-and-rerun-ci

[yuan-luo]

Thanks for the cleanup effort. I agree the sync fallback path is dead code and the ThreadPoolExecutor safety concern is valid. However, I think this PR throws out the baby with the bathwater — specifically the asyncio.wait_for timeout and Semaphore concurrency guard. I'd like to raise some concerns before we merge.

1. The analysis conflates two very different phases of process_mm_data_async

The PR description focuses on the CPU-bound preprocessing phase (HuggingFace image processor, tokenizer) being "fake async," and that's correct for that phase. But a multimodal request's lifecycle has two distinct phases:

Phase	Nature	Latency bound	asyncio.wait_for effective?
Remote resource loading (download image/video from URL)	Network I/O	Unbounded (server hang, network partition)	Depends on implementation
CPU preprocessing (image decode, tokenize)	CPU-bound	Bounded (typically <1s)	No (no yield points)

The remote loading phase is where timeout protection matters most — a hung remote server can block indefinitely, while CPU preprocessing has a predictable upper bound. The PR's argument that "most processors are fake async" only addresses phase 2 and does not consider phase 1.

2. asyncio.wait_for does work for processors with real await points

Several processors in this codebase use await loop.run_in_executor(...) for image loading. For example, LLaVA's processor offloads work to a thread/process pool:

# Evidence from stack traces (sglang/srt/managers/image_processor.py):
image_inputs = await self._process_single_image(images, input_text)
# which internally does:
image_inputs = await loop.run_in_executor(...)

For any processor that uses run_in_executor or genuine async I/O (aiohttp, etc.), the await points are real, and asyncio.wait_for can successfully cancel the coroutine at those points. The original PR #12066 was motivated by a real production issue with Qwen2.5-VL-72B where /metrics was blocked for 10+ seconds — this wasn't a hypothetical concern.

3. Even for "fake async" processors, timeout still prevents GPU resource waste

Consider this scenario without AsyncMMDataProcessor:

1. User sends a VLM request with an image URL pointing to a slow/hung server
2. process_mm_data_async blocks the event loop for 60 seconds downloading the image (synchronous requests.get / urllib.request.urlopen inside async def)
3. After 60 seconds, the download finally completes (or errors)
4. The request proceeds into the prefill queue and consumes GPU compute
5. But the HTTP client has long since timed out and disconnected

With asyncio.wait_for(timeout=10), step 4 would be skipped — the request would be discarded as a TimeoutError immediately after the blocking call returns. This "delayed timeout" doesn't save the event loop (that requires real async), but it does prevent wasting GPU cycles on already-abandoned requests. This matters at scale.

4. The Semaphore provides genuine backpressure for eligible processors

For processors that have even a single await point (during image download), the asyncio.Semaphore(max_concurrent_calls=32) provides real concurrency limiting:

• Prevents flooding the outbound network with too many concurrent image downloads
• Provides graceful degradation under load (queuing excess requests at the semaphore)
• Prevents remote image servers from rate-limiting the sglang instance

After this PR, there is zero backpressure mechanism for multimodal preprocessing. All concurrent requests will attempt to download and process images simultaneously.

5. Removing server args breaks existing deployments

Users who have tuned --mm-max-concurrent-calls and --mm-per-request-timeout for their production VLM workloads will have their configurations silently ignored after upgrading. This is a breaking change that deserves at minimum a deprecation period.

[yuan-luo]

Given the above analysis, instead of a full revert of #12066, I'd suggest a more targeted cleanup:

1. Remove the dead code: Drop the sync fallback path (ThreadPoolExecutor, process_mm_data detection, shutdown(), __del__). This addresses the legitimate design-flaw concern.
2. Keep wait_for + Semaphore: These provide real value for processors with await points and "delayed timeout" value for others.

The simplified AsyncMMDataProcessor would be like following:

class AsyncMMDataProcessor:
    def __init__(self, mm_processor, *, max_concurrent_calls=None, timeout_s=None):
        self.mm_processor = mm_processor
        self.timeout_s = timeout_s
        self.semaphore = (
            asyncio.Semaphore(max_concurrent_calls) if max_concurrent_calls else None
        )

    async def process(self, **kwargs):
        async def _invoke():
            return await self.mm_processor.process_mm_data_async(**kwargs)

        if self.semaphore is not None:
            async with self.semaphore:
                if self.timeout_s is not None:
                    return await asyncio.wait_for(_invoke(), timeout=self.timeout_s)
                return await _invoke()

        if self.timeout_s is not None:
            return await asyncio.wait_for(_invoke(), timeout=self.timeout_s)
        return await _invoke()

This keeps the safety net while removing all the dead code. Removing a partially-effective safety mechanism in favor of no safety mechanism doesn't seem like the right trade-off for a system serving trillions of tokens daily in production.

[yuan-luo]

Here's a concrete example demonstrating where AsyncMMDataProcessor's timeout provides real value.

Setup

Any processor that has a real await point — whether from run_in_executor, aiohttp, asyncio.gather, or even multiple await self._process_single(...) calls — is protected by asyncio.wait_for.

The Scenario

A sglang server serving a VLM model whose processor uses run_in_executor for image loading (e.g., the LLaVA family). A client sends a request with a remote image URL.

# Client request
curl http://sglang-server:30000/v1/chat/completions \
  -d '{
    "model": "llava-v1.6-34b",
    "messages": [{
      "role": "user",
      "content": [
        {"type": "image_url", "image_url": {"url": "https://slow-cdn.example.com/huge-image.jpg"}},
        {"type": "text", "text": "Describe this image"}
      ]
    }]
  }'

The remote CDN is experiencing issues — it accepts the TCP connection but responds extremely slowly (trickles data at ~1KB/s for a 5MB image, so ~80 minutes to complete).

What Happens Inside TokenizerManager

The request enters _tokenize_one_request, which calls process_mm_data_async. Inside the processor:

# Simplified view of what happens inside a processor using run_in_executor
async def process_mm_data_async(self, image_data, ...):
    loop = asyncio.get_running_loop()
    # This is a REAL await point - the event loop is free while the thread runs
    image = await loop.run_in_executor(
        self._executor,
        self._download_and_decode,  # calls requests.get(url) internally
        image_data[0]
    )
    # ... CPU preprocessing ...
    return result

Timeline: WITHOUT AsyncMMDataProcessor (after this PR)

T=0s      Client sends request. TokenizerManager creates async task.
T=0s      process_mm_data_async starts. run_in_executor spawns a thread
          to download the image. The event loop is FREE (this is a real await).
T=0s~     Event loop happily serves other requests. So far so good.

T=30s     Client's HTTP timeout fires. Client disconnects.
          But the download thread is still running. The coroutine is still
          awaiting the Future from run_in_executor. Nobody cancels it.

T=4800s   (~80 min later) Download thread finally finishes.
          The Future resolves. The coroutine resumes.
          CPU preprocessing runs. Result is returned.
          TokenizerManager sends the request to the Scheduler.
          Scheduler allocates KV cache, queues it for prefill.
          GPU runs prefill (1-3 seconds of H100 compute).
          GPU runs decode, generates full response.
          TokenizerManager tries to send response... but the client
          is long gone. All GPU compute wasted.

Impact: The request consumed GPU time and KV cache memory for a response that nobody will ever read. Under load, orphaned requests like this accumulate and degrade throughput for legitimate requests.

Timeline: WITH AsyncMMDataProcessor (before this PR)

T=0s      Client sends request. TokenizerManager creates async task.
T=0s      AsyncMMDataProcessor.process() wraps the call:
            await asyncio.wait_for(_invoke(), timeout=10.0)
T=0s      _invoke() calls process_mm_data_async.
          run_in_executor spawns a thread. Event loop is FREE.

T=10s     asyncio.wait_for timeout fires.
          Since the coroutine is suspended at a real `await` point
          (waiting on the run_in_executor Future), the event loop
          CAN cancel the Task right now.
          → asyncio.TimeoutError is raised.
          → The request is immediately rejected with an error.
          → The Semaphore is released.
          → NO GPU compute is wasted.
          → NO KV cache is allocated for this request.

          (The download thread continues in the background, but its
          result will be discarded when the Future is garbage collected.)

Impact: The system spends 0 GPU seconds on this doomed request. The error propagates back quickly (even though the client already disconnected at T=30s, the server-side cleanup happened at T=10s, freeing resources 4790 seconds earlier).

[yhyang201]

Here's a concrete example demonstrating where AsyncMMDataProcessor's timeout provides real value.

Setup

Any processor that has a real await point — whether from run_in_executor, aiohttp, asyncio.gather, or even multiple await self._process_single(...) calls — is protected by asyncio.wait_for.

The Scenario

A sglang server serving a VLM model whose processor uses run_in_executor for image loading (e.g., the LLaVA family). A client sends a request with a remote image URL.

# Client request
curl http://sglang-server:30000/v1/chat/completions \
  -d '{
    "model": "llava-v1.6-34b",
    "messages": [{
      "role": "user",
      "content": [
        {"type": "image_url", "image_url": {"url": "https://slow-cdn.example.com/huge-image.jpg"}},
        {"type": "text", "text": "Describe this image"}
      ]
    }]
  }'

The remote CDN is experiencing issues — it accepts the TCP connection but responds extremely slowly (trickles data at ~1KB/s for a 5MB image, so ~80 minutes to complete).

What Happens Inside TokenizerManager

The request enters _tokenize_one_request, which calls process_mm_data_async. Inside the processor:

# Simplified view of what happens inside a processor using run_in_executor
async def process_mm_data_async(self, image_data, ...):
    loop = asyncio.get_running_loop()
    # This is a REAL await point - the event loop is free while the thread runs
    image = await loop.run_in_executor(
        self._executor,
        self._download_and_decode,  # calls requests.get(url) internally
        image_data[0]
    )
    # ... CPU preprocessing ...
    return result

Timeline: WITHOUT AsyncMMDataProcessor (after this PR)

T=0s      Client sends request. TokenizerManager creates async task.
T=0s      process_mm_data_async starts. run_in_executor spawns a thread
          to download the image. The event loop is FREE (this is a real await).
T=0s~     Event loop happily serves other requests. So far so good.

T=30s     Client's HTTP timeout fires. Client disconnects.
          But the download thread is still running. The coroutine is still
          awaiting the Future from run_in_executor. Nobody cancels it.

T=4800s   (~80 min later) Download thread finally finishes.
          The Future resolves. The coroutine resumes.
          CPU preprocessing runs. Result is returned.
          TokenizerManager sends the request to the Scheduler.
          Scheduler allocates KV cache, queues it for prefill.
          GPU runs prefill (1-3 seconds of H100 compute).
          GPU runs decode, generates full response.
          TokenizerManager tries to send response... but the client
          is long gone. All GPU compute wasted.

Impact: The request consumed GPU time and KV cache memory for a response that nobody will ever read. Under load, orphaned requests like this accumulate and degrade throughput for legitimate requests.

Timeline: WITH AsyncMMDataProcessor (before this PR)

T=0s      Client sends request. TokenizerManager creates async task.
T=0s      AsyncMMDataProcessor.process() wraps the call:
            await asyncio.wait_for(_invoke(), timeout=10.0)
T=0s      _invoke() calls process_mm_data_async.
          run_in_executor spawns a thread. Event loop is FREE.

T=10s     asyncio.wait_for timeout fires.
          Since the coroutine is suspended at a real `await` point
          (waiting on the run_in_executor Future), the event loop
          CAN cancel the Task right now.
          → asyncio.TimeoutError is raised.
          → The request is immediately rejected with an error.
          → The Semaphore is released.
          → NO GPU compute is wasted.
          → NO KV cache is allocated for this request.

          (The download thread continues in the background, but its
          result will be discarded when the Future is garbage collected.)

Impact: The system spends 0 GPU seconds on this doomed request. The error propagates back quickly (even though the client already disconnected at T=30s, the server-side cleanup happened at T=10s, freeing resources 4790 seconds earlier).

I have a question I’d like to ask: after simplification, is the AsyncMMDataProcessor only applicable to the LLaVA processor?

[yuan-luo]

Thanks for the digging. I did a code audit and want to share what I found.

Code Audit Results

I ran grep -rn "await " python/sglang/srt/multimodal/processors/*.py and examined all await expressions across every processor. Here's the breakdown:

Processors with real yield points (where asyncio.wait_for IS effective):

• llava.py — uses await loop.run_in_executor(...) and await asyncio.gather(...). These are genuine yield points.

Processors with fake-async await (where asyncio.wait_for is NOT effective):

• qwen_vl.py, ernie45_vl.py, interns1pro.py — all await preprocess_video(...), but preprocess_video is declared async def with zero internal await expressions. Pure synchronous CPU work (numpy, video decoding, torch ops) inside an async shell.

So your analysis is largely correct for the current codebase — the timeout/semaphore only provides real protection for llava.py.

The Deeper Problem

However, this audit also revealed something concerning: preprocess_video in qwen_vl.py is async def but performs heavy synchronous video decoding (vr.get_frames_as_tensor(...)) with no await — it blocks the event loop during the entire decode. The same pattern exists in ernie45_vl.py and interns1pro.py. For a long video, this can block the TokenizerManager event loop for seconds, causing the exact same /metrics stall you described in #12066's motivation.

This is the root cause that #12066 tried to address. The AsyncMMDataProcessor wrapper was a mitigation layer, but the real fix should be making these functions truly async — wrapping the heavy work in run_in_executor, like llava.py already does.

My Concern

This PR removes the mitigation layer without fixing the root cause. After this lands:

1. llava.py loses its working timeout/semaphore protection
2. The fake-async processors continue to block the event loop (unchanged)
3. When someone later fixes the fake-async processors (wrapping video decode in run_in_executor), there will be no timeout/semaphore infrastructure to protect them

Would you consider one of these alternatives?

Option A: Keep a minimal wrapper (just wait_for + Semaphore, drop all the dead sync-fallback code), so llava.py stays protected and future processor fixes automatically get protection.

Option B: If you want to remove the wrapper entirely, could we first fix preprocess_video in qwen_vl.py/ernie45_vl.py/interns1pro.py to use run_in_executor (making them truly async like llava.py), and then add timeout protection at the call site in tokenizer_manager.py directly? That way we're not removing protection without also fixing the underlying event-loop-blocking issue.

Either way, the preprocess_video blocking issue should probably be tracked as a separate issue — Qwen-VL video requests are blocking the TokenizerManager event loop today, which is a latent production risk.

[yhyang201]

Thanks for the digging. I did a code audit and want to share what I found.

Code Audit Results

I ran grep -rn "await " python/sglang/srt/multimodal/processors/*.py and examined all await expressions across every processor. Here's the breakdown:

Processors with real yield points (where asyncio.wait_for IS effective):

• llava.py — uses await loop.run_in_executor(...) and await asyncio.gather(...). These are genuine yield points.

Processors with fake-async await (where asyncio.wait_for is NOT effective):

• qwen_vl.py, ernie45_vl.py, interns1pro.py — all await preprocess_video(...), but preprocess_video is declared async def with zero internal await expressions. Pure synchronous CPU work (numpy, video decoding, torch ops) inside an async shell.

So your analysis is largely correct for the current codebase — the timeout/semaphore only provides real protection for llava.py.

The Deeper Problem

However, this audit also revealed something concerning: preprocess_video in qwen_vl.py is async def but performs heavy synchronous video decoding (vr.get_frames_as_tensor(...)) with no await — it blocks the event loop during the entire decode. The same pattern exists in ernie45_vl.py and interns1pro.py. For a long video, this can block the TokenizerManager event loop for seconds, causing the exact same /metrics stall you described in #12066's motivation.

This is the root cause that #12066 tried to address. The AsyncMMDataProcessor wrapper was a mitigation layer, but the real fix should be making these functions truly async — wrapping the heavy work in run_in_executor, like llava.py already does.

My Concern

This PR removes the mitigation layer without fixing the root cause. After this lands:

1. llava.py loses its working timeout/semaphore protection
2. The fake-async processors continue to block the event loop (unchanged)
3. When someone later fixes the fake-async processors (wrapping video decode in run_in_executor), there will be no timeout/semaphore infrastructure to protect them

Would you consider one of these alternatives?

Option A: Keep a minimal wrapper (just wait_for + Semaphore, drop all the dead sync-fallback code), so llava.py stays protected and future processor fixes automatically get protection.

Option B: If you want to remove the wrapper entirely, could we first fix preprocess_video in qwen_vl.py/ernie45_vl.py/interns1pro.py to use run_in_executor (making them truly async like llava.py), and then add timeout protection at the call site in tokenizer_manager.py directly? That way we're not removing protection without also fixing the underlying event-loop-blocking issue.

Either way, the preprocess_video blocking issue should probably be tracked as a separate issue — Qwen-VL video requests are blocking the TokenizerManager event loop today, which is a latent production risk.

Thanks for the thorough audit — the findings align with ours.

Two points:

1. LLaVA timeout protection is preserved in this PR. We added asyncio.wait_for directly inside LlavaImageProcessor._process_single_image, wrapping the run_in_executor call. This is more precise than the external wrapper — it applies the timeout exactly at the real await point, per-image rather than per-request. See the latest commit.

2. The preprocess_video blocking issue is orthogonal to this PR. AsyncMMDataProcessor cannot mitigate it either — qwen_vl.process_mm_data_async() blocks the event loop on self.load_mm_data() → future.result() before preprocess_video is even reached, and preprocess_video itself has zero await points. With or without the wrapper, asyncio.wait_for has no opportunity to fire. The real fix is to make preprocess_video (and load_mm_data) truly async via run_in_executor, which is a separate effort worth tracking as its own issue.

[yuan-luo]

Thanks. Opinions are aligned. Let's remove this wrapper.

[mickqian]

/tag-and-rerun-ci