asyncio.gather 的顺序版本

Question

我尝试创建一个类似于 asyncio.gather 的方法，但它会按顺序而不是异步执行任务列表：

async def in_sequence(*tasks):
    """Executes tasks in sequence"""
    for task in tasks:
        await task

接下来这个方法应该是这样使用的：

async def some_work(work_name):
    """Do some work"""
    print(f"Start {work_name}")
    await asyncio.sleep(1)
    if raise_exception:
        raise RuntimeError(f"{work_name} raise an exception")
    print(f"Finish {work_name}")

async def main():
    try:
        await asyncio.gather(
            some_work("work1"),         # work1, work2, in_sequence and work5 executed in concurrently
            some_work("work2"),
            in_sequence(
                some_work("work3"),     # work3 and work4 executed in sequence
                some_work("work4")
            ),
            some_work("work5"),


    except RuntimeError as error:
        print(error)                    # raise an exception at any point to terminate

在我尝试在 some_work 中抛出异常之前一切正常：

async def main():
    try:
        await asyncio.gather(
            some_work("work1"),
            some_work("work2"),
            in_sequence(
                some_work("work3", raise_exception=True),       # raise an exception here
                some_work("work4")
            ),
            some_work("work5"),


    except RuntimeError as error:
        print(error)

紧接着，我收到以下错误消息：

RuntimeWarning: coroutine 'some_work' was never awaited

我阅读了文档并继续实验：

async def in_sequence(*tasks):
    """Executes tasks in sequence"""
    _tasks = []
    for task in tasks:
        _tasks.append(asyncio.create_task(task))

    for _task in _tasks:
        await _task

这个版本按预期工作！

对此，我有下一个问题：

1. 为什么第二个版本有效而​​第一个无效？
2. asyncio 是否已经拥有执行任务列表的工具 顺序？
3. 我选择了正确的实现方法还是有更好的实现方法 选项？

Answer 1

1. 第一个版本不起作用，因为in_sequence 没有捕获可以在await task 上引发的异常。第二个有效，因为create_task 创建了一个运行协程的类似未来的Task 对象。该对象不返回/传播包装协程的结果。 当您await 对象时，它会挂起，直到有a result 或an exception 设置 或直到it is canceled。

2. 好像没有。

3. 第二个版本会并发执行传递过来的协程，所以是不正确的实现。如果你真的想使用一些 in_sequence 函数，你可以：
   • 不知何故延迟了协程的创建。
   • 在async 函数中分组顺序执行

例如：

async def in_sequence(*fn_and_args):
    for fn, args, kwargs in fn_and_args:
        await fn(*args, **kwargs)  # create a coro and await it in place

in_sequence(
    (some_work, ("work3",), {'raise_exception': True}),
    (some_work, ("work4",), {}),
)

async def in_sequence():
    await some_work("work3", raise_exception=True)
    await some_work("work4")

Answer 2

您说 in_sequence 的版本有效（使用 asyncio.create_task），但我认为它没有。来自文档

将 coro 协程包装到一个 Task 中并安排其执行。返回 任务对象。

好像是并行运行协程，但需要按顺序运行。

于是试验并找到了两种方法来解决这个问题

使用您原来的 in_sequence 函数并添加此代码，以隐藏该错误：

import warnings
warnings.filterwarnings(
    'ignore',
    message=r'^coroutine .* was never awaited$',
    category=RuntimeWarning
)

修复 in_sequence 函数，如下所示：

async def in_sequence(*tasks):
    for index, task in enumerate(tasks):
        try:
            await task
        except Exception as e:
            for task in tasks[index + 1:]:
                task.close()
            raise e

其他问题的答案：

1. 当您在协程上没有链接时，C++ 代码会触发警告。只需简单的代码就可以向您展示这个想法（在终端中）：

---

async def test():
    return 1

f = test()
f = None # after that you will get that error

1. 我不知道
2. 见上文

Answer 3

这个版本按预期工作！

第二个版本的问题在于它实际上并没有顺序运行协同程序，而是并行运行它们。这是因为asyncio.create_task() 调度协程与当前协程并行运行。因此，当您在循环中等待任务时，实际上是在等待第一个任务时允许所有任务运行。尽管看起来，整个循环只会运行最长的任务。 （详情请参阅here。）

您的第一个版本显示的警告旨在防止您意外创建您从不等待的协程，例如只写asyncio.sleep(1) 而不是await asyncio.sleep(1)。就 asyncio 而言，main 正在实例化协程对象并将它们传递给 in_sequence，后者“忘记”等待其中一些对象。

抑制警告消息的一种方法是允许协程旋转，但立即取消它。例如：

async def in_sequence(*coros):
    remaining = iter(coros)
    for coro in remaining:
        try:
            await coro
        except Exception:
            for c in remaining:
                asyncio.create_task(c).cancel()
            raise

请注意，以下划线开头的变量名表示未使用的变量，因此您不应该为变量命名，因此如果您确实使用它们。

Answer 4

从 user4815162342 和 Anton Pomieshchenko 的解决方案中汲取灵感，我想出了这个变体：

async def in_sequence(*storm):
    twister = iter(storm)
    for task in twister:
        task = task() # if it's a regular function, it's done here.
        if inspect.isawaitable(task):
            try:
                await task # if it's also awaitable, await it
            except BaseException as e:
                task.throw(e) # if an error occurs, throw it into the coroutine
            finally:
                task.close() # to ensure coroutine closer

    assert not any(twister) # optionally verify that the iterator is now empty

通过这种方式，您可以使用 in_sequence 将常规函数与协程结合起来。但一定要这样称呼它：

await in_sequence(*[b.despawn, b.release])

请注意缺少() (__call__())，否则常规函数将立即被调用，协程将抛出一个RuntimeWarning，因为它从未被等待。 （b.despawn 是协程，b.release 不适用于我的示例）

您还可以在调用 task() 之前对 callable(task) 进行额外检查，但这取决于您。