Цикл событий: один поток, очередь и шесть шагов, которые повторяются

Один человек за стойкой

Представьте стойку выдачи. За ней один человек. Перед ним лежит стопка бумажек — дела, которые пора сделать. Рядом будильник: его можно завести на любое время. И есть окошко, куда просовывают новые заявки.

Правило у него одно, и оно важнее всего остального:

Взял стопку — сделай всё, что в ней было в момент, когда взял. Бумажки, подложенные пока работал, — в следующий заход.

Когда стопка пуста и будильник ещё не звонит, человек спит. Не сидит, глядя в стену, а спит по-настоящему — и просыпается либо от будильника, либо когда в окошко просунут заявку.

Вот и весь цикл событий. Всё остальное — подробности.

Из чего он сделан

Если открыть исходники, окажется, что это буквально три вещи:

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

Никакого «планировщика», который решает, кому дать поработать, там нет. Есть очередь и правило «делай всё, что в ней лежало».

Переключите вкладку «Из чего состоит» — там эти три части нарисованы с их настоящими именами и стрелками, кто куда попадает.

Что делает await

await — это не «подожди здесь». Это «мне нужен результат; если его пока нет, положи меня спать и займись другими».

async def получить_данные():
    ответ = await запрос_к_серверу()   # <- вот здесь функция уходит спать
    return ответ.text                  # <- а сюда вернётся, когда ответ придёт

Между этими двумя строками может пройти секунда, и всю секунду человек за стойкой занимается другими делами. Функция не «висит» — её просто отложили в сторону, а когда результат появился, положили обратно в стопку.

Важная деталь: если результат уже готов, никто никуда не уходит. Функция просто продолжается. Поэтому три вложенных await подряд, где никто ничего не ждёт, стоят ровно столько же, сколько пустая программа, — это измерено.

Почему один time.sleep ломает всё

Теперь понятно и главное правило асинхронного кода.

async def плохо():
    time.sleep(0.1)          # человек за стойкой замер на 100 мс
 
async def хорошо():
    await asyncio.sleep(0.1) # человек завёл будильник и занялся другими

Разница не в том, что первое «медленнее». Разница в том, что первое занимает единственного работника, и всё остальное просто не происходит.

Измерено на CPython 3.13.7 — пятьдесят задач, каждая занята 100 мс:

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

Второй замер на той же вкладке нагляднее. Фоновая задача просит будить её каждые 10 миллисекунд — за секунду она должна тикнуть примерно сто раз.

• Работник блокирует: тикнула 4 раза.
• Работник уступает: тикнула 99 раз.

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

Что делать, если блокирующее неизбежно

Иногда нужно вызвать библиотеку, которая про асинхронность не знает. Для этого есть готовый способ — отдать её отдельному потоку:

результат = await asyncio.to_thread(медленная_функция, аргумент)

Человек за стойкой заводит будильник и возвращается к другим делам, а медленная функция работает в стороне. Это единственный правильный ответ на «мне нужно вызвать блокирующее».

Три вещи, которые стоит запомнить

Работник один. Всё остальное — следствия.

await — это не пауза, а отметка «здесь можно отложить». Если откладывать нечего, ничего и не происходит.

Медленно не то, что долго ждёт, а то, что долго занимает. Задача, которая десять секунд ждёт ответа по сети, не мешает никому. Задача, которая сто миллисекунд считает, мешает всем.

Зачем это знать?

Асинхронный код в Python пишут все, а объяснить, что происходит между await и продолжением функции, могут немногие. Обычно это не мешает — пока не случается одно из трёх: сервис необъяснимо «залипает» на секунду, asyncio.sleep(0.01) почему-то спит 30 мс, или добавление ещё одной задачи роняет отзывчивость всего приложения.

Все три случая объясняются одним механизмом, и он умещается в одну функцию на восемьдесят строк. Эта статья разбирает её целиком — и проверяет каждое утверждение замером на живом интерпретаторе.

Модель в голове

