Концептуальный обзор asyncio¶A Conceptual Overview of asyncio

Цикл событий¶Event Loop

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

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

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

import asyncio

# Это создаёт цикл событий и бесконечно перебирает
# свою коллекцию задач.
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():
    # Это был бы плохой генератор случайных чисел!
    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)
# Это создаёт объект задачи и планирует его выполнение через цикл событий.
task = asyncio.create_task(coroutine)

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

import asyncio

async def main():
    # Выполняются всевозможные безумные асинхронные действия...
    ...

if __name__ == "__main__":
    asyncio.run(main())
    # Программа не дойдёт до следующего оператора print, пока
    # корутина main() не завершится.
    print("coroutine main() is done!")

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

 1async def hello():
 2    print("hello!")
 3
 4async def main():
 5    asyncio.create_task(hello())
 6    # Другие асинхронные инструкции, которые выполняются некоторое время
 7    # и передают управление циклу событий...
 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

    # Другие инструкции, связанные с посадкой дерева.
    ...

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

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

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 объект, который yields в своём методе __await__.

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

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

Future¶Futures

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

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

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

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

Самодельный asyncio.sleep¶A homemade 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():
    # Добавьте несколько других задач в цикл событий, чтобы было чем заняться
    # во время асинхронного ожидания.
    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 фактически ожидает каждую задачу в коллекции.
    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
    # Добавьте задачу-наблюдатель в цикл событий.
    watcher_task = asyncio.create_task(_sleep_watcher(future, time_to_wake))
    # Блокируется до тех пор, пока future не будет отмечен как завершённый.
    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:
            # Это помечает future как завершённый.
            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.