PEP 525 asyncgen 钩子

PEP 525 提出了一组由新的 sys.{get/set}_asyncgen_hooks() 函数管理的全局线程局部钩子，允许事件循环与垃圾回收器集成以运行异步生成器的清理。原则上，本提案和PEP 525 是互补的，就像 with 块和 __del__ 是互补的一样：本提案负责确保在大多数情况下进行确定性清理，而PEP 525 的 GC 钩子会清理任何遗漏的内容。__aiterclose__ 相比仅使用 GC 钩子，提供了许多优势

• GC 钩子语义不是抽象异步迭代器协议的一部分，而是专门限制在异步生成器具体类型。如果您使用类实现了一个异步迭代器，例如

  class MyAsyncIterator:
      async def __anext__():
          ...
  

  那么您无法将此重构为异步生成器而不会更改其语义，反之亦然。这看起来很不 Pythonic。（它也留下了基于类的异步迭代器究竟应该做什么的问题，因为它们面临着与异步生成器完全相同的清理问题。）另一方面，__aiterclose__ 在协议级别定义，因此它对鸭子类型友好，并且适用于所有迭代器，而不仅仅是生成器。

• 希望在非 CPython 实现（如 PyPy）上运行的代码通常不能依赖 GC 进行清理。如果没有 __aiterclose__，那么开发人员在 CPython 上开发和测试时几乎肯定会产生在 PyPy 上使用时会泄漏资源的库。希望针对替代实现的开发人员要么必须采取防御性方法，将每个 for 循环包装在 with 块中，要么仔细检查他们的代码以确定哪些生成器可能包含清理代码，并在这些生成器周围添加 with 块。有了 __aiterclose__，编写可移植代码变得简单自然。
• 构建健壮软件的重要部分是确保异常始终正确传播，而不会丢失。与传统的基于回调的系统相比，async/await 最令人兴奋的事情之一是，运行时现在可以承担传播错误的重任，而不是需要手动链式调用，这使得编写健壮代码变得 *非常* 容易。但是，这幅美好的新图景存在一个主要的差距：如果我们依赖 GC 进行生成器清理，那么在清理过程中引发的异常就会丢失。因此，再次强调，对于关心这种健壮性的开发人员来说，使用 __aiterclose__，要么必须采取防御性方法，将每个 for 循环包装在 with 块中，要么仔细检查他们的代码以确定哪些生成器可能包含清理代码。 __aiterclose__ 通过在调用方的上下文中执行清理来填补这个漏洞，因此编写更健壮的代码成为最简单的途径。
• WSGI 的经验表明，存在一些重要的基于迭代器的 API 需要及时清理，即使在 CPython 中也无法依赖 GC。例如，考虑一个基于 async/await 和异步迭代器的假设 WSGI 类 API，其中响应处理程序是一个异步生成器，它接收请求头 + 请求正文上的异步迭代器，并生成响应头 + 响应正文。（这实际上是我最初对异步生成器感兴趣的用例，即这不是假设的。）如果我们遵循 WSGI 中的要求，即必须正确关闭子迭代器，那么如果没有 __aiterclose__，我们系统中最小的中间件看起来就像

  async def noop_middleware(handler, request_header, request_body):
      async with aclosing(handler(request_body, request_body)) as aiter:
          async for response_item in aiter:
              yield response_item
  

  可以说，在普通代码中，可以避免在 for 循环周围使用 with 块，具体取决于一个人对理解生成器内部实现的信心程度。但在这里，我们必须处理任意的响应处理程序，因此如果没有 __aiterclose__，这种 with 结构是每个中间件的强制部分。

  __aiterclose__ 允许我们消除每个中间件中的强制样板代码和额外的缩进级别。

  async def noop_middleware(handler, request_header, request_body):
      async for response_item in handler(request_header, request_body):
          yield response_item
  

因此，__aiterclose__ 方法比 GC 钩子提供了实质性的优势。

这留下了一个问题，即我们是否希望将 GC 钩子和 __aiterclose__ 结合起来，或者只使用 __aiterclose__。由于绝大多数生成器都是使用 for 循环或等效方法进行迭代的，因此 __aiterclose__ 在 GC 有机会介入之前处理了大多数情况。GC 钩子提供额外价值的情况是在执行手动迭代的代码中，例如