Представьте одного человека за стойкой. Перед ним лежит стопка бумажек — это очередь готовых дел. Рядом стоит будильник, который можно завести на любое время. И есть окошко, в которое иногда просовывают новые заявки.

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

Пока стопка пуста, а будильник ещё не звонит, он спит — не «крутится вхолостую», а действительно спит, и просыпается либо от будильника, либо от заявки в окошке.

Всё остальное в этой статье — уточнение этой картинки до уровня, на котором её можно проверить кодом. Одна деталь важна сразу: человек один. Пока он занят одной бумажкой, он не смотрит ни на будильник, ни в окошко. Не потому, что не хочет, а потому что он один.

Из чего состоит цикл

Прежде чем разбирать, что цикл делает, стоит увидеть, из чего он сделан. Это буквально объект с тремя контейнерами, и всё происходящее в асинхронной программе — перекладывание работы между ними.

# base_events.py, BaseEventLoop.__init__ — тег v3.13.7
class BaseEventLoop(events.AbstractEventLoop):
    def __init__(self):                                    # :419
        self._timer_cancelled_count = 0                    # :420
        self._ready = collections.deque()                  # :423
        self._scheduled = []                               # :424
        self._clock_resolution = time.get_clock_info('monotonic').resolution  # :430

# selector_events.py, BaseSelectorEventLoop.__init__ — тот же тег
        if selector is None:
            selector = selectors.DefaultSelector()         # :63
        self._selector = selector                          # :65

Пять строк — и это весь состав.

_ready — очередь того, что пора делать прямо сейчас. Обычная deque. Это единственный вход в исполнение: и сработавший таймер, и готовый сокет, и call_soon — всё сначала оказывается здесь и только потом выполняется.

_scheduled — куча таймеров, отсортированная по времени срабатывания (обычный список, но работают с ним через heapq, поэтому в голове всегда ближайший дедлайн). Сюда попадает всё, что просили сделать «потом»: call_later, call_at и внутренности asyncio.sleep.

_selector — обёртка над epoll (или kqueue, или select — по системе). Единственное, что знает про сокеты, и единственное место, где процесс действительно спит.

Над этими тремя — слой задач и футур, а под ними ровно одна функция, которая что-либо вызывает. Причём цикл о задачах не знает вообще: Task и Future живут поверх него и общаются с ним только через call_soon и call_later. Для цикла всё сущее — это колбэки.

Вкладка «Из чего состоит» ниже показывает эту схему целиком, с настоящими именами полей и стрелками, кто куда попадает. Переключите на «Схему оборота» — и та же картинка начнёт подсвечивать те части, которые работают на каждом шаге.

Шесть шагов одного оборота

Цикл событий в CPython — это метод _run_once в Lib/asyncio/base_events.py. Его вызывает run_forever, и вся конструкция буквально такая:

# base_events.py:678–687, тег v3.13.7
def run_forever(self):
    ...
    self._run_forever_setup()
    try:
        while True:
            self._run_once()
            if self._stopping:
                break
    finally:
        self._run_forever_cleanup()

Внутри _run_once — шесть шагов. Вернитесь к визуализации выше и переключите её в режим «Схема оборота»: она подсветит, какая часть работает на каждом шаге. В режиме «Запись прогона» те же шаги показаны на настоящей программе.

Шаг, ради которого стоит остановиться, — последний. В исходниках рядом с ним стоит комментарий, который объясняет устройство планировщика лучше любого пересказа:

И следом — то, что делает планировщик честным:

# base_events.py:2033–2035
ntodo = len(self._ready)
for i in range(ntodo):
    handle = self._ready.popleft()

ntodo считается до цикла. Колбэк, который добавят в _ready во время этого прохода, в него уже не попадёт — он дождётся следующего оборота. Без этой строчки задача, которая бесконечно ставит себе продолжение через call_soon, заняла бы цикл навсегда.

Сколько оборотов делает ваша программа

Здесь начинаются измерения. Считать обороты можно точно: подменить BaseEventLoop._run_once обёрткой со счётчиком до создания цикла. Это не оценка — это ровно те итерации, которые были.

