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

Асинхронное программирование отличается от классического последовательного программирования.

На этой странице перечислены распространенные ошибки, ловушки и описаности, а также способы их избежания.

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

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

Существует несколько способов активации режима асинхронной отладки:

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

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

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

    logging.basicConfig(level=logging.DEBUG)

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

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

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

Параллелизм и многопоточность

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

Чтобы запланировать колбэк из другого потока ОС, следует использовать метод loop.call_soon_threadsafe(). Пример:

loop.call_soon_threadsafe(callback, *args)

Почти все asyncio объекты не являются потокобезопасными, что обычно не является проблемой, если нет кода, который работает с ними извне Task или обратного вызова. Если необходимо, чтобы такий код вызывал низкоуровневый asyncio API, то следует использовать метод 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)

# Позже в другом потоке  ОС

future = asyncio.run_coroutine_threadsafe(coro_func(), loop)
# Ожидание результата:
result = future.result()

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

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

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

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

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

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

Логгирование

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

По умолчанию уровень журналирования logging.INFO, который может быть легко изменён:

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

Обнаружение never-awaited корутин

Когда вызывается функция корутина, без await (например, 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()

Фикситс это добавлением await корутине, либо вызвать функцию asyncio.create_task():

async def main():
    await test()

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

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

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

import asyncio

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

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

asyncio.run(main())

Вывод:

Исключение задачи никогда не будет получено
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