agen = fetch_newline_separated_json_from_url(...)
while True:
    document = await type(agen).__anext__(agen)
    if document["id"] == needle:
        break
# doesn't do 'await agen.aclose()'

如果我们使用 GC 钩子和 __aiterclose__ 方法，这个生成器最终将由 GC 调用生成器 __del__ 方法进行清理，然后该方法将使用钩子回调到事件循环以运行清理代码。

如果我们使用不带 GC 钩子的方法，这个生成器最终将被垃圾回收，并产生以下影响

• 其 __del__ 方法将发出警告，表明生成器未关闭（类似于现有的“协程从未等待”警告）。
• 所涉及的基础资源仍将被清理，因为生成器帧仍将被垃圾回收，导致它放弃对所持有的任何文件句柄或套接字的引用，然后这些对象的 __del__ 方法将释放实际的操作系统资源。
• 但是，生成器本身中的任何清理代码（例如日志记录、缓冲区刷新）都将没有机会运行。

这里的解决方案——正如警告所示——是修复代码，使其调用 __aiterclose__，例如通过使用 with 块。

async with aclosing(fetch_newline_separated_json_from_url(...)) as agen:
    while True:
        document = await type(agen).__anext__(agen)
        if document["id"] == needle:
            break

基本上，在这种方法中，规则是，如果你想手动实现迭代器协议，那么你有责任实现所有内容，现在包括 __(a)iterclose__。

GC 钩子增加了不小的复杂性，形式为 (a) 新的全局解释器状态，(b) 稍微复杂的控制流（例如，异步生成器 GC 始终涉及复活，因此 PEP 442 的细节很重要），以及 (c) asyncio 中的新公共 API（await loop.shutdown_asyncgens()），用户必须记住在适当的时间调用它。（尤其是最后一点在某种程度上削弱了 GC 钩子提供安全备份以保证清理的说法，因为如果 shutdown_asyncgens() 没有正确调用，那么我 *认为* 生成器有可能被静默丢弃，而不会调用其清理代码；将其与仅使用 __aiterclose__ 的方法进行比较，在最坏的情况下，我们至少会打印出一个警告。这可能是可以修复的。）考虑到所有这些，GC 钩子可以说是得不偿失的，因为它们唯一帮助的人是那些想要手动调用 __anext__ 却又不想手动调用 __aiterclose__ 的人。但尤里不同意我的看法:-)。并且这两种选择都是可行的。

始终注入资源，并在顶层执行所有清理

python-dev 和 python-ideas 上的一些评论者建议，避免这些问题的模式是从上到下始终传递资源，例如 read_newline_separated_json 应该接收文件对象而不是路径，并在顶层处理清理。

def read_newline_separated_json(file_handle):
    for line in file_handle:
        yield json.loads(line)

def read_users(file_handle):
    for document in read_newline_separated_json(file_handle):
        yield User.from_json(document)

with open(path) as file_handle:
    for user in read_users(file_handle):
        ...

这在简单情况下效果很好；在这里，它让我们避免了“N+1 with 块问题”。但不幸的是，当事情变得更复杂时，它很快就会失效。考虑一下，如果我们的生成器不是从文件读取，而是从流式 HTTP GET 请求读取——同时处理重定向和通过 OAUTH 进行身份验证。然后，我们确实希望套接字由 HTTP 客户端库内部管理，而不是在顶层管理。此外，还有其他一些情况，生成器内部嵌入的 finally 块本身就很重要：数据库事务管理、在清理期间发出日志信息（WSGI close 的主要动机用例之一），等等。因此，这实际上是针对简单情况的解决方法，而不是通用解决方案。

更复杂的 __(a)iterclose__ 变体

__(a)iterclose__ 的语义在某种程度上受到 with 块的启发，但上下文管理器功能更强大：__(a)exit__ 可以区分正常退出和异常展开，并且在发生异常的情况下，它可以检查异常详细信息并选择性地抑制传播。__(a)iterclose__ 在这里提出的设计中不具备这些功能，但可以想象一个它具备这些功能的替代设计。

