[Frontend][Performance] Resolve async media across modalities concurrently (#54537)
Signed-off-by: waizuichougou <2082431897@qq.com>
waizuichougou committed
Aug 31, 2026 at 18:50 UTC
810bc3250c945829c64a745b4f695ddfd8f9a598
2 files changed
+78
-13
tests/entrypoints/unit_tests/test_chat_utils.py
+55
-1
@@ -2933,10 +2933,62 @@ def test_postprocess_messages_null_arguments_string():
2933
assert tool_calls[0]["function"]["arguments"] == {}
2934
2935
2936
+@pytest.mark.asyncio
2937
+async def test_resolve_items_runs_modalities_concurrently_and_preserves_order():
2938
+ """Media fetches overlap while modality and item order are preserved."""
2939
+
2940
+ active_fetches = 0
2941
+ max_active_fetches = 0
2942
+
2943
+ async def _fetch(name: str, delay: float):
2944
+ nonlocal active_fetches, max_active_fetches
2945
+ active_fetches += 1
2946
+ max_active_fetches = max(max_active_fetches, active_fetches)
2947
+ try:
2948
+ await asyncio.sleep(delay)
2949
+ return name, f"{name}-uuid"
2950
+ finally:
2951
+ active_fetches -= 1
2952
+
2953
+ tracker = AsyncMultiModalItemTracker(MagicMock())
2954
+ tracker._model_config.is_multimodal_model = True
2955
+ tracker.__dict__["mm_processor"] = MagicMock()
2956
+ tracker._items_by_modality["video"] = [
2957
+ lambda: _fetch("video-0", 0.02),
2958
+ lambda: _fetch("video-1", 0.01),
2959
+ ]
2960
+ tracker._items_by_modality["image"] = [
2961
+ lambda: _fetch("image-0", 0.02),
2962
+ lambda: _fetch("image-1", 0.01),
2963
+ ]
2964
+ tracker._items_by_modality["audio"] = [
2965
+ lambda: _fetch("audio-0", 0.02),
2966
+ lambda: _fetch("audio-1", 0.01),
2967
+ ]
2968
+
2969
+ mm_data, mm_uuids = await tracker.resolve_items()
2970
+
2971
+ assert mm_data is not None
2972
+ assert mm_uuids is not None
2973
+ assert max_active_fetches == 6
2974
+ assert list(mm_data) == ["image", "audio", "video"]
2975
+ assert list(mm_uuids) == ["image", "audio", "video"]
2976
+ assert mm_data == {
2977
+ "image": ["image-0", "image-1"],
2978
+ "audio": ["audio-0", "audio-1"],
2979
+ "video": ["video-0", "video-1"],
2980
+ }
2981
+ assert mm_uuids == {
2982
+ "image": ["image-0-uuid", "image-1-uuid"],
2983
+ "audio": ["audio-0-uuid", "audio-1-uuid"],
2984
+ "video": ["video-0-uuid", "video-1-uuid"],
2985
+ }
2986
+
2987
+
2988
@pytest.mark.asyncio
2989
async def test_resolve_items_does_not_leak_tasks_on_partial_failure():
2990
"""Regression test: one failing media fetch must not abandon the other
2939
- still-in-flight fetches in the same modality batch.
2991
+ still-in-flight fetches across modality batches.
2992
2993
Before the fix, `resolve_items` gathered per-modality fetches with plain
2994
`asyncio.gather`, so the first exception propagated immediately while
@@ -2957,6 +3009,8 @@ async def test_resolve_items_does_not_leak_tasks_on_partial_failure():
3009
lambda: _fetch(False, 0.2),
3010
lambda: _fetch(False, 0.2),
3011
]
3012
+ tracker._items_by_modality["audio"] = [lambda: _fetch(False, 0.2)]
3013
+ tracker._items_by_modality["video"] = [lambda: _fetch(False, 0.2)]
3014
3015
tasks_before = asyncio.all_tasks()
3016
with pytest.raises(ValueError, match="simulated fetch failure"):
vllm/entrypoints/chat_utils.py
+23
-12
@@ -871,19 +871,30 @@ class AsyncMultiModalItemTracker(BaseMultiModalItemTracker[_AsyncMultiModalItem]
871
if not self._items_by_modality:
872
return None, None
873
874
+ # Fetch all modalities together. Each tracked item is already an
875
+ # independent awaitable, and the async connector offloads blocking
876
+ # decode work, so waiting for one modality before starting the next
877
+ # needlessly adds their latency.
878
+ # Keep the original group and item order when rebuilding the result.
879
+ item_groups = list(self._items_by_modality.items())
880
+ items = [item for _, group in item_groups for item in group]
881
+ results = await asyncio.gather(
882
+ *(item() for item in items), return_exceptions=True
883
+ )
884
+ for result in results:
885
+ if isinstance(result, BaseException):
886
+ # Gathering with return_exceptions=True lets every task finish
887
+ # (or itself fail) before we raise, instead of abandoning
888
+ # still-in-flight fetches (real network/thread-pool work) the
889
+ # moment the first one fails.
890
+ raise result
891
+
892
resolved_items_by_modality: dict[str, list[Any]] = {}
875
- for modality, items in self._items_by_modality.items():
876
- results = await asyncio.gather(
877
- *(item() for item in items), return_exceptions=True
878
- )
879
- for result in results:
880
- if isinstance(result, BaseException):
881
- # Gathering with return_exceptions=True lets every task in
882
- # this modality finish (or itself fail) before we raise,
883
- # instead of abandoning still-in-flight fetches (real
884
- # network/thread-pool work) the moment the first one fails.
885
- raise result
886
- resolved_items_by_modality[modality] = results
893
+ result_idx = 0
894
+ for modality, group in item_groups:
895
+ next_result_idx = result_idx + len(group)
896
+ resolved_items_by_modality[modality] = results[result_idx:next_result_idx]
897
+ result_idx = next_result_idx
898
899
mm_processor = (
900
self.mm_processor if self._model_config.is_multimodal_model else None