Три строки в этой таблице стоит прочитать дважды.

Пустая программа делает шесть оборотов, а не один. Причина — не в цикле, а в asyncio.run: это три последовательных run_until_complete, и только первый исполняет ваш код. Второй закрывает асинхронные генераторы (shutdown_asyncgens), третий — пул потоков по умолчанию (shutdown_default_executor). Каждый даёт пару оборотов. В режиме «Запись прогона» выше эти четыре оборота видны поимённо в конце.

Три вложенных await стоят столько же, сколько пустая программа. То есть нисколько. await сам по себе цикла не касается — об этом ниже.

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

Что такое await на самом деле

Корутина — это генератор с другим протоколом. Её двигают вызовом .send(None), а return прилетает наружу как StopIteration.value. Это можно проверить руками, вообще без цикла событий:

async def inner():
    return "значение из return"
 
coro = inner()
try:
    coro.send(None)          # первый шаг — и сразу последний
except StopIteration as e:
    print(e.value)           # значение из return

Теперь то же самое, но с приостановкой. Весь протокол умещается в пять строк Future.__await__:

# futures.py:283–289, тег v3.13.7
def __await__(self):
    if not self.done():
        self._asyncio_future_blocking = True
        yield self  # This tells Task to wait for completion.
    if not self.done():
        raise RuntimeError("await wasn't used with future")
    return self.result()  # May raise too.

yield self пробрасывается вверх через всю цепочку await и вылезает наружу как результат coro.send(None). Флаг _asyncio_future_blocking — это условный знак «я настоящий awaitable, паркуй задачу»; задача проверяет его в tasks.py:324–325.

Отсюда следствие, которое объясняет строку «три вложенных await» из таблицы: если по пути не встретилось ни одного приостанавливающегося Future, вся цепочка await схлопывается без единого захода в цикл. Она стоит ровно столько, сколько стоит создать объекты корутин.

Что это в наносекундах:

Вторая и третья строки почти совпадают, и это ответ на вопрос «дорог ли await». Из 56,6 нс около 54 — это создание самого объекта корутины. Собственно машинерия await стоит единицы наносекунд. Дорога не она, а настоящая приостановка с возвратом в цикл: 1,7 мкс, в тридцать раз больше.

Как футура будит задачу

Это место, где чаще всего ошибаются, и ошибка звучит невинно: «set_result возобновляет корутину». Не возобновляет.

# futures.py:158–170, тег v3.13.7
def __schedule_callbacks(self):
    callbacks = self._callbacks[:]
    if not callbacks:
        return
    self._callbacks[:] = []
    for callback, ctx in callbacks:
        self._loop.call_soon(callback, self, context=ctx)

set_result, set_exception и cancel — все три заканчиваются вызовом этой функции, а она ставит колбэк в очередь. Корутина продолжится не здесь, а в шестом шаге следующего оборота.

Полная петля выглядит так:

1. await fut → Future.__await__ делает yield self;
2. задача видит флаг, вешает add_done_callback(self.__wakeup) и запоминает футуру в _fut_waiter (tasks.py:341–343) — с этого момента цикл о задаче не знает;
3. когда-то потом кто-то вызывает fut.set_result(v);
4. __schedule_callbacks кладёт Task.__wakeup в _ready через call_soon;
5. шестой шаг оборота вызывает __wakeup, тот вызывает __step, тот — coro.send(None);
6. __await__ возвращает self.result(), и функция продолжается со следующей строки.

asyncio.sleep(delay) — самый короткий полный пример этой петли:

# tasks.py:713–720, тег v3.13.7
future = loop.create_future()
h = loop.call_later(delay, futures._set_result_unless_cancelled, future, result)
try:
    return await future
finally:
    h.cancel()

Футура, таймер, ожидание — и всё. Никакой «магии сна» в asyncio нет: есть таймер, который через delay секунд положит результат в футуру, и это разбудит задачу по петле выше.

Почему один блокирующий вызов останавливает всё

Теперь механизм собран, и главное следствие можно не декларировать, а показать.

