asyncio.as_completed 如何工作

Question

阅读这个答案，我遇到了 asyncio.tasks.as_completed。我不明白该功能实际上是如何工作的。它被记录为一个非异步例程，按照它们完成的顺序返回期货。它创建一个与事件循环关联的队列，为每个未来添加一个完成回调，然后尝试从队列中获取与期货一样多的项目。

代码的核心如下：

    def _on_completion(f):
        if not todo:
            return  # _on_timeout() was here first.
        todo.remove(f)
        done.put_nowait(f)
        if not todo and timeout_handle is not None:
            timeout_handle.cancel()

    @coroutine
    def _wait_for_one():
        f = yield from done.get()
        if f is None:
            # Dummy value from _on_timeout().
            raise futures.TimeoutError
        return f.result()  # May raise f.exception().

    for f in todo:
        f.add_done_callback(_on_completion)
    if todo and timeout is not None:
        timeout_handle = loop.call_later(timeout, _on_timeout)
    for _ in range(len(todo)):
        yield _wait_for_one()

我想了解这段代码是如何工作的。我最大的问题是：

• 循环实际上在哪里运行。我没有看到对 loop.run_until_cobmplete 或 loop.run_forever 的任何调用。那么循环如何取得进展呢？

• 方法文档说该方法返回期货。你可以称之为

  for f in as_completed(futures): result = yield from f

我无法将其与 _wait_for_one 中的返回 f.result 行进行协调。记录的调用约定是否正确？如果是这样，那么收益从何而来？

Answer 1

您复制的代码缺少标题部分，这很重要。

# This is *not* a @coroutine!  It is just an iterator (yielding Futures).
def as_completed(fs, *, loop=None, timeout=None):
    """Return an iterator whose values are coroutines.

    When waiting for the yielded coroutines you'll get the results (or
    exceptions!) of the original Futures (or coroutines), in the order
    in which and as soon as they complete.

    This differs from PEP 3148; the proper way to use this is:

        for f in as_completed(fs):
            result = yield from f  # The 'yield from' may raise.
            # Use result.

    If a timeout is specified, the 'yield from' will raise
    TimeoutError when the timeout occurs before all Futures are done.

    Note: The futures 'f' are not necessarily members of fs.
    """
    if futures.isfuture(fs) or coroutines.iscoroutine(fs):
        raise TypeError("expect a list of futures, not %s" % type(fs).__name__)
    loop = loop if loop is not None else events.get_event_loop()
    todo = {ensure_future(f, loop=loop) for f in set(fs)}
    from .queues import Queue  # Import here to avoid circular import problem.
    done = Queue(loop=loop)
    timeout_handle = None

    def _on_timeout():
        for f in todo:
            f.remove_done_callback(_on_completion)
            done.put_nowait(None)  # Queue a dummy value for _wait_for_one().
        todo.clear()  # Can't do todo.remove(f) in the loop.

    def _on_completion(f):
        if not todo:
            return  # _on_timeout() was here first.
        todo.remove(f)
        done.put_nowait(f)
        if not todo and timeout_handle is not None:
            timeout_handle.cancel()

    @coroutine
    def _wait_for_one():
        f = yield from done.get()
        if f is None:
            # Dummy value from _on_timeout().
            raise futures.TimeoutError
        return f.result()  # May raise f.exception().

    for f in todo:
        f.add_done_callback(_on_completion)
    if todo and timeout is not None:
        timeout_handle = loop.call_later(timeout, _on_timeout)
    for _ in range(len(todo)):
        yield _wait_for_one()

[循环实际上在哪里运行？]

为简单起见，假设超时设置为无。

as_completed 期望可迭代的期货，而不是协程。所以这个期货已经绑定到循环并安排执行。换句话说，这些期货是 loop.create_task 或 asyncio.ensure_futures 的输出（并且没有明确写出）。所以循环已经在“运行”它们，当它们完成时，它们未来的 .done() 方法将返回 True。

然后创建“完成”队列。请注意，“完成”队列是 asyncio.queue 的一个实例，即实现阻塞方法（.get、.put）»使用循环«的队列。

通过“todo = { ...”这一行，每个协程的未来（即 fs 的一个元素）被包装在另一个未来»绑定到循环«中，并且这个最后一个未来的 done_callback 被设置为调用 _on_completion 函数。

_on_completion 函数将在循环完成协程的执行时被调用，协程的期货在设置为 as_completed 函数的“fs”中传递。

_on_completion 函数从待办事项集中删除“我们的未来”，并将其结果（即未来在“fs”集中的协程）放入完成队列。换句话说， as_completed 函数所做的就是通过 done_callback 附加这些期货，以便将原始期货的结果移动到完成队列中。

然后，对于 len(fs) == len(todo) 次，as_completed 函数产生一个协程，该协程阻止“yield from done.get()”，等待 _on_completed（或 _on_timeout）函数将结果放入 done完成队列。

由 as_completed 调用者执行的“yield from”将等待结果出现在完成队列中。

[收益从何而来？]

它来自于 todo 是 asyncio.queue 的事实，所以你可以（asyncio-）阻塞直到队列中的值是 .put() 。