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

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

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

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

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

• Включите режим отладки asyncio глобально, установив переменную окружения PYTHONASYNCIODEBUG в 1 или вызвав AbstractEventLoop.set_debug().

• Установите уровень журнала для логгера asyncio в logging.DEBUG. Например, вызовите logging.basicConfig(level=logging.DEBUG) при запуске.

• Настройте модуль warnings для отображения предупреждений ResourceWarning. Например, используйте параметр командной строки -Wdefault Python для их отображения.

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

• Журналировать корутины, определённые, но никогда не вызывавшиеся через "yield from"

• Методы call_soon() и call_at() вызывают исключение, если они вызываются из неправильного потока.

• Журналировать время выполнения селектора

• Журналировать колбэки, выполнение которых занимает более 100 мс. Атрибут AbstractEventLoop.slow_callback_duration – это минимальная продолжительность «медленных» колбэков в секундах.

• Предупреждения ResourceWarning выдаются, когда транспорты и циклы событий не закрыты явно.

Смотрите также

Метод AbstractEventLoop.set_debug() и логгер asyncio.

18.5.9.2. Отмена

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

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

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

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

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

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

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

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

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

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

Для планирования колбэка из другого потока следует использовать метод AbstractEventLoop.call_soon_threadsafe(). Пример:

loop.call_soon_threadsafe(callback, *args)

Большинство объектов asyncio не являются потокобезопасными. Беспокоиться об этом стоит только при доступе к объектам вне цикла событий. Например, чтобы отменить future, не вызывайте напрямую его метод 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)  # Ожидать результат с таймаутом

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

Смотрите также

Раздел Примитивы синхронизации описывает способы синхронизации задач.

Раздел Подпроцессы и потоки перечисляет ограничения asyncio на запуск подпроцессов из разных потоков.

18.5.9.4. Корректная обработка блокирующих функций

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

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

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

Смотрите также

Раздел Отложенные вызовы подробно описывает, как цикл событий обрабатывает время.

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

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

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

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

18.5.9.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() с корутинным объектом.

Смотрите также

Ожидающая задача уничтожена.

18.5.9.7. Обнаружение исключений, которые никогда не были обработаны

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

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

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")

Смотрите также

Метод Future.exception().

18.5.9.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()

18.5.9.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, чтобы получить tрассировку, где была создана задача. Пример лога в режиме отладки:

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>

Смотрите также

Обнаружение корутинных объектов, которые никогда не были запланированы.

18.5.9.10. Закрыть транспорты и циклы событий

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

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