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

Асинхронное программирование отличается от классического «последовательного» программирования. На этой странице перечислены распространённые ловушки и объяснено, как их избежать.

1. Режим отладки asyncio

Реализация asyncio написана с учётом производительности. Чтобы облегчить разработку асинхронного кода, вы можете включить режим отладки.

Чтобы включить все проверки отладки для приложения:

• Включите режим отладки asyncio глобально, установив переменную среды PYTHONASYNCIODEBUG в 1, или вызвав AbstractEventLoop.set_debug().
• Установите уровень логирования журнала asyncio в logging.DEBUG. Например, вызовите logging.basicConfig(level=logging.DEBUG) при запуске.
• Настройте модуль warnings для отображения предупреждений ResourceWarning. Например, используйте командную строку -Wdefault Python для их отображения.

Примеры проверок отладки:

• Регистрация кортежин, определённых, но ни разу не использованных
• call_soon() и call_at() методы генерируют исключение, если вызываются из неправильной нити.
• Регистрация времени выполнения селектора
• Регистрация обратных вызовов, выполняющихся более 100 мс. Атрибут AbstractEventLoop.slow_callback_duration — минимальная продолжительность «медленных» обратных вызовов в секундах.
• ResourceWarning предупреждения генерируются, когда транспорты и циклы событий не закрываются явно.

2. Отмена

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

Задачу и фьючеры можно отменить явно с помощью метода Future.cancel(). Функция wait_for() отменяет ожидаемую задачу при истечении таймаута. Есть и другие ситуации, когда задача может быть отменена косвенно.

Не вызывайте метод set_result() или set_exception() фьючера Future, если он отменён: это приведёт к ошибке. Например, напишите:

if not fut.cancelled():
    fut.set_result('done')

Не планируйте напрямую вызов метода set_result() или set_exception() фьючера с помощью AbstractEventLoop.call_soon(): фьючер может быть отменён до вызова его метода.

Если вы ждёте фьючер, следует предварительно проверить, не был ли он отменён, чтобы избежать бесполезных операций. Пример:

@coroutine
def slow_operation(fut):
    if fut.cancelled():
        return
    # ... slow computation ...
    yield from fut
    # ...

Функция shield() также может использоваться для игнорирования отмены.

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

Цикл событий работает в потоке и выполняет все обратные вызовы и задачи в одном потоке. Пока задача выполняется в цикле событий, в том же потоке другие задачи не выполняются. Но когда задача использует yield from, она приостанавливается, и цикл событий выполняет следующую задачу.

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

loop.call_soon_threadsafe(callback, *args)

Большинство объектов asyncio не потокобезопасны. Вам следует об этом беспокоиться только если вы обращаетесь к объектам вне цикла событий. Например, чтобы отменить фьючер, не вызывайте напрямую его метод Future.cancel(), а:

loop.call_soon_threadsafe(fut.cancel)

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

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

future = asyncio.run_coroutine_threadsafe(coro_func(), loop)
result = future.result(timeout)  # Wait for the result with a timeout

Метод AbstractEventLoop.run_in_executor() может использоваться с пулом потоков для выполнения обратного вызова в другом потоке, чтобы не блокировать поток цикла событий.

4. Правильная обработка блокирующих функций

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

Для сетевых операций и подпроцессов модуль asyncio предоставляет API высокого уровня, такие как протоколы.

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

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

Модуль asyncio регистрирует информацию с помощью модуля logging в логере 'asyncio'.

Уровень журнала по умолчанию для модуля asyncio — logging.INFO. Для тех, кому не нужна такая подробность от asyncio, уровень журнала можно изменить. Например, чтобы изменить уровень на logging.WARNING:

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

6. Обнаружение не запланированных корутин

Когда функция-корутина вызывается, а её результат не передаётся в ensure_future() или метод AbstractEventLoop.create_task(), выполнение корутины никогда не будет запланировано, что, скорее всего, является ошибкой. Включите режим отладки asyncio, чтобы записать предупреждение для его обнаружения.

Пример с ошибкой:

import asyncio

@asyncio.coroutine
def test():
    print("never scheduled")

test()

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

Coroutine test() at test.py:3 was never yielded from
Coroutine object created at (most recent call last):
  File "test.py", line 7, in <module>
    test()

Исправление заключается в вызове функции ensure_future() или метода AbstractEventLoop.create_task() с объектом корутины.

7. Обнаружение исключений, которые не обрабатываются

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

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

import asyncio

@asyncio.coroutine
def bug():
    raise Exception("not consumed")

loop = asyncio.get_event_loop()
asyncio.ensure_future(bug())
loop.run_forever()
loop.close()

Вывод:

