为什么 asyncio.wait 尽管超过超时但仍保留带有引用的任务？

Question

我最近发现并重现了由于使用asyncio.wait导致的内存泄漏。具体来说，我的程序定期执行某些函数，直到stop_event设置为止。我将程序简化为下面的代码片段（减少超时以更好地演示问题）：

async def main():
  stop_event = asyncio.Event()

  while True:
    # Do stuff here
    await asyncio.wait([stop_event.wait()], timeout=0.0001)

asyncio.run(main())

虽然这对我来说看起来无害，但事实证明这里存在内存泄漏。如果执行上面的代码，您将看到内存使用量在几分钟内增长到数百 MB。这让我很惊讶，并花了很长时间才找到。我期望在超时后，我正在等待的任何内容都会被清理（因为我自己没有保留任何对它的引用）。然而，事实证明并非如此。

使用gc.get_referrers，我能够推断每次调用时asyncio.wait(...)，都会创建一个新任务，该任务保存对返回的对象的引用stop_event.wait()，并且该任务将永远保留。具体来说，len(asyncio.all_tasks())随着时间的推移不断增加。即使超时了，任务仍然存在。只有在调用时，stop_event.set()这些任务才会立即完成，并且内存使用量才会急剧减少。

发现这一点后，文档中的这条注释让我尝试使用asyncio.wait_for代替：

与 wait_for() 不同，wait() 在发生超时时不会取消 future。

事实证明，它的表现确实如我所料。超时后不会保留任何引用，内存使用量和任务数量保持不变。这是没有内存泄漏的代码：

async def main():
  stop_event = asyncio.Event()

  while True:
    # Do stuff here
    try:
      await asyncio.wait_for(event.stop_event(), timeout=0.0001)
    except asyncio.TimeoutError:
      pass

asyncio.run(main())

虽然我很高兴这个问题现在已经解决了，但我不太理解这种行为。如果超过了超时时间，为什么要让这个任务保留一个引用呢？这似乎是造成内存泄漏的一个秘诀。关于不取消期货的说明我也不清楚。如果我们不明确取消 future，但我们只是不让任务在超时后保留引用，该怎么办？那不是也可以吗？

如果有人能对此有所启发，我们将不胜感激。多谢！

Answer 1

这里要理解的关键概念是任务的返回值wait()是一个元组(completed, pending)。

使用wait()基于代码的典型方法是这样的：

async def main():
    stop_event = asyncio.Event()

    pending = [... add things to wait ...]

    while pending:
        completed, pending = await asyncio.wait(pending, timeout=0.0001)

        process(completed) # e.g. update progress bar

        pending.extend(more_tasks_to_wait)

wait()with timeout 并不用于让一个协程等待另一个协程/任务完成，而是它的主要用例是定期刷新已完成的任务，同时让未完成的任务“在后台”继续，从而取消未完成的任务自动并不是真正理想的，因为您通常希望在下一次迭代中再次继续等待那些挂起的任务。

这种使用模式类似于select()系统调用。

另一方面， 的使用模式await wait_for(xyz, )基本上就像await xyz超时一样。这是一个常见且简单得多的用例。