但是，这似乎是不必要的复杂性：经验表明，可迭代对象通常具有 close 方法，甚至具有调用 self.close() 的 __exit__ 方法，但我不知道有任何常见的用例利用了 __exit__ 的全部功能。我也想不出任何在这种情况下有用的例子。并且允许迭代器通过吞掉异常来影响流程控制似乎是不必要的混乱——如果你处于确实需要这种情况的境地，那么你可能应该使用真正的 with 块。

指导原则

通常，__(a)iterclose__ 实现应该

• 是幂等的，
• 执行任何在假设迭代器在调用 __(a)iterclose__ 后不再使用的基础上进行的清理。特别是，一旦调用了 __(a)iterclose__，则调用 __(a)next__ 会产生未定义的行为。

并且通常，任何以耗尽为目的开始遍历可迭代对象的代码都应该安排确保最终调用 __(a)iterclose__，无论迭代器是否真的被耗尽。

迭代的更改

核心提案是 for 循环的行为变化。给定这段 Python 代码

for VAR in ITERABLE:
    LOOP-BODY
else:
    ELSE-BODY

我们将它转换为等效的

_iter = iter(ITERABLE)
_iterclose = getattr(type(_iter), "__iterclose__", lambda: None)
try:
    traditional-for VAR in _iter:
        LOOP-BODY
    else:
        ELSE-BODY
finally:
    _iterclose(_iter)

这里的“传统 for 语句”是指 3.5 及更早版本的经典 for 循环语义的简写。

除了顶级 for 语句之外，Python 还包含其他一些使用迭代器的地方。为了保持一致性，这些地方也应该使用与上述等效的语义调用 __iterclose__。这包括

• 推导式内的 for 循环
• * 解包
• 接受并完全使用可迭代对象的函数，如 list(it)、tuple(it)、itertools.product(it1, it2, ...) 等。

此外，一个成功耗尽被调用生成器的yield from应该作为最后一步调用其__iterclose__方法。（理由：yield from已经将调用生成器的生命周期与被调用生成器关联起来；如果在yield from执行到一半时调用生成器被关闭，那么这将自动关闭被调用生成器。）

异步迭代的更改

我们还对异步迭代结构进行了类似的更改，除了新的槽称为__aiterclose__，它是一个被await的异步方法。

基本迭代器类型的修改

生成器对象（包括由生成器推导式创建的那些）

• __iterclose__调用self.close()
• __del__调用self.close()（与现在相同），并且如果生成器没有被耗尽，则还会发出ResourceWarning。此警告默认情况下是隐藏的，但可以为那些想要确保他们没有无意中依赖于CPython特定GC语义的人启用。

异步生成器对象（包括由异步生成器推导式创建的那些）

• __aiterclose__调用self.aclose()
• __del__如果aclose尚未被调用，则会发出RuntimeWarning，因为这可能表示一个潜在的错误，类似于“协程从未等待”警告。

问题：文件对象是否应该实现__iterclose__来关闭文件？一方面，这会使此更改更具破坏性；另一方面，人们确实喜欢编写for line in open(...): ...，如果我们习惯了迭代器自行处理清理，那么如果文件不这样做，可能会变得非常奇怪。

新的便捷函数

operator模块获得了两个新函数，其语义等效于以下内容

def iterclose(it):
    if not isinstance(it, collections.abc.Iterator):
        raise TypeError("not an iterator")
    if hasattr(type(it), "__iterclose__"):
        type(it).__iterclose__(it)

async def aiterclose(ait):
    if not isinstance(it, collections.abc.AsyncIterator):
        raise TypeError("not an iterator")
    if hasattr(type(ait), "__aiterclose__"):
        await type(ait).__aiterclose__(ait)

itertools模块获得了一个新的迭代器包装器，可用于有选择地禁用新的__iterclose__行为

# QUESTION: I feel like there might be a better name for this one?
class preserve(iterable):
    def __init__(self, iterable):
        self._it = iter(iterable)

    def __iter__(self):
        return self

    def __next__(self):
        return next(self._it)

    def __iterclose__(self):
        # Swallow __iterclose__ without passing it on
        pass