Task exception was never retrieved
future: <Task finished coro=<coro() done, defined at asyncio/coroutines.py:139> exception=Exception('not consumed',)>
Traceback (most recent call last):
  File "asyncio/tasks.py", line 237, in _step
    result = next(coro)
  File "asyncio/coroutines.py", line 141, in coro
    res = func(*args, **kw)
  File "test.py", line 5, in bug
    raise Exception("not consumed")
Exception: not consumed

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

Task exception was never retrieved
future: <Task finished coro=<bug() done, defined at test.py:3> exception=Exception('not consumed',) created at test.py:8>
source_traceback: Object created at (most recent call last):
  File "test.py", line 8, in <module>
    asyncio.ensure_future(bug())
Traceback (most recent call last):
  File "asyncio/tasks.py", line 237, in _step
    result = next(coro)
  File "asyncio/coroutines.py", line 79, in __next__
    return next(self.gen)
  File "asyncio/coroutines.py", line 141, in coro
    res = func(*args, **kw)
  File "test.py", line 5, in bug
    raise Exception("not consumed")
Exception: not consumed

Существует несколько способов исправить эту проблему. Первый вариант — объединить корутину в другую корутину и использовать классический блок try/except:

@asyncio.coroutine
def handle_exception():
    try:
        yield from bug()
    except Exception:
        print("exception consumed")

loop = asyncio.get_event_loop()
asyncio.ensure_future(handle_exception())
loop.run_forever()
loop.close()

Другой вариант — использовать функцию AbstractEventLoop.run_until_complete():

task = asyncio.ensure_future(bug())
try:
    loop.run_until_complete(task)
except Exception:
    print("exception consumed")

8. Правильно объединяйте корутины

Когда функция корутины вызывает другие функции корутин и задачи, их следует явно объединять с помощью yield from. В противном случае последовательность выполнения не гарантируется.

Пример с разными ошибками, использующий asyncio.sleep() для имитации медленных операций:

import asyncio

@asyncio.coroutine
def create():
    yield from asyncio.sleep(3.0)
    print("(1) create file")

@asyncio.coroutine
def write():
    yield from asyncio.sleep(1.0)
    print("(2) write into file")

@asyncio.coroutine
def close():
    print("(3) close file")

@asyncio.coroutine
def test():
    asyncio.ensure_future(create())
    asyncio.ensure_future(write())
    asyncio.ensure_future(close())
    yield from asyncio.sleep(2.0)
    loop.stop()

loop = asyncio.get_event_loop()
asyncio.ensure_future(test())
loop.run_forever()
print("Pending tasks at exit: %s" % asyncio.Task.all_tasks(loop))
loop.close()

Ожидаемый вывод:

(1) create file
(2) write into file
(3) close file
Pending tasks at exit: set()

Фактический вывод:

(3) close file
(2) write into file
Pending tasks at exit: {<Task pending create() at test.py:7 wait_for=<Future pending cb=[Task._wakeup()]>>}
Task was destroyed but it is pending!
task: <Task pending create() done at test.py:5 wait_for=<Future pending cb=[Task._wakeup()]>>

Цикл остановился до завершения create(), close() была вызвана до write(), тогда как функции корутин были вызваны в этом порядке: create(), write(), close().

Чтобы исправить пример, необходимо пометить задачи с помощью yield from:

@asyncio.coroutine
def test():
    yield from asyncio.ensure_future(create())
    yield from asyncio.ensure_future(write())
    yield from asyncio.ensure_future(close())
    yield from asyncio.sleep(2.0)
    loop.stop()

Или без asyncio.ensure_future():

@asyncio.coroutine
def test():
    yield from create()
    yield from write()
    yield from close()
    yield from asyncio.sleep(2.0)
    loop.stop()

9. Уничтожение ожидающей задачи

Если ожидающая задача уничтожается, выполнение её обернутой корутины не завершилось. Вероятно, это ошибка, поэтому генерируется предупреждение.

Пример лога:

Task was destroyed but it is pending!
task: <Task pending coro=<kill_me() done, defined at test.py:5> wait_for=<Future pending cb=[Task._wakeup()]>>

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

Task was destroyed but it is pending!
source_traceback: Object created at (most recent call last):
  File "test.py", line 15, in <module>
    task = asyncio.ensure_future(coro, loop=loop)
task: <Task pending coro=<kill_me() done, defined at test.py:5> wait_for=<Future pending cb=[Task._wakeup()] created at test.py:7> created at test.py:15>

10. Закрытие транспортов и циклов событий

Когда транспорт больше не нужен, вызовите его метод close() для освобождения ресурсов. Циклы событий также должны быть закрыты явно.

Если транспорт или цикл событий не закрыты явно, в его деструкторе будет выведено предупреждение ResourceWarning. По умолчанию, предупреждения ResourceWarning игнорируются. Раздел Отладочный режим asyncio описывает, как их отображать.