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

Цикл событий

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

Асинхронные функции и сопрограммы

Вот обычная, ничем не примечательная функция 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

Задачи

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

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

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

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

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

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

В отличие от ожидания задачи, ожидание сопрограммы не передаёт управление обратно циклу событий! Если сначала обернуть сопрограмму в задачу, а затем ожидать её, управление будет передано циклу событий. По сути, поведение 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 и глубоко вложенным стеком вызовов такие накладные расходы могут существенно снизить производительность.

Внутреннее устройство сопрограмм

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

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 тоже было запрещено, но затем его снова разрешили для поддержки асинхронных генераторов. Несмотря на это, yield from и await фактически выполняют одно и то же.

Объекты Future

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

У объекта Future есть несколько важных атрибутов. Один из них — состояние, которое может быть «ожидающим», «отменённым» или «завершённым». Другой — результат, который устанавливается при переходе в состояние «завершён». В отличие от сопрограммы, объект Future не представляет само вычисление; он представляет состояние и результат этого вычисления — примерно как сигнальная лампа (красная, жёлтая или зелёная) или индикатор.

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

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

Самодельная asyncio.sleep

Рассмотрим пример создания собственного варианта асинхронной задержки (async_sleep) с помощью объекта Future. Он будет имитировать asyncio.sleep().

Этот фрагмент регистрирует несколько задач в цикле событий, а затем ожидает сопрограмму, обёрнутую в задачу: 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. Поскольку эта вспомогательная задача вызывается лишь один раз за цикл событий, вы правильно заметите, что асинхронная задержка продлится не менее трёх секунд, а не ровно три секунды. Обратите внимание, что это справедливо и для 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.

Возможно, вам кажется, что реализация асинхронной задержки получилась излишне запутанной. Так и есть. Этот пример должен был показать универсальность объектов Future на простом примере, который можно адаптировать для более сложных задач. Для сравнения, вот как можно реализовать это без объектов Future:

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.