保证退出其上下文管理器的Coroutine

Question

我想在协同程序中使用上下文管理器。这个协程应该处理未知数量的步骤。然而，由于步骤数量未知，尚不清楚上下文管理器何时应该退出。当协同例程超出范围/被垃圾回收时，我希望它退出；然而，在下面的例子中似乎没有发生这种情况：

import contextlib

@contextlib.contextmanager
def cm():
print("STARTED")
yield
print("ENDED")

def coro(a: str):
with cm():
print(a)
while True:
val1, val2 = yield
print(val1, val2)

c = coro("HI")

c.send(None)
print("---")
c.send((1, 2))
print("---!")

该程序的输出：

STARTED
HI
---
1 2
---!

• 上下文管理器从不打印"；ENDED">

如何制作一个支持任意数量步骤并保证优雅退出的协同程序？我不想让打电话的人对此负责。

Answer 1

TLDR：所以问题是当在with块内引发(并且没有处理)异常时。调用上下文管理器的__exit__方法时出现该异常。对于contextmanager修饰的生成器，这将导致生成器的异常为thrown。cm不处理此异常，因此不运行清理代码。当coro被垃圾回收时，它的close方法被调用，其中throw是GeneratorExit到coro(然后被抛出到cm)。以下是对上述步骤的详细描述。

close方法throw从GeneratorExit到coro，这意味着GeneratorExit在yield的点处被提升。coro不处理GeneratorExit，因此它通过错误退出上下文。这导致使用错误和错误信息来调用上下文的__exit__方法。来自contextmanager修饰生成器的__exit__方法做什么？如果使用异常调用它，它会将该异常抛出到基础生成器。

此时，GeneratorExit是从上下文管理器主体中的yield语句中引发的。该未处理的异常导致清理代码无法运行。该未处理的异常由上下文管理器引发，并传递回contextmanager装饰器的__exit__。作为引发的同一错误，__exit__返回False，表示发送给__exit__的原始错误未得到处理。

最后，这将继续GeneratorExit在with块之外的传播，在coro中继续对其进行未处理但是，不处理GeneratorExits对于生成器来说是常规的，因此原始的close方法会抑制GeneratorExit。

参见yield文档的这一部分：

如果生成器在完成之前没有恢复(通过达到零引用计数或垃圾收集)，则将调用生成器迭代器的close()方法，允许执行任何挂起的finally子句。

查看close文档，我们看到：

在生成器函数暂停的点处引发GeneratorExit。如果生成器函数随后正常退出、已关闭或引发GeneratorExit(通过不捕获异常)，那么close将返回给其调用方。

with语句文档的这一部分：

6. 执行套件。

7. 将调用上下文管理器的出口()方法。如果异常导致套件退出，则其类型、值和回溯将作为参数传递给exit()。否则，将提供三个None参数。

以及contextmanager装饰器的__exit__方法的代码。

因此，对于所有这些上下文(投篮)，我们获得所需行为的最简单方法是尝试，除非最后在上下文管理器的定义中。这是contextlib文档中建议的方法。他们所有的例子都遵循这个形式

因此，您可以使用try…except…finally语句来捕获错误(如果有)，或者确保进行一些清理。

import contextlib

@contextlib.contextmanager
def cm():
try:
print("STARTED")
yield
except Exception:
raise
finally:
print("ENDED")

def coro(a: str):
with cm():
print(a)
while True:
val1, val2 = yield
print(val1, val2)

c = coro("HI")

c.send(None)
print("---")
c.send((1, 2))
print("---!")

现在的输出是：

STARTED
HI
---
1 2
---!
ENDED

根据需要。

我们也可以用传统的方式定义我们的上下文管理器：作为一个具有__enter__和__exit__方法的类，仍然得到正确的行为：

class CM:
def __enter__(self):
print('STARTED')
def __exit__(self, exc_type, exc_value, traceback):
print('ENDED')
return False

情况稍微简单一些，因为我们可以准确地看到__exit__方法是什么，而无需查看源代码。GeneratorExit被发送(作为一个参数)到__exit__，在那里__exit__愉快地运行它的清理代码，然后返回False。这不是严格意义上的所必需的，否则会返回None(另一个Falsey值)，但这表明发送到__exit__的任何异常都没有得到处理。(如果没有异常，__exit__的返回值无关紧要)。

Answer 2

您可以通过向协程发送一些会导致它脱离循环并返回的消息来告诉协程关闭，如下所示。这样做会导致在执行此操作时引发StopIteration异常，因此我添加了另一个上下文管理器以允许抑制它。注意，我还添加了一个coroutine装饰器，使它们在第一次调用时自动启动，但该部分是严格可选的。

import contextlib
from typing import Callable

QUIT = 'quit'
def coroutine(func: Callable):
""" Decorator to make coroutines automatically start when called. """
def start(*args, **kwargs):
cr = func(*args, **kwargs)
next(cr)
return cr
return start
@contextlib.contextmanager
def ignored(*exceptions):
try:
yield
except exceptions:
pass

@contextlib.contextmanager
def cm():
print("STARTED")
yield
print("ENDED")
@coroutine
def coro(a: str):
with cm():
print(a)
while True:
value = (yield)
if value == QUIT:
break
val1, val2 = value
print(val1, val2)
print("---")
with ignored(StopIteration):
c = coro("HI")
#c.send(None)  # No longer needed.
c.send((1, 2))
c.send((3, 5))
c.send(QUIT)  # Tell coroutine to clean itself up and exit.
print("---!")

输出：

STARTED
HI
---
1 2
3 5
ENDED
---!