在典型的aiohttp web服务器中应该如何处理CancelledError？

Question

我有一个带有如下处理程序的aiohttp web服务器应用程序：

async def handler(request):
    async with request.app["db"].acquire() as db:
        row = await query(db)

    return aiohttp.web.json_response(row)

其中app["db"]是某种池化资源(aiopg、aioredis，现在已经无关紧要了)。直到今天，这一切都运行得很好。所有客户端都无缘无故地因超时而开始断开连接，并且应用程序日志中充满了如下跟踪

[2017-05-21 17:58:24,254] ERROR    [aiohttp.server] Error handling request
Traceback (most recent call last):
  File "/virtualenv/lib/python3.6/site-packages/aiohttp/web_server.py", line 61, in handle_request
    resp = yield from self._handler(request)
  File "/virtualenv/lib/python3.6/site-packages/aiohttp/web.py", line 249, in _handle
    resp = yield from handler(request)
  File "/visio-longer/visio_longer/views/communicate/__init__.py", line 81, in legacy_communicate
    async with request.app["db"].acquire() as db:
  File "/virtualenv/src/aiopg/aiopg/utils.py", line 140, in __aenter__
    self._conn = yield from self._coro
  File "/virtualenv/src/aiopg/aiopg/sa/engine.py", line 162, in _acquire
    raw = yield from self._pool.acquire()
  File "/virtualenv/src/aiopg/aiopg/utils.py", line 67, in __iter__
    resp = yield from self._coro
  File "/virtualenv/src/aiopg/aiopg/pool.py", line 168, in _acquire
    with (yield from self._cond):
  File "/usr/lib/python3.6/asyncio/locks.py", line 67, in __iter__
    yield from self.acquire()
  File "/usr/lib/python3.6/asyncio/locks.py", line 176, in acquire
    yield from fut
concurrent.futures._base.CancelledError

这里的关键点是，当从池中获取数据库连接时，它正在接收CancelledError (客户端超时断开连接)：

  File "/visio-longer/visio_longer/views/communicate/__init__.py", line 81, in legacy_communicate
    async with request.app["db"].acquire() as db:

我运行了一个协程程序，它每5秒打印一次池状态(size和freesize)，此时池中有大量空闲连接！

经过几个小时的调查，理论上，在执行池的上下文管理器__atexit__时接收CancelledError会中止将连接返回到池的过程，这会导致池故障。我找到了一个在asyncpg中修复这种行为的commit，aioredis包含类似的代码，我还创建了一个指向aiopg的awkward-looking fix。所有这些都无济于事--我仍然从aioredis和aiopg得到相同的错误。

该情况已通过替换

async def handler(request):
    async with request.app["db"].acquire() as db:
        row = await query(db)

    return aiohttp.web.json_response(row)

通过使用asyncio.shield包装使用连接池的每一段代码

async def handler(request):
    async def process():
        async with request.app["db"].acquire() as db:
            row = await query(db)

    return aiohttp.web.json_response(asyncio.shield(process(row)))

因此，中止的请求仍然被处理到它们的末端(包括将获取的资源返回到池)。

应该是这样的吗？现在我的代码看起来很糟糕，而且不能保证下一次我不会忘记用asyncio.shield包装我的池。解决这个问题的正确方法是什么(很明显，库本身不能解决这个问题)？

Answer 1

我看不出这与其他任何异常有什么不同。CanceledError只是另一个错误。所有的上下文管理器出口函数都会运行，因此它们应该有机会释放资源。它看起来就像是一个与连接池中的锁相关的未来正在被取消。这似乎是有问题的，看起来应该与客户端连接被取消无关。我假设有一个对应于正在进行的请求的协程，当连接丢失时，协程应该被取消。对Task.cancel的调用将在协程中引发CanceledError。假设CanceledError最终从协程中被提升，Task的结果将是CanceledError。然而，我不明白与锁相关的未来是如何被取消的。

如果您正在等待上下文管理器的exit函数中的某些内容，或者finally块，或者其他一些清理代码，您可能需要更加小心。例如，如果您需要在exit函数中使用锁来返回某些资源，那么您可能需要一个捕获CanceledError (并最终重新引发它)或使用async.shield的while循环。我只会在exit函数/清理处理程序中的特定代码中这样做，而不是在整个请求中。此外，考虑使用回调来返回资源(通过loop.call_soon)是否比在端清理代码中等待更好。如果清理代码不等待，则不能在其中引发新的CanceledError。上下文管理器退出函数和最终块可以由CanceledError在其他地方触发，即使它们不等待，但这与任何其他异常没有什么不同