| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | import asyncio |
| 4 | from collections.abc import AsyncIterator |
| 5 | |
| 6 | import pytest |
| 7 | |
| 8 | from vllm.utils.async_utils import merge_async_iterators |
| 9 | |
| 10 | |
| 11 | async def _mock_async_iterator(idx: int): |
| 12 | try: |
| 13 | while True: |
| 14 | yield f"item from iterator {idx}" |
| 15 | await asyncio.sleep(0.1) |
| 16 | except asyncio.CancelledError: |
| 17 | print(f"iterator {idx} cancelled") |
| 18 | |
| 19 | |
| 20 | @pytest.mark.asyncio |
| 21 | async def test_merge_async_iterators(): |
| 22 | iterators = [_mock_async_iterator(i) for i in range(3)] |
| 23 | merged_iterator = merge_async_iterators(*iterators) |
| 24 | |
| 25 | async def stream_output(generator: AsyncIterator[tuple[int, str]]): |
| 26 | async for idx, output in generator: |
| 27 | print(f"idx: {idx}, output: {output}") |
| 28 | |
| 29 | task = asyncio.create_task(stream_output(merged_iterator)) |
| 30 | await asyncio.sleep(0.5) |
| 31 | task.cancel() |
| 32 | with pytest.raises(asyncio.CancelledError): |
| 33 | await task |
| 34 | |
| 35 | for iterator in iterators: |
| 36 | try: |
| 37 | await asyncio.wait_for(anext(iterator), 1) |
| 38 | except StopAsyncIteration: |
| 39 | # All iterators should be cancelled and print this message. |
| 40 | print("Iterator was cancelled normally") |
| 41 | except (Exception, asyncio.CancelledError) as e: |
| 42 | raise AssertionError() from e |
| 43 | |
| 44 | |
| 45 | @pytest.mark.asyncio |
| 46 | async def test_merge_async_iterators_single_closes_underlying(): |
| 47 | # The single-iterator fast path must close the underlying generator when |
| 48 | # the merged generator is closed, matching the multi-iterator path. On the |
| 49 | # buggy fast path the underlying generator is left running. |
| 50 | closed = False |
| 51 | |
| 52 | async def gen(): |
| 53 | nonlocal closed |
| 54 | try: |
| 55 | while True: |
| 56 | yield "x" |
| 57 | await asyncio.sleep(0.01) |
| 58 | finally: |
| 59 | closed = True |
| 60 | |
| 61 | merged = merge_async_iterators(gen()) |
| 62 | assert await anext(merged) == (0, "x") |
| 63 | await merged.aclose() |
| 64 | assert closed |