Концептуальный обзор asyncio

Эта статья HOWTO помогает сформировать устойчивую ментальную модель того, как на фундаментальном уровне работает asyncio, помогая понять причины и логику рекомендуемых шаблонов.

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

• Что происходит за кулисами, когда ожидается объект?

• Как asyncio различает задачу, не требующую процессорного времени (например, сетевой запрос или чтение файла), и задачу, которая требует (например, вычисление n-факториал)?

• Как написать асинхронный вариант операции, например, async sleep или запрос к базе данных.

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

• Руководство, вдохновившее эту статью HOWTO, автора Александра Нордина.

• Этот подробный цикл видеоуроков на YouTube по asyncio, созданный участником основной команды Python, Лукашем Лангой.

• 500 Lines or Less: A Web Crawler With asyncio Coroutines, авторы А. Джесси Джирю Дэвис и Гвидо ван Россум.

A conceptual overview part 1: the high-levelКонцептуальный обзор, часть 1: высокоуровневый обзор

В части 1 мы рассмотрим основные высокоуровневые строительные блоки asyncio: цикл событий, корутинные функции, объекты-корутины, задачи и await.

Event loopЦикл событий

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

Если говорить более технически, цикл событий содержит набор задач для выполнения. Некоторые задачи добавляются напрямую вами, а некоторые — косвенно через asyncio. Цикл событий берёт задачу из своего накопителя работы и вызывает её (или «передаёт ей управление»), аналогично вызову функции, после чего задача выполняется. Как только она приостанавливается или завершается, она возвращает управление циклу событий. Затем цикл событий выберет другую задачу из своего пула и вызовет её. Вы можете примерно представлять набор задач как очередь: задачи добавляются и затем обрабатываются по одной, как правило (но не всегда), по порядку. Этот процесс повторяется бесконечно, цикл событий крутится без остановки. Если задач, ожидающих выполнения, больше нет, цикл событий достаточно умён, чтобы отдыхать и избегать ненужной траты тактов процессора, и вернётся, когда появится новая работа, например, когда завершатся операции ввода-вывода или истечёт таймер.

Эффективная работа зависит от того, насколько хорошо задачи делят ресурсы и сотрудничают; жадная задача может захватить управление и оставить другие задачи голодать, что сделает подход с циклом событий в целом бесполезным.

import asyncio

# This creates an event loop and indefinitely cycles through
# its collection of jobs.
event_loop = asyncio.new_event_loop()
event_loop.run_forever()

Asynchronous functions and coroutinesАсинхронные функции и корутины

Это обычная, скучная функция Python:

def hello_printer():
    print(
        "Hi, I am a lowly, simple printer, though I have all I "
        "need in life -- \nfresh paper and my dearly beloved octopus "
        "partner in crime."
    )

Вызов обычной функции запускает её логику или тело:

>>> hello_printer()
Hi, I am a lowly, simple printer, though I have all I need in life --
fresh paper and my dearly beloved octopus partner in crime.

Конструкция async def, в отличие от обычного def, делает функцию асинхронной (или «корутинной функцией»). Её вызов создаёт и возвращает объект корутины.

async def loudmouth_penguin(magic_number: int):
    print(
     "I am a super special talking penguin. Far cooler than that printer. "
     f"By the way, my lucky number is: {magic_number}."
    )

Вызов асинхронной функции loudmouth_penguin не выполняет оператор print; вместо этого он создаёт объект-корутину:

>>> loudmouth_penguin(magic_number=3)
<coroutine object loudmouth_penguin at 0x104ed2740>

Термины «корутинная функция» и «объект-корутина» часто смешивают, называя и то, и другое просто «корутиной». Это может сбивать с толку! В этой статье под корутиной понимается именно объект-корутина, или точнее, экземпляр types.CoroutineType (нативная корутина). Обратите внимание, что корутины могут также существовать как экземпляры collections.abc.Coroutine – это различие имеет значение для проверки типов.

