python - aiohttp.web + aiopg + sqlalchemy : random "cursor.execute() called while another coroutine is already waiting for incoming data" under load

Question

我有一个简单的 aiohttp.web 应用程序，它通过 aiopg sqlalchemy 集成执行 SQL 请求。很简单:

import aiohttp.web
from aiopg.sa import create_engine

app = aiohttp.web.Application()


async def rows(request):
    async with request.app["db"].acquire() as db:
        return aiohttp.web.json_response(list(await db.execute("SELECT * FROM table")))

app.router.add_route("GET", "/rows", rows)


async def init(app):
    app["db"] = await create_engine(host="postgres", user="visio", password="visio", database="visio")

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    handler = app.make_handler()

    loop.run_until_complete(init(app))

    loop.run_until_complete(loop.create_server(handler, "0.0.0.0", 80))
    loop.run_forever()

当服务器负载达到 100 rps 时，此错误开始随机出现:

RuntimeError: cursor.execute() called while another coroutine is already waiting for incoming data
  File "aiohttp/server.py", line 261, in start
    yield from self.handle_request(message, payload)
  File "aiohttp/web.py", line 88, in handle_request
    resp = yield from handler(request)
  File "visio_longer/views/communicate/__init__.py", line 72, in legacy_communicate
    device = await query_device(db, access_token)
  File "visio_longer/views/communicate/__init__.py", line 31, in query_device
    (Device.access_token == access_token)
  File "aiopg/utils.py", line 72, in __await__
    resp = yield from self._coro
  File "aiopg/sa/connection.py", line 103, in _execute
    yield from cursor.execute(str(compiled), post_processed_params[0])
  File "aiopg/cursor.py", line 103, in execute
    waiter = self._conn._create_waiter('cursor.execute')
  File "aiopg/connection.py", line 186, in _create_waiter
    'already waiting for incoming data' % func_name)

它会在随机时间随机查询发生，几天一次，有时这些错误会以 4 或 2 个的形式出现。我的代码有问题吗？ aiohttp 和 aiopg 版本是最新的 pip

Answer 1

它认为你忘记了获取光标

 async with conn.cursor() as cur:
     await cur.execute("SELECT 1")