Many concurrent requests block the event loop

首先使用 asyncio 和 uvloop 运行提供的 25,000-task reproduction，然后检查 asyncpg/protocol.pyx 中 _on_result__simple_query 和 waiter.set_result(...) 附近的代码，报告指出延迟出现在这里。完成的标准是识别并修复并发请求下的事件循环阻塞问题，然后重新运行 reproduction，以确认报告的停顿已得到解决。

• asyncpg version: 0.28.0
• PostgreSQL version: 14.7
• Do you use a PostgreSQL SaaS? If so, which? Can you reproduce
  the issue with a local PostgreSQL install?: Issue occurs on RDS and local Docker Postgres.
• Python version: 3.9.16. Reproduction also works on 3.10.11.
• Platform: Ubuntu 20.04.5 LTS
• Do you use pgbouncer?: No.
• Did you install asyncpg with pip?: Yes. The issue occurs with both pip install asyncpg==0.28.0 and install from source.
• If you built asyncpg locally, which version of Cython did you use?: 0.29.32
• Can the issue be reproduced under both asyncio and
  uvloop?: Can be reproduced under both asyncio and uvloop.

Reproduction (for simplicity, against a local Docker instance of Postgres).

import asyncpg

import asyncio
import time


async def main():
    asyncio.get_running_loop().slow_callback_duration = 0.05

    pg = await asyncpg.create_pool(
        user="postgres",
        password="",
        database="defaultdb",
        host="localhost",
        port="5432",
    )

    await pg.execute(
        """
        CREATE TABLE IF NOT EXISTS my_table (
            id VARCHAR(20) PRIMARY KEY,
            value VARCHAR(255)
        );
        INSERT INTO my_table (id, value) VALUES ('id-123', '4')
        ON CONFLICT (id) DO NOTHING;
        """
    )

    query = "UPDATE my_table SET value = '4';"

    for i in range(25000):

        async def go():
            for _ in range(10):
                async with pg.acquire() as conn:
                    async with conn.transaction():
                        await conn.execute(query)

        asyncio.create_task(go())

        t0 = time.time()
        await asyncio.sleep(0.001)
        elapsed_ms = (time.time() - t0) * 1000

        if elapsed_ms > 50:
            print(f">>> {i} took {elapsed_ms}ms")


# import uvloop

# asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

asyncio.run(main(), debug=True)

Output:

Executing <Task pending name='Task-1' coro=<main() running at /home/ubuntu/modal/analytics/asyncpg_blocking.py:42> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f975b2e7d90>()] created at /usr/lib/python3.9/asyncio/base_events.py:429> cb=[_run_until_complete_cb() at /usr/lib/python3.9/asyncio/base_events.py:184] created at /usr/lib/python3.9/asyncio/base_events.py:626> took 0.061 seconds
>>> 3851 took 62.56461143493652ms

Executing <Task pending name='Task-9969' coro=<main.<locals>.go() running at /home/ubuntu/modal/analytics/asyncpg_blocking.py:35> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f975bb1f190>()] created at /usr/lib/python3.9/asyncio/base_events.py:429> created at /usr/lib/python3.9/asyncio/tasks.py:361> took 0.095 seconds
>>> 4921 took 96.0381031036377ms

Executing <Handle _SelectorSocketTransport._read_ready() created at /usr/lib/python3.9/asyncio/selector_events.py:254> took 0.126 seconds
>>> 5990 took 128.0958652496338ms

Executing <Task pending name='Task-1' coro=<main() running at /home/ubuntu/modal/analytics/asyncpg_blocking.py:42> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f9759c54490>()] created at /usr/lib/python3.9/asyncio/base_events.py:429> cb=[_run_until_complete_cb() at /usr/lib/python3.9/asyncio/base_events.py:184] created at /usr/lib/python3.9/asyncio/base_events.py:626> took 0.138 seconds
>>> 7064 took 139.1909122467041ms

Executing <Task pending name='Task-1' coro=<main() running at /home/ubuntu/modal/analytics/asyncpg_blocking.py:42> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f975bfd2580>()] created at /usr/lib/python3.9/asyncio/base_events.py:429> cb=[_run_until_complete_cb() at /usr/lib/python3.9/asyncio/base_events.py:184] created at /usr/lib/python3.9/asyncio/base_events.py:626> took 0.166 seconds
>>> 8125 took 167.76275634765625ms