Корутина представляет тело или логику функции. Корутину необходимо явно запускать; ещё раз: просто создание корутины не запускает её. Примечательно, что корутину можно приостанавливать и возобновлять в разных точках тела функции. Именно эта способность приостанавливаться и возобновляться обеспечивает асинхронное поведение!

Корутины и корутинные функции были построены с использованием функциональности генераторов и генераторных функций. Напомним, генераторная функция — это функция, которая yield, например:

def get_random_number():
    # This would be a bad random number generator!
    print("Hi")
    yield 1
    print("Hello")
    yield 7
    print("Howdy")
    yield 4
    ...

Подобно корутинной функции, вызов генераторной функции не запускает её. Вместо этого она создаёт объект-генератор:

>>> get_random_number()
<generator object get_random_number at 0x1048671c0>

Перейти к следующему yield генератора можно с помощью встроенной функции next(). Иными словами, генератор выполняется, затем приостанавливается. Например:

>>> generator = get_random_number()
>>> next(generator)
Hi
1
>>> next(generator)
Hello
7

TasksЗадачи

Грубо говоря, задачи — это корутины (не корутинные функции), привязанные к циклу событий. Задача также поддерживает список функций обратного вызова (колбэков), важность которых станет ясна через минуту, когда мы обсудим await.

Создание задачи автоматически планирует её выполнение (путём добавления колбэка для её запуска в список дел цикла событий, то есть коллекцию заданий). Рекомендуемый способ создания задач — через asyncio.create_task().

asyncio автоматически связывает задачи с циклом событий. Эта автоматическая связь была намеренно заложена в asyncio ради простоты. Без неё пришлось бы отслеживать объект цикла событий и передавать его любой корутинной функции, которая хочет создавать задачи, добавляя избыточный мусор в код.

coroutine = loudmouth_penguin(magic_number=5)
# This creates a Task object and schedules its execution via the event loop.
task = asyncio.create_task(coroutine)

Ранее мы вручную создавали цикл событий и запускали его на бесконечное выполнение. На практике рекомендуется использовать (и часто встречается) asyncio.run(), который берёт на себя управление циклом событий и гарантирует, что переданная корутина завершится до продолжения. Например, многие асинхронные программы следуют такой схеме:

import asyncio

async def main():
    # Perform all sorts of wacky, wild asynchronous things...
    ...

if __name__ == "__main__":
    asyncio.run(main())
    # The program will not reach the following print statement until the
    # coroutine main() finishes.
    print("coroutine main() is done!")

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

 1async def hello():
 2    print("hello!")
 3
 4async def main():
 5    asyncio.create_task(hello())
 6    # Other asynchronous instructions which run for a while
 7    # and cede control to the event loop...
 8    ...
 9
10asyncio.run(main())

Поскольку нет ссылки на объект задачи, созданный на строке 5, он может быть удалён сборщиком мусора до того, как цикл событий его вызовет. Последующие инструкции в корутине main() возвращают управление циклу событий, чтобы тот мог выполнить другие задания. Когда цикл событий в конце концов попытается запустить задачу, он может потерпеть неудачу и обнаружить, что объекта задачи не существует! Это может произойти даже если корутина хранит ссылку на задачу, но завершается до того, как задача закончится. При выходе из корутины локальные переменные выходят из области видимости и могут быть подвержены сборке мусора. На практике asyncio и сборщик мусора Python прилагают немало усилий, чтобы такого не случалось. Но это не повод быть беспечным!

await

await — это ключевое слово Python, которое чаще всего используется одним из двух способов:

await task
await coroutine

В решающей степени поведение await зависит от типа ожидаемого объекта.

Awaiting tasksОжидание задач

Ожидание задачи передаёт управление от текущей задачи или корутины циклу событий. В процессе передачи управления происходит несколько важных вещей. Для иллюстрации воспользуемся следующим примером кода:

async def plant_a_tree():
    dig_the_hole_task = asyncio.create_task(dig_the_hole())
    await dig_the_hole_task

    # Other instructions associated with planting a tree.
    ...

