Function cancel_async_tasks

lmdeploy/pytorch/utils.py:223–231

Cancel async tasks.

(tasks: Sequence[asyncio.Task])

async def cancel_async_tasks(tasks: Sequence[asyncio.Task]):
    """Cancel async tasks."""
    if isinstance(tasks, asyncio.Task):
        tasks = [tasks]

    tasks = list(task for task in tasks if not task.done())
    for task in tasks:
        task.cancel()
    return await asyncio.gather(*tasks, return_exceptions=True)