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

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

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

Включить режим отладки asyncio можно несколькими способами:

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

Помимо включения режима отладки, рекомендуется также:

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

  logging.basicConfig(level=logging.DEBUG)
  

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

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

• asyncio проверяет корутины, для которых не был выполнен await, и записывает их в журнал; это помогает избежать ошибки «забытый 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, чтобы выполнять код в другом процессе.

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

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

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

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

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

По умолчанию установлен уровень ведения журнала logging.INFO; его можно легко изменить:

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

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

Обнаружение корутин, для которых не был выполнен await

Если функция-корутина вызвана, но для неё не выполнен 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 так и не ожидается, исключение не будет передано пользовательскому коду. В этом случае 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