В этом примере представьте, что цикл событий передал управление началу корутины plant_a_tree(). Как видно выше, корутина создаёт задачу и затем ожидает её. Инструкция await dig_the_hole_task добавляет колбэк (который возобновит plant_a_tree()) в список колбэков объекта dig_the_hole_task. Затем инструкция передаёт управление циклу событий. Некоторое время спустя цикл событий передаст управление dig_the_hole_task, и задача завершит всё, что ей нужно сделать. Как только задача завершится, она добавит свои различные колбэки в цикл событий; в данном случае это вызов для возобновления plant_a_tree().

В общем случае, когда ожидаемая задача завершается (dig_the_hole_task), исходная задача или корутина (plant_a_tree()) добавляется обратно в список дел цикла событий для возобновления.

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

Awaiting coroutinesОжидание корутин

В отличие от задач, ожидание корутины не возвращает управление циклу событий! Сначала обернуть корутину в задачу, а затем ожидать её — это бы передало управление. Поведение await coroutine фактически аналогично вызову обычной синхронной функции Python. Рассмотрим такую программу:

import asyncio

async def coro_a():
   print("I am coro_a(). Hi!")

async def coro_b():
   print("I am coro_b(). I sure hope no one hogs the event loop...")

async def main():
   task_b = asyncio.create_task(coro_b())
   num_repeats = 3
   for _ in range(num_repeats):
      await coro_a()
   await task_b

asyncio.run(main())

Первый оператор в корутине main() создаёт task_b и планирует его выполнение через цикл событий. Затем coro_a() многократно ожидается. Управление никогда не передаётся циклу событий, поэтому мы видим вывод всех трёх вызовов coro_a() до вывода coro_b():

I am coro_a(). Hi!
I am coro_a(). Hi!
I am coro_a(). Hi!
I am coro_b(). I sure hope no one hogs the event loop...

Если изменить await coro_a() на await asyncio.create_task(coro_a()), поведение меняется. Корутина main() передаёт управление циклу событий этим оператором. Затем цикл событий проходит по своему накопившемуся списку задач, вызывая task_b, а затем задачу, которая оборачивает coro_a(), прежде чем возобновить корутину main().

I am coro_b(). I sure hope no one hogs the event loop...
I am coro_a(). Hi!
I am coro_a(). Hi!
I am coro_a(). Hi!

Такое поведение await coroutine может сбить с толку многих! Этот пример показывает, как использование только await coroutine может непреднамеренно захватить управление у других задач и фактически заблокировать цикл событий. asyncio.run() может помочь обнаружить такие случаи с помощью флага debug=True, который включает режим отладки. Среди прочего, он будет регистрировать все корутины, которые монополизируют выполнение на 100 мс или дольше.

Дизайн намеренно жертвует некоторой концептуальной ясностью в использовании await в пользу повышения производительности. Каждый раз, когда ожидается задача, управление должно быть передано вверх по стеку вызовов к циклу событий. Затем цикл событий должен управлять своим внутренним состоянием и выполнить свою логику обработки, чтобы возобновить следующее задание. Это может показаться незначительным, но в большой программе с множеством await эти накладные расходы могут привести к заметному снижению производительности.

A conceptual overview part 2: the nuts and boltsКонцептуальный обзор, часть 2: механизмы работы

Часть 2 подробно рассматривает механизмы, которые asyncio использует для управления потоком управления. Вот где происходит волшебство. Из этого раздела вы узнаете, что await делает за кулисами и как создавать собственные асинхронные операторы.

The inner workings of coroutinesВнутреннее устройство корутин

asyncio использует четыре компонента Python для передачи управления.

