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

18.5.9.1. Режим отладки asyncio¶Debug mode of 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 выводятся, когда транспорты и циклы событий не закрыты явно.

18.5.9.2. Отмена¶Cancellation

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

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, следует заранее проверить, не был ли future отменён, чтобы избежать бесполезных операций. Пример:

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

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

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

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

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

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

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

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

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

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

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

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

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

Когда вызывается корутинная функция и её результат не передаётся в 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. Обнаружение непоглощённых исключений¶Detect exceptions never consumed

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

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