异步上下文管理器是否需要保护其清理代码不被取消？

Question

问题（我认为）

{a1}文档给出了以下示例：

@asynccontextmanager
async def get_connection():
    conn = await acquire_db_connection()
    try:
        yield conn
    finally:
        await release_db_connection(conn)

在我看来，这可能会泄露资源。如果此代码的任务是cancelled，而此代码位于其await release_db_connection(conn)行上，则发布可能会中断。asyncio.CancelledError将从finally块内的某个地方向上传播，从而阻止后续清理代码运行

因此，实际上，如果您正在实现一个处理超时请求的web服务器，那么在错误的时间触发超时可能会导致数据库连接泄漏

完整的可运行示例

import asyncio
from contextlib import asynccontextmanager

async def acquire_db_connection():
    await asyncio.sleep(1)
    print("Acquired database connection.")
    return "<fake connection object>"

async def release_db_connection(conn):
    await asyncio.sleep(1)
    print("Released database connection.")

@asynccontextmanager
async def get_connection():
    conn = await acquire_db_connection()
    try:
        yield conn
    finally:
        await release_db_connection(conn)

async def do_stuff_with_connection():
    async with get_connection() as conn:
        await asyncio.sleep(1)
        print("Did stuff with connection.")

async def main():
    task = asyncio.create_task(do_stuff_with_connection())

    # Cancel the task just as the context manager running
    # inside of it is executing its cleanup code.
    await asyncio.sleep(2.5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

    print("Done.")

asyncio.run(main())

Python 3.7.9上的输出：

Acquired database connection.
Did stuff with connection.
Done.

请注意Released database connection从不打印

我的问题

• 这是个问题，对吧？直觉上，我认为.cancel()的意思是“优雅地取消，清理沿途使用的任何资源。”（否则，他们为什么要将取消实现为异常传播？）。也许，例如，.cancel()意味着快速而不是优雅。是否有权威人士澄清.cancel()在这里应该做什么
• 如果这确实是一个问题，我该如何解决它

Answer 1

专注于保护清理工作不被取消是一种转移注意力的做法。有许多事情可能出错，而上下文管理器无法知道

• 可能发生哪些错误，以及
• 必须防止哪些错误

正确处理错误是资源处理实用程序的责任

• 如果release_db_connection不能被取消，它必须保护自己不被取消
• 如果acquire/release必须成对运行，则它必须是单个async with上下文管理器。内部也可能涉及进一步的保护，例如防止取消

async def release_db_connection(conn):
    """
    Cancellation safe variant of `release_db_connection`

    Internally protects against cancellation by delaying it until cleanup.
    """
    # cleanup is run in separate task so that it
    # cannot be cancelled from the outside.
    shielded_release = asyncio.create_task(asyncio.sleep(1))
    # Wait for cleanup completion – unlike `asyncio.shield`,
    # delay any cancellation until we are done.
    try:
        await shielded_release
    except asyncio.CancelledError:
        await shielded_release
        # propagate cancellation when we are done
        raise
    finally:
        print("Released database connection.")

注意：异步清理很棘手。例如，a simple ^{} is not sufficient if the event loop does not wait for shielded tasks.避免发明自己的保护，依靠底层框架做正确的事情

---

任务的取消是a）仍然允许异步操作且b）可能被延迟/抑制的正常关机。明确允许准备好处理CancelledError以进行清理的协程

Task.cancel

The coroutine then has a chance to clean up or even deny the request by suppressing the exception with a try … … except CancelledError … finally block. […] Task.cancel() does not guarantee that the Task will be cancelled, although suppressing cancellation completely is not common and is actively discouraged.

强制关机是coroutine.close/GeneratorExit。这对应于立即同步关机，并禁止通过await、async for或async with进行暂停

coroutine.close

[…] it raises GeneratorExit at the suspension point, causing the coroutine to immediately clean itself up.

Answer 2

您可以使用asyncio.shield保护任务为了保证上下文管理器正常关闭，我只在main()中做了更改：

async def main():
    task = asyncio.create_task(do_stuff_with_connection())
    # shield context manager from cancellation
    sh_task = asyncio.shield(task)
    # Cancel the task just as the context manager running
    # inside of it is executing its cleanup code.
    await asyncio.sleep(2.5)
    sh_task.cancel()  # cancel shielded task
    try:
        await sh_task
    except asyncio.CancelledError:
        pass

    await asyncio.sleep(5)  # wait till shielded task is done

    print("Done.")