coroutine.send(arg) — это метод, используемый для запуска или возобновления корутины. Если корутина была приостановлена и теперь возобновляется, аргумент arg будет отправлен как возвращаемое значение оператора yield, который изначально её приостановил. Если корутина используется впервые (а не возобновляется), arg должно быть None.

 1class Rock:
 2    def __await__(self):
 3        value_sent_in = yield 7
 4        print(f"Rock.__await__ resuming with value: {value_sent_in}.")
 5        return value_sent_in
 6
 7async def main():
 8    print("Beginning coroutine main().")
 9    rock = Rock()
10    print("Awaiting rock...")
11    value_from_rock = await rock
12    print(f"Coroutine received value: {value_from_rock} from rock.")
13    return 23
14
15coroutine = main()
16intermediate_result = coroutine.send(None)
17print(f"Coroutine paused and returned intermediate value: {intermediate_result}.")
18
19print(f"Resuming coroutine and sending in value: 42.")
20try:
21    coroutine.send(42)
22except StopIteration as e:
23    returned_value = e.value
24print(f"Coroutine main() finished and provided value: {returned_value}.")

yield, как обычно, приостанавливает выполнение и возвращает управление вызывающему. В приведённом выше примере yield на строке 3 вызывается ... = await rock на строке 11. В более широком смысле, await вызывает метод __await__() данного объекта. await также делает ещё одну очень важную вещь: он распространяет (или «передаёт») любые yield, которые он получает, вверх по цепочке вызовов. В данном случае это обратно к ... = coroutine.send(None) на строке 16.

Корутина возобновляется вызовом coroutine.send(42) на строке 21. Корутина продолжает с того места, где она yield (или приостановилась) на строке 3, и выполняет оставшиеся операторы своего тела. Когда корутина завершается, она возбуждает исключение StopIteration с возвращаемым значением, прикреплённым в атрибуте value.

Этот фрагмент даёт следующий вывод:

Beginning coroutine main().
Awaiting rock...
Coroutine paused and returned intermediate value: 7.
Resuming coroutine and sending in value: 42.
Rock.__await__ resuming with value: 42.
Coroutine received value: 42 from rock.
Coroutine main() finished and provided value: 23.

Здесь стоит на мгновение остановиться и убедиться, что вы проследили различные способы, которыми передавались управление и значения. Было рассмотрено много важных идей, и стоит убедиться, что ваше понимание прочно.

Единственный способ выполнить yield (или фактически уступить управление) из корутины — await объект, который yield в своём методе __await__. Возможно, это звучит странно. Вы, вероятно, думаете:

1. А как насчёт yield непосредственно внутри корутинной функции? Корутинная функция становится асинхронной генераторной функцией, совершенно другой сущностью.

2. А как насчёт yield from внутри корутинной функции к (обычному) генератору? Это вызывает ошибку: SyntaxError: yield from not allowed in a coroutine. Это было намеренно сделано для простоты — предписан только один способ использования корутин. Несмотря на это, yield from и await фактически делают то же самое. Изначально yield также был запрещён, но был повторно принят, чтобы разрешить асинхронные генераторы.

FuturesFuture

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

У future есть несколько важных атрибутов. Один из них — его состояние, которое может быть «pending», «cancelled» или «done». Другой — его результат, который устанавливается, когда состояние переходит в done. В отличие от корутины, future не представляет само вычисление; вместо этого он представляет состояние и результат этого вычисления, что-то вроде индикатора (красный, жёлтый, зелёный) или индикатора состояния.

asyncio.Task является подклассом asyncio.Future для получения этих различных возможностей. В предыдущем разделе говорилось, что задачи хранят список колбэков, что было не совсем точно. На самом деле класс Future реализует эту логику, которую наследует Task.

Объекты future также могут использоваться напрямую (не через задачи). Задачи помечают себя как выполненные, когда их корутина завершена. Объекты future гораздо более универсальны и будут помечены как выполненные, когда вы это укажете. Таким образом, они представляют собой гибкий интерфейс для создания собственных условий ожидания и возобновления.

A homemade asyncio.sleepСамодельный asyncio.sleep

Мы рассмотрим пример того, как можно использовать future для создания собственного варианта асинхронного sleep (async_sleep), который имитирует asyncio.sleep().

