Как создать монки-патч для режима отладки AsyncIO для получения более подробной информации?

Я хочу выяснить, что именно блокирует мой event loop. Я запускаю режим отладки с помощью <code class="python">asyncio.run(main(), debug=True)</code> и добавил <code class="python">time.sleep()</code> для примера. В логах я вижу следующую запись: <pre><code class="python">Executing &lt;Task pending name='Task-73' coro=&lt;Dispatcher._process_update() running at /usr/local/lib/python3.12/site-packages/aiogram/dispatcher/dispatcher.py:309&gt; wait_for=&lt;Future pending cb=[Protocol._on_waiter_completed(), Task.task_wakeup()] created at /usr/local/lib/python3.12/site-packages/asyncpg/connection.py:1945&gt; cb=[set.discard()] created at /usr/local/lib/python3.12/asyncio/tasks.py:420&gt; took 2.108 seconds</code></pre> Он фиксирует, что есть задержки, но логирование выглядит неудовлетворительно. Как мне реализовать monkey patch для получения строки кода, которая вызывает блокировку, в данном случае <code class="python">time.sleep</code>? Я использую uvloop и Python 3.12.
  • 15 сентября 2024 г. 17:43
Ответы на вопрос 2
Для создания monkey-patch для вашего AsyncIO-кода, чтобы улучшить логирование и отладку блокировок в вашем event loop, можно заменить стандартные blocking функции, такие как `time.sleep()`, на асинхронные эквиваленты или обернуть их в такую функцию, которая будет фиксировать стек вызовов (call stack). В вашем случае `time.sleep()` блокирует весь event loop, и это может быть трудно отследить.

Вот пример, как сделать это:

1. **Замените blocking функции.** Вы можете создать свою обертку для `time.sleep()`, которая будет регистрировать стек вызовов перед вызовом блочащей функции.

2. **Используйте Python's `traceback` модуль.** Он поможет вам получить информацию о стеке вызовов.

Вот пример кода, который показывает, как это сделать:

```python
import time
import asyncio
import traceback
import functools

# Сохраним оригинальную функцию
original_sleep = time.sleep

def monkey_patched_sleep(seconds):
    stack = traceback.format_stack()
    print(f"Blocking call to time.sleep({seconds})\n" + ''.join(stack))
    original_sleep(seconds)

# Заменяем time.sleep на наш monkey-patch
time.sleep = monkey_patched_sleep

async def main():
    print("Start of main")
    await asyncio.sleep(1)  # это не блокирует event loop
    time.sleep(2)            # это будет блокировать event loop
    print("End of main")

if __name__ == "__main__":
    asyncio.run(main(), debug=True)
```

### Как это работает:
- Мы сохраняем оригинальную функцию `time.sleep`.
- Мы создаем новую функцию `monkey_patched_sleep`, которая вызывает `traceback.format_stack()` для получения информации о стеке вызовов и выводим её в консоль (или обрабатываем по-другому, в зависимости от ваших потребностей).
- Затем мы переназначаем `time.sleep` на нашу новую функцию.

### Ограничения:
- Этот метод будет работать только для `time.sleep`. Если у вас есть другие блокирующие функции, такие как `threading.sleep`, вам нужно будет проделать то же самое с ними.
- Это решение может создавать много вывода, если у вас часто происходят блокирующие вызовы, так что вам, возможно, стоит сделать логирование более избирательным, основанным на уровне или других критериях.

Используйте этот метод с осторожностью и тестируйте его в ваших условиях, так как его использование может значительно увеличить вывод в консоли.
Поскольку судя по описанию задача найти проблему, порекомендую готовый инструмент <a href="https://pypi.org/project/yappi/" rel="nofollow">yappi</a> для профилирования асинк кода, чтобы не городить свой.
Похожие вопросы