fix(entrypoints): stop resolve_items leaking in-flight media fetch tasks on partial failure (#48333)
Signed-off-by: ErenAta16 <erena6466@gmail.com>
ErenAta16 committed
Jul 11, 2026 at 18:42 UTC
006731153693b01320feb93ab165de24fc364b0c
2 files changed
+52
-4
tests/entrypoints/unit_tests/test_chat_utils.py
+39
@@ -1,9 +1,11 @@
1
# SPDX-License-Identifier: Apache-2.0
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4
+import asyncio
5
import warnings
6
from collections.abc import Mapping
7
from typing import Literal
8
+from unittest.mock import MagicMock
9
10
import pytest
11
import torch
@@ -13,6 +15,7 @@ from vllm.assets.image import ImageAsset
15
from vllm.assets.video import VideoAsset
16
from vllm.config import ModelConfig
17
from vllm.entrypoints.chat_utils import (
18
+ AsyncMultiModalItemTracker,
19
ConversationMessage,
20
_postprocess_messages,
21
parse_chat_messages,
@@ -2742,3 +2745,39 @@ def test_postprocess_messages_null_arguments_string():
2745
tool_calls = messages[0]["tool_calls"]
2746
assert tool_calls is not None
2747
assert tool_calls[0]["function"]["arguments"] == {}
2748
+
2749
+
2750
+@pytest.mark.asyncio
2751
+async def test_resolve_items_does_not_leak_tasks_on_partial_failure():
2752
+ """Regression test: one failing media fetch must not abandon the other
2753
+ still-in-flight fetches in the same modality batch.
2754
+
2755
+ Before the fix, `resolve_items` gathered per-modality fetches with plain
2756
+ `asyncio.gather`, so the first exception propagated immediately while
2757
+ sibling fetches (real network/thread-pool work in production) kept
2758
+ running detached, with nothing left to await or cancel them.
2759
+ """
2760
+
2761
+ async def _fetch(should_fail: bool, delay: float):
2762
+ if should_fail:
2763
+ await asyncio.sleep(0.01)
2764
+ raise ValueError("simulated fetch failure")
2765
+ await asyncio.sleep(delay)
2766
+ return ("decoded", None)
2767
+
2768
+ tracker = AsyncMultiModalItemTracker(MagicMock())
2769
+ tracker._items_by_modality["image"] = [
2770
+ lambda: _fetch(True, 0),
2771
+ lambda: _fetch(False, 0.2),
2772
+ lambda: _fetch(False, 0.2),
2773
+ ]
2774
+
2775
+ tasks_before = asyncio.all_tasks()
2776
+ with pytest.raises(ValueError, match="simulated fetch failure"):
2777
+ await tracker.resolve_items()
2778
+
2779
+ leaked_tasks = asyncio.all_tasks() - tasks_before
2780
+ assert not leaked_tasks, (
2781
+ f"resolve_items left {len(leaked_tasks)} task(s) running after "
2782
+ f"raising: {leaked_tasks}"
2783
+ )
vllm/entrypoints/chat_utils.py
+13
-4
@@ -821,10 +821,19 @@ class AsyncMultiModalItemTracker(BaseMultiModalItemTracker[_AsyncMultiModalItem]
821
if not self._items_by_modality:
822
return None, None
823
824
- resolved_items_by_modality = {
825
- modality: await asyncio.gather(*(item() for item in items))
826
- for modality, items in self._items_by_modality.items()
827
- }
824
+ resolved_items_by_modality: dict[str, list[Any]] = {}
825
+ for modality, items in self._items_by_modality.items():
826
+ results = await asyncio.gather(
827
+ *(item() for item in items), return_exceptions=True
828
+ )
829
+ for result in results:
830
+ if isinstance(result, BaseException):
831
+ # Gathering with return_exceptions=True lets every task in
832
+ # this modality finish (or itself fail) before we raise,
833
+ # instead of abandoning still-in-flight fetches (real
834
+ # network/thread-pool work) the moment the first one fails.
835
+ raise result
836
+ resolved_items_by_modality[modality] = results
837
838
mm_processor = (
839
self.mm_processor if self._model_config.is_multimodal_model else None