Этот фрагмент регистрирует несколько задач в цикле событий, а затем ожидает задачу, созданную asyncio.create_task, которая оборачивает корутину async_sleep(3). Мы хотим, чтобы эта задача завершилась только после того, как пройдёт три секунды, но не препятствуя выполнению других задач.

async def other_work():
    print("I like work. Work work.")

async def main():
    # Add a few other tasks to the event loop, so there's something
    # to do while asynchronously sleeping.
    work_tasks = [
        asyncio.create_task(other_work()),
        asyncio.create_task(other_work()),
        asyncio.create_task(other_work())
    ]
    print(
        "Beginning asynchronous sleep at time: "
        f"{datetime.datetime.now().strftime("%H:%M:%S")}."
    )
    await asyncio.create_task(async_sleep(3))
    print(
        "Done asynchronous sleep at time: "
        f"{datetime.datetime.now().strftime("%H:%M:%S")}."
    )
    # asyncio.gather effectively awaits each task in the collection.
    await asyncio.gather(*work_tasks)

Ниже мы используем future, чтобы обеспечить собственный контроль над тем, когда эта задача будет помечена как выполненная. Если future.set_result() (метод, отвечающий за пометку этого future как выполненного) никогда не будет вызван, то эта задача никогда не завершится. Мы также привлекли помощь другой задачи, которую мы увидим через мгновение, которая будет отслеживать, сколько времени прошло, и соответственно вызывать future.set_result().

async def async_sleep(seconds: float):
    future = asyncio.Future()
    time_to_wake = time.time() + seconds
    # Add the watcher-task to the event loop.
    watcher_task = asyncio.create_task(_sleep_watcher(future, time_to_wake))
    # Block until the future is marked as done.
    await future

Ниже мы используем довольно простой объект YieldToEventLoop() для yield из его метода __await__, уступая управление циклу событий. Это практически то же самое, что и вызов asyncio.sleep(0), но такой подход обеспечивает большую ясность, не говоря уже о том, что использовать asyncio.sleep при демонстрации реализации — это своего рода обман!

Как обычно, цикл событий перебирает свои задачи, передавая им управление и получая его обратно, когда они приостанавливаются или завершаются. watcher_task, который выполняет корутину _sleep_watcher(...), будет вызываться один раз за полный цикл событий. При каждом возобновлении он будет проверять время, и если прошло недостаточно времени, то снова приостановится и вернёт управление циклу событий. Когда пройдёт достаточно времени, _sleep_watcher(...) помечает future как выполненный и завершается, выходя из своего бесконечного цикла while. Учитывая, что эта вспомогательная задача вызывается только один раз за цикл событий, вы будете правы, заметив, что этот асинхронный sleep будет спать как минимум три секунды, а не ровно три секунды. Обратите внимание, что это также верно для asyncio.sleep.

class YieldToEventLoop:
    def __await__(self):
        yield

async def _sleep_watcher(future, time_to_wake):
    while True:
        if time.time() >= time_to_wake:
            # This marks the future as done.
            future.set_result(None)
            break
        else:
            await YieldToEventLoop()

Вот вывод полной программы:

$ python custom-async-sleep.py
Beginning asynchronous sleep at time: 14:52:22.
I like work. Work work.
I like work. Work work.
I like work. Work work.
Done asynchronous sleep at time: 14:52:25.

Возможно, вы считаете, что эта реализация асинхронного sleep была излишне запутанной. И, да, так и было. Пример был предназначен для демонстрации универсальности futures на простом примере, который можно повторить для более сложных задач. Для справки, вы могли бы реализовать это без futures, например так:

async def simpler_async_sleep(seconds):
    time_to_wake = time.time() + seconds
    while True:
        if time.time() >= time_to_wake:
            return
        else:
            await YieldToEventLoop()

Но на этом пока всё. Надеемся, вы готовы более уверенно погрузиться в асинхронное программирование или изучить продвинутые темы в rest of the documentation.