aiohttp.web + aiopg + SQLAlchemy: случайный «cursor.execute() вызывается, а другие сопрограммной уже ждут поступающих данных» под нагрузкой

Question

У меня есть тривиальное aiohttp.web приложения, которое выполняет запросы SQL с помощью aiopg интеграции SQLAlchemy. Это так же просто, как:

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 версий являются последними из пипа

Answer 1

Это кажется, что вы забыли приобрести курсор

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