---
jupytext:
  formats: md:myst
  text_representation:
    extension: .md
    format_name: myst
kernelspec:
  display_name: Python 3.14
  language: python
  name: python3
---

```{code-cell} python
:tags: [remove-cell]

import os
import sys

os.environ["PATH"] = f"{os.path.dirname(sys.executable)}:{os.environ['PATH']}"
```

# Лекция 7. Асинхронность, корутины и event loop

В лекциях про потоки и мультипроцессинг мы обсуждали, как ОС делит
процессорное время. На одном ядре ОС запускает готовый поток, а когда тот
блокируется на вводе-выводе — другой.
ОС может и прервать вычисления по таймеру, отдав ядро другому потоку.
Это **вытесняющее планирование**.

Теперь перенесём похожую схему внутрь одного потока. **event loop** выбирает готовую **корутину (coroutine)**. Когда она
ждёт данные или срабатывание таймера, event loop запускает другую:

```text
ОС:       готовый поток    -> выполняется -> ждёт ввод-вывод -> снова готов
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()` возобновляет выполнение.

```{code-cell} python
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")
```

### `send` — передаём данные в генератор

`yield` — **выражение**: при возобновлении оно получает значение,
переданное через `.send()`.

```{code-cell} python
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()
```

### `yield from` — делегирование

`yield from` передаёт управление вложенному генератору:

```{code-cell} python
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)
```

Значения из `inner()` передаются вызывающему коду, а `return "done"`
становится результатом `yield from`. Делегирование также перенаправляет
вызовы `send`, `throw` и `close` вложенному генератору.

### От `yield` к `await`

