Разработка с 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 для их отображения.

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

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

18.5.9.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() также может использоваться для игнорирования отмены.

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

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

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

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

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

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 объясняет, как их отобразить.