Executing <Handle _SelectorSocketTransport._read_ready() created at /usr/lib/python3.9/asyncio/selector_events.py:254> took 0.193 seconds
>>> 9282 took 194.34309005737305ms
Executing <Task pending name='Task-21463' coro=<main.<locals>.go() running at /home/ubuntu/modal/analytics/asyncpg_blocking.py:35> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f975902e3a0>()] created at /usr/lib/python3.9/asyncio/base_events.py:429> created at /usr/lib/python3.9/asyncio/tasks.py:361> took 0.232 seconds
>>> 10635 took 232.7420711517334ms

Executing <Handle _SelectorSocketTransport._read_ready() created at /usr/lib/python3.9/asyncio/selector_events.py:254> took 0.256 seconds
>>> 12092 took 257.3966979980469ms

Executing <Task pending name='Task-27750' coro=<PoolConnectionHolder.release() running at /home/ubuntu/modal/asyncpg/asyncpg/pool.py:198> wait_for=<Future pending cb=[Protocol._on_waiter_completed(), <TaskWakeupMethWrapper object at 0x7f975b3fc8b0>()] created at /usr/lib/python3.9/asyncio/base_events.py:429> cb=[shield.<locals>._inner_done_callback() at /usr/lib/python3.9/asyncio/tasks.py:890] created at /usr/lib/python3.9/asyncio/tasks.py:883> took 0.292 seconds
>>> 13746 took 293.03622245788574ms

Executing <Task pending name='Task-31496' coro=<main.<locals>.go() running at /home/ubuntu/modal/analytics/asyncpg_blocking.py:35> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f975bb1fe20>()] created at /usr/lib/python3.9/asyncio/base_events.py:429> created at /usr/lib/python3.9/asyncio/tasks.py:361> took 0.321 seconds
>>> 15671 took 321.78592681884766ms

Executing <Task pending name='Task-1' coro=<main() running at /home/ubuntu/modal/analytics/asyncpg_blocking.py:42> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f975ae52ca0>()] created at /usr/lib/python3.9/asyncio/base_events.py:429> cb=[_run_until_complete_cb() at /usr/lib/python3.9/asyncio/base_events.py:184] created at /usr/lib/python3.9/asyncio/base_events.py:626> took 0.375 seconds
>>> 17901 took 376.0251998901367ms

Executing <Task pending name='Task-21270' coro=<main.<locals>.go() running at /home/ubuntu/modal/analytics/asyncpg_blocking.py:37> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /usr/lib/python3.9/asyncio/tasks.py:907, <TaskWakeupMethWrapper object at 0x7f975787ac40>()] created at /usr/lib/python3.9/asyncio/base_events.py:429> created at /usr/lib/python3.9/asyncio/tasks.py:361> took 0.418 seconds
>>> 20313 took 418.8547134399414ms

Executing <Task pending name='Task-46404' coro=<PoolConnectionHolder.release() running at /home/ubuntu/modal/asyncpg/asyncpg/pool.py:198> wait_for=<Future pending cb=[Protocol._on_waiter_completed(), <TaskWakeupMethWrapper object at 0x7f975504dc70>()] created at /usr/lib/python3.9/asyncio/base_events.py:429> cb=[shield.<locals>._inner_done_callback() at /usr/lib/python3.9/asyncio/tasks.py:890] created at /usr/lib/python3.9/asyncio/tasks.py:883> took 0.486 seconds
>>> 23108 took 487.2567653656006m

Further investigation

Adding verbose prints to protocol.pyx led to me chasing down one particular 80ms+ blocking execution, which ended at waiter.set_result(...) in _on_result__simple_query, which took up the majority (150ms out of 151ms, for example) of a slow callback. After this, I wasn't sure how to continue debugging -- open to suggestions or ideas here.

Removing debug=True

The issue is still present, albeit less frequent, without debug mode on

>>> 11807 took 86.72428131103516ms
>>> 16438 took 117.77758598327637ms
>>> 21042 took 172.47319221496582ms

Thanks all!