Разработка с asyncio

Режим отладки

По умолчанию asyncio работает в режиме производства. Для облегчения разработки asyncio имеет режим отладки.

Существует несколько способов включения режима отладки asyncio:

• Установить переменную окружения PYTHONASYNCIODEBUG на значение 1.
• Использование режима разработки Python (Режим разработки Python).
• Передача значения debug=True в asyncio.run().
• Вызов loop.set_debug().

В дополнение к включению режима отладки, также следует рассмотреть:

• установку уровня логирования журнала asyncio на logging.DEBUG, например, следующий фрагмент кода можно запустить при запуске приложения:

  logging.basicConfig(level=logging.DEBUG)
  

• настройка модуля warnings для отображения предупреждений ResourceWarning. Один из способов сделать это — использование командной строки с опцией -W default.

Когда режим отладки включен:

• asyncio проверяет наличие коротин, которые не были ожидаемыми и регистрирует их; это помогает избежать ловушки «забытого await».
• Многие небезопасные для потоков API asyncio (например, методы loop.call_soon() и loop.call_at()) генерируют исключение, если они вызываются из неправильного потока.
• Время выполнения селектора ввода-вывода регистрируется, если для выполнения операции ввода-вывода требуется слишком много времени.
• Обработчики, занимающие более 100 миллисекунд, регистрируются. Атрибут loop.slow_callback_duration можно использовать для установки минимальной продолжительности выполнения, которая считается «медленной», в секундах.

Конкурентность и многопоточность

Цикл событий работает в потоке (обычно в главном потоке) и выполняет все обратные вызовы и задачи в своём потоке. Пока задача выполняется в цикле событий, другие задачи не могут выполняться в том же потоке. Когда задача выполняет выражение await, текущая задача приостанавливается, и цикл событий выполняет следующую задачу.

Для планирования обратного вызова (обратного вызова) из другого операционной системы потока, следует использовать метод loop.call_soon_threadsafe(). Пример:

loop.call_soon_threadsafe(callback, *args)

Практически все объекты asyncio не потокобезопасны, что обычно не проблема, если нет кода, работающего с ними извне задачи или обратного вызова. Если для такого кода необходимо вызвать низкоуровневый API asyncio, следует использовать метод loop.call_soon_threadsafe(), например:

loop.call_soon_threadsafe(fut.cancel)

Для планирования объекта корутины из другого потока ОС, следует использовать функцию run_coroutine_threadsafe(). Она возвращает concurrent.futures.Future для доступа к результату:

async def coro_func():
     return await asyncio.sleep(1, 42)

# Later in another OS thread:

future = asyncio.run_coroutine_threadsafe(coro_func(), loop)
# Wait for the result:
result = future.result()

Для обработки сигналов цикл событий должен выполняться в главном потоке.

Метод loop.run_in_executor() может использоваться с concurrent.futures.ThreadPoolExecutor для выполнения блокирующего кода в другом потоке ОС без блокировки потока ОС, в котором выполняется цикл событий.

В настоящее время нет способа запланировать корутины или обратные вызовы непосредственно из другого процесса (такого, как тот, который был запущен с помощью multiprocessing). В разделе «Методы цикла событий» (Методы цикла событий) перечислены API, которые могут читать из труб и наблюдать за дескрипторами файлов без блокировки цикла событий. Кроме того, API подсистемы asyncio для запуска процесса (Подсистема процесса) предоставляют способ запуска процесса и связи с ним из цикла событий. Наконец, упомянутый выше метод loop.run_in_executor() также может быть использован с concurrent.futures.ProcessPoolExecutor для выполнения кода в другом процессе.

Выполнение блокирующего кода

Блокирующий (ёмкий для процессора) код не должен вызываться напрямую. Например, если функция выполняет интенсивные вычисления в процессоре в течение 1 секунды, все конкурентные задачи asyncio и операции ввода-вывода будут задерживаться на 1 секунду.

Для избежания блокировки потока ОС с циклом событий можно использовать исполняемый модуль для выполнения задачи в другом потоке или даже в другом процессе. Подробнее см. метод loop.run_in_executor().

Ведение журнала

asyncio использует модуль logging, и все логирование выполняется через журнал "asyncio".

Уровень логирования по умолчанию — logging.INFO, который легко настроить:

logging.getLogger("asyncio").setLevel(logging.WARNING)

Логирование сети может заблокировать цикл событий. Рекомендуется использовать отдельный поток для обработки логов или использовать неблокирующий ввод-вывод. Например, см. Обработка блокирующих обработчиков.

Обнаружение неожидаемых корутин

Когда функция корутины вызывается, но не ожидается (например, coro() вместо await coro()) или корутина не запланирована с помощью asyncio.create_task(), asyncio выведет RuntimeWarning:

import asyncio

async def test():
    print("never scheduled")

async def main():
    test()

asyncio.run(main())

Вывод:

test.py:7: RuntimeWarning: coroutine 'test' was never awaited
  test()

Вывод в режиме отладки:

test.py:7: RuntimeWarning: coroutine 'test' was never awaited
Coroutine created at (most recent call last)
  File "../t.py", line 9, in <module>
    asyncio.run(main(), debug=True)

  < .. >

  File "../t.py", line 7, in main
    test()
  test()

Обычно решение заключается в ожидании корутины или вызове функции asyncio.create_task():

async def main():
    await test()

Обнаружение необработанных исключений

Если вызывается Future.set_exception(), но объект Future никогда не ожидается, исключение никогда не будет распространено в код пользователя. В этом случае asyncio выведет сообщение в журнал при сборке мусора объекта Future.

Пример необработанного исключения:

import asyncio

async def bug():
    raise Exception("not consumed")

async def main():
    asyncio.create_task(bug())

asyncio.run(main())

Вывод:

Task exception was never retrieved
future: <Task finished coro=<bug() done, defined at test.py:3>
  exception=Exception('not consumed')>

Traceback (most recent call last):
  File "test.py", line 4, in bug
    raise Exception("not consumed")
Exception: not consumed

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

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

Вывод в режиме отладки:

Task exception was never retrieved
future: <Task finished coro=<bug() done, defined at test.py:3>
    exception=Exception('not consumed') created at asyncio/tasks.py:321>

source_traceback: Object created at (most recent call last):
  File "../t.py", line 9, in <module>
    asyncio.run(main(), debug=True)

< .. >

Traceback (most recent call last):
  File "../t.py", line 4, in bug
    raise Exception("not consumed")
Exception: not consumed