Возьмём N задач, каждая занята 100 мс. В одном варианте — time.sleep(0.1), в другом — await asyncio.sleep(0.1). Замеряется общее время, число оборотов и длительность самого долгого оборота.

Смотреть надо не на время, а на число оборотов. У блокирующего варианта оно не растёт вовсе: и одна задача, и пятьдесят дают два оборота. Растёт длительность одного оборота — до 5011 мс.

Это ровно то, что предсказывает ntodo = len(self._ready). Все пятьдесят задач были готовы одновременно, попали в очередь и были съедены одним проходом шестого шага. Их блокировки сложились.

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

Вкладка «Отзывчивость» переводит это на язык, который чувствуется: фоновая задача просит будить её каждые 10 мс, пока воркер делает десять порций работы по 100 мс. При блокирующем воркере она тикнула 4 раза за секунду с медианой интервала 300,9 мс. При кооперативном — 99 раз с медианой 10,2 мс.

Почему 300, а не 100 — вопрос, на который стоит ответить, потому что ответ целиком выводится из шестого шага. Пульс теряет оборот дважды: таймер, истёкший во время блокировки, снимается только в начале следующего оборота (шаг 5), а колбэк, попавший в _ready во время оборота, выполняется лишь оборотом позже — из-за ntodo. Три оборота по 100 мс и дают 300. Проверено трассировкой: в записи видно TaskStepMethWrapper в трёх оборотах подряд, прежде чем появляется Task.task_wakeup пульса.

Что изменилось в 3.12, 3.13 и 3.14

Само ядро цикла — не менялось. Тело _run_once в тегах v3.13.7 и v3.14.0 совпадает побайтово, все 83 строки; сдвинулся только номер первой строки, 1970 → 1966. Числа оборотов в таблице выше одинаковы во всех трёх версиях — планировщик считает одинаково, различается только скорость.

Менялось вокруг.

3.12 — eager tasks. asyncio.eager_task_factory заставляет корутину начинать выполняться синхронно прямо в конструкторе задачи. Документация тут предельно точна: «Tasks are only scheduled on the event loop if they block». Отсюда свойство, которое стоит знать до того, как включать: ускорение получают только задачи, завершающиеся без единой приостановки. Если тело приостанавливается, выигрыша нет — задача всё равно регистрируется в цикле.

3.13 — корректность отмены. Исправлена ситуация, в которой две вложенные TaskGroup с одновременными исключениями могли повиснуть; заодно группы стали сохранять счётчик отмен (Task.cancelling()). Появились asyncio.Queue.shutdown, Server.close_clients().

3.14 — интроспекция и свободные потоки. Появился python -m asyncio ps PID и pstree PID — просмотр дерева задач живого процесса, и модуль asyncio.graph с print_call_graph(). Оптимизация, заявленная в релиз-ноутсах, — «improved by 10-20%» за счёт нового потокового двусвязного списка задач. И самое ломающее: asyncio.get_event_loop теперь бросает RuntimeError, если цикла нет, а не создаёт его молча.

Что с этим делать

Не вызывайте блокирующее в корутине. Это единственное правило, из которого следуют остальные, и теперь понятно почему: поток один, а шестой шаг — обычный for без прерываний.

Если блокирующее неизбежно — await loop.run_in_executor(...) или asyncio.to_thread(...). Работа уедет в поток, а цикл вернётся в select().

Уступать между порциями работы недостаточно. В замере выше воркер честно делал await asyncio.sleep(0) после каждой стомиллисекундной порции — и пульс всё равно потерял 95 тиков из 99. Одна уступка на 100 мс работы не спасает.

asyncio.run в цикле — антипаттерн. Из ~53 мкс, которые он стоит, полезной работы там 6: остальное — создание и закрытие цикла плюс те самые четыре оборота завершения. Для повторных запусков есть asyncio.Runner.

Не оптимизируйте await. Он стоит 56 нс и почти весь этот бюджет уходит на создание объекта корутины. Оптимизировать имеет смысл число приостановок, а не число await.