PEP 525 非同步生成器鉤子

PEP 525 提議了一組由新的 sys.{get/set}_asyncgen_hooks() 函式管理的全域性執行緒區域性鉤子，允許事件迴圈與垃圾回收器整合，以執行非同步生成器的清理。原則上，本提案和 PEP 525 是互補的，就像 with 塊和 __del__ 是互補的一樣：本提案負責確保在大多數情況下的確定性清理，而 PEP 525 的 GC 鉤子清理任何遺漏的部分。但 __aiterclose__ 相比單獨的 GC 鉤子有許多優點

• GC 鉤子語義不屬於抽象非同步迭代器協議的一部分，而是專門限制在非同步生成器具體型別上。如果你有一個使用類實現的非同步迭代器，像這樣

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

  那麼你無法在不改變其語義的情況下將其重構為非同步生成器，反之亦然。這看起來非常不符合 Python 的風格。（它還留下了這樣一個問題：鑑於基於類的非同步迭代器面臨與非同步生成器完全相同的清理問題，它們到底應該怎麼做。）另一方面，__aiterclose__ 是在協議級別定義的，因此它對鴨子型別友好，並且適用於所有迭代器，而不僅僅是生成器。

• 希望在非 CPython 實現（如 PyPy）上執行的程式碼通常不能依賴 GC 進行清理。如果沒有 __aiterclose__，幾乎可以肯定的是，在 CPython 上開發和測試的開發人員會編寫在 PyPy 上使用時會洩漏資源的庫。希望針對其他實現的開發人員將不得不採取防禦性方法，將每個 for 迴圈包裝在 with 塊中，或者仔細審計他們的程式碼，找出哪些生成器可能包含清理程式碼，並僅在這些生成器周圍新增 with 塊。有了 __aiterclose__，編寫可移植程式碼變得簡單自然。
• 構建健壯軟體的一個重要部分是確保異常始終正確傳播而不會丟失。與傳統的基於回撥的系統相比，async/await 最令人興奮的一點是，它不再需要手動連結，執行時現在可以承擔傳播錯誤的繁重工作，使得編寫健壯程式碼*容易得多*。但是，這個美好的新景象有一個主要缺陷：如果我們依賴 GC 進行生成器清理，那麼清理過程中引發的異常就會丟失。因此，再次強調，有了 __aiterclose__，關注這種健壯性的開發人員將不得不採取防禦性方法，將每個 for 迴圈包裝在 with 塊中，或者仔細審計他們的程式碼，找出哪些生成器可能包含清理程式碼。__aiterclose__ 透過在呼叫者的上下文中執行清理來彌補這個漏洞，因此編寫更健壯的程式碼成為阻力最小的路徑。
• WSGI 經驗表明，存在重要的基於迭代器的 API，它們需要即時清理，不能依賴 GC，即使在 CPython 中也是如此。例如，考慮一個基於 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 迴圈或類似的方式進行迭代，因此在 GC 有機會介入之前，__aiterclose__ 處理了大多數情況。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__ 的人。但 Yury 在這一點上與我意見相左 :-)。兩種選擇都可行。

始終注入資源，並在頂層執行所有清理

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()
• 如果尚未呼叫 aclose，則 __del__ 會發出 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__ 之前，在底層迭代器上呼叫 __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 塊呼叫 map 物件的 __iterclose__()。
3. map.__iterclose__() 呼叫生成器推導式物件的 __iterclose__()。
4. 這會將一個 GeneratorExit 異常注入到生成器推導式主體中，該主體目前暫停在推導式的 for 迴圈主體內部。
5. 異常從 for 迴圈中傳播出來，觸發 for 迴圈的隱式 finally 塊，該塊在表示對 read_newline_separated_json 呼叫的生成器物件上呼叫 __iterclose__。
6. 這會在 read_newline_separated_json 的主體中注入一個內部 GeneratorExit 異常，目前暫停在 yield 處。
7. 內部 GeneratorExit 從 for 迴圈中繼續傳播，命中生成器函式的邊界，並導致 read_newline_separated_json 的 __iterclose__() 方法成功返回。
8. 檔案物件已關閉。
9. 內部 GeneratorExit 恢復傳播，到達生成器函式的邊界，並導致 read_newline_separated_json 的 __iterclose__() 方法成功返回。
10. 控制返回到生成器推導式主體，外部 GeneratorExit 繼續傳播，允許推導式的 __iterclose__() 成功返回。
11. 其餘的 __iterclose__() 呼叫順利展開，返回到 list 的主體。
12. 原始的 AttributeError 恢復傳播。

（上述細節假設我們實現了 file.__iterclose__；如果不是，則向 read_newline_separated_json 新增一個 with 塊，並且基本相同的邏輯也會透過。）

當然，從使用者的角度來看，這可以簡化為

1. int.upper() 引發 AttributeError 1. 檔案物件已關閉。 2. AttributeError 從 list 中傳播出來

所以我們已經實現了我們的目標，讓這一切“just work”，而使用者無需考慮。