[Performance]Qwen3-Omni performance optimization

[amy-why-3459]

PLEASE FILL IN THE PR DESCRIPTION HERE ENSURING ALL CHECKLIST ITEMS (AT THE BOTTOM) HAVE BEEN CONSIDERED.

Purpose

1. Code2Wav does not require _update_request_states during the idle phase.
2. Only pass thinker_hidden_states during the prefill phase.
3. Convert the save_async operation to asynchronous.
4. Modify the method of _update_request_information

#1377

Test Plan

vllm serve Qwen/Qwen3-Omni-30B-A3B-Instruct 
     --omni 
     --port 8091 
     --stage-configs-path /vllm_omni/model_executor/stage_configs/qwen3_omni_moe_async_chunk.yaml

Test Result

vllm bench serve \
    --omni \
  --dataset-name random \
  --port 28889 \
  --max-concurrency 1 \
  --model Qwen/Qwen3-Omni-30B-A3B-Instruct \
  --endpoint /v1/chat/completions \
  --backend openai-chat-omni \
  --num-prompts 1 \
  --random-input-len 2500 \
  --ignore-eos \
  --percentile-metrics ttft,tpot,itl,e2el,audio_ttfp,audio_rtf \
  --random-output-len 900 \
  --extra_body '{"modalities": ["text", "audio"]}'

============ Serving Benchmark Result ============
Successful requests:                     1
Failed requests:                         0
Maximum request concurrency:             1
Benchmark duration (s):                  54.17
Request throughput (req/s):              0.02
Peak concurrent requests:                1.00
----------------End-to-end Latency----------------
Mean E2EL (ms):                          54166.65
Median E2EL (ms):                        54166.65
P99 E2EL (ms):                           54166.65
================== Text Result ===================
Total input tokens:                      2500
Total generated tokens:                  900
Output token throughput (tok/s):         16.61
Peak output token throughput (tok/s):    85.00
Peak concurrent requests:                1.00
Total Token throughput (tok/s):          62.77
---------------Time to First Token----------------
Mean TTFT (ms):                          308.13
Median TTFT (ms):                        308.13
P99 TTFT (ms):                           308.13
-----Time per Output Token (excl. 1st token)------
Mean TPOT (ms):                          13.28
Median TPOT (ms):                        13.28
P99 TPOT (ms):                           13.28
---------------Inter-token Latency----------------
Mean ITL (ms):                           13.28
Median ITL (ms):                         12.24
P99 ITL (ms):                            133.63
================== Audio Result ==================
Total audio duration generated(s):       211.66
Total audio frames generated:            5079900
Audio throughput(audio duration/s):      3.91
---------------Time to First Packet---------------
Mean AUDIO_TTFP (ms):                    1370.63
Median AUDIO_TTFP (ms):                  1370.63
P99 AUDIO_TTFP (ms):                     1370.63
-----------------Real Time Factor-----------------
Mean AUDIO_RTF:                          0.26
Median AUDIO_RTF:                        0.26
P99 AUDIO_RTF:                           0.26
==================================================

---

Essential Elements of an Effective PR Description Checklist

• The purpose of the PR, such as "Fix some issue (link existing issues this PR will resolve)".
• The test plan. Please providing the test scripts & test commands. Please state the reasons if your codes don't require additional test scripts. For test file guidelines, please check the test style doc
• The test results. Please pasting the results comparison before and after, or e2e results.
• (Optional) The necessary documentation update, such as updating supported_models.md and examples for a new model. Please run mkdocs serve to sync the documentation editions to ./docs.
• (Optional) Release notes update. If your change is user facing, please update the release notes draft.

BEFORE SUBMITTING, PLEASE READ https://github.com/vllm-project/vllm-omni/blob/main/CONTRIBUTING.md (anything written below this line will be removed by GitHub Actions)

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 88d82dc03a

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Replace invalid emptiness check on token-id list

In the chunk_id != 0 path, output_token_ids is normalized by _ensure_list(...) immediately before this check, so it is typically a Python list; calling output_token_ids.empty() will raise AttributeError on subsequent async chunks (after the first payload is sent), which aborts thinker2talker_async_chunk and prevents later chunk payloads from being produced.

Useful? React with 👍 / 👎.

[lishunyang12]

Interesting set of changes! The core idea of deferring payload processing from save_async to _send_single_request makes sense -- it keeps the hot path lighter and moves heavy work to the sender thread.

I noticed the PR description is still empty. Could you fill in the Purpose/Test Plan sections? It would really help reviewers understand the motivation and verify correctness. A few comments below.

[lishunyang12]

I was tracing through the change and I think this is a nice improvement -- save_async now only captures the raw pooling_output and request under the lock, and the expensive custom_process_next_stage_input_func call moves to _send_single_request.

