Как сделать Monkey patch для AsyncIO debug mode для детальной информации?

Question

[Tayrus0]

Я хочу отследить что блокирует мой event loop, я запускаю режим дебага asyncio.run(main(), debug=True), и поставил time.sleep() ради примера

Executing <Task pending name='Task-73' coro=<Dispatcher._process_update() running at /usr/local/lib/python3.12/site-packages/aiogram/dispatcher/dispatcher.py:309> wait_for=<Future pending cb=[Protocol._on_waiter_completed(), Task.task_wakeup()] created at /usr/local/lib/python3.12/site-packages/asyncpg/connection.py:1945> cb=[set.discard()] created at /usr/local/lib/python3.12/asyncio/tasks.py:420> took 2.108 seconds

он фиксирует, что есть тормоза, но логирует ужасно, как сделать monkey patch для того чтобы конкретно получить строку кода, которая тормозит, в нашем случае time.sleep? Я использую uvloop и python 3.12

Comments

[Tayrus0]

«Пользователь удалён», Мне нужен именно патч для этого механизма обнаружения блокировок

[Tayrus0]

«Пользователь удалён», поэтому вопрос и задан

Answer 1

[fenrir]

Поскольку судя по описанию задача найти проблему, порекомендую готовый инструмент yappi для профилирования асинк кода, чтобы не городить свой.

Comments

[Tayrus0]

Пробовал его, замудренный, так и не понятно как в нем определять блокировки, если покажите буду рад, для меня пока что встроенный дебаг мод проще

[fenrir]

Tayrus0,

Пример

import time
import asyncio
import yappi

def burn_cpu(secs):
    t0 = time.process_time()
    elapsed = 0
    while (elapsed <= secs):
        for _ in range(1000):
            pass
        elapsed = time.process_time() - t0


async def burn_async_io(secs):
    await asyncio.sleep(secs)


def burn_io(secs):
    time.sleep(secs)


async def main():
    # yappi.set_clock_type("wall")
    yappi.start()
    burn_cpu(1.0)
    await burn_async_io(1.0)
    burn_io(1.0)
    await burn_async_io(1.0)
    yappi.get_func_stats().print_all()

if __name__ == '__main__':
    asyncio.run(main())

Результат