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

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

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

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