However, I noticed that in the original code, save_async would return early if payload_data was falsy (i.e., the processor returned nothing). With the new code, we always enqueue the task, and the early return happens later in _send_single_request. This means the chunk_id is incremented and a task is queued even when the processor will ultimately produce no payload. Could this leave gaps in the chunk_id sequence that downstream consumers might not expect?

[lishunyang12]

I wanted to make sure I understand this change correctly. Previously, thinker_hidden_states and thinker_sequences were sent on every chunk. Now thinker_hidden_states is only sent when output_token_ids is empty (prefill), and thinker_sequences is removed entirely.

Does this mean the talker no longer needs thinker_sequences at all? And for thinker_hidden_states, the talker only needs it once during prefill and can reuse it across subsequent chunks? A brief comment explaining this would help future readers understand the performance rationale.

[amy-why-3459]

We have already assigned all_token_ids to thinker_sequences, so there is no need to add it again later.

[lishunyang12]

I see that _update_request_information replaces calls to _merge_additional_information_update. I was curious about the difference -- does _merge_additional_information_update do a deeper merge (e.g., nested dict merging), while this new method does a shallow dict.update()? If so, could there be cases where nested keys get overwritten instead of merged? If the payloads are always flat dicts, this is fine, but it would be helpful to add a comment noting this assumption.

[amy-why-3459]

We found that _merge_additional_information_update was time-consuming with long inputs, so we replaced it with _update_request_information.

[lishunyang12]

Follow-up pass -- a few more things I noticed on the concurrency side after looking at this again.

[lishunyang12]

The task now stores a live reference to request. When _send_single_request runs later, it reads request.is_finished(), request.all_token_ids, request.prompt_token_ids, etc. Since the request object is mutable, those values could have changed between enqueue time and send time. The old code processed everything eagerly in save_async so this was not an issue.

Is _send_single_request always called quickly enough that this does not matter in practice? Or should we snapshot the fields we need here?

[amy-why-3459]

The request.is_finished() function has been added to the task. prompt_token_ids and all_token_ids are cumulative, so updates to them are fine.

[lishunyang12]

What is the scenario where async_chunk is true but num_scheduled_tokens is 0? Is this guarding against a specific crash, or is it a perf skip? If the scheduler output carries metadata updates (e.g., finished status) but no tokens, would skipping _update_request_states cause staleness?

[amy-why-3459]

This is for performance optimization. When num_scheduled_tokens is 0, it means the model is idling and _update_request_states is unnecessary.

[lishunyang12]

Nit: hasattr + getattr on the same attribute back to back -- you could just do:

info_dict = getattr(req_state, "additional_information_cpu", None)
if info_dict is None:
    return

[hsliuustc0106]

@vllm-omni-reviewer

[github-actions]

🤖 VLLM-Omni PR Review

Code Review: Qwen3-Omni Performance Optimization

1. Overview

This PR aims to optimize performance for Qwen3-Omni by refactoring the transfer adapter's threading model. The key changes include:

• Replacing dictionary-based pending request tracking with deque-based queues
• Removing threading.Lock() synchronization
• Moving payload processing from the main thread to the save loop thread
• Optimizing data transfer by conditionally including thinker_hidden_states

Assessment: Concerns - The removal of locks introduces potential race conditions that need to be addressed before merging.

2. Code Quality

Thread Safety Issues

Critical Issue: Removing the lock without proper analysis of all shared data structures:

1. _finished_load_reqs race condition:

   • Modified in recv_loop thread via _poll_single_request() (line 135)
   • Read in process_pending_chunks() on main thread (line 228-229)
   • Modified via .remove() in _process_chunk_queue() (line 299)
   • No synchronization between these accesses

2. put_req_chunk race condition:

   • chunk_id = self.put_req_chunk[request_id] in _send_single_request (save_loop thread, line 191)
   • self.put_req_chunk[request_id] += 1 in _send_single_request (line 208)
   • Potentially accessed elsewhere without synchronization

3. requests_with_ready_chunks race condition:

   • Modified in _process_chunk_queue() (main thread)
   • Accessed in _clear_chunk_ready()
   • No synchronization

Code Style

• The refactoring improves code readability by simplifying the queue management
• However, the removal of thread safety mechanisms is concerning

3. Architecture & Design

Positive Changes

1. Simpler queue management: Using deque with popleft() is cleaner than the previous dictionary-of-deques approach
2. Deferred processing: Moving payload construction to _send_single_request reduces main thread work

Design Concerns

1. Lock removal strategy: While deque.append() and deque.popleft() are atomic in CPython due to the GIL, this is an implementation detail, not a language guarantee. More importantly, other shared data structures (_finished_load_reqs, put_req_chunk, requests_with_ready_chunks) are not protected.

2. request_ids_mapping assignment moved:

   • Before: Assigned in load_async (main thread) before adding to queue
   • After: Assigned in recv_loop when processing
   • This changes the timing semantics

Recommended Fix

