[Frontend][Performance] Resolve async media across modalities concurrently (vllm-project#54537)

Signed-off-by: waizuichougou <2082431897@qq.com>

Author: waizuichougou

tests/entrypoints/unit_tests/test_chat_utils.py

@@ -2933,10 +2933,62 @@ def test_postprocess_messages_null_arguments_string():
     assert tool_calls[0]["function"]["arguments"] == {}
 
 
+@pytest.mark.asyncio
+async def test_resolve_items_runs_modalities_concurrently_and_preserves_order():
+    """Media fetches overlap while modality and item order are preserved."""
+
+    active_fetches = 0
+    max_active_fetches = 0
+
+    async def _fetch(name: str, delay: float):
+        nonlocal active_fetches, max_active_fetches
+        active_fetches += 1
+        max_active_fetches = max(max_active_fetches, active_fetches)
+        try:
+            await asyncio.sleep(delay)
+            return name, f"{name}-uuid"
+        finally:
+            active_fetches -= 1
+
+    tracker = AsyncMultiModalItemTracker(MagicMock())
+    tracker._model_config.is_multimodal_model = True
+    tracker.__dict__["mm_processor"] = MagicMock()
+    tracker._items_by_modality["video"] = [
+        lambda: _fetch("video-0", 0.02),
+        lambda: _fetch("video-1", 0.01),
+    ]
+    tracker._items_by_modality["image"] = [
+        lambda: _fetch("image-0", 0.02),
+        lambda: _fetch("image-1", 0.01),
+    ]
+    tracker._items_by_modality["audio"] = [
+        lambda: _fetch("audio-0", 0.02),
+        lambda: _fetch("audio-1", 0.01),
+    ]
+
+    mm_data, mm_uuids = await tracker.resolve_items()
+
+    assert mm_data is not None
+    assert mm_uuids is not None
+    assert max_active_fetches == 6
+    assert list(mm_data) == ["image", "audio", "video"]
+    assert list(mm_uuids) == ["image", "audio", "video"]
+    assert mm_data == {
+        "image": ["image-0", "image-1"],
+        "audio": ["audio-0", "audio-1"],
+        "video": ["video-0", "video-1"],
+    }
+    assert mm_uuids == {
+        "image": ["image-0-uuid", "image-1-uuid"],
+        "audio": ["audio-0-uuid", "audio-1-uuid"],
+        "video": ["video-0-uuid", "video-1-uuid"],
+    }
+
+
 @pytest.mark.asyncio
 async def test_resolve_items_does_not_leak_tasks_on_partial_failure():
     """Regression test: one failing media fetch must not abandon the other
-    still-in-flight fetches in the same modality batch.
+    still-in-flight fetches across modality batches.
 
     Before the fix, `resolve_items` gathered per-modality fetches with plain
     `asyncio.gather`, so the first exception propagated immediately while
@@ -2957,6 +3009,8 @@ async def _fetch(should_fail: bool, delay: float):
         lambda: _fetch(False, 0.2),
         lambda: _fetch(False, 0.2),
     ]
+    tracker._items_by_modality["audio"] = [lambda: _fetch(False, 0.2)]
+    tracker._items_by_modality["video"] = [lambda: _fetch(False, 0.2)]
 
     tasks_before = asyncio.all_tasks()
     with pytest.raises(ValueError, match="simulated fetch failure"):

vllm/entrypoints/chat_utils.py

@@ -871,19 +871,30 @@ async def resolve_items(
         if not self._items_by_modality:
             return None, None
 
+        # Fetch all modalities together. Each tracked item is already an
+        # independent awaitable, and the async connector offloads blocking
+        # decode work, so waiting for one modality before starting the next
+        # needlessly adds their latency.
+        # Keep the original group and item order when rebuilding the result.
+        item_groups = list(self._items_by_modality.items())
+        items = [item for _, group in item_groups for item in group]
+        results = await asyncio.gather(
+            *(item() for item in items), return_exceptions=True
+        )
+        for result in results:
+            if isinstance(result, BaseException):
+                # Gathering with return_exceptions=True lets every task finish
+                # (or itself fail) before we raise, instead of abandoning
+                # still-in-flight fetches (real network/thread-pool work) the
+                # moment the first one fails.
+                raise result
+
         resolved_items_by_modality: dict[str, list[Any]] = {}
-        for modality, items in self._items_by_modality.items():
-            results = await asyncio.gather(
-                *(item() for item in items), return_exceptions=True
-            )
-            for result in results:
-                if isinstance(result, BaseException):
-                    # Gathering with return_exceptions=True lets every task in
-                    # this modality finish (or itself fail) before we raise,
-                    # instead of abandoning still-in-flight fetches (real
-                    # network/thread-pool work) the moment the first one fails.
-                    raise result
-            resolved_items_by_modality[modality] = results
+        result_idx = 0
+        for modality, group in item_groups:
+            next_result_idx = result_idx + len(group)
+            resolved_items_by_modality[modality] = results[result_idx:next_result_idx]
+            result_idx = next_result_idx
 
         mm_processor = (
             self.mm_processor if self._model_config.is_multimodal_model else None