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

18.5.9.1. Режим отладки asyncio¶Debug mode of asyncio

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

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

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

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

• Журналируйте корутины, которые определены, но никогда не вызывались через yield from
• Методы call_soon() и call_at() вызывают исключение, если они вызваны из неправильного потока.
• Журналируйте время выполнения селектора
• Журналировать колбэки, выполнение которых занимает более 100 мс. Атрибут BaseEventLoop.slow_callback_duration задаёт минимальную длительность в секундах для «медленных» колбэков.
• Предупреждения ResourceWarning выводятся, когда транспорты и циклы событий не закрыты явно.

18.5.9.2. Отмена¶Cancellation

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

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

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

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

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

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

@coroutine
def slow_operation(fut):
    if fut.cancelled():
        return
    # ... медленное вычисление ...
    yield from fut
    # ...

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

18.5.9.3. Конкурентность и многопоточность¶Concurrency and multithreading

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

Чтобы запланировать колбэк из другого потока, следует использовать метод BaseEventLoop.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)  # Ожидание результата с тайм-аутом

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

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

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

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

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

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

18.5.9.5. Журналирование¶Logging

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

18.5.9.6. Обнаружение корутинных объектов, которые никогда не были запланированы¶Detect coroutine objects never scheduled

Когда вызывается корутинная функция, а её результат не передаётся в async() или метод BaseEventLoop.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()

Исправление – вызвать функцию async() или метод BaseEventLoop.create_task() с объектом корутины.

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

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

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

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

18.5.9.8. Правильное построение цепочек корутин¶Chain coroutines correctly

Когда корутинная функция вызывает другие корутинные функции и задачи, их следует явно связывать с помощью 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. Уничтожение ожидающей задачи¶Pending task destroyed

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

Пример записи в журнале:

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 transports and event loops

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

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