If performance is the goal, consider using queue.Queue which is explicitly thread-safe, or keep minimal locking for non-queue shared state:

# Option 1: Use queue.Queue for thread-safe communication
from queue import Queue
self._pending_load_reqs = Queue()
self._pending_save_reqs = Queue()

# Option 2: Keep lock for shared state, use lock-free for queues
self._state_lock = threading.Lock()  # For _finished_load_reqs, etc.

4. Security & Safety

Resource Management

• The removal of get_finished_requests() changes how finished requests are tracked
• The .remove() operation on _finished_load_reqs in _process_chunk_queue could raise KeyError if the request ID doesn't exist

Suggested fix (chunk_transfer_adapter.py:299):

if request.request_id in finished_load_reqs:
    request.status = target_status
    finished_load_reqs.discard(request.request_id)  # Use discard for safety
    self.requests_with_ready_chunks.add(request.request_id)
    continue

5. Testing & Documentation

Missing Information

The PR description is incomplete:

• ❌ No test plan provided
• ❌ No test results or performance benchmarks
• ❌ No explanation of how correctness is maintained without locks

Required Before Merge

1. Performance benchmarks: What improvement does this optimization provide?
2. Stress testing: Multi-threaded stress tests to verify no race conditions
3. Concurrency tests: Tests that specifically exercise the threading code paths

6. Specific Suggestions

base.py:26-27

# BEFORE
self._pending_load_reqs = deque()
# AFTER - Consider adding a comment about thread safety assumptions
self._pending_load_reqs = deque()  # Thread-safe for append/popleft in CPython

base.py:37

# Removed lock - need to protect other shared state
# Consider adding back a minimal lock for _finished_load_reqs
self._state_lock = threading.Lock()  # For protecting finished sets

chunk_transfer_adapter.py:135

# This modifies _finished_load_reqs without lock
self._finished_load_reqs.add(req_id)
# Should use synchronization if accessed from multiple threads

chunk_transfer_adapter.py:228-229

# Reading _finished_load_reqs from main thread while recv_loop modifies it
self._process_chunk_queue(
    waiting_queue, ..., self._finished_load_reqs  # Race condition here
)

qwen3_omni.py:139-142

This optimization looks reasonable - only include thinker_hidden_states when needed:

if not output_token_ids:
    # When prefilling a chunked thinker, thinker_hidden_states needs to be updated.
    talker_additional_info["thinker_hidden_states"] = pooling_output.get("24").detach().cpu()

However, the comment could be clearer about why this is safe.

gpu_generation_model_runner.py:103

if self.model_config.async_chunk and num_scheduled_tokens:

Good defensive check - prevents unnecessary work when no tokens are scheduled.

7. Approval Status

Changes Requested

The PR introduces potential race conditions by removing locks without ensuring all shared data structures are properly synchronized. While the performance optimization goal is valid, the implementation needs to address:

1. Critical: Thread safety for _finished_load_reqs, put_req_chunk, and requests_with_ready_chunks
2. Important: Add test plan and performance benchmarks to the PR description
3. Minor: Use discard() instead of remove() for safer set operations

The optimization in qwen3_omni.py and the guard in gpu_generation_model_runner.py look good, but the threading changes in the transfer adapter need revision before this can be merged safely.

---

This review was generated automatically by the VLLM-Omni PR Reviewer Bot
using glm-5.

[hsliuustc0106]

This PR optimizes Qwen3-Omni performance by refactoring the async chunk transfer mechanism. Key improvements include moving expensive processing outside locks and optimizing data transfer between stages. However, there are important concerns raised in the review comments about race conditions with mutable request objects and the removal of thinker_sequences that need to be addressed before merging.

[hsliuustc0106]

As noted in the review, storing a live reference to the mutable request object could lead to race conditions. The request's state (is_finished, all_token_ids, etc.) may change between enqueue and send time. Consider capturing immutable snapshots of needed fields at enqueue time instead.

[hsliuustc0106]

Good optimization - moving the expensive custom_process_next_stage_input_func call outside the lock in save_async reduces contention. This should improve throughput.

[hsliuustc0106]

As flagged by Codex, this emptiness check may be problematic. After _ensure_list(), output_token_ids is a list, so not output_token_ids checks for empty list. Verify this is the intended behavior vs checking for None.

[hsliuustc0106]

As suggested in the review, you can simplify this with info_dict = getattr(req_state, "additional_information_cpu", None) and then check if info_dict is None: return. This avoids the redundant hasattr/getattr pattern.

[amy-why-3459]

@Shirley125 @R2-Y PTAL

[amy-why-3459]

@lishunyang12 Please check again.

[amy-why-3459]

@Sy0307 PTAL

[hsliuustc0106]

lgtm

[Sy0307]

This sleep step can be triggered? When 'not ready', your code does this: it calls popleft() (which temporarily empties the queue), the poll fails, and then it immediately calls append() to put the same request back. Sleep can not be triggered forever.