Разработка с 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.

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

• Многие небезопасные для потоков 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 или InterpreterPoolExecutor, чтобы выполнять блокирующий код в другом потоке ОС, не блокируя поток ОС, в котором работает цикл событий.

В настоящее время нет способа напрямую планировать сопрограммы или обратные вызовы из другого процесса (например, запущенного с помощью 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

Рекомендации по использованию асинхронных генераторов

Для написания корректного и эффективного кода asyncio необходимо знать о некоторых подводных камнях. В этом разделе изложены основные рекомендации, которые помогут сэкономить часы отладки.

import asyncio
import contextlib

async def gen():
  yield 1
  yield 2

async def func():
  async with contextlib.aclosing(gen()) as g:
    async for x in g:
      break  # Don't iterate until the end

asyncio.run(func())

import asyncio
work_done = False

async def cursor():
    try:
        yield 1
    finally:
        assert work_done

async def rows():
    global work_done
    try:
        yield 2
    finally:
        await asyncio.sleep(0.1) # immitate some async work
        work_done = True


async def main():
    async for c in cursor():
        async for r in rows():
            break
        break

asyncio.run(main())

unhandled exception during asyncio.run() shutdown
task: <Task finished name='Task-3' coro=<<async_generator_athrow without __name__>()> exception=AssertionError()>
Traceback (most recent call last):
  File "example.py", line 6, in cursor
    yield 1
asyncio.exceptions.CancelledError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "example.py", line 8, in cursor
    assert work_done
           ^^^^^^^^^
AssertionError

async def main():
    async with contextlib.aclosing(cursor()) as cursor_gen:
        async for c in cursor_gen:
            async with contextlib.aclosing(rows()) as rows_gen:
                async for r in rows_gen:
                    break
            break

import asyncio

async def agenfn():
    try:
        yield 10
    finally:
        await asyncio.sleep(0)


with asyncio.Runner() as runner:
    agen = agenfn()
    print(runner.run(anext(agen)))
    del agen

10
Exception ignored while closing generator <async_generator object agenfn at 0x000002F71CD10D70>:
Traceback (most recent call last):
  File "example.py", line 13, in <module>
    del agen
        ^^^^
RuntimeError: async generator ignored GeneratorExit

import asyncio

async def agenfn():
    try:
        yield 10
    finally:
        await asyncio.sleep(0)

async def main():
    agen = agenfn()
    print(await anext(agen))
    del agen

asyncio.run(main())

import asyncio

async def consumer():
    for idx in range(100):
        await asyncio.sleep(0)
        message = yield idx
        print('received', message)

async def amain():
    agenerator = consumer()
    await agenerator.asend(None)

    fa = asyncio.create_task(agenerator.asend('A'))
    fb = asyncio.create_task(agenerator.asend('B'))
    await fa
    await fb

asyncio.run(amain())

received A
Traceback (most recent call last):
  File "test.py", line 38, in <module>
    asyncio.run(amain())
    ~~~~~~~~~~~^^^^^^^^^
  File "Lib/asyncio/runners.py", line 204, in run
    return runner.run(main)
           ~~~~~~~~~~^^^^^^
  File "Lib/asyncio/runners.py", line 127, in run
    return self._loop.run_until_complete(task)
           ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  File "Lib/asyncio/base_events.py", line 719, in run_until_complete
    return future.result()
           ~~~~~~~~~~~~~^^
  File "test.py", line 36, in amain
    await fb
RuntimeError: anext(): asynchronous generator is already running