В первых версиях `asyncio` использовали генераторные корутины:
`@asyncio.coroutine` и `yield from future`
([PEP 3156](https://peps.python.org/pep-3156/)).

[PEP 492](https://peps.python.org/pep-0492/) добавил в Python 3.5
`async def` и `await`. Идея приостановки сохранилась, но появились
отдельные объекты корутин и отдельный синтаксический сахар для ожидания.

---

## 2. От генераторов к корутинам

Вызов функции, объявленной через `async def` без `yield`, возвращает
объект корутины. Проверим его тип через `inspect`:

```{code-cell} python
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()
```

Вызов `greet("world")` создал объект, но **не выполнил тело функции**.
Чтобы выполнить корутину, передаём её event loop или используем
`await` внутри другой корутины.

![Выполнение обычной функции и корутины с остановками и возобновлениями](img/coroutine-suspend-resume.png)

Нижняя схема соответствует выполнению `greet()`: два `await asyncio.sleep(...)`
дают две остановки (`suspend`) и два возобновления (`resume`).
После каждой паузы корутина продолжает выполнение со следующего шага.

### Что делает event loop

Упрощённо event loop работает так:

1. Хранит очередь готовых колбэков, в том числе шагов задач.
2. Запускает очередной шаг: корутина выполняется до
   приостановки или завершения.
3. Пока задача ждёт результат операции, её следующие шаги не запускаются.
4. Когда операция завершилась или сработал таймер, снова ставит шаг задачи
   в очередь.

Если готовых колбэков нет, event loop ждёт событий ввода-вывода
или ближайшего таймера.

На Linux `asyncio` по умолчанию использует `SelectorEventLoop` с `epoll`.
Подробности — в
[документации `SelectorEventLoop`](https://docs.python.org/3.14/library/asyncio-eventloop.html#asyncio.SelectorEventLoop).

### Запускаем корутину

В обычном скрипте используем `asyncio.run()`

```{code-cell} python
%%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)
```

```{code-cell} python
!python /tmp/async_hello.py
```

[`asyncio.run()`](https://docs.python.org/3.14/library/asyncio-runner.html#asyncio.run)
создаёт event loop, выполняет переданную корутину и закрывает его.
Обычно это одна точка входа: `asyncio.run(main())`.

### `await` — ждём результат

```{code-cell} python
result = await greet("world")
print(result)
```

`await greet(...)` выполняет вложенную корутину в текущей
задаче и получает её результат. Отдельная задача здесь не создаётся.
Две остановки происходят внутри вызовов `asyncio.sleep()`.

`await` **не гарантирует переключения**: если операция
завершается без ожидания, выполнение продолжается в той же задаче.
Через `await` можно ждать корутины, `Task`, `Future` и другие
объекты, поддерживающие `await` (awaitable).

---

## 3. Основные функции `asyncio`

### `asyncio.create_task` — создаём отдельную задачу

```{code-cell} python
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()
```

[`create_task(coro)`](https://docs.python.org/3.14/library/asyncio-task.html#asyncio.create_task)
возвращает `Task`. При стандартных настройках код корутины начнёт
выполняться, когда текущая задача отдаст управление event loop.
Задача может работать и до `await task`. Через `await task` мы ждём
её завершения и получаем результат или исключение.

В Python 3.14 параметр `eager_start=True` позволяет начать выполнение
уже внутри `create_task()`. Выше используется стандартный отложенный запуск.
Ссылку на созданную задачу сохраняем до её завершения.

### `asyncio.gather` — собираем результаты нескольких задач

Имитируем чтение трёх файлов с медленного диска через `sleep()`.
При последовательном ожидании понадобилось бы около `0.6` с:

```{code-cell} python
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()
```

[`gather()`](https://docs.python.org/3.14/library/asyncio-task.html#asyncio.gather)
планирует переданные корутины как задачи и возвращает
[`Future`](https://docs.python.org/3.14/library/asyncio-future.html#asyncio.Future) —
объект с будущим результатом операции. Через `await future` получаем список
результатов **в порядке аргументов**, независимо от порядка завершения.
Вручную создавать `Future` в прикладном коде обычно не нужно.

По умолчанию первая ошибка передаётся вызывающему коду, а остальные
задачи автоматически не отменяются.

Удержим вторую задачу на `Event`, пока первая падает:

```{code-cell} python
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()
```

После ошибки `gather()` задача `other_task` продолжает ждать.
Открываем `Event` и забираем её результат — `42`.

### `asyncio.wait_for` — таймаут и отмена

```{code-cell} python
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()
```

Когда таймаут истекает,
[`wait_for()`](https://docs.python.org/3.14/library/asyncio-task.html#asyncio.wait_for)
запрашивает отмену операции. В примере `slow_op()` получает
`asyncio.CancelledError` и выполняет `finally`, а вызывающий код — `TimeoutError`.

`task.cancel()` запрашивает отмену. Задача получает `CancelledError`, когда
event loop снова передаёт ей управление. Если задача заняла поток
вычислениями или блокирующим вызовом, отмена задержится.
Если перехватываем `CancelledError`, после освобождения ресурсов обычно
нужно пробросить его дальше. `wait_for()` дожидается обработки отмены,
поэтому фактическое время ожидания может превысить таймаут.

---

## 4. Асинхронные итераторы и контекстные менеджеры

### `async for`

`async for` получает итератор через `__aiter__()` и ожидает результат
каждого `__anext__()`. Завершение обозначается `StopAsyncIteration`:

```{code-cell} python
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}")
```

`__anext__()` может ждать данные или таймер, отдавая управление другим
задачам. А `__aiter__()` сразу возвращает асинхронный итератор.

### `async with`

`async with` вызывает `__aenter__()` при входе в блок и `__aexit__()`
при выходе, дожидаясь обоих методов. Откроем файл и прочитаем текст.
Открытие, чтение и закрытие выполняем через `asyncio.to_thread()`:

```{code-cell} python
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())
```

Вместо класса можно использовать
[`@contextlib.asynccontextmanager`](https://docs.python.org/3.14/library/contextlib.html#contextlib.asynccontextmanager):
написать асинхронный генератор с одним `yield`, а освобождение ресурса
поместить в `finally`.

---

## 5. Structured concurrency

Запустили задачу, но не стали ждать её через `await`. Её исключение
не передастся вызывающему коду автоматически:

```{code-cell} python
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()
```

Ошибка сохранилась в задаче: её видно через `task.exception()`,
но `main_orphan()` продолжила работу.
`try/except` вокруг `create_task()` такую ошибку не перехватит.

### `TaskGroup`

При `structured concurrency` дочерние задачи завершаются до выхода
из блока, который их запустил. В `asyncio` для этого есть
[`TaskGroup`](https://docs.python.org/3.14/library/asyncio-task.html#task-groups)
(добавлен в Python 3.11):

```{code-cell} python
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()
```

При выходе из `async with` группа ждёт завершения всех задач,
созданных через `tg.create_task()`.
Если одна завершилась с ошибкой, отличной от `CancelledError`, группа
отменяет остальные, ждёт их завершения и передаёт ошибки наружу.
Для обычных исключений используется `ExceptionGroup` из
[PEP 654](https://peps.python.org/pep-0654/).

Посмотрим, как ошибка одной задачи отменяет соседнюю:

```{code-cell} python
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()
```

`except* ValueError` обрабатывает исключения типа `ValueError` внутри группы.
`KeyboardInterrupt` и `SystemExit` после завершения остальных задач
пробрасываются напрямую.

Если дочерние задачи составляют одну операцию и должны завершаться вместе,
удобно использовать `TaskGroup`. `gather()` собирает результаты в порядке
аргументов, но иначе обрабатывает ошибки.

---

## 6. `race condition` и дедлоки

### `race condition` в одном потоке

`race condition` может возникнуть и в одном потоке. Например, если задача
приостановилась между чтением общих данных и записью результата:

```{code-cell} python
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()
```

Обе задачи прочитали `0`, затем каждая записала `1`. Защитим всю операцию
блокировкой `asyncio.Lock`:

```{code-cell} python
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()
```

Задачи одного event loop чередуются, когда приостанавливаются.
Поэтому проверяем всю операцию "прочитать, дождаться, записать",
а не только присваивание.

### Дедлок в одном потоке

Дедлоку тоже не нужен параллелизм: достаточно циклического ожидания.
Один поток может зависнуть при повторном захвате обычного `threading.Lock`.
А две асинхронные задачи — при захвате блокировок в разном порядке:

```{code-cell} python
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()
```

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

[`asyncio.timeout()`](https://docs.python.org/3.14/library/asyncio-task.html#asyncio.timeout)
ограничивает время выполнения блока: отменяет текущую задачу,
а при выходе преобразует отмену в `TimeoutError`. Таймаут спасает
демонстрацию от зависания, но не исправляет порядок захвата блокировок.

Асинхронность позволяет перекрывать ожидания. Вычисления на Python
от этого не распараллеливаются, а гонки и дедлоки остаются.

---

## 7. Тестирование: `pytest-asyncio`

Для теста с `await` нужны `async def` и плагин `pytest-asyncio`.
Плагин запускает тест в event loop. Он уже входит в набор зависимостей
`test` нашего `pyproject.toml`.

```{code-cell} python
%%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]
```

```{code-cell} python
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)
```

### Ключевые настройки

Выше использовали явные маркеры `@pytest.mark.asyncio` и режим `strict`.
Для проекта только с `asyncio` можно включить режим `auto`:

```toml
[tool.pytest.ini_options]
asyncio_mode = "auto"  # выполнять все async def test_* через pytest-asyncio
```

В этом режиме маркеры перед каждым асинхронным тестом не нужны.

### Асинхронные фикстуры

Фикстура может использовать `await` при подготовке и освобождении ресурса.
Создадим временный файл, прочитаем его в тесте и удалим после `yield`.
Файловые операции выполняем через `to_thread()`, чтобы не блокировать event loop:

```{code-cell} python
%%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"
```

```{code-cell} python
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)
```

В режиме `strict` асинхронную фикстуру объявляем через
[`@pytest_asyncio.fixture`](https://pytest-asyncio.readthedocs.io/en/stable/reference/decorators/index.html).
В режиме `auto` плагин также подхватывает асинхронные фикстуры,
объявленные через обычный `@pytest.fixture`.

### Время жизни event loop

По умолчанию у каждого теста свой event loop. Для дорогой фикстуры
на всю сессию, например пула соединений, может понадобиться общий event loop:

```toml
[tool.pytest.ini_options]
asyncio_default_test_loop_scope = "session"
asyncio_default_fixture_loop_scope = "session"
```

Подробнее — в документации по настройке event loop
[для тестов](https://pytest-asyncio.readthedocs.io/en/stable/how-to-guides/change_default_test_loop.html)
и [для фикстур](https://pytest-asyncio.readthedocs.io/en/stable/how-to-guides/change_default_fixture_loop.html).
`scope` задаёт, как долго переиспользуется сама фикстура.

Для отдельного теста или фикстуры можно указать `loop_scope` в декораторе.
event loop должен жить не меньше самой фикстуры. Незавершённые фоновые задачи
в общем event loop могут мешать следующим тестам.

## 8. Сводка: конкурентное выполнение в Python

В лекции про потоки разбирали `threading`, в лекции про мультипроцессинг —
процессы, субинтерпретаторы и Python без GIL. Сравним их с `asyncio`:

| Механизм | Выполнение | Память и обмен | Завершение |
|---|---|---|---|
| `threading.Thread`, обычная сборка | ожидания перекрываются; GIL ограничивает параллельное выполнение Python-кода | общие объекты | остановка через флаг или `Event` |
| `threading.Thread`, сборка без GIL | Python-код может выполняться параллельно при выключенном GIL | общие объекты, нужна синхронизация | остановка через флаг или `Event` |
| `multiprocessing.Process` | отдельный процесс, возможно другое ядро | раздельная память, явный обмен | `join()`, при необходимости `terminate()` или `kill()` |
| `multiprocessing.Pool` | несколько процессов обрабатывают джобы | аргументы и результаты передаются через `pickle` | `close()` и `join()`; `terminate()` останавливает весь пул |
| `InterpreterPoolExecutor` | несколько потоков с отдельными интерпретаторами и GIL | отдельные состояния Python; аргументы и результаты передаются через `pickle` | ожидание через `shutdown()`; выполняющуюся джобу нельзя отменить через `Future.cancel()` |
| `asyncio.Task` | задачи выполняются по очереди в потоке event loop | общие объекты внутри event loop | `cancel()` запрашивает отмену, `await` ждёт завершения |
| `subprocess.run` | внешняя программа, вызывающий поток ждёт | отдельный процесс, `stdin`/`stdout`/`stderr` | ожидание завершения; `timeout` завершает дочерний процесс |
| `subprocess.Popen` | внешняя программа, ожидание задаём сами | отдельный процесс, каналы ввода-вывода | `wait()`/`communicate()`, `terminate()`/`kill()` |

Субинтерпретаторы живут в одном процессе: сбой в C-расширении может уронить
его целиком. А
[`InterpreterPoolExecutor`](https://docs.python.org/3.14/library/concurrent.futures.html#interpreterpoolexecutor)
передаёт аргументы и результаты через `pickle`, несмотря на общий PID.

---

## Разбираем на семинаре

### Практика с `asyncio`

Соберём конвейер: одна задача получает блоки данных, другая обрабатывает
и сохраняет их. Передавать блоки будем через `asyncio.Queue(maxsize=...)`.
Когда очередь заполнена, источник должен ждать освобождения места.

Запустим задачи через `TaskGroup`, добавим таймаут и остановку.
Проверим, что при ошибке или отмене дочерние задачи завершаются,
а ресурсы закрываются. Напишем асинхронную фикстуру и тесты для конвейера.

### `anyio` и `trio`

[`trio`](https://trio.readthedocs.io/en/stable/reference-core.html) —
другая библиотека для асинхронного кода. Дочерние задачи запускаются
внутри `nursery`.
[`anyio`](https://anyio.readthedocs.io/en/stable/) даёт общий API для
`asyncio` и `trio`. Чтобы код работал с обеими библиотеками, используем
API `anyio`. Вызовы, привязанные к `asyncio`, придётся заменить.

#### Тестирование через `anyio`

[`anyio` уже содержит плагин для `pytest`](https://anyio.readthedocs.io/en/stable/testing.html).
С `@pytest.mark.anyio` тесты можно запускать на `asyncio` или `trio`, если
код совместим с выбранной библиотекой и она установлена.
Не включаем `auto` одновременно у `anyio` и `pytest-asyncio`:
оба плагина попытаются подхватить одни и те же тесты.

### event loop на генераторах — `mini_asyncio`

#### Генератор как задача, которую можно приостановить

Вспомним аналогию с ОС. Чтобы чередовать задачи внутри одного потока,
нам нужен способ вернуть управление планировщику посреди выполнения.
Генератор умеет это делать в точке `yield`.

Пусть у нас три воркера. После каждого шага возвращаем воркера
в конец очереди:

```{code-cell} python
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)])
```

Задачи выполняются по очереди, по одному шагу за раз, в одном потоке.

Добавим таймеры. У задачи будет два состояния:
готова к выполнению или ждёт срабатывания таймера.

#### Запрос к планировщику через `yield`

Генератор выдаёт запрос `Sleep`, а планировщик решает, когда его возобновить.
Это похоже на системный вызов, но `yield Sleep(...)` в ядро ОС не обращается.

В очереди храним идентификатор задачи, генератор и значение для `send()`.
Текущий планировщик сохраняем в `ContextVar` на время `run()`.

```{code-cell} python
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()` до срабатывания ближайшего таймера. Постоянный опрос не нужен.

Добавим функции, которые вызываются из генераторов:

```{code-cell} python
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)
```

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

```{code-cell} python
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")
```

В начале все три воркера печатают шаг `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 также ждёт сетевые события; наш планировщик работает только с таймерами.

## Материалы

- [`asyncio`: обзор и основные функции](https://docs.python.org/3.14/library/asyncio.html)
- [Корутины, задачи, отмена и `TaskGroup`](https://docs.python.org/3.14/library/asyncio-task.html)
- [PEP 3156: устройство `asyncio`](https://peps.python.org/pep-3156/)
- [PEP 492: `async def` и `await`](https://peps.python.org/pep-0492/)
- [PEP 654: группы исключений и `except*`](https://peps.python.org/pep-0654/)
- [`pytest-asyncio`: настройки и примеры](https://pytest-asyncio.readthedocs.io/en/stable/)
- [Тестирование с `anyio`](https://anyio.readthedocs.io/en/stable/testing.html)