在异步生成器函数中从托儿所内部进行屈服是否不好？

Question

我被告知以下代码不安全，因为不允许有一个从托儿所内部生成的异步生成器，除非它是异步上下文管理器。

T = TypeVar('T')

async def delay(interval: float, source: AsyncIterable[T]) -> AsyncIterable[T]:
    """Delays each item in source by an interval.

    Received items are temporarily stored in an unbounded queue, along with a timestamp, using
    a background task. The foreground task takes items from the queue, and waits until the
    item is older than the given interval and then yields it."""

    send_channel, receive_channel = trio.open_memory_channel(math.inf)

    async def pull_task():
        async with aclosing(source) as agen:
            async for item in agen:
                send_channel.send_nowait((item, trio.current_time() + interval))

    async with trio.open_nursery() as nursery:
        nursery.start_soon(pull_task)
        async with receive_channel:
            async for item, timestamp in receive_channel:
                now = trio.current_time()
                if timestamp > now:
                    await trio.sleep(timestamp - now)
                yield item

我很难理解这怎么可能会被打破。如果任何人都可以提供使用这个确切的生成器函数的示例代码，这证明了不安全性，我们将不胜感激和奖励。

上述代码的目标是延迟异步序列的处理，而不施加任何反压。如果您能证明这段代码并不像我期望的那样工作，那也将不胜感激。

谢谢。

Answer 1

yield不幸的是，不支持托儿所内的正确 \xe2\x80\x93或取消范围，除非在@contextlib.asynccontextmanager使用创建异步上下文管理器或编写异步 pytest 固定装置的狭隘情况下。

\n

有几个原因。其中一些是技术性的：Trio 必须跟踪堆栈上当前“活动”的托儿所/取消作用域，当您yield超出其中一个时，它就会破坏嵌套，并且 Trio 无法知道您已经做了这个。（库无法检测yield上下文管理器之外的情况。）

\n

但还有一个根本性的、无法解决的原因，那就是 Trio 和结构化并发的整个思想是，每个任务“属于”一个父任务，如果子任务崩溃，父任务可以收到通知。但是，当您yield在生成器中时，生成器框架会被冻结并与当前任务 \xe2\x80\x93 分离，它可能会在另一个任务中恢复，或者根本不会恢复。因此yield，当您这样做时，就会破坏托儿所中所有子任务与其父母之间的联系。没有办法将其与结构化并发的原则相协调。

\n

在 Trio 聊天中，Joshua Oreman给出了一个适合您的案例的具体示例：

\n

\n

如果我运行以下命令

\n

async def arange(*args):\n    for val in range(*args):\n        yield val\n\nasync def break_it():\n    async with aclosing(delay(0, arange(3))) as aiter:\n        with trio.move_on_after(1):\n            async for value in aiter:\n                await trio.sleep(0.4)\n                print(value)\n\ntrio.run(break_it)\n

\n

然后我得到

\n

RuntimeError: Cancel scope stack corrupted: attempted to exit\n<trio.CancelScope at 0x7f364621c280, active, cancelled> in <Task\n\'__main__.break_it\' at 0x7f36462152b0> that\'s still within its child\n<trio.CancelScope at 0x7f364621c400, active>\n\nThis is probably a bug in your code, that has caused Trio\'s internal\nstate to become corrupted. We\'ll do our best to recover, but from now\non there are no guarantees.\n\nTypically this is caused by one of the following:\n  - yielding within a generator or async generator that\'s opened a cancel\n    scope or nursery (unless the generator is a @contextmanager or\n    @asynccontextmanager); see https://github.com/python-trio/trio/issues/638 [...]\n

\n

通过更改超时和延迟，使超时在生成器内部而不是在生成器外部过期，我还能够得到不同的错误：trio.MultiError: Cancelled(), GeneratorExit() raised out of aclosing()

\n

\n

这里还有关于所有这些问题的长时间讨论，这就是我们发现这不能被支持的地方： https: //github.com/python-trio/trio/issues/264

\n

这是一种不幸的情况，因为我们无法支持它，这很遗憾，更糟糕的是它看起来可以在简单的情况下工作，所以人们最终可能会编写大量使用它的代码在意识到它不起作用之前先尝试一下:-(

\n

我们的计划是让非法的情况在你尝试的时候立即给出明显的错误yield，至少避免第二个问题。但是，这需要一段时间，因为它需要向 Python 解释器添加一些额外的钩子。

\n

还可以创建一个几乎与异步生成器一样易于编写和使用的构造，但可以避免此问题。这个想法是，您不需要将生成器从使用它的任务的堆栈中推送和弹出，而是将“生成器”代码作为第二个任务运行，为消费者任务提供值。有关更多详细信息，请参阅从此处开始的线程。

\n