Лекция 7. Асинхронность, корутины и event loop#
В лекциях про потоки и мультипроцессинг мы обсуждали, как ОС делит процессорное время. На одном ядре ОС запускает готовый поток, а когда тот блокируется на вводе-выводе — другой. ОС может и прервать вычисления по таймеру, отдав ядро другому потоку. Это вытесняющее планирование.
Теперь перенесём похожую схему внутрь одного потока. event loop выбирает готовую корутину (coroutine). Когда она ждёт данные или срабатывание таймера, event loop запускает другую:
ОС: готовый поток -> выполняется -> ждёт ввод-вывод -> снова готов
asyncio: готовая корутина -> выполняется -> ждёт событие -> снова готова
ОС может вытеснить поток, а корутина сама отдаёт управление event loop. Сами корутины выполняются в одном потоке. Если корутина заняла его вычислениями или блокирующим вызовом, остальные ждут.
План:#
Генераторы: приостановка,
send,yield from.Корутины,
async def,awaitи event loop.Задачи, ожидание результатов, таймауты и отмена в
asyncio.Асинхронные итераторы и контекстные менеджеры.
Structured concurrency:
TaskGroup.Блокирующие вызовы,
race conditionи дедлоки.Тестирование с
pytest-asyncioи выбор модели конкурентного выполнения.
На семинаре попишем код с asyncio: соберём конвейер с очередью,
ограничим буфер и разберём остановку задач. Ещё познакомимся с anyio
и trio и соберём свой event loop на генераторах — mini_asyncio.
1. Вспоминаем генераторы#
Вызов функции с yield возвращает генератор. Он сохраняет состояние
между yield: yield выдаёт значение и возвращает управление
вызывающему коду, а next() возобновляет выполнение.
from collections.abc import Generator
def counter(start: int, stop: int) -> Generator[int]:
i = start
while i < stop:
yield i
i += 1
c = counter(0, 3)
print(next(c), next(c), next(c))
try:
next(c)
except StopIteration:
print("generator finished")
0 1 2
generator finished
send — передаём данные в генератор#
yield — выражение: при возобновлении оно получает значение,
переданное через .send().
def echo() -> Generator[str, str, None]:
msg = "start"
while True:
received = yield msg
msg = f"got: {received}"
g = echo()
print(next(g)) # доходим до первого yield — печатает "start"
print(g.send("hello")) # печатает "got: hello"
print(g.send("world")) # печатает "got: world"
g.close()
start
got: hello
got: world
yield from — делегирование#
yield from передаёт управление вложенному генератору:
def inner() -> Generator[int, None, str]:
yield 1
yield 2
return "done"
def outer() -> Generator[int | str]:
result = yield from inner()
yield f"inner returned: {result}"
for x in outer():
print(x)
1
2
inner returned: done
Значения из inner() передаются вызывающему коду, а return "done"
становится результатом yield from. Делегирование также перенаправляет
вызовы send, throw и close вложенному генератору.
От yield к await#
В первых версиях asyncio использовали генераторные корутины:
@asyncio.coroutine и yield from future
(PEP 3156).
PEP 492 добавил в Python 3.5
async def и await. Идея приостановки сохранилась, но появились
отдельные объекты корутин и отдельный синтаксический сахар для ожидания.
2. От генераторов к корутинам#
Вызов функции, объявленной через async def без yield, возвращает
объект корутины. Проверим его тип через inspect:
import asyncio
import inspect
async def greet(name: str) -> str:
print("greet: step 1")
await asyncio.sleep(0.05)
print("greet: step 2")
await asyncio.sleep(0.05)
print("greet: step 3")
return f"hello {name}"
coro = greet("world")
print(type(coro))
print("generator:", inspect.isgenerator(coro))
print("coroutine:", inspect.iscoroutine(coro))
print("state:", inspect.getcoroutinestate(coro))
coro.close()
<class 'coroutine'>
generator: False
coroutine: True
state: CORO_CREATED
Вызов greet("world") создал объект, но не выполнил тело функции.
Чтобы выполнить корутину, передаём её event loop или используем
await внутри другой корутины.
Нижняя схема соответствует выполнению greet(): два await asyncio.sleep(...)
дают две остановки (suspend) и два возобновления (resume).
После каждой паузы корутина продолжает выполнение со следующего шага.
Что делает event loop#
Упрощённо event loop работает так:
Хранит очередь готовых колбэков, в том числе шагов задач.
Запускает очередной шаг: корутина выполняется до приостановки или завершения.
Пока задача ждёт результат операции, её следующие шаги не запускаются.
Когда операция завершилась или сработал таймер, снова ставит шаг задачи в очередь.
Если готовых колбэков нет, event loop ждёт событий ввода-вывода или ближайшего таймера.
На Linux asyncio по умолчанию использует SelectorEventLoop с epoll.
Подробности — в
документации SelectorEventLoop.
Запускаем корутину#
В обычном скрипте используем asyncio.run()
%%writefile /tmp/async_hello.py
import asyncio
async def greet(name: str) -> str:
print("greet: step 1")
await asyncio.sleep(0.05)
print("greet: step 2")
await asyncio.sleep(0.05)
print("greet: step 3")
return f"hello {name}"
result = asyncio.run(greet("world"))
print(result)
Writing /tmp/async_hello.py
!python /tmp/async_hello.py
greet: step 1
greet: step 2
greet: step 3
hello world
asyncio.run()
создаёт event loop, выполняет переданную корутину и закрывает его.
Обычно это одна точка входа: asyncio.run(main()).
await — ждём результат#
result = await greet("world")
print(result)
greet: step 1
greet: step 2
greet: step 3
hello world
await greet(...) выполняет вложенную корутину в текущей
задаче и получает её результат. Отдельная задача здесь не создаётся.
Две остановки происходят внутри вызовов asyncio.sleep().
await не гарантирует переключения: если операция
завершается без ожидания, выполнение продолжается в той же задаче.
Через await можно ждать корутины, Task, Future и другие
объекты, поддерживающие await (awaitable).
3. Основные функции asyncio#
asyncio.create_task — создаём отдельную задачу#
async def background(name: str) -> None:
for i in range(3):
await asyncio.sleep(0.05)
print(f"[{name}] step {i}")
async def task_demo() -> None:
task = asyncio.create_task(background("bg"))
print("main: background task scheduled, doing other work")
await asyncio.sleep(0.1)
print("main: waiting for background task")
await task
print("both tasks finished")
await task_demo()
main: background task scheduled, doing other work
[bg] step 0
main: waiting for background task
[bg] step 1
[bg] step 2
both tasks finished
create_task(coro)
возвращает Task. При стандартных настройках код корутины начнёт
выполняться, когда текущая задача отдаст управление event loop.
Задача может работать и до await task. Через await task мы ждём
её завершения и получаем результат или исключение.
В Python 3.14 параметр eager_start=True позволяет начать выполнение
уже внутри create_task(). Выше используется стандартный отложенный запуск.
Ссылку на созданную задачу сохраняем до её завершения.
asyncio.gather — собираем результаты нескольких задач#
Имитируем чтение трёх файлов с медленного диска через sleep().
При последовательном ожидании понадобилось бы около 0.6 с:
import asyncio
import time
async def read_file(name: str, delay: float) -> str:
await asyncio.sleep(delay)
return f"{name}: contents"
async def gather_demo() -> None:
start = time.perf_counter()
future = asyncio.gather(
read_file("a.txt", 0.2),
read_file("b.txt", 0.1),
read_file("c.txt", 0.3),
)
print("is Future:", isinstance(future, asyncio.Future))
results = await future
elapsed = time.perf_counter() - start
print("results:", results)
print(f"elapsed: {elapsed:.2f}s (max delay: 0.3s)")
await gather_demo()
is Future: True
results: ['a.txt: contents', 'b.txt: contents', 'c.txt: contents']
elapsed: 0.30s (max delay: 0.3s)
gather()
планирует переданные корутины как задачи и возвращает
Future —
объект с будущим результатом операции. Через await future получаем список
результатов в порядке аргументов, независимо от порядка завершения.
Вручную создавать Future в прикладном коде обычно не нужно.
По умолчанию первая ошибка передаётся вызывающему коду, а остальные задачи автоматически не отменяются.
Удержим вторую задачу на Event, пока первая падает:
async def gather_error_demo() -> None:
release = asyncio.Event()
async def broken() -> None:
raise ValueError("operation failed")
async def other() -> int:
await release.wait()
print("other: finished")
return 42
other_task = asyncio.create_task(other())
try:
await asyncio.gather(broken(), other_task)
except ValueError as error:
print("caught:", error)
print("other done:", other_task.done())
print("other cancelled:", other_task.cancelled())
release.set()
print("other result:", await other_task)
await gather_error_demo()
caught: operation failed
other done: False
other cancelled: False
other: finished
other result: 42
После ошибки gather() задача other_task продолжает ждать.
Открываем Event и забираем её результат — 42.
asyncio.wait_for — таймаут и отмена#
async def slow_op() -> str:
try:
await asyncio.sleep(1.0)
return "done"
finally:
print("slow_op: cleaning up")
async def timeout_demo() -> None:
try:
await asyncio.wait_for(slow_op(), timeout=0.2)
except TimeoutError:
print("timed out")
await timeout_demo()
slow_op: cleaning up
timed out
Когда таймаут истекает,
wait_for()
запрашивает отмену операции. В примере slow_op() получает
asyncio.CancelledError и выполняет finally, а вызывающий код — TimeoutError.
task.cancel() запрашивает отмену. Задача получает CancelledError, когда
event loop снова передаёт ей управление. Если задача заняла поток
вычислениями или блокирующим вызовом, отмена задержится.
Если перехватываем CancelledError, после освобождения ресурсов обычно
нужно пробросить его дальше. wait_for() дожидается обработки отмены,
поэтому фактическое время ожидания может превысить таймаут.
4. Асинхронные итераторы и контекстные менеджеры#
async for#
async for получает итератор через __aiter__() и ожидает результат
каждого __anext__(). Завершение обозначается StopAsyncIteration:
from typing import Self
class Ticker:
def __init__(self, n: int, interval: float) -> None:
self.n = n
self.interval = interval
self.i = 0
def __aiter__(self) -> Self:
return self
async def __anext__(self) -> int:
if self.i >= self.n:
raise StopAsyncIteration()
await asyncio.sleep(self.interval)
self.i += 1
return self.i
async for tick in Ticker(3, 0.05):
print(f"tick {tick}")
tick 1
tick 2
tick 3
__anext__() может ждать данные или таймер, отдавая управление другим
задачам. А __aiter__() сразу возвращает асинхронный итератор.
async with#
async with вызывает __aenter__() при входе в блок и __aexit__()
при выходе, дожидаясь обоих методов. Откроем файл и прочитаем текст.
Открытие, чтение и закрытие выполняем через asyncio.to_thread():
from pathlib import Path
from tempfile import TemporaryDirectory
from types import TracebackType
from typing import Self
class AsyncTextFile:
def __init__(self, path: Path) -> None:
self._path = path
async def __aenter__(self) -> Self:
self._file = await asyncio.to_thread(self._path.open, "r", encoding="utf-8")
print("opened")
return self
async def __aexit__(
self,
exc_type: type[BaseException] | None,
exc: BaseException | None,
tb: TracebackType | None,
) -> None:
await asyncio.to_thread(self._file.close)
print("closed")
async def read(self) -> str:
return await asyncio.to_thread(self._file.read)
with TemporaryDirectory() as directory:
path = Path(directory) / "example.txt"
await asyncio.to_thread(path.write_text, "hello from disk", encoding="utf-8")
async with AsyncTextFile(path) as file:
print(await file.read())
opened
hello from disk
closed
Вместо класса можно использовать
@contextlib.asynccontextmanager:
написать асинхронный генератор с одним yield, а освобождение ресурса
поместить в finally.
5. Structured concurrency#
Запустили задачу, но не стали ждать её через await. Её исключение
не передастся вызывающему коду автоматически:
import asyncio
async def buggy() -> None:
await asyncio.sleep(0.05)
raise ValueError("oops")
async def main_orphan() -> None:
# Антипаттерн: запускаем задачу, но не ждём её завершения.
task = asyncio.create_task(buggy())
await asyncio.sleep(0.2)
print("main_orphan: continued after the child task failed")
print("child done:", task.done())
print("child exception:", repr(task.exception()))
await main_orphan()
main_orphan: continued after the child task failed
child done: True
child exception: ValueError('oops')
Ошибка сохранилась в задаче: её видно через task.exception(),
но main_orphan() продолжила работу.
try/except вокруг create_task() такую ошибку не перехватит.
TaskGroup#
При structured concurrency дочерние задачи завершаются до выхода
из блока, который их запустил. В asyncio для этого есть
TaskGroup
(добавлен в Python 3.11):
async def worker(name: str, delay: float) -> str:
await asyncio.sleep(delay)
return f"{name} done"
async def taskgroup_demo() -> None:
async with asyncio.TaskGroup() as tg:
t1 = tg.create_task(worker("A", 0.1))
t2 = tg.create_task(worker("B", 0.2))
# автоматически ждём всех при выходе из async with
print(f"results: {t1.result()}, {t2.result()}")
await taskgroup_demo()
results: A done, B done
При выходе из async with группа ждёт завершения всех задач,
созданных через tg.create_task().
Если одна завершилась с ошибкой, отличной от CancelledError, группа
отменяет остальные, ждёт их завершения и передаёт ошибки наружу.
Для обычных исключений используется ExceptionGroup из
PEP 654.
Посмотрим, как ошибка одной задачи отменяет соседнюю:
async def fail() -> None:
await asyncio.sleep(0.01)
raise ValueError("operation failed")
async def long_running() -> None:
try:
await asyncio.sleep(1)
finally:
print("long_running: cleaning up")
async def taskgroup_error_demo() -> None:
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(long_running())
tg.create_task(fail())
except* ValueError as errors:
print("caught exception group:", errors.exceptions)
await taskgroup_error_demo()
long_running: cleaning up
caught exception group: (ValueError('operation failed'),)
except* ValueError обрабатывает исключения типа ValueError внутри группы.
KeyboardInterrupt и SystemExit после завершения остальных задач
пробрасываются напрямую.
Если дочерние задачи составляют одну операцию и должны завершаться вместе,
удобно использовать TaskGroup. gather() собирает результаты в порядке
аргументов, но иначе обрабатывает ошибки.
6. race condition и дедлоки#
race condition в одном потоке#
race condition может возникнуть и в одном потоке. Например, если задача
приостановилась между чтением общих данных и записью результата:
async def lost_update_demo() -> None:
value = 0
async def increment() -> None:
nonlocal value
before = value
await asyncio.sleep(0) # другая задача тоже успеет прочитать value
value = before + 1
await asyncio.gather(increment(), increment())
print("expected 2, got:", value)
await lost_update_demo()
expected 2, got: 1
Обе задачи прочитали 0, затем каждая записала 1. Защитим всю операцию
блокировкой asyncio.Lock:
async def locked_update_demo() -> None:
value = 0
lock = asyncio.Lock()
async def increment() -> None:
nonlocal value
async with lock:
before = value
await asyncio.sleep(0)
value = before + 1
await asyncio.gather(increment(), increment())
print("with lock:", value)
await locked_update_demo()
with lock: 2
Задачи одного event loop чередуются, когда приостанавливаются. Поэтому проверяем всю операцию «прочитать, дождаться, записать», а не только присваивание.
Дедлок в одном потоке#
Дедлоку тоже не нужен параллелизм: достаточно циклического ожидания.
Один поток может зависнуть при повторном захвате обычного threading.Lock.
А две асинхронные задачи — при захвате блокировок в разном порядке:
async def deadlock_demo() -> None:
lock_a = asyncio.Lock()
lock_b = asyncio.Lock()
async def worker_ab() -> None:
async with lock_a:
print("worker_ab: holding A, acquiring B")
await asyncio.sleep(0.01) # даём worker_ba взять lock_b
async with lock_b:
print("worker_ab: holding both locks")
async def worker_ba() -> None:
async with lock_b:
print("worker_ba: holding B, acquiring A")
await asyncio.sleep(0.01) # даём worker_ab взять lock_a
async with lock_a:
print("worker_ba: holding both locks")
# Ограничиваем ожидание, чтобы демонстрация не зависла навсегда.
try:
async with asyncio.timeout(1.0):
await asyncio.gather(worker_ab(), worker_ba())
except TimeoutError:
print("deadlock: worker_ab holds A and waits for B; worker_ba holds B and waits for A")
await deadlock_demo()
worker_ab: holding A, acquiring B
worker_ba: holding B, acquiring A
deadlock: worker_ab holds A and waits for B; worker_ba holds B and waits for A
Каждая задача ждёт блокировку, которую держит другая. Планировщик такой цикл не разорвёт. Помогает единый порядок захвата блокировок, как и в примерах с потоками.
asyncio.timeout()
ограничивает время выполнения блока: отменяет текущую задачу,
а при выходе преобразует отмену в TimeoutError. Таймаут спасает
демонстрацию от зависания, но не исправляет порядок захвата блокировок.
Асинхронность позволяет перекрывать ожидания. Вычисления на Python от этого не распараллеливаются, а гонки и дедлоки остаются.
7. Тестирование: pytest-asyncio#
Для теста с await нужны async def и плагин pytest-asyncio.
Плагин запускает тест в event loop. Он уже входит в набор зависимостей
test нашего pyproject.toml.
%%writefile /tmp/test_async.py
import asyncio
import pytest
async def fetch_delayed(x: int, delay: float) -> int:
await asyncio.sleep(delay)
return x * 2
@pytest.mark.asyncio
async def test_fetch() -> None:
result = await fetch_delayed(21, 0.01)
assert result == 42
@pytest.mark.asyncio
async def test_gather() -> None:
results = await asyncio.gather(
fetch_delayed(1, 0.01),
fetch_delayed(2, 0.01),
fetch_delayed(3, 0.01),
)
assert results == [2, 4, 6]
Writing /tmp/test_async.py
import subprocess
result = subprocess.run(
[sys.executable, "-m", "pytest", "/tmp/test_async.py", "-v", "--asyncio-mode=strict"],
capture_output=True,
text=True,
check=True,
)
print(result.stdout)
============================= test session starts ==============================
platform darwin -- Python 3.14.8, pytest-9.1.1, pluggy-1.6.0 -- /Users/chopik/advanced-python-ami-lectures-2026/course-content/.venv/bin/python3
cachedir: .pytest_cache
hypothesis profile 'default'
rootdir: /tmp
plugins: mock-3.15.1, cov-7.1.0, xdist-3.8.0, asyncio-1.4.0, anyio-4.14.2, hypothesis-6.165.2
asyncio: mode=Mode.STRICT, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collecting ... collected 2 items
../../../../../../tmp/test_async.py::test_fetch PASSED [ 50%]
../../../../../../tmp/test_async.py::test_gather PASSED [100%]
============================== 2 passed in 0.12s ===============================
Ключевые настройки#
Выше использовали явные маркеры @pytest.mark.asyncio и режим strict.
Для проекта только с asyncio можно включить режим auto:
[tool.pytest.ini_options]
asyncio_mode = "auto" # выполнять все async def test_* через pytest-asyncio
В этом режиме маркеры перед каждым асинхронным тестом не нужны.
Асинхронные фикстуры#
Фикстура может использовать await при подготовке и освобождении ресурса.
Создадим временный файл, прочитаем его в тесте и удалим после yield.
Файловые операции выполняем через to_thread(), чтобы не блокировать event loop:
%%writefile /tmp/test_async_file.py
import asyncio
from collections.abc import AsyncGenerator
from pathlib import Path
import pytest
import pytest_asyncio
@pytest_asyncio.fixture
async def text_file(tmp_path: Path) -> AsyncGenerator[Path]:
path = tmp_path / "example.txt"
await asyncio.to_thread(path.write_text, "hello from disk\n", encoding="utf-8")
try:
yield path
finally:
await asyncio.to_thread(path.unlink)
@pytest.mark.asyncio
async def test_read_file(text_file: Path) -> None:
content = await asyncio.to_thread(text_file.read_text, encoding="utf-8")
assert content == "hello from disk\n"
Writing /tmp/test_async_file.py
result = subprocess.run(
[sys.executable, "-m", "pytest", "/tmp/test_async_file.py", "-v", "--asyncio-mode=strict"],
capture_output=True,
text=True,
check=True,
)
print(result.stdout)
============================= test session starts ==============================
platform darwin -- Python 3.14.8, pytest-9.1.1, pluggy-1.6.0 -- /Users/chopik/advanced-python-ami-lectures-2026/course-content/.venv/bin/python3
cachedir: .pytest_cache
hypothesis profile 'default'
rootdir: /tmp
plugins: mock-3.15.1, cov-7.1.0, xdist-3.8.0, asyncio-1.4.0, anyio-4.14.2, hypothesis-6.165.2
asyncio: mode=Mode.STRICT, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collecting ... collected 1 item
../../../../../../tmp/test_async_file.py::test_read_file PASSED [100%]
============================== 1 passed in 0.09s ===============================
В режиме strict асинхронную фикстуру объявляем через
@pytest_asyncio.fixture.
В режиме auto плагин также подхватывает асинхронные фикстуры,
объявленные через обычный @pytest.fixture.
Время жизни event loop#
По умолчанию у каждого теста свой event loop. Для дорогой фикстуры на всю сессию, например пула соединений, может понадобиться общий event loop:
[tool.pytest.ini_options]
asyncio_default_test_loop_scope = "session"
asyncio_default_fixture_loop_scope = "session"
Подробнее — в документации по настройке event loop
для тестов
и для фикстур.
scope задаёт, как долго переиспользуется сама фикстура.
Для отдельного теста или фикстуры можно указать loop_scope в декораторе.
event loop должен жить не меньше самой фикстуры. Незавершённые фоновые задачи
в общем event loop могут мешать следующим тестам.
8. Сводка: конкурентное выполнение в Python#
В лекции про потоки разбирали threading, в лекции про мультипроцессинг —
процессы, субинтерпретаторы и Python без GIL. Сравним их с asyncio:
Механизм |
Выполнение |
Память и обмен |
Завершение |
|---|---|---|---|
|
ожидания перекрываются; GIL ограничивает параллельное выполнение Python-кода |
общие объекты |
остановка через флаг или |
|
Python-код может выполняться параллельно при выключенном GIL |
общие объекты, нужна синхронизация |
остановка через флаг или |
|
отдельный процесс, возможно другое ядро |
раздельная память, явный обмен |
|
|
несколько процессов обрабатывают джобы |
аргументы и результаты передаются через |
|
|
несколько потоков с отдельными интерпретаторами и GIL |
отдельные состояния Python; аргументы и результаты передаются через |
ожидание через |
|
задачи выполняются по очереди в потоке event loop |
общие объекты внутри event loop |
|
|
внешняя программа, вызывающий поток ждёт |
отдельный процесс, |
ожидание завершения; |
|
внешняя программа, ожидание задаём сами |
отдельный процесс, каналы ввода-вывода |
|
Субинтерпретаторы живут в одном процессе: сбой в C-расширении может уронить
его целиком. А
InterpreterPoolExecutor
передаёт аргументы и результаты через pickle, несмотря на общий PID.
Разбираем на семинаре#
Практика с asyncio#
Соберём конвейер: одна задача получает блоки данных, другая обрабатывает
и сохраняет их. Передавать блоки будем через asyncio.Queue(maxsize=...).
Когда очередь заполнена, источник должен ждать освобождения места.
Запустим задачи через TaskGroup, добавим таймаут и остановку.
Проверим, что при ошибке или отмене дочерние задачи завершаются,
а ресурсы закрываются. Напишем асинхронную фикстуру и тесты для конвейера.
anyio и trio#
trio —
другая библиотека для асинхронного кода. Дочерние задачи запускаются
внутри nursery.
anyio даёт общий API для
asyncio и trio. Чтобы код работал с обеими библиотеками, используем
API anyio. Вызовы, привязанные к asyncio, придётся заменить.
Тестирование через anyio#
anyio уже содержит плагин для pytest.
С @pytest.mark.anyio тесты можно запускать на asyncio или trio, если
код совместим с выбранной библиотекой и она установлена.
Не включаем auto одновременно у anyio и pytest-asyncio:
оба плагина попытаются подхватить одни и те же тесты.
event loop на генераторах — mini_asyncio#
Генератор как задача, которую можно приостановить#
Вспомним аналогию с ОС. Чтобы чередовать задачи внутри одного потока,
нам нужен способ вернуть управление планировщику посреди выполнения.
Генератор умеет это делать в точке yield.
Пусть у нас три воркера. После каждого шага возвращаем воркера в конец очереди:
from collections import deque
from collections.abc import Generator, Iterable
def worker(name: str, steps: int) -> Generator[None]:
for i in range(steps):
print(f"[{name}] step {i}")
yield # добровольно возвращаем управление планировщику
def scheduler(tasks: Iterable[Generator[None]]) -> None:
q = deque(tasks)
while q:
task = q.popleft()
try:
next(task) # запускаем до следующего yield
q.append(task) # снова в очередь
except StopIteration:
pass # задача завершилась
scheduler([worker("A", 3), worker("B", 3), worker("C", 2)])
[A] step 0
[B] step 0
[C] step 0
[A] step 1
[B] step 1
[C] step 1
[A] step 2
[B] step 2
Задачи выполняются по очереди, по одному шагу за раз, в одном потоке.
Добавим таймеры. У задачи будет два состояния: готова к выполнению или ждёт срабатывания таймера.
Запрос к планировщику через yield#
Генератор выдаёт запрос Sleep, а планировщик решает, когда его возобновить.
Это похоже на системный вызов, но yield Sleep(...) в ядро ОС не обращается.
В очереди храним идентификатор задачи, генератор и значение для send().
Текущий планировщик сохраняем в ContextVar на время run().
from collections import deque
from collections.abc import Generator
from contextvars import ContextVar
from dataclasses import dataclass
import time
type Task = Generator[object, object, object]
@dataclass
class Sleep:
duration: float
_current: ContextVar[Scheduler | None] = ContextVar("mini_asyncio.current", default=None)
class Scheduler:
def __init__(self) -> None:
self.ready: deque[tuple[int, Task, object]] = deque()
self.sleeping: list[tuple[float, int, Task]] = []
self.results: dict[int, object] = {}
self.next_id = 0
def _new_tid(self) -> int:
self.next_id += 1
return self.next_id
def create_task(self, coro: Task) -> int:
tid = self._new_tid()
self.ready.append((tid, coro, None))
return tid
def run(self, main: Task) -> None:
token = _current.set(self)
try:
self.create_task(main)
while self._has_work():
self._wake_sleeping()
if not self.ready:
self._idle_wait_or_deadlock()
continue
self._step_one()
finally:
_current.reset(token)
def _has_work(self) -> bool:
return bool(self.ready or self.sleeping)
def _wake_sleeping(self) -> None:
now = time.monotonic()
still: list[tuple[float, int, Task]] = []
for wake_at, tid, task in self.sleeping:
if wake_at <= now:
self.ready.append((tid, task, None))
else:
still.append((wake_at, tid, task))
self.sleeping = still
def _idle_wait_or_deadlock(self) -> None:
if self.sleeping:
next_wake = min(w for w, _, _ in self.sleeping)
time.sleep(max(0.0, next_wake - time.monotonic()))
def _step_one(self) -> None:
tid, task, value = self.ready.popleft()
try:
syscall = task.send(value)
except StopIteration as e:
self._finish_task(tid, e.value)
return
self._handle_syscall(tid, task, syscall)
def _finish_task(self, tid: int, result: object) -> None:
self.results[tid] = result
def _handle_syscall(self, tid: int, task: Task, syscall: object) -> None:
if isinstance(syscall, Sleep):
self.sleeping.append((time.monotonic() + syscall.duration, tid, task))
else:
raise RuntimeError(f"unknown syscall: {syscall!r}")
Когда готовых задач нет, _idle_wait_or_deadlock() приостанавливает поток через
time.sleep() до срабатывания ближайшего таймера. Постоянный опрос не нужен.
Добавим функции, которые вызываются из генераторов:
def sleep(duration: float) -> Generator[Sleep, object]:
"""Запрашивает приостановку на заданное число секунд."""
yield Sleep(duration)
def create_task(coro: Task) -> int:
"""Регистрирует задачу в текущем планировщике без приостановки."""
scheduler = _current.get()
if scheduler is None:
raise RuntimeError("no running event loop; call create_task() inside run()")
return scheduler.create_task(coro)
def run(main: Task) -> None:
"""Запускает планировщик до завершения всех зарегистрированных задач."""
Scheduler().run(main)
Используем те же три воркера, но теперь у каждого своя задержка. Время отсчитываем от начала работы планировщика:
def worker(name: str, steps: int, delay: float) -> Generator[Sleep, object]:
for i in range(steps):
print(f"[{name}] step {i}, t={time.monotonic() - started:.2f}")
yield from sleep(delay) # похоже на await asyncio.sleep(delay)
def main() -> Generator[Sleep, object]:
create_task(worker("A", steps=3, delay=0.10))
create_task(worker("B", steps=4, delay=0.05))
create_task(worker("C", steps=2, delay=0.15))
yield from sleep(0.01)
print("[main] all tasks scheduled")
started = time.monotonic()
run(main())
print(f"scheduler finished in {time.monotonic() - started:.2f}s")
[A] step 0, t=0.00
[B] step 0, t=0.00
[C] step 0, t=0.00
[main] all tasks scheduled
[B] step 1, t=0.05
[A] step 1, t=0.11
[B] step 2, t=0.11
[C] step 1, t=0.15
[B] step 3, t=0.16
[A] step 2, t=0.21
scheduler finished in 0.31s
В начале все три воркера печатают шаг 0. Затем их шаги чередуются:
B ждёт по 0.05 с, A — по 0.10 с, C — по 0.15 с.
Около t=0.10 готовы и A, и B. Порядок срабатывания близких таймеров
и точное время пробуждения зависят от момента их установки и нагрузки.
Каждый шаг, включая последний, заканчивается паузой. При последовательном
выполнении ожидания заняли бы 0.30 + 0.20 + 0.30 = 0.80 с.
Здесь они перекрываются: общее время близко к max(0.30, 0.20, 0.30) = 0.30 с
плюс накладные расходы.
Наш run() ждёт завершения всех задач. Это отличие от
asyncio.run(main()): тот ждёт main, а при завершении отменяет
оставшиеся задачи. В приложении нужно явно дождаться дочерних задач.
Планировщик выдаёт идентификаторы задач и сохраняет их результаты,
но одна задача пока не может дождаться другой. В ДЗ mini_asyncio
из группы async_intro нужно добавить AwaitTask, await_task(tid)
и gather(*coros). Результат ожидаемой задачи передадим через task.send(value).
У asyncio также есть отмена и таймауты.
event loop также ждёт сетевые события; наш планировщик работает только с таймерами.