示例用法（假设文件对象实现了__iterclose__）

with open(...) as handle:
    # Iterate through the same file twice:
    for line in itertools.preserve(handle):
        ...
    handle.seek(0)
    for line in itertools.preserve(handle):
        ...

@contextlib.contextmanager
def iterclosing(iterable):
    it = iter(iterable)
    try:
        yield preserve(it)
    finally:
        iterclose(it)

迭代器包装器的 __iterclose__ 实现

Python提供了一些充当其他迭代器包装器的迭代器类型：map、zip、itertools.accumulate、csv.reader等。这些迭代器应该定义一个__iterclose__方法，该方法依次在其底层迭代器上调用__iterclose__。例如，map可以实现为

# Helper function
map_chaining_exceptions(fn, items, last_exc=None):
    for item in items:
        try:
            fn(item)
        except BaseException as new_exc:
            if new_exc.__context__ is None:
                new_exc.__context__ = last_exc
            last_exc = new_exc
    if last_exc is not None:
        raise last_exc

class map:
    def __init__(self, fn, *iterables):
        self._fn = fn
        self._iters = [iter(iterable) for iterable in iterables]

    def __iter__(self):
        return self

    def __next__(self):
        return self._fn(*[next(it) for it in self._iters])

    def __iterclose__(self):
        map_chaining_exceptions(operator.iterclose, self._iters)

def chain(*iterables):
    try:
        while iterables:
            for element in iterables.pop(0):
                yield element
    except BaseException as e:
        def iterclose_iterable(iterable):
            operations.iterclose(iter(iterable))
        map_chaining_exceptions(iterclose_iterable, iterables, last_exc=e)

在某些情况下，这需要一些巧妙的方法；例如，itertools.tee不应在底层迭代器上调用__iterclose__，直到它在所有克隆迭代器上都被调用。

示例/原理

所有这些的回报是，我们现在可以编写像这样的简单代码

def read_newline_separated_json(path):
    for line in open(path):
        yield json.loads(line)

并确信文件将接收确定性的清理而无需最终用户做任何特殊努力，即使在复杂的情况下也是如此。例如，考虑这个愚蠢的管道

list(map(lambda key: key.upper(),
         doc["key"] for doc in read_newline_separated_json(path)))

如果我们的文件包含一个文档，其中doc["key"]恰好是一个整数，那么将发生以下事件序列

1. key.upper()引发AttributeError，它从map传播出来并触发list内部的隐式finally块。
2. list中的finally块在映射对象上调用__iterclose__()。
3. map.__iterclose__()在生成器推导式对象上调用__iterclose__()。
4. 这将GeneratorExit异常注入到生成器推导式主体中，该主体当前挂起在推导式的for循环体内部。
5. 异常从for循环传播出来，触发for循环的隐式finally块，该块在表示对read_newline_separated_json的调用的生成器对象上调用__iterclose__。
6. 这将一个内部GeneratorExit异常注入到read_newline_separated_json的主体中，目前在yield处挂起。
7. 内部GeneratorExit从for循环传播出来，触发for循环的隐式finally块，该块在文件对象上调用__iterclose__()。
8. 文件对象已关闭。
9. 内部GeneratorExit继续传播，到达生成器函数的边界，并导致read_newline_separated_json的__iterclose__()方法成功返回。
10. 控制权返回到生成器推导式主体，并且外部GeneratorExit继续传播，允许推导式的__iterclose__()成功返回。
11. 其余的__iterclose__()调用在没有事件的情况下展开，返回到list的主体中。
12. 原始AttributeError继续传播。

（以上细节假设我们实现了file.__iterclose__；如果不是，则将with块添加到read_newline_separated_json中，并且基本上相同的逻辑会贯穿始终。）

当然，从用户的角度来看，这可以简化为

1. int.upper()引发AttributeError 1. 文件对象已关闭。 2. AttributeError从list传播出来

因此，我们已经实现了我们的目标，即让它“正常工作”，而无需用户考虑。