Async Rust: от Future до продакшена

О проекте

• Курс основан на оригинальных материалах Rust Training. Лицензии: MIT (код) и CC-BY-4.0 (текст).
• Русский перевод и адаптация выполнены участниками проекта RustTraining-Rus.
• Замечания и исправления присылайте через Issues и Pull Requests репозитория.

Подробное руководство по асинхронному программированию в Rust. В отличие от большинства туториалов по async, которые начинаются с tokio::main и обходят внутреннее устройство стороной, эта книга строит понимание с первых принципов: трейт Future, опрос (polling), конечные автоматы — и только потом переходит к практическим паттернам, выбору рантайма и продакшен-ловушкам.

Для кого эта книга

• Разработчики на Rust, которые уверенно пишут синхронный код, но путаются в async
• Разработчики из C#, Go, Python или JavaScript, которые знают async/await, но не знают модели Rust
• Все, кто натыкался на Future is not Send, Pin<Box<dyn Future>> или вопрос «почему программа зависла?»

Предварительные требования

Вы должны уверенно владеть:

• Владением, заимствованием и временами жизни (lifetimes)
• Трейтами и обобщёнными типами (включая impl Trait)
• Типом Result<T, E> и оператором ?
• Базовой многопоточностью (std::thread::spawn, Arc, Mutex)

Предыдущий опыт асинхронного Rust не нужен.

Как пользоваться этой книгой

При первом чтении идите по порядку. Части I–III опираются друг на друга. Каждая глава содержит:

В каждой главе есть:

• Блок «Что вы узнаете» в начале
• Диаграммы Mermaid для тех, кто лучше воспринимает визуально
• Встроенное упражнение со скрытым решением
• Ключевые выводы, которые суммируют основные идеи
• Перекрёстные ссылки на связанные главы

План изучения

Общее расчётное время: 22–30 часов

Работа с упражнениями

Каждая содержательная глава содержит встроенное упражнение. Итоговый проект (гл. 17) объединяет всё в одном приложении. Для максимальной пользы:

1. Сначала решите упражнение, и только потом раскрывайте решение — именно в процессе борьбы с задачей и происходит обучение
2. Набирайте код, а не копируйте его — для синтаксиса Rust важна мышечная память
3. Запускайте каждый пример — cargo new async-exercises и проверяйте по ходу чтения

Содержание

Часть I: Как работает async

• 1. Почему async в Rust устроен иначе 🟢 — Фундаментальное отличие: в Rust нет встроенного рантайма
• 2. Трейт Future 🟡 — poll(), Waker и контракт, на котором всё держится
• 3. Как работает poll 🟡 — Конечный автомат опроса и минимальный исполнитель
• 4. Pin и Unpin 🔴 — Зачем самоссылающимся структурам нужен pinning
• 5. Раскрываем конечный автомат 🟢 — Что компилятор на самом деле генерирует из async fn

Часть II: Экосистема

• 6. Создаём фьючи вручную 🟡 — TimerFuture, Join, Select с нуля
• 7. Исполнители и рантаймы 🟡 — tokio, smol, async-std, embassy — как выбрать
• 8. Глубокое погружение в Tokio 🟡 — Варианты рантайма, spawn, каналы, примитивы синхронизации
• 9. Когда Tokio не подходит 🟡 — LocalSet, FuturesUnordered, дизайн, не привязанный к рантайму
• 10. Async-трейты 🟡 — RPITIT, динамическая диспетчеризация, trait_variant, async-замыкания

Часть III: Async в продакшене

• 11. Потоки и AsyncIterator 🟡 — Асинхронная итерация, AsyncRead/Write, комбинаторы потоков
• 12. Типичные ловушки 🔴 — 9 продакшен-багов и как их избежать
• 13. Продакшен-паттерны 🔴 — Graceful shutdown, обратное давление, middleware Tower
• 14. Async — это оптимизация, а не архитектура 🔴 — Синхронное ядро / асинхронная оболочка, налог на раскраску функций

Приложения

• Итоги и справочная карточка — Таблицы для быстрого поиска и деревья решений
• Итоговый проект: асинхронный чат-сервер — Создайте полноценное асинхронное приложение

1. Почему async в Rust устроен иначе 🟢

Что вы узнаете:

• Почему в Rust нет встроенного асинхронного рантайма (и что это значит для вас)
• Три ключевых свойства: ленивое выполнение, отсутствие рантайма, абстракции с нулевой стоимостью
• Когда async — правильный инструмент (а когда он медленнее)
• Как модель Rust соотносится с C#, Go, Python и JavaScript

Фундаментальное отличие

Большинство языков с async/await скрывают внутренний механизм. В C# есть пул потоков CLR. В JavaScript есть цикл событий (event loop). В Go есть горутины и планировщик, встроенный в рантайм. В Python есть asyncio.

В Rust ничего этого нет.

Нет встроенного рантайма, нет пула потоков, нет цикла событий. Ключевое слово async — это стратегия компиляции с нулевой стоимостью: она превращает вашу функцию в конечный автомат, реализующий трейт Future. Кто-то другой (исполнитель, executor) должен продвигать этот автомат вперёд.

Три ключевых свойства async в Rust

graph LR
    subgraph "C# / JS / Go"
        EAGER["Жадное выполнение<br/>Задача стартует сразу"]
        BUILTIN["Встроенный рантайм<br/>Пул потоков в комплекте"]
        GC["Управляется GC<br/>Никаких забот о временах жизни"]
    end

    subgraph "Rust (и Python*)"
        LAZY["Ленивое выполнение<br/>Ничего не происходит, пока не вызван poll/await"]
        BYOB["Свой рантайм<br/>Исполнитель выбираете вы"]
        OWNED["Действует владение<br/>Важны времена жизни, Send, Sync"]
    end

    EAGER -. "противоположность" .-> LAZY
    BUILTIN -. "противоположность" .-> BYOB
    GC -. "противоположность" .-> OWNED

    style LAZY fill:#e8f5e8,color:#000
    style BYOB fill:#e8f5e8,color:#000
    style OWNED fill:#e8f5e8,color:#000
    style EAGER fill:#e3f2fd,color:#000
    style BUILTIN fill:#e3f2fd,color:#000
    style GC fill:#e3f2fd,color:#000

* Корутины Python, как и фьючи Rust, ленивые — они не выполняются, пока их не await-ят или не запланируют. Однако Python по-прежнему использует GC и не знает о владении и временах жизни.

Нет встроенного рантайма

// Это компилируется, но НИЧЕГО не делает:
async fn fetch_data() -> String {
    "hello".to_string()
}

fn main() {
    let future = fetch_data(); // Создаёт Future, но не выполняет его
    // future — это просто структура на стеке
    // Никакого вывода, никаких побочных эффектов, ничего не происходит
    drop(future); // Молча уничтожена — работа так и не начиналась
}

Сравните с C#, где Task стартует сразу:

// C# — это немедленно начинает выполняться:
async Task<string> FetchData() => "hello";

var task = FetchData(); // Уже выполняется!
var result = await task; // Только ждём завершения

Ленивые фьючи против жадных задач

Это самая важная смена мышления:

// Чтобы РЕАЛЬНО выполнить фьючу, нужен исполнитель:
#[tokio::main]
async fn main() {
    let result = fetch_data().await; // ТЕПЕРЬ оно выполняется
    println!("{result}");
}

Когда использовать async (и когда не стоит)

graph TD
    START["Какой характер работы?"]

    IO["Ограничено вводом-выводом?<br/>(сеть, файлы, БД)"]
    CPU["Ограничено процессором?<br/>(вычисления, парсинг)"]
    MANY["Много одновременных соединений?<br/>(100+)"]
    FEW["Мало одновременных задач?<br/>(<10)"]

    USE_ASYNC["✅ Используйте async/await"]
    USE_THREADS["✅ Используйте std::thread или rayon"]
    USE_SPAWN_BLOCKING["✅ Используйте spawn_blocking()"]
    MAYBE_SYNC["Рассмотрите синхронный код<br/>(проще, меньше накладных расходов)"]

    START -->|Сеть, файлы, БД| IO
    START -->|Вычисления| CPU
    IO -->|Да, много| MANY
    IO -->|Всего несколько| FEW
    MANY --> USE_ASYNC
    FEW --> MAYBE_SYNC
    CPU -->|Распараллелить| USE_THREADS
    CPU -->|Внутри async-контекста| USE_SPAWN_BLOCKING

    style USE_ASYNC fill:#c8e6c9,color:#000
    style USE_THREADS fill:#c8e6c9,color:#000
    style USE_SPAWN_BLOCKING fill:#c8e6c9,color:#000
    style MAYBE_SYNC fill:#fff3e0,color:#000

Практическое правило: async нужен для конкурентности ввода-вывода (делать много дел одновременно, пока ждём), а не для параллелизма вычислений (ускорение одной операции). Если у вас 10 000 сетевых соединений, async проявляет себя во всей красе. Если вы считаете числа, используйте rayon или потоки ОС.

Когда async может быть медленнее

Async не бесплатен. При низкой конкурентности синхронный код может работать быстрее async:

Рекомендация по бенчмаркам: если одновременных операций ввода-вывода меньше ~10, сначала профилируйте, и только потом переходите на async. Простой std::thread::spawn на соединение хорошо масштабируется до сотен потоков на современном Linux.

Упражнение: когда использовать async?

Ключевые выводы — почему async устроен иначе

• Фьючи в Rust ленивые — они ничего не делают, пока исполнитель их не опросит
• Встроенного рантайма нет — вы выбираете (или создаёте) свой
• Async — это стратегия компиляции с нулевой стоимостью, которая порождает конечные автоматы
• Async проявляет себя в конкурентности ввода-вывода; для вычислений используйте потоки или rayon

См. также: Гл. 2 — Трейт Future — трейт, на котором всё это держится, Гл. 7 — Исполнители и рантаймы — как выбрать рантайм

2. Трейт Future 🟡

Что вы узнаете:

• Трейт Future: Output, poll(), Context, Waker
• Как waker сообщает исполнителю «опроси меня снова»
• Контракт: если никогда не вызвать wake(), программа молча зависнет
• Реализация настоящего фьючи вручную (Delay)

Анатомия Future

Всё в асинхронном Rust в конечном счёте реализует этот трейт:

#![allow(unused)]
fn main() {
pub trait Future {
    type Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

pub enum Poll<T> {
    Ready(T),   // Future завершился со значением T
    Pending,    // Future ещё не готов — вызови меня позже
}
}

Вот и всё. Future — это всё, что можно опросить (polled): у него спрашивают «ты уже готов?», и он отвечает либо «да, вот результат», либо «пока нет, я разбужу тебя, когда буду готов».

Output, poll(), Context, Waker

sequenceDiagram
    participant E as Исполнитель (Executor)
    participant F as Future (Задача)
    participant OS as Операционная система<br/>(например, epoll/kqueue)
    participant R as Реактор (Runtime)

    E->>F: Вызывает poll(cx)
    Note right of F: Future пытается выполнить операцию
    F->>OS: Системный вызов (например, read из TCP-сокета)
    OS-->>F: Возвращает ошибку: не готово
    
    F->>R: Регистрирует: (Waker)
    F-->>E: Возвращает Poll::Pending
    Note left of E: Задача убирается<br/>из очереди на выполнение

    E->>E: (Исполнитель выполняет другие задачи ИЛИ засыпает)
    R->>OS: epoll_wait() / опрашивает ОС на события

    Note right of OS: (Спустя некоторое время) приходят новые данные
    OS-->>R: Будит реактор: данные УЖЕ ГОТОВЫ
    
    R->>R: Реактор находит Waker
    R->>E: Вызывает Waker::wake()
    Note right of E: Задача возвращается<br/>в очередь исполнителя

    E->>F: Снова вызывает poll(cx)
    Note right of F: Future снова пытается выполнить операцию
    F->>OS: Системный вызов (например, read из TCP-сокета)
    OS-->>F: Успех: возвращает буфер с данными
    F-->>E: Возвращает Poll::Ready(Data)

Разберём каждую часть:

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

// Future, который сразу возвращает 42
struct Ready42;

impl Future for Ready42 {
    type Output = i32; // Что в итоге вернёт future

    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<i32> {
        Poll::Ready(42) // Всегда готов — ожидание не нужно
    }
}
}

Компоненты:

• Output — тип значения, которое получится по завершении future
• poll() — вызывается исполнителем, чтобы проверить прогресс; возвращает Ready(value) или Pending
• Pin<&mut Self> — гарантирует, что future не переместят в памяти (почему это важно, разберём в гл. 4)
• Context — содержит Waker, чтобы future мог сообщить исполнителю, что готов двигаться дальше

Контракт Waker

Waker — это механизм обратного вызова. Когда future возвращает Pending, он обязан обеспечить последующий вызов waker.wake() — иначе исполнитель никогда не опросит его снова, и программа зависнет.

#![allow(unused)]
fn main() {
use std::task::{Context, Poll, Waker};
use std::pin::Pin;
use std::future::Future;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

/// Future, который завершается через заданное время (упрощённая реализация)
struct Delay {
    completed: Arc<Mutex<bool>>,
    waker_stored: Arc<Mutex<Option<Waker>>>,
    duration: Duration,
    started: bool,
}

impl Delay {
    fn new(duration: Duration) -> Self {
        Delay {
            completed: Arc::new(Mutex::new(false)),
            waker_stored: Arc::new(Mutex::new(None)),
            duration,
            started: false,
        }
    }
}

impl Future for Delay {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        // Проверяем, не завершился ли уже, до того как сохранить waker
        if *self.completed.lock().unwrap() {
            return Poll::Ready(());
        }

        // Сохраняем waker — исполнитель может передавать новый при каждом poll
        *self.waker_stored.lock().unwrap() = Some(cx.waker().clone());

        // Запускаем фоновый таймер при первом poll
        if !self.started {
            self.started = true;
            let completed = Arc::clone(&self.completed);
            let waker = Arc::clone(&self.waker_stored);
            let duration = self.duration;

            thread::spawn(move || {
                thread::sleep(duration);
                *completed.lock().unwrap() = true;

                // ВАЖНО: будим исполнитель, чтобы он снова опросил нас
                if let Some(w) = waker.lock().unwrap().take() {
                    w.wake(); // «Эй, исполнитель, я готов — опроси меня снова!»
                }
            });
        }

        // Перепроверяем завершение после сохранения waker (защита от гонки)
        if *self.completed.lock().unwrap() {
            return Poll::Ready(());
        }

        Poll::Pending // Ещё не готово
    }
}
}

Ключевая мысль: в C# за пробуждение отвечает TaskScheduler автоматически. В Rust вы (или библиотека ввода-вывода, которую вы используете) отвечаете за вызов waker.wake(). Забудете — и программа молча зависнет.

Упражнение: реализуйте CountdownFuture

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

struct CountdownFuture {
    count: u32,
}

impl CountdownFuture {
    fn new(start: u32) -> Self {
        CountdownFuture { count: start }
    }
}

impl Future for CountdownFuture {
    type Output = &'static str;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if self.count == 0 {
            println!("Liftoff!");
            Poll::Ready("Liftoff!")
        } else {
            println!("{}...", self.count);
            self.count -= 1;
            cx.waker().wake_by_ref(); // Немедленно планируем повторный poll
            Poll::Pending
        }
    }
}
}

Ключевые выводы — трейт Future

• Future::poll() возвращает Poll::Ready(value) или Poll::Pending
• Перед возвратом Pending future обязан зарегистрировать Waker — исполнитель использует его, чтобы понять, когда опрашивать снова
• Pin<&mut Self> гарантирует, что future не переместят в памяти (нужно для самоссылающихся конечных автоматов — см. гл. 4)
• Всё в асинхронном Rust — async fn, .await, комбинаторы — построено на этом одном трейте

См. также: Гл. 3 — Как работает poll — цикл исполнителя, Гл. 6 — Создаём фьючи вручную — более сложные реализации

3. Как работает poll 🟡

Что вы узнаете:

• Цикл опроса исполнителя: poll → Pending → wake → снова poll
• Как собрать минимальный исполнитель с нуля
• Правила «ложных» пробуждений и почему они важны
• Вспомогательные функции: poll_fn() и yield_now()

Конечный автомат опроса

Исполнитель работает в цикле: опрашивает future, и если он возвращает Pending, откладывает его до срабатывания waker, а затем опрашивает снова. Это принципиально отличается от потоков ОС, где планированием занимается ядро.

stateDiagram-v2
    [*] --> Idle : Future создан
    Idle --> Polling : исполнитель вызывает poll()
    Polling --> Complete : Ready(value)
    Polling --> Waiting : Pending
    Waiting --> Polling : вызван waker.wake()
    Complete --> [*] : Значение возвращено

Важно: находясь в состоянии Waiting, future обязан был зарегистрировать waker у источника ввода-вывода. Без регистрации — зависание навсегда.

Минимальный исполнитель

Чтобы развеять магию исполнителей, построим самый простой из возможных:

use std::future::Future;
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
use std::pin::Pin;

/// Простейший исполнитель: крутим poll в цикле, пока не получим Ready
fn block_on<F: Future>(mut future: F) -> F::Output {
    // Закрепляем future на стеке
    // SAFETY: `future` не перемещается после этого момента — мы обращаемся
    // к нему только через закреплённую ссылку до самого завершения.
    let mut future = unsafe { Pin::new_unchecked(&mut future) };

    // Создаём пустой waker (он просто продолжает опрос — неэффективно, зато просто)
    fn noop_raw_waker() -> RawWaker {
        fn no_op(_: *const ()) {}
        fn clone(_: *const ()) -> RawWaker { noop_raw_waker() }
        let vtable = &RawWakerVTable::new(clone, no_op, no_op, no_op);
        RawWaker::new(std::ptr::null(), vtable)
    }

    // SAFETY: noop_raw_waker() возвращает корректный RawWaker с правильной таблицей vtable.
    let waker = unsafe { Waker::from_raw(noop_raw_waker()) };
    let mut cx = Context::from_waker(&waker);

    // Крутим цикл, пока future не завершится
    loop {
        match future.as_mut().poll(&mut cx) {
            Poll::Ready(value) => return value,
            Poll::Pending => {
                // Настоящий исполнитель усыпил бы поток здесь
                // и ждал бы waker.wake() — а мы просто крутимся
                std::thread::yield_now();
            }
        }
    }
}

// Использование:
fn main() {
    let result = block_on(async {
        println!("Hello from our mini executor!");
        42
    });
    println!("Got: {result}");
}

Не используйте это в продакшене! Здесь активный цикл, он тратит CPU впустую. Настоящие исполнители (tokio, smol) используют epoll/kqueue/io_uring, чтобы спать до готовности ввода-вывода. Но этот пример показывает главную идею: исполнитель — это просто цикл, который вызывает poll().

Уведомления о пробуждении

Настоящий исполнитель управляется событиями. Когда все future находятся в Pending, исполнитель засыпает. Waker работает как механизм прерывания:

#![allow(unused)]
fn main() {
// Концептуальная модель главного цикла настоящего исполнителя:
fn executor_loop(tasks: &mut TaskQueue) {
    loop {
        // 1. Опрашиваем все задачи, которые были разбужены
        while let Some(task) = tasks.get_woken_task() {
            match task.poll() {
                Poll::Ready(result) => task.complete(result),
                Poll::Pending => { /* задача остаётся в очереди, ждём пробуждения */ }
            }
        }

        // 2. Спим, пока что-нибудь нас не разбудит (epoll_wait, kevent и т. п.)
        //    Вот где основную работу делает mio/polling
        tasks.wait_for_events(); // блокируется до события ввода-вывода или срабатывания waker
    }
}
}

Ложные пробуждения

Future может быть опрошен, даже когда его ввод-вывод ещё не готов. Это называется ложным пробуждением (spurious wake). Future должен корректно с этим справляться:

#![allow(unused)]
fn main() {
impl Future for MyFuture {
    type Output = Data;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Data> {
        // ✅ ПРАВИЛЬНО: всегда перепроверяем реальное условие
        if let Some(data) = self.try_read_data() {
            Poll::Ready(data)
        } else {
            // Регистрируем waker заново (он мог измениться!)
            self.register_waker(cx.waker());
            Poll::Pending
        }

        // ❌ НЕПРАВИЛЬНО: считать, что poll означает готовность данных
        // let data = self.read_data(); // может заблокироваться или упасть с паникой
        // Poll::Ready(data)
    }
}
}

Правила реализации poll():

1. Никогда не блокируйтесь — если данных нет, сразу возвращайте Pending
2. Всегда регистрируйте waker заново — он мог измениться между опросами
3. Обрабатывайте ложные пробуждения — проверяйте реальное условие, а не полагайтесь на готовность
4. Не опрашивайте после Ready — поведение не определено (может вызвать панику, вернуть Pending или снова Ready). Гарантированно безопасный опрос после завершения даёт только FusedFuture

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicBool, Ordering};
use std::task::{Context, Poll, Waker};

struct FlagFuture {
    flag: Arc<AtomicBool>,
    waker_slot: Arc<Mutex<Option<Waker>>>,
}

impl FlagFuture {
    fn new(flag: Arc<AtomicBool>, waker_slot: Arc<Mutex<Option<Waker>>>) -> Self {
        FlagFuture { flag, waker_slot }
    }
}

impl Future for FlagFuture {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // Всегда перепроверяем реальное условие — никогда не доверяем одному только пробуждению
        if self.flag.load(Ordering::Acquire) {
            return Poll::Ready(());
        }

        // Сохраняем/обновляем waker, чтобы нас уведомили
        let mut slot = self.waker_slot.lock().unwrap();
        *slot = Some(cx.waker().clone());

        // Перепроверяем после сохранения waker, чтобы избежать гонки:
        // флаг мог быть установлен между нашей первой проверкой
        // и сохранением waker
        if self.flag.load(Ordering::Acquire) {
            Poll::Ready(())
        } else {
            Poll::Pending
        }
    }
}

// Сторона, которая устанавливает флаг (например, другой поток или задача):
fn set_flag(flag: &AtomicBool, waker_slot: &Mutex<Option<Waker>>) {
    flag.store(true, Ordering::Release);
    if let Some(waker) = waker_slot.lock().unwrap().take() {
        waker.wake();
    }
}

// Эквивалент через poll_fn:
// async fn wait_for_flag(flag: Arc<AtomicBool>, waker_slot: Arc<Mutex<Option<Waker>>>) {
//     std::future::poll_fn(|cx| {
//         if flag.load(Ordering::Acquire) {
//             return Poll::Ready(());
//         }
//         *waker_slot.lock().unwrap() = Some(cx.waker().clone());
//         if flag.load(Ordering::Acquire) { Poll::Ready(()) } else { Poll::Pending }
//     }).await
// }
}

Полезные утилиты: poll_fn и yield_now

Две утилиты из стандартной библиотеки и tokio, которые избавляют от написания полных реализаций Future:

#![allow(unused)]
fn main() {
use std::future::poll_fn;
use std::task::Poll;

// poll_fn: создаём разовый future из замыкания
let value = poll_fn(|cx| {
    // Делаем что-нибудь с cx.waker(), возвращаем Ready или Pending
    Poll::Ready(42)
}).await;

// Пример из практики: адаптируем API на основе колбэков к async
async fn read_when_ready(source: &MySource) -> Data {
    poll_fn(|cx| source.poll_read(cx)).await
}
}

#![allow(unused)]
fn main() {
// yield_now: добровольно отдаём управление исполнителю
// Полезно в тяжёлых по CPU асинхронных циклах, чтобы не морить голодом другие задачи
async fn cpu_heavy_work(items: &[Item]) {
    for (i, item) in items.iter().enumerate() {
        process(item); // работа с CPU

        // Каждые 100 элементов уступаем управление, чтобы могли выполниться другие задачи
        if i % 100 == 0 {
            tokio::task::yield_now().await;
        }
    }
}
}

Когда использовать yield_now(): если асинхронная функция выполняет работу с CPU в цикле без единой точки .await, она монополизирует поток исполнителя. Периодически вставляйте yield_now().await, чтобы включить кооперативную многозадачность.

Ключевые выводы — как работает poll

• Исполнитель многократно вызывает poll() для future, которые были разбужены
• Future должны обрабатывать ложные пробуждения — всегда перепроверяйте реальное условие
• poll_fn() позволяет создавать разовые future из замыканий
• yield_now() — это лаз для кооперативного планирования в async-коде с тяжёлыми вычислениями

См. также: Гл. 2 — Трейт Future — определение трейта, Гл. 5 — Раскрываем конечный автомат — что генерирует компилятор

4. Pin и Unpin 🔴

Что вы узнаете:

• Почему самоссылающиеся структуры ломаются при перемещении в памяти
• Что гарантирует Pin<P> и как он предотвращает перемещение
• Три практических паттерна закрепления: Box::pin(), tokio::pin!(), Pin::new()
• Когда Unpin даёт вам лазейку

Зачем нужен Pin

Это самое запутанное понятие в асинхронном Rust. Построим интуицию шаг за шагом.

Проблема: самоссылающиеся структуры

Когда компилятор превращает async fn в конечный автомат, этот автомат может содержать ссылки на собственные поля. Так возникает самоссылающаяся структура — и её перемещение в памяти сделает внутренние ссылки недействительными.

#![allow(unused)]
fn main() {
// Что компилятор генерирует (упрощённо) для:
// async fn example() {
//     let data = vec![1, 2, 3];
//     let reference = &data;       // Указывает на data выше
//     use_ref(reference).await;
// }

// Превращается во что-то вроде:
enum ExampleStateMachine {
    State0 {
        data: Vec<i32>,
        // reference: &Vec<i32>,  // ПРОБЛЕМА: указывает на `data` выше
        //                        // Если структура переместится, указатель повиснет!
    },
    State1 {
        data: Vec<i32>,
        reference: *const Vec<i32>, // Внутренний указатель на поле data
    },
    Complete,
}
}

graph LR
    subgraph "До перемещения (корректно)"
        A["data: [1,2,3]<br/>по адресу 0x1000"]
        B["reference: 0x1000<br/>(указывает на data)"]
        B -->|"корректно"| A
    end

    subgraph "После перемещения (НЕКОРРЕКТНО)"
        C["data: [1,2,3]<br/>по адресу 0x2000"]
        D["reference: 0x1000<br/>(всё ещё указывает на СТАРОЕ место!)"]
        D -->|"висячий указатель!"| E["💥 0x1000<br/>(освобождено / мусор)"]
    end

    style E fill:#ffcdd2,color:#000
    style D fill:#ffcdd2,color:#000
    style B fill:#c8e6c9,color:#000

Самоссылающиеся структуры

Это не академическая проблема. Любая async fn, которая удерживает ссылку через точку .await, создаёт самоссылающийся конечный автомат:

#![allow(unused)]
fn main() {
async fn problematic() {
    let data = String::from("hello");
    let slice = &data[..]; // slice заимствует data

    some_io().await; // <-- точка .await: автомат хранит и data, и slice

    println!("{slice}"); // использует ссылку после await
}
// Сгенерированный автомат содержит `data: String` и `slice: &str`,
// где slice указывает ВНУТРЬ data. Перемещение автомата = висячий указатель.
}

Pin на практике

Pin<P> — это обёртка, которая запрещает перемещать значение, на которое указывает указатель:

#![allow(unused)]
fn main() {
use std::pin::Pin;

let mut data = String::from("hello");

// Закрепляем — теперь его нельзя переместить
let pinned: Pin<&mut String> = Pin::new(&mut data);

// Всё ещё можно использовать:
println!("{}", pinned.as_ref().get_ref()); // "hello"

// Но получить обратно &mut String нельзя (это позволило бы mem::swap):
// let mutable: &mut String = Pin::into_inner(pinned); // Только если String: Unpin
// String — это Unpin, поэтому здесь это на самом деле работает.
// Но для самоссылающихся конечных автоматов (они !Unpin) это заблокировано.
}

В реальном коде Pin встречается в основном в трёх местах:

#![allow(unused)]
fn main() {
// 1. Сигнатура poll() — все future опрашиваются через Pin
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Output>;

// 2. Box::pin() — разместить future в куче и закрепить его
let future: Pin<Box<dyn Future<Output = i32>>> = Box::pin(async { 42 });

// 3. tokio::pin!() — закрепить future на стеке
tokio::pin!(my_future);
// Теперь my_future: Pin<&mut impl Future>
}

Лазейка Unpin

Большинство типов в Rust — Unpin: они не содержат самоссылок, поэтому закрепление для них ничего не меняет. !Unpin — только автоматы, сгенерированные компилятором из async fn.

#![allow(unused)]
fn main() {
// Все эти типы — Unpin, закрепление для них ничего особого не делает:
// i32, String, Vec<T>, HashMap<K,V>, Box<T>, &T, &mut T

// Эти типы — !Unpin, их ОБЯЗАТЕЛЬНО нужно закрепить перед опросом:
// Конечные автоматы, сгенерированные из `async fn` и `async {}`

// Практический вывод:
// Если вы пишете Future вручную и в нём НЕТ самоссылок,
// реализуйте Unpin, чтобы с ним было проще работать:
impl Unpin for MySimpleFuture {} // «Его безопасно перемещать, поверьте мне»
}

Краткая справка

#![allow(unused)]
fn main() {
// Фрагмент A
let fut = async { 42 };
let pinned = Box::pin(fut);
let moved = pinned; // Перемещаем Box
let result = moved.await;

// Фрагмент B
let fut = async { 42 };
tokio::pin!(fut);
let moved = fut; // Перемещаем закреплённый future
let result = moved.await;

// Фрагмент C
use std::pin::Pin;
let mut fut = async { 42 };
let pinned = Pin::new(&mut fut);
}

#![allow(unused)]
fn main() {
let fut = async { 42 };
tokio::pin!(fut);
let moved = fut;        // Перемещаем обёртку Pin<&mut> — OK
// fut.await;           // ❌ Ошибка: fut уже перемещён
let result = moved.await; // ✅ Используем moved
}

#![allow(unused)]
fn main() {
let fut = async { 42 };
let pinned = Box::pin(fut); // Закрепление в куче — работает с !Unpin
}

Ключевые выводы — Pin и Unpin

• Pin<P> — это обёртка, которая запрещает перемещать то, на что указывает; без неё невозможны самоссылающиеся конечные автоматы
• Box::pin() — безопасный и простой вариант по умолчанию для закрепления future в куче
• tokio::pin!() закрепляет на стеке — обёртку Pin<&mut> можно перемещать, но сам future остаётся на месте
• Unpin — это автотрейт-исключение: типы, реализующие Unpin, можно перемещать даже будучи закреплёнными (большинство типов — Unpin; async-блоки — нет)

См. также: Гл. 2 — Трейт Future — Pin<&mut Self> в poll, Гл. 5 — Раскрываем конечный автомат — почему async-автоматы самоссылающиеся

5. Раскрываем конечный автомат 🟢

Что вы узнаете:

• Как компилятор превращает async fn в перечисление (enum) — конечный автомат
• Сравнение бок о бок: исходный код и сгенерированные состояния
• Почему большие стековые выделения в async fn раздувают размер future
• Оптимизация уничтожения: значения освобождаются, как только больше не нужны

Что компилятор генерирует на самом деле

Когда вы пишете async fn, компилятор превращает ваш последовательно выглядящий код в конечный автомат на основе enum. Понимание этого преобразования — ключ к пониманию производительности асинхронного Rust и многих его особенностей.

Бок о бок: async fn и конечный автомат

#![allow(unused)]
fn main() {
// Что вы пишете:
async fn fetch_two_pages() -> String {
    let page1 = http_get("https://example.com/a").await;
    let page2 = http_get("https://example.com/b").await;
    format!("{page1}\n{page2}")
}
}

Компилятор генерирует примерно такой код:

#![allow(unused)]
fn main() {
enum FetchTwoPagesStateMachine {
    // Состояние 0: собираемся вызвать http_get для page1
    Start,

    // Состояние 1: ждём page1, удерживая future
    WaitingPage1 {
        fut1: HttpGetFuture,
    },

    // Состояние 2: получили page1, ждём page2
    WaitingPage2 {
        page1: String,
        fut2: HttpGetFuture,
    },

    // Конечное состояние
    Complete,
}

impl Future for FetchTwoPagesStateMachine {
    type Output = String;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<String> {
        loop {
            match self.as_mut().get_mut() {
                Self::Start => {
                    let fut1 = http_get("https://example.com/a");
                    *self.as_mut().get_mut() = Self::WaitingPage1 { fut1 };
                }
                Self::WaitingPage1 { fut1 } => {
                    let page1 = match Pin::new(fut1).poll(cx) {
                        Poll::Ready(v) => v,
                        Poll::Pending => return Poll::Pending,
                    };
                    let fut2 = http_get("https://example.com/b");
                    *self.as_mut().get_mut() = Self::WaitingPage2 { page1, fut2 };
                }
                Self::WaitingPage2 { page1, fut2 } => {
                    let page2 = match Pin::new(fut2).poll(cx) {
                        Poll::Ready(v) => v,
                        Poll::Pending => return Poll::Pending,
                    };
                    let result = format!("{page1}\n{page2}");
                    *self.as_mut().get_mut() = Self::Complete;
                    return Poll::Ready(result);
                }
                Self::Complete => panic!("polled after completion"),
            }
        }
    }
}
}

Примечание: это преобразование концептуальное. Реальный вывод компилятора использует unsafe-проекции Pin — вызовы get_mut(), показанные здесь, требуют Unpin, а async-автоматы являются !Unpin. Цель — проиллюстрировать переходы между состояниями, а не получить код, который компилируется.

stateDiagram-v2
    [*] --> Start
    Start --> WaitingPage1: Создаём future http_get #1
    WaitingPage1 --> WaitingPage1: poll() → Pending
    WaitingPage1 --> WaitingPage2: poll() → Ready(page1)
    WaitingPage2 --> WaitingPage2: poll() → Pending
    WaitingPage2 --> Complete: poll() → Ready(page2)
    Complete --> [*]: Возвращаем format!("{page1}\\n{page2}")

Содержимое состояний:

• WaitingPage1 — хранит fut1: HttpGetFuture (page2 ещё не выделен)
• WaitingPage2 — хранит page1: String, fut2: HttpGetFuture (fut1 уже уничтожен)

Почему это важно для производительности

Нулевая стоимость: конечный автомат — это enum на стеке. Никаких выделений в куче на каждый future, никакого сборщика мусора, никакого боксинга — если вы явно не используете Box::pin().

Размер: размер enum равен максимуму размеров всех его вариантов. Каждая точка .await создаёт новый вариант. Это означает:

#![allow(unused)]
fn main() {
async fn small() {
    let a: u8 = 0;
    yield_now().await;
    let b: u8 = 0;
    yield_now().await;
}
// Размер ≈ max(size_of(u8), size_of(u8)) + дискриминант + размеры future
//       ≈ маленький!

async fn big() {
    let buf: [u8; 1_000_000] = [0; 1_000_000]; // 1 МБ на стеке!
    some_io().await;
    process(&buf);
}
// Размер ≈ 1 МБ + размеры вложенных future
// ⚠️ Не размещайте огромные буферы на стеке в async-функциях!
// Используйте Vec<u8> или Box<[u8]>.
}

Оптимизация уничтожения: при переходе между состояниями автомат уничтожает значения, которые больше не нужны. В примере выше fut1 уничтожается при переходе из WaitingPage1 в WaitingPage2 — компилятор вставляет уничтожение автоматически.

Практическое правило: большие стековые выделения в async fn раздувают размер future. Если вы видите переполнение стека в async-коде, проверьте большие массивы или глубоко вложенные future. При необходимости используйте Box::pin(), чтобы разместить вложенные future в куче.

Упражнение: предскажите конечный автомат

#![allow(unused)]
fn main() {
async fn pipeline(url: &str) -> Result<usize, Error> {
    let response = fetch(url).await?;
    let body = response.text().await?;
    let parsed = parse(body).await?;
    Ok(parsed.len())
}
}

Ключевые выводы — раскрываем конечный автомат

• async fn компилируется в enum с одним вариантом на каждую точку .await
• Размер future = максимум размеров всех вариантов — большие значения на стеке его раздувают
• Компилятор автоматически вставляет уничтожение значений при переходах между состояниями
• Используйте Box::pin() или выделение в куче, когда размер future становится проблемой

См. также: Гл. 4 — Pin и Unpin — почему сгенерированный enum нужно закреплять, Гл. 6 — Создаём фьючи вручную — построить такие автоматы самостоятельно

6. Создаём фьючи вручную 🟡

Что вы узнаете:

• Реализация TimerFuture с пробуждением через поток
• Построение комбинатора Join: одновременный запуск двух фьючей
• Построение комбинатора Select: гонка двух фьючей
• Как комбинаторы компонуются — фьючи внутри фьючей до самого низа

Простой таймер-фьюча

Теперь построим настоящие, полезные фьючи с нуля. Это закрепит теорию из глав 2–5.

TimerFuture: полный пример

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};
use std::thread;
use std::time::{Duration, Instant};

pub struct TimerFuture {
    shared_state: Arc<Mutex<SharedState>>,
}

struct SharedState {
    completed: bool,
    waker: Option<Waker>,
}

impl TimerFuture {
    pub fn new(duration: Duration) -> Self {
        let shared_state = Arc::new(Mutex::new(SharedState {
            completed: false,
            waker: None,
        }));

        // Запускаем поток, который через duration выставит completed=true
        let thread_shared_state = Arc::clone(&shared_state);
        thread::spawn(move || {
            thread::sleep(duration);
            let mut state = thread_shared_state.lock().unwrap();
            state.completed = true;
            if let Some(waker) = state.waker.take() {
                waker.wake(); // Уведомляем исполнитель
            }
        });

        TimerFuture { shared_state }
    }
}

impl Future for TimerFuture {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let mut state = self.shared_state.lock().unwrap();
        if state.completed {
            Poll::Ready(())
        } else {
            // Сохраняем waker, чтобы поток таймера смог нас разбудить
            // ВАЖНО: всегда обновляйте waker — исполнитель мог заменить его
            // между опросами
            state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }
}

// Использование:
// async fn example() {
//     println!("Запускаем таймер...");
//     TimerFuture::new(Duration::from_secs(2)).await;
//     println!("Таймер сработал!");
// }
//
// ⚠️ Здесь на каждый таймер создаётся системный поток — для обучения это нормально,
// но в продакшене используйте `tokio::time::sleep`: он работает на общем
// колесе таймеров и не требует дополнительных потоков.
}

Join: одновременный запуск двух фьючей

Join опрашивает два future и завершается, когда оба закончились. Так tokio::join! работает внутри:

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

/// Опрашивает два future одновременно и возвращает оба результата в кортеже
pub struct Join<A, B>
where
    A: Future,
    B: Future,
{
    a: MaybeDone<A>,
    b: MaybeDone<B>,
}

enum MaybeDone<F: Future> {
    Pending(F),
    Done(F::Output),
    Taken, // Результат уже забран
}

// MaybeDone<F> хранит F::Output, и компилятор не может доказать,
// что он Unpin, даже когда F: Unpin. Поскольку мы используем Join только с Unpin-
// future и никогда не проецируем закрепление на поля, ручная реализация
// Unpin безопасна и позволяет вызывать self.get_mut() в poll().
impl<A: Future + Unpin, B: Future + Unpin> Unpin for Join<A, B> {}

impl<A, B> Join<A, B>
where
    A: Future,
    B: Future,
{
    pub fn new(a: A, b: B) -> Self {
        Join {
            a: MaybeDone::Pending(a),
            b: MaybeDone::Pending(b),
        }
    }
}

impl<A, B> Future for Join<A, B>
where
    A: Future + Unpin,
    B: Future + Unpin,
{
    type Output = (A::Output, B::Output);

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();

        // Опрашиваем A, если он ещё не завершён
        if let MaybeDone::Pending(ref mut fut) = this.a {
            if let Poll::Ready(val) = Pin::new(fut).poll(cx) {
                this.a = MaybeDone::Done(val);
            }
        }

        // Опрашиваем B, если он ещё не завершён
        if let MaybeDone::Pending(ref mut fut) = this.b {
            if let Poll::Ready(val) = Pin::new(fut).poll(cx) {
                this.b = MaybeDone::Done(val);
            }
        }

        // Оба готовы?
        match (&this.a, &this.b) {
            (MaybeDone::Done(_), MaybeDone::Done(_)) => {
                // Забираем оба результата
                let a_val = match std::mem::replace(&mut this.a, MaybeDone::Taken) {
                    MaybeDone::Done(v) => v,
                    _ => unreachable!(),
                };
                let b_val = match std::mem::replace(&mut this.b, MaybeDone::Taken) {
                    MaybeDone::Done(v) => v,
                    _ => unreachable!(),
                };
                Poll::Ready((a_val, b_val))
            }
            _ => Poll::Pending, // Хотя бы один ещё ожидает
        }
    }
}

// Использование (async-блоки являются !Unpin, поэтому оборачиваем их в Box::pin):
// let (page1, page2) = Join::new(
//     Box::pin(http_get("https://example.com/a")),
//     Box::pin(http_get("https://example.com/b")),
// ).await;
// Оба запроса выполняются одновременно!
}

Ключевая мысль: «одновременно» здесь означает чередование на одном потоке. Join не создаёт потоков — он опрашивает оба future в одном и том же вызове poll(). Это кооперативная конкурентность, а не параллелизм.

graph LR
    subgraph "Комбинаторы фьючей"
        direction TB
        TIMER["TimerFuture<br/>Один future, пробуждение через задержку"]
        JOIN["Join&lt;A, B&gt;<br/>Ждём ОБА"]
        SELECT["Select&lt;A, B&gt;<br/>Ждём ПЕРВЫЙ"]
        RETRY["RetryFuture<br/>Пересоздаём при ошибке"]
    end

    TIMER --> JOIN
    TIMER --> SELECT
    SELECT --> RETRY

    style TIMER fill:#d4efdf,stroke:#27ae60,color:#000
    style JOIN fill:#e8f4f8,stroke:#2980b9,color:#000
    style SELECT fill:#fef9e7,stroke:#f39c12,color:#000
    style RETRY fill:#fadbd8,stroke:#e74c3c,color:#000

Select: гонка двух фьючей

Select завершается, когда любой из future закончился первым (второй уничтожается):

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

pub enum Either<A, B> {
    Left(A),
    Right(B),
}

/// Возвращает тот future, который завершился первым; второй уничтожает
pub struct Select<A, B> {
    a: A,
    b: B,
}

impl<A, B> Select<A, B>
where
    A: Future + Unpin,
    B: Future + Unpin,
{
    pub fn new(a: A, b: B) -> Self {
        Select { a, b }
    }
}

impl<A, B> Future for Select<A, B>
where
    A: Future + Unpin,
    B: Future + Unpin,
{
    type Output = Either<A::Output, B::Output>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // Сначала опрашиваем A
        if let Poll::Ready(val) = Pin::new(&mut self.a).poll(cx) {
            return Poll::Ready(Either::Left(val));
        }

        // Затем опрашиваем B
        if let Poll::Ready(val) = Pin::new(&mut self.b).poll(cx) {
            return Poll::Ready(Either::Right(val));
        }

        Poll::Pending
    }
}

// Использование с таймаутом:
// match Select::new(http_get(url), TimerFuture::new(timeout)).await {
//     Either::Left(response) => println!("Получен ответ: {}", response),
//     Either::Right(()) => println!("Запрос просрочен по таймауту!"),
// }
}

Замечание о справедливости: наш Select всегда опрашивает A первым — если готовы оба, всегда побеждает A. Макрос select! в tokio рандомизирует порядок опроса, чтобы обеспечить справедливость.

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

pub struct RetryFuture<F, Fut, T, E>
where
    F: Fn() -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    factory: F,
    current: Option<Pin<Box<Fut>>>,
    remaining: usize,
    last_error: Option<E>,
}

impl<F, Fut, T, E> RetryFuture<F, Fut, T, E>
where
    F: Fn() -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    pub fn new(max_attempts: usize, factory: F) -> Self {
        let current = Some(Box::pin((factory)()));
        RetryFuture {
            factory,
            current,
            remaining: max_attempts.saturating_sub(1),
            last_error: None,
        }
    }
}

impl<F, Fut, T, E> Future for RetryFuture<F, Fut, T, E>
where
    F: Fn() -> Fut + Unpin,
    Fut: Future<Output = Result<T, E>>,
    E: Unpin,
{
    type Output = Result<T, E>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // Pin<Box<Fut>> всегда Unpin, поэтому структура Unpin, когда Unpin F и E.
        // Это позволяет безопасно использовать get_mut() без unsafe-кода.
        loop {
            if let Some(ref mut fut) = self.current {
                match fut.as_mut().poll(cx) {
                    Poll::Ready(Ok(val)) => return Poll::Ready(Ok(val)),
                    Poll::Ready(Err(e)) => {
                        self.last_error = Some(e);
                        if self.remaining > 0 {
                            self.remaining -= 1;
                            self.current = Some(Box::pin((self.factory)()));
                            // Возвращаемся в цикл, чтобы сразу опросить новый future
                        } else {
                            return Poll::Ready(Err(self.last_error.take().unwrap()));
                        }
                    }
                    Poll::Pending => return Poll::Pending,
                }
            } else {
                return Poll::Ready(Err(self.last_error.take().unwrap()));
            }
        }
    }
}

// Использование:
// let result = RetryFuture::new(3, || async {
//     http_get("https://flaky-server.com/api").await
// }).await;
}

Ключевые выводы — создаём фьючи вручную

• Future нужны три вещи: состояние, реализация poll() и регистрация waker
• Join опрашивает оба вложенных future; Select возвращает тот, что завершится первым
• Комбинаторы — это сами фьючи, оборачивающие другие фьючи: и так до самого низа
• Построение фьючей вручную даёт глубокое понимание, но в продакшене используйте tokio::join!/select!

См. также: Гл. 2 — Трейт Future — определение трейта, Гл. 8 — Глубокое погружение в Tokio — продакшен-аналоги

7. Исполнители и рантаймы 🟡

Что вы узнаете:

• Что делает исполнитель: опрашивает future и эффективно засыпает
• Шесть основных рантаймов: mio, io_uring, tokio, async-std, smol, embassy
• Дерево решений для выбора подходящего рантайма
• Почему важен дизайн библиотек, не привязанных к рантайму

Что делает исполнитель

У исполнителя две задачи:

1. Опрашивать future, когда они готовы продвинуться дальше
2. Эффективно засыпать, когда ни один future не готов (используя API уведомлений ввода-вывода ОС)

graph TB
    subgraph Executor["Исполнитель (например, tokio)"]
        QUEUE["Очередь задач"]
        POLLER["Опрос I/O<br/>(epoll/kqueue/io_uring)"]
        THREADS["Пул рабочих потоков"]
    end

    subgraph Tasks["Задачи"]
        T1["Задача 1<br/>(HTTP-запрос)"]
        T2["Задача 2<br/>(запрос к БД)"]
        T3["Задача 3<br/>(чтение файла)"]
    end

    subgraph OS["Операционная система"]
        NET["Сетевой стек"]
        DISK["Дисковый ввод-вывод"]
    end

    T1 --> QUEUE
    T2 --> QUEUE
    T3 --> QUEUE
    QUEUE --> THREADS
    THREADS -->|"poll()"| T1
    THREADS -->|"poll()"| T2
    THREADS -->|"poll()"| T3
    POLLER <-->|"регистрация/уведомление"| NET
    POLLER <-->|"регистрация/уведомление"| DISK
    POLLER -->|"пробуждение задач"| QUEUE

    style Executor fill:#e3f2fd,color:#000
    style OS fill:#f3e5f5,color:#000

mio: базовый слой

mio (Metal I/O) — это не исполнитель, а самая низкоуровневая кроссплатформенная библиотека уведомлений о вводе-выводе. Она оборачивает epoll (Linux), kqueue (macOS/BSD) и IOCP (Windows).

#![allow(unused)]
fn main() {
// Условное использование mio (упрощённо):
use mio::{Events, Interest, Poll, Token};
use mio::net::TcpListener;

let mut poll = Poll::new()?;
let mut events = Events::with_capacity(128);

let mut server = TcpListener::bind("0.0.0.0:8080")?;
poll.registry().register(&mut server, Token(0), Interest::READABLE)?;

// Цикл событий — блокируется, пока что-нибудь не произойдёт
loop {
    poll.poll(&mut events, None)?; // Спим до события ввода-вывода
    for event in events.iter() {
        match event.token() {
            Token(0) => { /* у сервера новое соединение */ }
            _ => { /* другой ввод-вывод готов */ }
        }
    }
}
}

Большинство разработчиков никогда не работают с mio напрямую — tokio и smol строятся поверх него.

io_uring: завершение как основа

io_uring в Linux (ядро 5.1+) — это принципиальный сдвиг по сравнению с моделью ввода-вывода на основе готовности, которую используют mio/epoll:

На основе готовности (epoll / mio / tokio):
  1. Спрашиваем: «Можно ли читать из этого сокета?»  → epoll_wait()
  2. Ядро: «Да, готов»                                → событие EPOLLIN
  3. Приложение: read(fd, buf)                        → может всё ещё ненадолго заблокироваться!

На основе завершения (io_uring):
  1. Отправляем: «Прочитай из этого сокета в этот буфер»  → SQE
  2. Ядро: выполняет чтение асинхронно
  3. Приложение: получает готовый результат с данными     → CQE

graph LR
    subgraph "Модель готовности (epoll)"
        A1["Приложение: готово?"] --> K1["Ядро: да"]
        K1 --> A2["Приложение: теперь read()"]
        A2 --> K2["Ядро: вот данные"]
    end

    subgraph "Модель завершения (io_uring)"
        B1["Приложение: прочитай это за меня"] --> K3["Ядро: работает..."]
        K3 --> B2["Приложение: получил результат + данные"]
    end

    style B1 fill:#c8e6c9,color:#000
    style B2 fill:#c8e6c9,color:#000

Проблема владения: io_uring требует, чтобы ядро владело буфером до завершения операции. Это конфликтует со стандартным трейтом AsyncRead в Rust, который заимствует буфер. Поэтому в tokio-uring другие трейты ввода-вывода:

#![allow(unused)]
fn main() {
// Стандартный tokio (на основе готовности) — заимствует буфер:
let n = stream.read(&mut buf).await?;  // buf заимствован

// tokio-uring (на основе завершения) — забирает владение буфером:
let (result, buf) = stream.read(buf).await;  // buf передан внутрь и возвращён обратно
let n = result?;
}

// Cargo.toml: tokio-uring = "0.5"
// ЗАМЕТЬТЕ: только Linux, требуется ядро 5.1+

fn main() {
    tokio_uring::start(async {
        let file = tokio_uring::fs::File::open("data.bin").await.unwrap();
        let buf = vec![0u8; 4096];
        let (result, buf) = file.read_at(buf, 0).await;
        let bytes_read = result.unwrap();
        println!("Прочитано байт: {}: {:?}", bytes_read, &buf[..bytes_read]);
    });
}

Когда использовать io_uring: высокопроизводительный файловый ввод-вывод или сеть, где узким местом становятся накладные расходы системных вызовов (базы данных, движки хранения, прокси на 100k+ соединений). Для большинства приложений правильный выбор — стандартный tokio с epoll.

tokio: рантайм «всё в комплекте»

Доминирующий асинхронный рантайм в экосистеме Rust. Используется в Axum, Hyper, Tonic и большинстве продакшен-серверов на Rust.

// Cargo.toml:
// [dependencies]
// tokio = { version = "1", features = ["full"] }

#[tokio::main]
async fn main() {
    // Запускает многопоточный рантайм с планировщиком work-stealing
    let handle = tokio::spawn(async {
        tokio::time::sleep(std::time::Duration::from_secs(1)).await;
        "done"
    });

    let result = handle.await.unwrap();
    println!("{result}");
}

Возможности tokio: таймеры, ввод-вывод, TCP/UDP, Unix-сокеты, обработка сигналов, примитивы синхронизации (Mutex, RwLock, Semaphore, каналы), fs, process, интеграция с tracing.

async-std: зеркало стандартной библиотеки

Повторяет API std в асинхронных версиях. Менее популярен, чем tokio, но проще для начинающих.

// Cargo.toml:
// [dependencies]
// async-std = { version = "1", features = ["attributes"] }

#[async_std::main]
async fn main() {
    use async_std::fs;
    let content = fs::read_to_string("hello.txt").await.unwrap();
    println!("{content}");
}

smol: минималистичный рантайм

Небольшой асинхронный рантайм без зависимостей. Отлично подходит для библиотек, которые хотят async, но не хотят тянуть tokio.

// Cargo.toml:
// [dependencies]
// smol = "2"

fn main() {
    smol::block_on(async {
        let result = smol::unblock(|| {
            // Выполняет блокирующий код в пуле потоков
            std::fs::read_to_string("hello.txt")
        }).await.unwrap();
        println!("{result}");
    });
}

embassy: async для встраиваемых систем (no_std)

Асинхронный рантайм для встраиваемых систем. Без выделения памяти в куче, std не нужен.

// Работает на микроконтроллерах (например, STM32, nRF52, RP2040)
#[embassy_executor::main]
async fn main(spawner: embassy_executor::Spawner) {
    // Мигаем светодиодом через async/await — RTOS не нужна!
    let mut led = Output::new(p.PA5, Level::Low, Speed::Low);
    loop {
        led.set_high();
        Timer::after(Duration::from_millis(500)).await;
        led.set_low();
        Timer::after(Duration::from_millis(500)).await;
    }
}

Дерево решений по выбору рантайма

graph TD
    START["Выбор рантайма"]

    Q1{"Пишете<br/>сетевой сервер?"}
    Q2{"Нужна экосистема tokio<br/>(Axum, Tonic, Hyper)?"}
    Q3{"Пишете<br/>библиотеку?"}
    Q4{"Встраиваемая система /<br/>no_std?"}
    Q5{"Нужны минимальные<br/>зависимости?"}

    TOKIO["🟢 tokio<br/>Лучшая экосистема, самый популярный"]
    SMOL["🔵 smol<br/>Минимальный, без привязки к экосистеме"]
    EMBASSY["🟠 embassy<br/>Ориентирован на встраиваемые, без выделений"]
    ASYNC_STD["🟣 async-std<br/>API как у std, хорош для обучения"]
    AGNOSTIC["🔵 независимый от рантайма<br/>Используйте только крейт futures"]

    START --> Q1
    Q1 -->|Да| Q2
    Q1 -->|Нет| Q3
    Q2 -->|Да| TOKIO
    Q2 -->|Нет| Q5
    Q3 -->|Да| AGNOSTIC
    Q3 -->|Нет| Q4
    Q4 -->|Да| EMBASSY
    Q4 -->|Нет| Q5
    Q5 -->|Да| SMOL
    Q5 -->|Нет| ASYNC_STD

    style TOKIO fill:#c8e6c9,color:#000
    style SMOL fill:#bbdefb,color:#000
    style EMBASSY fill:#ffe0b2,color:#000
    style ASYNC_STD fill:#e1bee7,color:#000
    style AGNOSTIC fill:#bbdefb,color:#000

Сравнительная таблица рантаймов

// ----- версия для tokio -----
// Cargo.toml: tokio = { version = "1", features = ["full"] }
#[tokio::main]
async fn main() {
    let (url_result, file_result) = tokio::join!(
        async {
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            "Ответ с URL"
        },
        async {
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
            "Содержимое файла"
        },
    );
    println!("URL: {url_result}, Файл: {file_result}");
}

// ----- версия для smol -----
// Cargo.toml: smol = "2", futures-lite = "2"
fn main() {
    smol::block_on(async {
        let (url_result, file_result) = futures_lite::future::zip(
            async {
                smol::Timer::after(std::time::Duration::from_millis(100)).await;
                "Ответ с URL"
            },
            async {
                smol::Timer::after(std::time::Duration::from_millis(50)).await;
                "Содержимое файла"
            },
        ).await;
        println!("URL: {url_result}, Файл: {file_result}");
    });
}

// ----- версия для async-std -----
// Cargo.toml: async-std = { version = "1", features = ["attributes"] }
#[async_std::main]
async fn main() {
    let (url_result, file_result) = futures::future::join(
        async {
            async_std::task::sleep(std::time::Duration::from_millis(100)).await;
            "Ответ с URL"
        },
        async {
            async_std::task::sleep(std::time::Duration::from_millis(50)).await;
            "Содержимое файла"
        },
    ).await;
    println!("URL: {url_result}, Файл: {file_result}");
}

Ключевые выводы — исполнители и рантаймы

• Работа исполнителя: опрашивать future при пробуждении и эффективно засыпать, используя API ввода-вывода ОС
• tokio — выбор по умолчанию для серверов; smol — для минимального отпечатка; embassy — для встраиваемых систем
• Ваша бизнес-логика должна зависеть от std::future::Future, а не от конкретного рантайма
• io_uring (Linux 5.1+) — будущее высокопроизводительного ввода-вывода, но экосистема ещё созревает

См. также: Гл. 8 — Глубокое погружение в Tokio — специфика tokio, Гл. 9 — Когда Tokio не подходит — альтернативы

8. Глубокое погружение в Tokio 🟡

Что вы узнаете:

• Варианты рантайма: multi-thread и current-thread и когда какой использовать
• tokio::spawn, требование 'static и JoinHandle
• Семантика отмены задач (отмена при уничтожении)
• Примитивы синхронизации: Mutex, RwLock, Semaphore и все четыре типа каналов

Варианты рантайма: multi-thread и current-thread

Tokio предлагает две конфигурации рантайма:

// Многопоточный (по умолчанию с #[tokio::main])
// Использует пул потоков с work-stealing — задачи могут переходить между потоками
#[tokio::main]
async fn main() {
    // N рабочих потоков (по умолчанию — количество ядер CPU)
    // Задачи должны быть Send + 'static
}

// Current-thread — всё выполняется в одном потоке
#[tokio::main(flavor = "current_thread")]
async fn main() {
    // Однопоточный — задачам не нужен Send
    // Легче по весу, подходит для простых утилит или WASM
}

// Ручное создание рантайма:
let rt = tokio::runtime::Builder::new_multi_thread()
    .worker_threads(4)
    .enable_all()
    .build()
    .unwrap();

rt.block_on(async {
    println!("Работаем на своём рантайме");
});

graph TB
    subgraph "Multi-Thread (по умолчанию)"
        MT_Q1["Поток 1<br/>Задача A, Задача D"]
        MT_Q2["Поток 2<br/>Задача B"]
        MT_Q3["Поток 3<br/>Задача C, Задача E"]
        STEAL["Кража работы:<br/>простаивающие потоки забирают задачи у занятых"]
        MT_Q1 <--> STEAL
        MT_Q2 <--> STEAL
        MT_Q3 <--> STEAL
    end

    subgraph "Current-Thread"
        ST_Q["Один поток<br/>Задача A → Задача B → Задача C → Задача D"]
    end

    style MT_Q1 fill:#c8e6c9,color:#000
    style MT_Q2 fill:#c8e6c9,color:#000
    style MT_Q3 fill:#c8e6c9,color:#000
    style ST_Q fill:#bbdefb,color:#000

tokio::spawn и требование ’static

tokio::spawn помещает future в очередь задач рантайма. Поскольку он может выполняться на любом рабочем потоке в любой момент, future должен быть Send + 'static:

#![allow(unused)]
fn main() {
use tokio::task;

async fn example() {
    let data = String::from("hello");

    // ✅ Работает: передаём владение в задачу
    let handle = task::spawn(async move {
        println!("{data}");
        data.len()
    });

    let len = handle.await.unwrap();
    println!("Длина: {len}");
}

async fn problem() {
    let data = String::from("hello");

    // ❌ НЕ РАБОТАЕТ: data заимствован, а не 'static
    // task::spawn(async {
    //     println!("{data}"); // заимствует `data` — не 'static
    // });

    // ❌ НЕ РАБОТАЕТ: Rc не является Send
    // let rc = std::rc::Rc::new(42);
    // task::spawn(async move {
    //     println!("{rc}"); // Rc — !Send, нельзя пересечь границу потока
    // });
}
}

Почему 'static? Порождённая задача выполняется независимо — она может пережить область видимости, в которой была создана. Компилятор не может доказать, что ссылки останутся действительными, поэтому требуются данные, которыми задача владеет.

Почему Send? Задачу могут возобновить на другом потоке, а не на том, где её приостановили. Все данные, которые хранятся через точки .await, должны безопасно передаваться между потоками.

#![allow(unused)]
fn main() {
// Типичный паттерн: клонируем общие данные в задачу
let shared = Arc::new(config);

for i in 0..10 {
    let shared = Arc::clone(&shared); // Клонируем Arc, а не данные
    tokio::spawn(async move {
        process_item(i, &shared).await;
    });
}
}

JoinHandle и отмена задач

#![allow(unused)]
fn main() {
use tokio::task::JoinHandle;
use tokio::time::{sleep, Duration};

async fn cancellation_example() {
    let handle: JoinHandle<String> = tokio::spawn(async {
        sleep(Duration::from_secs(10)).await;
        "completed".to_string()
    });

    // Отменить задачу, уничтожив handle? НЕТ — задача продолжит работать!
    // drop(handle); // Задача продолжает выполняться в фоне

    // Чтобы действительно отменить, вызовите abort():
    handle.abort();

    // Ожидание отменённой задачи возвращает JoinError
    match handle.await {
        Ok(val) => println!("Получено: {val}"),
        Err(e) if e.is_cancelled() => println!("Задача отменена"),
        Err(e) => println!("Задача завершилась паникой: {e}"),
    }
}
}

Важно: уничтожение JoinHandle НЕ отменяет задачу в tokio. Задача становится отсоединённой и продолжает работать. Чтобы её отменить, нужно явно вызвать .abort(). Это отличается от уничтожения Future напрямую, которое действительно отменяет (уничтожает) базовое вычисление.

Примитивы синхронизации Tokio

Tokio предоставляет примитивы синхронизации, которые понимают async. Ключевой принцип: не используйте std::sync::Mutex через точки .await.

#![allow(unused)]
fn main() {
use tokio::sync::{Mutex, RwLock, Semaphore, mpsc, oneshot, broadcast, watch};

// --- Mutex ---
// Асинхронный мьютекс: метод lock() асинхронный и не блокирует поток
let data = Arc::new(Mutex::new(vec![1, 2, 3]));
{
    let mut guard = data.lock().await; // Неблокирующая блокировка
    guard.push(4);
} // Guard уничтожен здесь — блокировка снята

// --- Каналы ---
// mpsc: несколько отправителей, один получатель
let (tx, mut rx) = mpsc::channel::<String>(100); // Буфер ограниченного размера

tokio::spawn(async move {
    tx.send("hello".into()).await.unwrap();
});

let msg = rx.recv().await.unwrap();

// oneshot: одно значение, один получатель
let (tx, rx) = oneshot::channel::<i32>();
tx.send(42).unwrap(); // await не нужен — либо отправляет, либо ошибка
let val = rx.await.unwrap();

// broadcast: несколько отправителей, несколько получателей (каждый получает все сообщения)
let (tx, _) = broadcast::channel::<String>(100);
let mut rx1 = tx.subscribe();
let mut rx2 = tx.subscribe();

// watch: одно значение, несколько получателей (только последнее значение)
let (tx, rx) = watch::channel(0u64);
tx.send(42).unwrap();
println!("Последнее значение: {}", *rx.borrow());
}

Примечание: .unwrap() в примерах с каналами используется для краткости. В продакшене обрабатывайте ошибки отправки и получения аккуратно: неудачный .send() означает, что получатель уже уничтожен, а неудачный .recv() — что канал закрыт.

graph LR
    subgraph "Типы каналов"
        direction TB
        MPSC["mpsc<br/>N→1<br/>Буферизованная очередь"]
        ONESHOT["oneshot<br/>1→1<br/>Одно значение"]
        BROADCAST["broadcast<br/>N→N<br/>Все получатели получают всё"]
        WATCH["watch<br/>1→N<br/>Только последнее значение"]
    end

    P1["Отправитель 1"] --> MPSC
    P2["Отправитель 2"] --> MPSC
    MPSC --> C1["Получатель"]

    P3["Отправитель"] --> ONESHOT
    ONESHOT --> C2["Получатель"]

    P4["Отправитель"] --> BROADCAST
    BROADCAST --> C3["Получатель 1"]
    BROADCAST --> C4["Получатель 2"]

    P5["Отправитель"] --> WATCH
    WATCH --> C5["Получатель 1"]
    WATCH --> C6["Получатель 2"]

Разбор кейса: выбор канала для сервиса уведомлений

Вы создаёте сервис уведомлений, где:

• Несколько обработчиков API порождают события
• Одна фоновая задача группирует их в пакеты и отправляет
• Наблюдатель за конфигурацией обновляет лимиты скорости во время работы
• Сигнал остановки должен дойти до всех компонентов

Какой канал для каждого случая?

graph LR
    subgraph "Сервис уведомлений"
        direction TB
        API1["Обработчик API 1"] -->|mpsc| BATCH["Batcher"]
        API2["Обработчик API 2"] -->|mpsc| BATCH
        CONFIG["Наблюдатель конфигурации"] -->|watch| RATE["Ограничитель скорости"]
        CTRL["Ctrl+C"] -->|broadcast| API1
        CTRL -->|broadcast| BATCH
        CTRL -->|broadcast| RATE
    end

    style API1 fill:#d4efdf,stroke:#27ae60,color:#000
    style API2 fill:#d4efdf,stroke:#27ae60,color:#000
    style BATCH fill:#e8f4f8,stroke:#2980b9,color:#000
    style CONFIG fill:#fef9e7,stroke:#f39c12,color:#000
    style RATE fill:#fef9e7,stroke:#f39c12,color:#000
    style CTRL fill:#fadbd8,stroke:#e74c3c,color:#000

#![allow(unused)]
fn main() {
use std::future::Future;
use std::sync::Arc;
use tokio::sync::Semaphore;

async fn run_with_limit<F, Fut, T>(tasks: Vec<F>, limit: usize) -> Vec<T>
where
    F: FnOnce() -> Fut + Send + 'static,
    Fut: Future<Output = T> + Send + 'static,
    T: Send + 'static,
{
    let semaphore = Arc::new(Semaphore::new(limit));
    let mut handles = Vec::new();

    for task in tasks {
        let permit = Arc::clone(&semaphore);
        let handle = tokio::spawn(async move {
            let _permit = permit.acquire().await.unwrap();
            // Разрешение удерживается, пока выполняется задача, затем освобождается
            task().await
        });
        handles.push(handle);
    }

    let mut results = Vec::new();
    for handle in handles {
        results.push(handle.await.unwrap());
    }
    results
}

// Использование:
// let tasks: Vec<_> = urls.into_iter().map(|url| {
//     move || async move { fetch(url).await }
// }).collect();
// let results = run_with_limit(tasks, 10).await; // Не более 10 одновременно
}

Ключевые выводы — глубокое погружение в Tokio

• Используйте multi_thread для серверов (по умолчанию); current_thread — для CLI-утилит, тестов или типов !Send
• tokio::spawn требует future типа 'static — для совместного использования данных применяйте Arc или каналы
• Уничтожение JoinHandle не отменяет задачу — вызывайте .abort() явно
• Выбирайте примитивы синхронизации по потребности: Mutex для общего состояния, Semaphore для лимитов конкурентности, mpsc/oneshot/broadcast/watch для обмена сообщениями

См. также: Гл. 9 — Когда Tokio не подходит — альтернативы spawn, Гл. 12 — Типичные ловушки — ошибки с MutexGuard через await

9. Когда Tokio не подходит 🟡

Что вы узнаете:

• Проблему 'static: когда tokio::spawn заставляет использовать Arc повсюду
• LocalSet для фьючей типа !Send
• FuturesUnordered для конкурентности, дружественной к заимствованиям (без spawn)
• JoinSet для управляемых групп задач
• Написание библиотек, не привязанных к рантайму

graph TD
    START["Нужны конкурентные фьючи?"] --> STATIC{"Фьючи могут быть 'static?"}
    STATIC -->|Да| SEND{"Фьючи являются Send?"}
    STATIC -->|Нет| FU["FuturesUnordered<br/>Выполняется в текущей задаче"]
    SEND -->|Да| SPAWN["tokio::spawn<br/>Многопоточный"]
    SEND -->|Нет| LOCAL["LocalSet<br/>Однопоточный"]
    SPAWN --> MANAGE{"Нужно отслеживать/прерывать задачи?"}
    MANAGE -->|Да| JOINSET["JoinSet / TaskTracker"]
    MANAGE -->|Нет| HANDLE["JoinHandle"]

    style START fill:#f5f5f5,stroke:#333,color:#000
    style FU fill:#d4efdf,stroke:#27ae60,color:#000
    style SPAWN fill:#e8f4f8,stroke:#2980b9,color:#000
    style LOCAL fill:#fef9e7,stroke:#f39c12,color:#000
    style JOINSET fill:#e8daef,stroke:#8e44ad,color:#000
    style HANDLE fill:#e8f4f8,stroke:#2980b9,color:#000

Проблема ’static у фьючей

spawn в Tokio требует future типа 'static. Это значит, что в порождённых задачах нельзя заимствовать локальные данные:

#![allow(unused)]
fn main() {
async fn process_items(items: &[String]) {
    // ❌ Так нельзя — items заимствован, а не 'static
    // for item in items {
    //     tokio::spawn(async {
    //         process(item).await;
    //     });
    // }

    // 😐 Обходной путь 1: клонировать всё
    for item in items {
        let item = item.clone();
        tokio::spawn(async move {
            process(&item).await;
        });
    }

    // 😐 Обходной путь 2: использовать Arc
    let items = Arc::new(items.to_vec());
    for i in 0..items.len() {
        let items = Arc::clone(&items);
        tokio::spawn(async move {
            process(&items[i]).await;
        });
    }
}
}

Это раздражает! В Go можно просто написать go func() { use(item) } с замыканием. В Rust система владения заставляет думать о том, кому что принадлежит и как долго это живёт.

Альтернативы tokio::spawn

Не каждая задача требует spawn. Вот три инструмента, каждый из которых решает другое ограничение:

#![allow(unused)]
fn main() {
// 1. FuturesUnordered — полностью обходит 'static (без spawn!)
use futures::stream::{FuturesUnordered, StreamExt};

async fn process_items(items: &[String]) {
    let futures: FuturesUnordered<_> = items
        .iter()
        .map(|item| async move {
            // ✅ Может заимствовать item — ни spawn, ни 'static не нужны!
            process(item).await
        })
        .collect();

    // Доводим все фьючи до завершения
    futures.for_each(|result| async move {
        println!("Результат: {result:?}");
    }).await;
}

// 2. tokio::task::LocalSet — запускает фьючи !Send в текущем потоке
//    ⚠️  Всё ещё требует 'static — решает Send, а не 'static
use tokio::task::LocalSet;

let local_set = LocalSet::new();
local_set.run_until(async {
    tokio::task::spawn_local(async {
        // Здесь можно использовать Rc, Cell и другие типы !Send
        let rc = std::rc::Rc::new(42);
        println!("{rc}");
    }).await.unwrap();
}).await;

// 3. JoinSet в tokio (tokio 1.21+) — управляемый набор порождённых задач
//    ⚠️  Всё ещё требует 'static + Send — решает управление задачами,
//    а не проблему 'static. Полезен для отслеживания, прерывания и
//    ожидания динамической группы задач.
use tokio::task::JoinSet;

async fn with_joinset() {
    let mut set = JoinSet::new();

    for i in 0..10 {
        // i — Copy и перемещается в замыкание, поэтому он уже 'static.
        // Для заимствованных данных всё равно понадобятся Arc или clone.
        set.spawn(async move {
            tokio::time::sleep(Duration::from_millis(100)).await;
            i * 2
        });
    }

    while let Some(result) = set.join_next().await {
        println!("Задача завершена: {:?}", result.unwrap());
    }
}
}

Какой инструмент решает какую проблему?

Ограничение, с которым вы столкнулись	Инструмент	Обходит 'static?	Обходит Send?
Нельзя сделать фьючи 'static	FuturesUnordered	✅ Да	✅ Да
Фьючи 'static, но !Send	LocalSet	❌ Нет	✅ Да
Нужно отслеживать / прерывать порождённые задачи	JoinSet	❌ Нет	❌ Нет

Лёгкие рантаймы для библиотек

Если вы пишете библиотеку — не заставляйте пользователей использовать tokio:

#![allow(unused)]
fn main() {
// ❌ ПЛОХО: библиотека навязывает tokio пользователям
pub async fn my_lib_function() {
    tokio::time::sleep(Duration::from_secs(1)).await;
    // Теперь твои пользователи ОБЯЗАНЫ использовать tokio
}

// ✅ ХОРОШО: библиотека не привязана к рантайму
pub async fn my_lib_function() {
    // Используем только типы из std::future и крейта futures
    do_computation().await;
}

// ✅ ХОРОШО: принимаем обобщённую фьючу для операций ввода-вывода
pub async fn fetch_with_retry<F, Fut, T, E>(
    operation: F,
    max_retries: usize,
) -> Result<T, E>
where
    F: Fn() -> Fut,
    Fut: Future<Output = Result<T, E>>,
{
    for attempt in 0..max_retries {
        match operation().await {
            Ok(val) => return Ok(val),
            Err(e) if attempt == max_retries - 1 => return Err(e),
            Err(_) => continue,
        }
    }
    unreachable!()
}
}

Эмпирическое правило: библиотеки должны зависеть от крейта futures, а не от tokio. Приложения должны зависеть от tokio (или от выбранного рантайма). Так экосистема остаётся компонуемой.

#![allow(unused)]
fn main() {
use futures::stream::{FuturesUnordered, StreamExt};
use tokio::time::{sleep, Duration};

// Версия 1: tokio::spawn — требует 'static, нужно клонирование
async fn lengths_with_spawn(items: &[String]) -> Vec<usize> {
    let mut handles = Vec::new();
    for item in items {
        let owned = item.clone(); // Обязательно клонируем — spawn требует 'static
        handles.push(tokio::spawn(async move {
            sleep(Duration::from_millis(10)).await;
            owned.len()
        }));
    }

    let mut results = Vec::new();
    for handle in handles {
        results.push(handle.await.unwrap());
    }
    results
}

// Версия 2: FuturesUnordered — заимствует данные, клонирование не нужно
async fn lengths_without_spawn(items: &[String]) -> Vec<usize> {
    let futures: FuturesUnordered<_> = items
        .iter()
        .map(|item| async move {
            sleep(Duration::from_millis(10)).await;
            item.len() // ✅ Заимствует item — без клонирования!
        })
        .collect();

    futures.collect().await
}

#[tokio::test]
async fn test_both_versions() {
    let items = vec!["hello".into(), "world".into(), "rust".into()];

    let v1 = lengths_with_spawn(&items).await;
    // Примечание: v1 сохраняет порядок вставки (последовательное ожидание)

    let mut v2 = lengths_without_spawn(&items).await;
    v2.sort(); // FuturesUnordered возвращает результаты в порядке завершения

    assert_eq!(v1, vec![5, 5, 4]);
    assert_eq!(v2, vec![4, 5, 5]);
}
}

Ключевые выводы — когда Tokio не подходит

• FuturesUnordered выполняет фьючи конкурентно в текущей задаче — требования 'static нет
• LocalSet позволяет запускать фьючи !Send в однопоточном исполнителе
• JoinSet (tokio 1.21+) предоставляет управляемые группы задач с автоматической очисткой
• Для библиотек: зависьте только от std::future::Future и крейта futures, а не напрямую от tokio

См. также: Гл. 8 — Глубокое погружение в Tokio — когда spawn — правильный инструмент, Гл. 11 — Потоки — buffer_unordered() как ещё один ограничитель конкурентности

10. Async-трейты 🟡

Что вы узнаете:

• Почему асинхронные методы в трейтах стабилизировались так долго
• RPITIT: нативные асинхронные методы трейтов (Rust 1.75+)
• Проблему динамической диспетчеризации и ограничения Send через trait_variant
• Асинхронные замыкания (Rust 1.85+): async Fn() и async FnOnce()

graph TD
    subgraph "Подходы к async-трейтам"
        direction TB
        RPITIT["RPITIT (Rust 1.75+)<br/>async fn в трейте<br/>Только статическая диспетчеризация"]
        VARIANT["trait_variant<br/>Автоматически генерирует вариант с Send<br/>Только статическая диспетчеризация"]
        BOXED["Box&lt;dyn Future&gt;<br/>Ручное боксирование<br/>Работает везде"]
        CLOSURE["Асинхронные замыкания (1.85+)<br/>async Fn() / async FnOnce()<br/>Колбэки и middleware"]
    end

    RPITIT -->|"Нужен Send?"| VARIANT
    RPITIT -->|"Нужен dyn?"| BOXED
    CLOSURE -->|"Заменяет"| BOXED

    style RPITIT fill:#d4efdf,stroke:#27ae60,color:#000
    style VARIANT fill:#e8f4f8,stroke:#2980b9,color:#000
    style BOXED fill:#fef9e7,stroke:#f39c12,color:#000
    style CLOSURE fill:#e8daef,stroke:#8e44ad,color:#000

История: почему это заняло так много времени

Асинхронные методы в трейтах многие годы были самой востребованной функцией Rust. Проблема:

#![allow(unused)]
fn main() {
// Это не компилировалось до Rust 1.75 (декабрь 2023):
trait DataStore {
    async fn get(&self, key: &str) -> Option<String>;
}
// Почему? Потому что async fn возвращает `impl Future<Output = T>`,
// а `impl Trait` в позиции возвращаемого значения трейта не поддерживался.
}

Основная сложность: когда метод трейта возвращает impl Future, каждый реализующий тип возвращает свой конкретный тип. Компилятору нужно знать размер возвращаемого типа, но методы трейтов диспетчеризуются динамически.

RPITIT: возвращаемый impl Trait в трейте

С Rust 1.75 это просто работает для статической диспетчеризации:

#![allow(unused)]
fn main() {
trait DataStore {
    async fn get(&self, key: &str) -> Option<String>;
    // Эквивалентно:
    // fn get(&self, key: &str) -> impl Future<Output = Option<String>>;
}

struct InMemoryStore {
    data: std::collections::HashMap<String, String>,
}

impl DataStore for InMemoryStore {
    async fn get(&self, key: &str) -> Option<String> {
        self.data.get(key).cloned()
    }
}

// ✅ Работает с обобщёнными типами (статическая диспетчеризация):
async fn lookup<S: DataStore>(store: &S, key: &str) {
    if let Some(val) = store.get(key).await {
        println!("{key} = {val}");
    }
}
}

Динамическая диспетчеризация и ограничения Send

Ограничение: нельзя напрямую использовать dyn DataStore, потому что компилятор не знает размер возвращаемого future:

#![allow(unused)]
fn main() {
// ❌ Не работает:
// async fn lookup_dyn(store: &dyn DataStore, key: &str) { ... }
// Ошибка: трейт `DataStore` не является dyn-совместимым, потому что метод `get`
//         является `async`

// ✅ Обходной путь: возвращаем упакованный (boxed) future
trait DynDataStore {
    fn get(&self, key: &str) -> Pin<Box<dyn Future<Output = Option<String>> + Send + '_>>;
}
}

Проблема Send: в многопоточных рантаймах порождённые задачи должны быть Send. Но асинхронные методы трейтов не добавляют ограничение Send автоматически:

#![allow(unused)]
fn main() {
trait Worker {
    async fn run(self); // Future может быть Send, а может и не быть
}

struct MyWorker;

impl Worker for MyWorker {
    async fn run(self) {
        // Если здесь используются типы !Send, future тоже !Send
        let rc = std::rc::Rc::new(42);
        some_work().await;
        println!("{rc}");
    }
}

// ❌ Это не сработает, потому что future — !Send (Rc — !Send):
// tokio::spawn(worker.run()); // Требует Send + 'static
//
// Примечание: здесь мы используем `self` (владение), потому что tokio::spawn
// также требует 'static — future, заимствующий &self, не может быть 'static.
// Даже без Rc `async fn run(&self)` нельзя было бы порождать.
}

Крейт trait_variant

Крейт trait_variant (из рабочей группы по async в Rust) автоматически генерирует вариант с Send:

#![allow(unused)]
fn main() {
// Cargo.toml: trait-variant = "0.1"

#[trait_variant::make(SendDataStore: Send)]
trait DataStore {
    async fn get(&self, key: &str) -> Option<String>;
    async fn set(&self, key: &str, value: String);
}

// Теперь у вас два трейта:
// - DataStore: без ограничения Send для future
// - SendDataStore: все future являются Send
// У обоих одинаковые методы; реализующие типы реализуют DataStore
// и получают SendDataStore бесплатно, если их future являются Send.

// Используйте SendDataStore, когда нужно порождать задачи:
async fn spawn_lookup<S: SendDataStore + 'static>(store: Arc<S>) {
    tokio::spawn(async move {
        store.get("key").await;
    });
}

// ⚠️ Важно: trait_variant НЕ включает динамическую диспетчеризацию.
// Сгенерированный трейт по-прежнему использует `impl Future`, поэтому `dyn SendDataStore`
// не является dyn-совместимым. Для dyn-диспетчеризации всё ещё нужно ручное боксирование
// (см. подход с Box::pin выше) или крейт `async-trait`.
}

Краткая справка: async-трейты

Рекомендация: для нового кода (Rust 1.75+) используйте нативные async-трейты. Добавляйте trait_variant, когда нужны ограничения Send для порождения задач. Для динамической диспетчеризации используйте ручной Box::pin или крейт async-trait. Нативный подход не имеет накладных расходов для статической диспетчеризации.

Асинхронные замыкания (Rust 1.85+)

С Rust 1.85 стабилизированы async closures — замыкания, которые захватывают окружение и возвращают future:

#![allow(unused)]
fn main() {
// До 1.85: неудобный обходной путь
let urls = vec!["https://a.com", "https://b.com"];
let fetchers: Vec<_> = urls.iter().map(|url| {
    let url = url.to_string();
    // Возвращаем обычное замыкание, которое возвращает async-блок
    move || async move { reqwest::get(&url).await }
}).collect();

// После 1.85: асинхронные замыкания работают напрямую
let fetchers: Vec<_> = urls.iter().map(|url| {
    async move || { reqwest::get(url).await }
    // ↑ Это асинхронное замыкание — захватывает url и возвращает Future
}).collect();
}

Асинхронные замыкания реализуют новые трейты AsyncFn, AsyncFnMut и AsyncFnOnce, которые повторяют Fn, FnMut, FnOnce:

#![allow(unused)]
fn main() {
// Обобщённая функция, принимающая асинхронное замыкание
async fn retry<F>(max: usize, f: F) -> Result<String, Error>
where
    F: AsyncFn() -> Result<String, Error>,
{
    for _ in 0..max {
        if let Ok(val) = f().await {
            return Ok(val);
        }
    }
    f().await
}
}

Совет по миграции: если у вас код с Fn() -> impl Future<Output = T>, рассмотрите переход на AsyncFn() -> T — сигнатуры станут чище.

use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::time::{sleep, Duration};

trait Cache {
    async fn get(&self, key: &str) -> Option<String>;
    async fn set(&self, key: &str, value: String);
}

// --- Реализация в памяти ---
struct MemoryCache {
    store: Mutex<HashMap<String, String>>,
}

impl MemoryCache {
    fn new() -> Self {
        MemoryCache {
            store: Mutex::new(HashMap::new()),
        }
    }
}

impl Cache for MemoryCache {
    async fn get(&self, key: &str) -> Option<String> {
        self.store.lock().await.get(key).cloned()
    }

    async fn set(&self, key: &str, value: String) {
        self.store.lock().await.insert(key.to_string(), value);
    }
}

// --- Имитация реализации Redis ---
struct RedisCache {
    store: Mutex<HashMap<String, String>>,
    latency: Duration,
}

impl RedisCache {
    fn new(latency_ms: u64) -> Self {
        RedisCache {
            store: Mutex::new(HashMap::new()),
            latency: Duration::from_millis(latency_ms),
        }
    }
}

impl Cache for RedisCache {
    async fn get(&self, key: &str) -> Option<String> {
        sleep(self.latency).await; // Имитируем сетевой round-trip
        self.store.lock().await.get(key).cloned()
    }

    async fn set(&self, key: &str, value: String) {
        sleep(self.latency).await;
        self.store.lock().await.insert(key.to_string(), value);
    }
}

// --- Обобщённая функция, работающая с любым Cache ---
async fn cache_demo<C: Cache>(cache: &C, label: &str) {
    cache.set("greeting", "Hello, async!".into()).await;
    let val = cache.get("greeting").await;
    println!("[{label}] greeting = {val:?}");
}

#[tokio::main]
async fn main() {
    let mem = MemoryCache::new();
    cache_demo(&mem, "memory").await;

    let redis = RedisCache::new(50);
    cache_demo(&redis, "redis").await;
}

Ключевые выводы — async-трейты

• С Rust 1.75 можно писать async fn прямо в трейтах (крейт #[async_trait] не нужен)
• trait_variant::make автоматически генерирует вариант с Send для порождения задач (только статическая диспетчеризация)
• Асинхронные замыкания (async Fn()) стабилизированы в 1.85 — используйте их для колбэков и middleware
• Для кода, критичного к производительности, предпочитайте статическую диспетчеризацию (<S: Service>) вместо dyn

См. также: Гл. 13 — Продакшен-паттерны — трейт Service из Tower, Гл. 6 — Создаём фьючи вручную — ручные реализации трейтов

11. Потоки и AsyncIterator 🟡

Что вы узнаете:

• Трейт Stream: асинхронная итерация по нескольким значениям
• Создание потоков: stream::iter, async_stream, unfold
• Комбинаторы потоков: map, filter, buffer_unordered, fold
• Асинхронные трейты ввода-вывода: AsyncRead, AsyncWrite, AsyncBufRead

Обзор трейта Stream

Stream относится к Iterator так же, как Future относится к одиночному значению, — он асинхронно выдаёт несколько значений:

#![allow(unused)]
fn main() {
// std::iter::Iterator (синхронный, несколько значений)
trait Iterator {
    type Item;
    fn next(&mut self) -> Option<Self::Item>;
}

// futures::Stream (асинхронный, несколько значений)
trait Stream {
    type Item;
    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>;
}
}

graph LR
    subgraph "Синхронно"
        VAL["Значение<br/>(T)"]
        ITER["Итератор<br/>(несколько T)"]
    end

    subgraph "Асинхронно"
        FUT["Future<br/>(async T)"]
        STREAM["Stream<br/>(async несколько T)"]
    end

    VAL -->|"сделать асинхронным"| FUT
    ITER -->|"сделать асинхронным"| STREAM
    VAL -->|"сделать множественным"| ITER
    FUT -->|"сделать множественным"| STREAM

    style VAL fill:#e3f2fd,color:#000
    style ITER fill:#e3f2fd,color:#000
    style FUT fill:#c8e6c9,color:#000
    style STREAM fill:#c8e6c9,color:#000

Создание потоков

#![allow(unused)]
fn main() {
use futures::stream::{self, StreamExt};
use tokio::time::{interval, Duration};
use tokio_stream::wrappers::IntervalStream;

// 1. Из итератора
let s = stream::iter(vec![1, 2, 3]);

// 2. Из асинхронного генератора (с помощью крейта async_stream)
// Cargo.toml: async-stream = "0.3"
use async_stream::stream;

fn countdown(from: u32) -> impl futures::Stream<Item = u32> {
    stream! {
        for i in (0..=from).rev() {
            tokio::time::sleep(Duration::from_millis(500)).await;
            yield i;
        }
    }
}

// 3. Из интервала tokio
let tick_stream = IntervalStream::new(interval(Duration::from_secs(1)));

// 4. Из приёмника канала (tokio_stream::wrappers)
let (tx, rx) = tokio::sync::mpsc::channel::<String>(100);
let rx_stream = tokio_stream::wrappers::ReceiverStream::new(rx);

// 5. Из unfold (генерация из асинхронного состояния)
let s = stream::unfold(0u32, |state| async move {
    if state >= 5 {
        None // Поток завершается
    } else {
        let next = state + 1;
        Some((state, next)) // выдаём `state`, новое состояние — `next`
    }
});
}

Потребление потоков

#![allow(unused)]
fn main() {
use futures::stream::{self, StreamExt};

async fn stream_examples() {
    let s = stream::iter(vec![1, 2, 3, 4, 5]);

    // for_each — обрабатываем каждый элемент
    s.for_each(|x| async move {
        println!("{x}");
    }).await;

    // map + collect
    let doubled: Vec<i32> = stream::iter(vec![1, 2, 3])
        .map(|x| x * 2)
        .collect()
        .await;

    // filter
    let evens: Vec<i32> = stream::iter(1..=10)
        .filter(|x| futures::future::ready(x % 2 == 0))
        .collect()
        .await;

    // buffer_unordered — обрабатываем N элементов одновременно
    let results: Vec<_> = stream::iter(vec!["url1", "url2", "url3"])
        .map(|url| async move {
            // Имитация HTTP-запроса
            tokio::time::sleep(Duration::from_millis(100)).await;
            format!("ответ от {url}")
        })
        .buffer_unordered(10) // До 10 одновременных запросов
        .collect()
        .await;

    // take, skip, zip, chain — точно как у Iterator
    let first_three: Vec<i32> = stream::iter(1..=100)
        .take(3)
        .collect()
        .await;
}
}

Сравнение с C# IAsyncEnumerable

#![allow(unused)]
fn main() {
// Rust: поток строк из базы данных
// ПРИМЕЧАНИЕ: при использовании ? внутри тела нужен try_stream! (а не stream!).
// stream! не пробрасывает ошибки — try_stream! выдаёт Err(e) и завершается.
fn get_users(db: &Database) -> impl Stream<Item = Result<User, DbError>> + '_ {
    try_stream! {
        let mut cursor = db.query("SELECT * FROM users").await?;
        while let Some(row) = cursor.next().await {
            yield User::from_row(row?);
        }
    }
}

// Потребление:
let mut users = pin!(get_users(&db));
while let Some(result) = users.next().await {
    match result {
        Ok(user) => println!("{}", user.name),
        Err(e) => eprintln!("Ошибка: {e}"),
    }
}
}

// Эквивалент на C#:
async IAsyncEnumerable<User> GetUsers() {
    await using var reader = await db.QueryAsync("SELECT * FROM users");
    while (await reader.ReadAsync()) {
        yield return User.FromRow(reader);
    }
}

// Потребление:
await foreach (var user in GetUsers()) {
    Console.WriteLine(user.Name);
}

#![allow(unused)]
fn main() {
use futures::stream::{self, StreamExt};

#[derive(Debug)]
struct Stats {
    count: usize,
    min: f64,
    max: f64,
    sum: f64,
}

impl Stats {
    fn average(&self) -> f64 {
        if self.count == 0 { 0.0 } else { self.sum / self.count as f64 }
    }
}

async fn compute_stats<S: futures::Stream<Item = f64>>(stream: S) -> Stats {
    stream
        .fold(
            Stats { count: 0, min: f64::INFINITY, max: f64::NEG_INFINITY, sum: 0.0 },
            |mut acc, value| async move {
                acc.count += 1;
                acc.min = acc.min.min(value);
                acc.max = acc.max.max(value);
                acc.sum += value;
                acc
            },
        )
        .await
}

#[tokio::test]
async fn test_stats() {
    let readings = stream::iter(vec![23.5, 24.1, 22.8, 25.0, 23.9]);
    let stats = compute_stats(readings).await;

    assert_eq!(stats.count, 5);
    assert!((stats.min - 22.8).abs() < f64::EPSILON);
    assert!((stats.max - 25.0).abs() < f64::EPSILON);
    assert!((stats.average() - 23.86).abs() < 0.01);
}
}

Асинхронные трейты ввода-вывода: AsyncRead, AsyncWrite, AsyncBufRead

Так же как std::io::Read/Write являются основой синхронного ввода-вывода, их асинхронные аналоги являются основой асинхронного ввода-вывода. Эти трейты предоставляются tokio::io (или futures::io для кода, не привязанного к рантайму):

#![allow(unused)]
fn main() {
// tokio::io — асинхронные версии трейтов std::io

/// Асинхронное чтение байтов из источника
pub trait AsyncRead {
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,  // Безопасная обёртка tokio над неинициализированной памятью
    ) -> Poll<io::Result<()>>;
}

/// Асинхронная запись байтов в приёмник
pub trait AsyncWrite {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>>;

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>>;
    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>>;
}

/// Буферизованное чтение с поддержкой строк
pub trait AsyncBufRead: AsyncRead {
    fn poll_fill_buf(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<&[u8]>>;
    fn consume(self: Pin<&mut Self>, amt: usize);
}
}

На практике вы редко вызываете эти методы poll_* напрямую. Вместо этого используйте трейты-расширения AsyncReadExt и AsyncWriteExt, которые предоставляют удобные вспомогательные методы для .await:

#![allow(unused)]
fn main() {
use tokio::io::{AsyncReadExt, AsyncWriteExt, AsyncBufReadExt};
use tokio::net::TcpStream;
use tokio::io::BufReader;

async fn io_examples() -> tokio::io::Result<()> {
    let mut stream = TcpStream::connect("127.0.0.1:8080").await?;

    // AsyncWriteExt: write_all, write_u32, write_buf и др.
    stream.write_all(b"GET / HTTP/1.0\r\n\r\n").await?;

    // AsyncReadExt: read, read_exact, read_to_end, read_to_string
    let mut response = Vec::new();
    stream.read_to_end(&mut response).await?;

    // AsyncBufReadExt: read_line, lines(), split()
    let file = tokio::fs::File::open("config.txt").await?;
    let reader = BufReader::new(file);
    let mut lines = reader.lines();
    while let Some(line) = lines.next_line().await? {
        println!("{line}");
    }

    Ok(())
}
}

Реализация собственного асинхронного ввода-вывода — оборачиваем протокол поверх сырого TCP:

#![allow(unused)]
fn main() {
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use std::pin::Pin;
use std::task::{Context, Poll};

/// Протокол с префиксом длины: [u32 длина][байты полезной нагрузки]
struct FramedStream<T> {
    inner: T,
}

impl<T: AsyncRead + AsyncReadExt + Unpin> FramedStream<T> {
    /// Прочитать один полный кадр
    async fn read_frame(&mut self) -> tokio::io::Result<Vec<u8>>
    {
        // Читаем 4-байтовый префикс длины
        let len = self.inner.read_u32().await? as usize;

        // Читаем ровно столько байт
        let mut payload = vec![0u8; len];
        self.inner.read_exact(&mut payload).await?;
        Ok(payload)
    }
}

impl<T: AsyncWrite + AsyncWriteExt + Unpin> FramedStream<T> {
    /// Записать один полный кадр
    async fn write_frame(&mut self, data: &[u8]) -> tokio::io::Result<()>
    {
        self.inner.write_u32(data.len() as u32).await?;
        self.inner.write_all(data).await?;
        self.inner.flush().await?;
        Ok(())
    }
}
}

Трейты ввода-вывода tokio и futures: они похожи, но не идентичны — AsyncRead из tokio использует ReadBuf (безопасно работает с неинициализированной памятью), а futures::AsyncRead использует &mut [u8]. Для конвертации между ними используйте tokio_util::compat.

Утилиты копирования: tokio::io::copy(&mut reader, &mut writer) — асинхронный аналог std::io::copy, полезен для прокси-серверов или передачи файлов. tokio::io::copy_bidirectional копирует данные в обоих направлениях одновременно.

#![allow(unused)]
fn main() {
use tokio::io::AsyncBufReadExt;

async fn count_non_empty_lines<R: tokio::io::AsyncBufRead + Unpin>(
    reader: R,
) -> tokio::io::Result<usize> {
    let mut lines = reader.lines();
    let mut count = 0;
    while let Some(line) = lines.next_line().await? {
        if !line.is_empty() {
            count += 1;
        }
    }
    Ok(count)
}

// Работает с любым AsyncBufRead:
// let file = tokio::io::BufReader::new(tokio::fs::File::open("data.txt").await?);
// let count = count_non_empty_lines(file).await?;
//
// let tcp = tokio::io::BufReader::new(TcpStream::connect("...").await?);
// let count = count_non_empty_lines(tcp).await?;
}

Ключевые выводы — потоки и AsyncIterator

• Stream — асинхронный аналог Iterator: выдаёт Poll::Ready(Some(item)) или Poll::Ready(None)
• .buffer_unordered(N) обрабатывает N элементов потока одновременно — главный инструмент конкурентности для потоков
• async_stream::stream! — самый простой способ создать собственный поток (использует yield)
• AsyncRead/AsyncBufRead позволяют писать обобщённый, переиспользуемый код ввода-вывода для файлов, сокетов и каналов

См. также: Гл. 9 — Когда Tokio не подходит — FuturesUnordered (родственный паттерн), Гл. 13 — Продакшен-паттерны — обратное давление с ограниченными каналами

12. Типичные ловушки 🔴

Что вы узнаете:

• 9 типичных ошибок асинхронного Rust и как исправить каждую
• Почему блокировка исполнителя — ошибка номер один (и как её исправляет spawn_blocking)
• Опасности отмены: что происходит, когда future уничтожается посреди await
• Отладка: tokio-console, tracing, #[instrument]
• Тестирование: #[tokio::test], time::pause(), моки на основе трейтов

Блокировка исполнителя

Ошибка номер один в асинхронном Rust: выполнение блокирующего кода в потоке исполнителя. Это морит голодом другие задачи.

#![allow(unused)]
fn main() {
// ❌ НЕПРАВИЛЬНО: блокирует весь поток исполнителя
async fn bad_handler() -> String {
    let data = std::fs::read_to_string("big_file.txt").unwrap(); // БЛОКИРУЕТ!
    process(&data)
}

// ✅ ПРАВИЛЬНО: переносим блокирующую работу в отдельный пул потоков
async fn good_handler() -> String {
    let data = tokio::task::spawn_blocking(|| {
        std::fs::read_to_string("big_file.txt").unwrap()
    }).await.unwrap();
    process(&data)
}

// ✅ ТОЖЕ ПРАВИЛЬНО: используем асинхронную fs из tokio
async fn also_good_handler() -> String {
    let data = tokio::fs::read_to_string("big_file.txt").await.unwrap();
    process(&data)
}
}

graph TB
    subgraph "❌ Блокирующий вызов в исполнителе"
        T1_BAD["Поток 1: std::fs::read()<br/>🔴 ЗАБЛОКИРОВАН на 500 мс"]
        T2_BAD["Поток 2: обрабатывает запросы<br/>🟢 Работает в одиночку"]
        TASKS_BAD["100 ожидающих задач<br/>⏳ Голодают"]
        T1_BAD -->|"не может опрашивать"| TASKS_BAD
    end

    subgraph "✅ spawn_blocking"
        T1_GOOD["Поток 1: опрашивает future<br/>🟢 Свободен"]
        T2_GOOD["Поток 2: опрашивает future<br/>🟢 Свободен"]
        BT["Поток пула для блокирующих задач:<br/>std::fs::read()<br/>🔵 Отдельный пул"]
        TASKS_GOOD["100 задач<br/>✅ Все продвигаются"]
        T1_GOOD -->|"опрашивает"| TASKS_GOOD
        T2_GOOD -->|"опрашивает"| TASKS_GOOD
    end

std::thread::sleep и tokio::time::sleep

#![allow(unused)]
fn main() {
// ❌ НЕПРАВИЛЬНО: блокирует поток исполнителя на 5 секунд
async fn bad_delay() {
    std::thread::sleep(Duration::from_secs(5)); // Поток не может опрашивать ничего другого!
}

// ✅ ПРАВИЛЬНО: уступаем управление исполнителю, другие задачи могут работать
async fn good_delay() {
    tokio::time::sleep(Duration::from_secs(5)).await; // Не блокирует!
}
}

Удержание MutexGuard через .await

#![allow(unused)]
fn main() {
use std::sync::Mutex; // std Mutex — НЕ учитывает async

// ⚠️ РИСКОВАННО: MutexGuard удерживается через .await
async fn bad_mutex(data: &Mutex<Vec<String>>) {
    let mut guard = data.lock().unwrap();
    guard.push("item".into());
    some_io().await; // Guard удерживается здесь — другие потоки не могут взять блокировку!
    guard.push("another".into());
}
// ПРИМЕЧАНИЕ: это компилируется! std::sync::MutexGuard — !Send, но компилятор требует Send
// для Future только тогда, когда вы передаёте его туда, где это требуется
// (например, в tokio::spawn). Прямой вызов bad_mutex(...).await компилируется без проблем.
// Однако tokio::spawn(bad_mutex(data)) упадёт с ошибкой ограничения Send.
}

Почему это обычно проблема — но не всегда:

Удержание std::sync::Mutex через .await блокирует поток ОС на всё время I/O, не давая исполнителю опрашивать другие задачи в этом потоке. Для коротких критических секций это расточительно; для длительного I/O — это ловушка производительности.

Однако бывают законные случаи, когда блокировку необходимо удерживать через .await — так же как транзакция базы данных удерживает блокировку между чтением и фиксацией. Если отпустить и заново захватить блокировку, возникает гонка TOCTOU (time-of-check to time-of-use): другая задача может изменить данные между вашими двумя критическими секциями. Правильное решение зависит от сценария:

#![allow(unused)]
fn main() {
// ВАРИАНТ 1: ограничиваем область видимости guard — подходит, когда операции независимы
async fn scoped_mutex(data: &Mutex<Vec<String>>) {
    {
        let mut guard = data.lock().unwrap();
        guard.push("item".into());
    } // Guard уничтожается здесь
    some_io().await; // Блокировка снята — другие задачи могут продолжать
    {
        let mut guard = data.lock().unwrap();
        guard.push("another".into());
    }
}
// ⚠️ Осторожно: другая задача может захватить блокировку и изменить Vec между двумя секциями.
//    Это нормально, если два push независимы, но неверно, если "another"
//    зависит от состояния, установленного "item".

// ВАРИАНТ 2: используем tokio::sync::Mutex — удерживает блокировку через .await,
//            не блокируя поток ОС. Лучше всего, когда нужна транзакционная
//            операция «прочитать-изменить-записать» через точку await.
use tokio::sync::Mutex as AsyncMutex;

async fn async_mutex(data: &AsyncMutex<Vec<String>>) {
    let mut guard = data.lock().await; // Асинхронная блокировка — не блокирует поток
    guard.push("item".into());
    some_io().await; // OK — guard tokio Mutex является Send
    guard.push("another".into());
    // Guard удерживался всё время — ни гонки TOCTOU, ни заблокированного потока.
}
}

Когда какой Mutex использовать:

• std::sync::Mutex: короткие критические секции без .await внутри
• tokio::sync::Mutex: когда нужно удерживать блокировку через точки .await (транзакционная семантика, предотвращение TOCTOU)
• parking_lot::Mutex: прямая замена std, быстрее и компактнее, но тоже без .await

Эмпирическое правило: не режьте критическую секцию вокруг .await не задумываясь. Спросите себя, действительно ли обе половины независимы. Если нет — если вторая половина зависит от состояния из первой, — используйте tokio::sync::Mutex или перепроектируйте поток данных.

Опасности отмены

Уничтожение future отменяет его — но это может оставить систему в несогласованном состоянии:

#![allow(unused)]
fn main() {
// ❌ ОПАСНО: утечка ресурсов при отмене
async fn transfer(from: &Account, to: &Account, amount: u64) {
    from.debit(amount).await;  // Если отменить ЗДЕСЬ...
    to.credit(amount).await;   // ...деньги пропадут!
}

// ✅ БЕЗОПАСНО: делаем операции атомарными или используем компенсацию
async fn safe_transfer(from: &Account, to: &Account, amount: u64) -> Result<(), Error> {
    // Используем транзакцию базы данных (всё или ничего)
    let tx = db.begin_transaction().await?;
    tx.debit(from, amount).await?;
    tx.credit(to, amount).await?;
    tx.commit().await?; // Фиксируется, только если всё прошло успешно
    Ok(())
}

// ✅ ТОЖЕ БЕЗОПАСНО: tokio::select! с учётом отмены
tokio::select! {
    result = transfer(from, to, amount) => {
        // Перевод завершён
    }
    _ = shutdown_signal() => {
        // Не прерываем перевод на полпути — даём ему завершиться
        // Или: явно откатываем
    }
}
}

Нет асинхронного Drop

Трейт Drop в Rust синхронный — вы не можете вызвать .await внутри drop(). Это частый источник путаницы:

#![allow(unused)]
fn main() {
struct DbConnection { /* ... */ }

impl Drop for DbConnection {
    fn drop(&mut self) {
        // ❌ Так нельзя — drop() синхронный!
        // self.connection.shutdown().await;

        // ✅ Обходной путь 1: порождаем задачу очистки (fire-and-forget)
        let conn = self.connection.take();
        tokio::spawn(async move {
            let _ = conn.shutdown().await;
        });

        // ✅ Обходной путь 2: синхронное закрытие
        // self.connection.blocking_close();
    }
}
}

Лучшая практика: предоставьте явный метод async fn close(self) и задокументируйте, что вызывающий код должен использовать его. Полагайтесь на Drop только как на страховочную сетку, а не как на основной путь очистки.

Справедливость и голодание в select!

#![allow(unused)]
fn main() {
use tokio::sync::mpsc;

// ❌ НЕСПРАВЕДЛИВО: busy_stream всегда побеждает, slow_stream голодает
async fn unfair(mut fast: mpsc::Receiver<i32>, mut slow: mpsc::Receiver<i32>) {
    loop {
        tokio::select! {
            Some(v) = fast.recv() => println!("fast: {v}"),
            Some(v) = slow.recv() => println!("slow: {v}"),
            // Если готовы оба, tokio выбирает случайно.
            // Но если `fast` ВСЕГДА готов, `slow` почти никогда не опрашивается.
        }
    }
}

// ✅ СПРАВЕДЛИВО: используем biased select или вычитываем пачками
async fn fair(mut fast: mpsc::Receiver<i32>, mut slow: mpsc::Receiver<i32>) {
    loop {
        tokio::select! {
            biased; // Всегда проверяем по порядку — явный приоритет

            Some(v) = slow.recv() => println!("slow: {v}"),  // Приоритет!
            Some(v) = fast.recv() => println!("fast: {v}"),
        }
    }
}
}

Случайное последовательное выполнение

#![allow(unused)]
fn main() {
// ❌ ПОСЛЕДОВАТЕЛЬНО: занимает в сумме 2 секунды
async fn slow() {
    let a = fetch("url_a").await; // 1 секунда
    let b = fetch("url_b").await; // 1 секунда (ждёт, пока завершится a!)
}

// ✅ КОНКУРЕНТНО: занимает в сумме 1 секунду
async fn fast() {
    let (a, b) = tokio::join!(
        fetch("url_a"), // Оба стартуют сразу
        fetch("url_b"),
    );
}

// ✅ ТОЖЕ КОНКУРЕНТНО: через let + join
async fn also_fast() {
    let fut_a = fetch("url_a"); // Создаём future (ленивый — ещё не запущен)
    let fut_b = fetch("url_b"); // Создаём future
    let (a, b) = tokio::join!(fut_a, fut_b); // ТЕПЕРЬ оба выполняются конкурентно
}
}

Ловушка: let a = fetch(url).await; let b = fetch(url).await; выполняется последовательно! Второй .await не начнётся, пока не завершится первый. Для конкурентности используйте join! или spawn.

Разбор кейса: отладка зависшего продакшен-сервиса

Реальный сценарий: сервис 10 минут обрабатывает запросы нормально, а потом перестаёт отвечать. В логах нет ошибок. Загрузка CPU — 0%.

Шаги диагностики:

1. Подключаем tokio-console — показывает 200+ задач, застрявших в состоянии Pending
2. Смотрим детали задач — все ждут одного и того же Mutex::lock().await
3. Первопричина — одна задача удерживала std::sync::MutexGuard через .await и упала с паникой, «отравив» мьютекс. Теперь все остальные задачи падают на lock().unwrap()

Исправление:

Чек-лист профилактики:

• Используйте tokio::sync::Mutex, если guard пересекает любой .await
• Добавляйте #[tracing::instrument] к асинхронным функциям для отслеживания span
• Запускайте tokio-console в staging, чтобы рано замечать зависшие задачи
• Добавляйте эндпоинты проверки работоспособности, которые проверяют отзывчивость задач

#![allow(unused)]
fn main() {
use std::sync::Mutex;

async fn process_requests(urls: Vec<String>) -> Vec<String> {
    let results = Mutex::new(Vec::new());
    
    for url in &urls {
        let response = reqwest::get(url).await.unwrap().text().await.unwrap();
        std::thread::sleep(std::time::Duration::from_millis(100)); // Ограничение частоты
        let mut guard = results.lock().unwrap();
        guard.push(response);
        expensive_parse(&guard).await; // Разбираем все результаты на данный момент
    }
    
    results.into_inner().unwrap()
}
}

#![allow(unused)]
fn main() {
use tokio::sync::Mutex;
use std::sync::Arc;
use futures::stream::{self, StreamExt};

async fn process_requests(urls: Vec<String>) -> Vec<String> {
    // Исправление 4: обрабатываем URL конкурентно с помощью buffer_unordered
    let results: Vec<String> = stream::iter(urls)
        .map(|url| async move {
            let response = reqwest::get(&url).await.unwrap().text().await.unwrap();
            // Исправление 2: используем tokio::time::sleep вместо std::thread::sleep
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            response
        })
        .buffer_unordered(10) // До 10 одновременных запросов
        .collect()
        .await;

    // Исправление 3: разбираем после сбора — мьютекс вообще не нужен!
    for result in &results {
        expensive_parse(result).await;
    }

    results
}
}

Отладка асинхронного кода

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

tokio-console: инспектор задач в реальном времени

tokio-console даёт вид, похожий на htop, для каждой порождённой задачи: её состояние, длительность опроса, активность waker и использование ресурсов.

# Cargo.toml
[dependencies]
console-subscriber = "0.4"
tokio = { version = "1", features = ["full", "tracing"] }

#[tokio::main]
async fn main() {
    console_subscriber::init(); // Заменяет стандартный подписчик tracing
    // ... остальная часть вашего приложения
}

Затем в другом терминале:

$ RUSTFLAGS="--cfg tokio_unstable" cargo run   # Обязательный флаг на этапе компиляции
$ tokio-console                                # Подключается к 127.0.0.1:6669

tracing + #[instrument]: структурированное логирование для async

Крейт tracing понимает времена жизни Future. Span остаются открытыми через точки .await, давая логический стек вызовов, даже когда поток ОС уже переключился на другую работу:

#![allow(unused)]
fn main() {
use tracing::{info, instrument};

#[instrument(skip(db_pool), fields(user_id = %user_id))]
async fn handle_request(user_id: u64, db_pool: &Pool) -> Result<Response> {
    info!("ищем пользователя");
    let user = db_pool.get_user(user_id).await?;  // span остаётся открытым через .await
    info!(email = %user.email, "пользователь найден");
    let orders = fetch_orders(user_id).await?;     // всё тот же span
    Ok(build_response(user, orders))
}
}

Вывод (с tracing_subscriber::fmt::json()):

{"timestamp":"...","level":"INFO","span":{"name":"handle_request","user_id":"42"},"message":"ищем пользователя"}
{"timestamp":"...","level":"INFO","span":{"name":"handle_request","user_id":"42"},"fields":{"email":"a@b.com"},"message":"пользователь найден"}

Чек-лист отладки

Совет: включите RUSTFLAGS="--cfg tokio_unstable", чтобы получить метрики на уровне задач в tokio-console. Это флаг времени компиляции, а не времени выполнения.

Тестирование асинхронного кода

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

Базовые асинхронные тесты с #[tokio::test]:

#![allow(unused)]
fn main() {
// Cargo.toml
// [dev-dependencies]
// tokio = { version = "1", features = ["full", "test-util"] }

#[tokio::test]
async fn test_basic_async() {
    let result = fetch_data().await;
    assert_eq!(result, "expected");
}

// Однопоточный тест (полезно для типов !Send):
#[tokio::test(flavor = "current_thread")]
async fn test_single_threaded() {
    let rc = std::rc::Rc::new(42);
    let val = async { *rc }.await;
    assert_eq!(val, 42);
}

// Многопоточный тест с явным количеством воркеров:
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn test_concurrent_behavior() {
    // Проверяет гонки при реальной конкурентности
    let counter = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0));
    let c1 = counter.clone();
    let c2 = counter.clone();
    let (a, b) = tokio::join!(
        tokio::spawn(async move { c1.fetch_add(1, std::sync::atomic::Ordering::SeqCst) }),
        tokio::spawn(async move { c2.fetch_add(1, std::sync::atomic::Ordering::SeqCst) }),
    );
    a.unwrap();
    b.unwrap();
    assert_eq!(counter.load(std::sync::atomic::Ordering::SeqCst), 2);
}
}

Управление временем — тестируем таймауты, не дожидаясь их:

#![allow(unused)]
fn main() {
use tokio::time::{self, Duration, Instant};

#[tokio::test]
async fn test_timeout_behavior() {
    // Останавливаем время — sleep() переходит вперёд мгновенно, без реальной задержки
    time::pause();

    let start = Instant::now();
    time::sleep(Duration::from_secs(3600)).await; // «ждём» 1 час — занимает 0 мс
    assert!(start.elapsed() >= Duration::from_secs(3600));
    // Тест выполнился за миллисекунды, а не за час!
}

#[tokio::test]
async fn test_retry_timing() {
    time::pause();

    // Проверяем, что логика повторов ждёт ожидаемое время
    let start = Instant::now();
    let result = retry_with_backoff(|| async {
        Err::<(), _>("simulated failure")
    }, 3, Duration::from_secs(1))
    .await;

    assert!(result.is_err());
    // 1с + 2с + 4с = 7с ожидания (экспоненциально)
    assert!(start.elapsed() >= Duration::from_secs(7));
}

#[tokio::test]
async fn test_deadline_exceeded() {
    time::pause();

    let result = tokio::time::timeout(
        Duration::from_secs(5),
        async {
            // Имитируем медленную операцию
            time::sleep(Duration::from_secs(10)).await;
            "done"
        }
    ).await;

    assert!(result.is_err()); // Таймаут
}
}

Мокирование асинхронных зависимостей — используйте трейт-объекты или дженерики:

#![allow(unused)]
fn main() {
// Определяем трейт для зависимости:
trait Storage {
    async fn get(&self, key: &str) -> Option<String>;
    async fn set(&self, key: &str, value: String);
}

// Продакшен-реализация:
struct RedisStorage { /* ... */ }
impl Storage for RedisStorage {
    async fn get(&self, key: &str) -> Option<String> {
        // Реальный вызов Redis
        todo!()
    }
    async fn set(&self, key: &str, value: String) {
        todo!()
    }
}

// Тестовый мок:
struct MockStorage {
    data: std::sync::Mutex<std::collections::HashMap<String, String>>,
}

impl MockStorage {
    fn new() -> Self {
        MockStorage { data: std::sync::Mutex::new(std::collections::HashMap::new()) }
    }
}

impl Storage for MockStorage {
    async fn get(&self, key: &str) -> Option<String> {
        self.data.lock().unwrap().get(key).cloned()
    }
    async fn set(&self, key: &str, value: String) {
        self.data.lock().unwrap().insert(key.to_string(), value);
    }
}

// Тестируемая функция обобщена по Storage:
async fn cache_lookup<S: Storage>(store: &S, key: &str) -> String {
    match store.get(key).await {
        Some(val) => val,
        None => {
            let val = "computed".to_string();
            store.set(key, val.clone()).await;
            val
        }
    }
}

#[tokio::test]
async fn test_cache_miss_then_hit() {
    let mock = MockStorage::new();

    // Первый вызов: промах → вычисляем и сохраняем
    let val = cache_lookup(&mock, "key1").await;
    assert_eq!(val, "computed");

    // Второй вызов: попадание → возвращаем сохранённое значение
    let val = cache_lookup(&mock, "key1").await;
    assert_eq!(val, "computed");
    assert!(mock.data.lock().unwrap().contains_key("key1"));
}
}

Тестирование каналов и обмена между задачами:

#![allow(unused)]
fn main() {
#[tokio::test]
async fn test_producer_consumer() {
    let (tx, mut rx) = tokio::sync::mpsc::channel(10);

    tokio::spawn(async move {
        for i in 0..5 {
            tx.send(i).await.unwrap();
        }
        // tx уничтожается здесь — канал закрывается
    });

    let mut received = Vec::new();
    while let Some(val) = rx.recv().await {
        received.push(val);
    }

    assert_eq!(received, vec![0, 1, 2, 3, 4]);
}
}

Ключевые выводы — типичные ловушки

• Никогда не блокируйте исполнитель — используйте spawn_blocking для работы с CPU или синхронного кода
• Никогда не удерживайте MutexGuard через .await — ограничивайте блокировки или используйте tokio::sync::Mutex
• Отмена мгновенно уничтожает future — для частично выполненных операций используйте паттерны, устойчивые к отмене (cancel-safe)
• Используйте tokio-console и #[tracing::instrument] для отладки асинхронного кода
• Тестируйте асинхронный код через #[tokio::test] и time::pause() для детерминированного тайминга

См. также: Гл. 8 — Глубокое погружение в Tokio — примитивы синхронизации, Гл. 13 — Продакшен-паттерны — graceful shutdown и структурированную конкурентность

13. Продакшен-паттерны 🔴

Что вы узнаете:

• Graceful shutdown с помощью каналов watch и select!
• Обратное давление: ограниченные каналы защищают от OOM
• Структурированная конкурентность: JoinSet и TaskTracker
• Таймауты, повторы и экспоненциальная задержка (backoff)
• Обработка ошибок: thiserror против anyhow, паттерн двойного ?
• Tower: паттерн middleware, который используют axum, tonic и hyper

Корректное завершение работы (Graceful Shutdown)

Продакшен-серверы должны корректно завершаться: доводить до конца текущие запросы, сбрасывать буферы, закрывать соединения:

use tokio::signal;
use tokio::sync::watch;

async fn main_server() {
    // Создаём канал для сигнала остановки
    let (shutdown_tx, shutdown_rx) = watch::channel(false);

    // Запускаем сервер
    let server_handle = tokio::spawn(run_server(shutdown_rx.clone()));

    // Ждём Ctrl+C
    signal::ctrl_c().await.expect("Не удалось подписаться на Ctrl+C");
    println!("Получен сигнал остановки, доводим текущие запросы...");

    // Уведомляем все задачи о завершении
    // ПРИМЕЧАНИЕ: .unwrap() используется для краткости. В продакшен-коде нужно обрабатывать
    // случай, когда все получатели уже уничтожены.
    shutdown_tx.send(true).unwrap();

    // Ждём завершения сервера (с таймаутом)
    match tokio::time::timeout(
        std::time::Duration::from_secs(30),
        server_handle,
    ).await {
        Ok(Ok(())) => println!("Сервер корректно остановлен"),
        Ok(Err(e)) => eprintln!("Ошибка сервера: {e}"),
        Err(_) => eprintln!("Таймаут остановки сервера — принудительный выход"),
    }
}

async fn run_server(mut shutdown: watch::Receiver<bool>) {
    loop {
        tokio::select! {
            // Принимаем новые соединения
            conn = accept_connection() => {
                let shutdown = shutdown.clone();
                tokio::spawn(handle_connection(conn, shutdown));
            }
            // Сигнал остановки
            _ = shutdown.changed() => {
                if *shutdown.borrow() {
                    println!("Прекращаем приём новых соединений");
                    break;
                }
            }
        }
    }
    // Активные соединения завершатся сами,
    // потому что у них есть собственный клон shutdown_rx
}

async fn handle_connection(conn: Connection, mut shutdown: watch::Receiver<bool>) {
    loop {
        tokio::select! {
            request = conn.next_request() => {
                // Обрабатываем запрос полностью — не бросаем его на полпути
                process_request(request).await;
            }
            _ = shutdown.changed() => {
                if *shutdown.borrow() {
                    // Завершаем текущий запрос и выходим
                    break;
                }
            }
        }
    }
}

sequenceDiagram
    participant OS as Сигнал ОС
    participant Main as Главная задача
    participant WCH as Канал watch
    participant W1 as Воркер 1
    participant W2 as Воркер 2

    OS->>Main: SIGINT (Ctrl+C)
    Main->>WCH: send(true)
    WCH-->>W1: changed()
    WCH-->>W2: changed()

    Note over W1: Доводим текущий запрос
    Note over W2: Доводим текущий запрос

    W1-->>Main: Задача завершена
    W2-->>Main: Задача завершена
    Main->>Main: Все воркеры закончили → выход

Обратное давление через ограниченные каналы

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

#![allow(unused)]
fn main() {
use tokio::sync::mpsc;

async fn backpressure_example() {
    // Ограниченный канал: не более 100 элементов в буфере
    let (tx, mut rx) = mpsc::channel::<WorkItem>(100);

    // Производитель: естественно замедляется, когда буфер заполнен
    let producer = tokio::spawn(async move {
        for i in 0..1_000_000 {
            // send() асинхронный — ждёт, если буфер полон
            // Это создаёт естественное обратное давление!
            tx.send(WorkItem { id: i }).await.unwrap();
        }
    });

    // Потребитель: обрабатывает элементы в своём темпе
    let consumer = tokio::spawn(async move {
        while let Some(item) = rx.recv().await {
            process(item).await; // Медленная обработка — нормально, производитель будет ждать
        }
    });

    let _ = tokio::join!(producer, consumer);
}

// Сравните с неограниченным — ОПАСНО:
// let (tx, rx) = mpsc::unbounded_channel(); // Никакого обратного давления!
// Производитель может бесконечно заполнять память
}

Структурированная конкурентность: JoinSet и TaskTracker

JoinSet объединяет связанные задачи и гарантирует, что все они завершатся:

#![allow(unused)]
fn main() {
use tokio::task::JoinSet;
use tokio::time::{sleep, Duration};

async fn structured_concurrency() {
    let mut set = JoinSet::new();

    // Порождаем пачку задач
    for url in get_urls() {
        set.spawn(async move {
            fetch_and_process(url).await
        });
    }

    // Собираем все результаты (порядок не гарантирован)
    let mut results = Vec::new();
    while let Some(result) = set.join_next().await {
        match result {
            Ok(Ok(data)) => results.push(data),
            Ok(Err(e)) => eprintln!("Ошибка задачи: {e}"),
            Err(e) => eprintln!("Задача завершилась паникой: {e}"),
        }
    }

    // ВСЕ задачи завершены — фоновой работы не осталось
    println!("Обработано элементов: {}", results.len());
}

// TaskTracker (tokio-util 0.7.9+) — ждём все порождённые задачи
use tokio_util::task::TaskTracker;

async fn with_tracker() {
    let tracker = TaskTracker::new();

    for i in 0..10 {
        tracker.spawn(async move {
            sleep(Duration::from_millis(100 * i)).await;
            println!("Задача {i} завершена");
        });
    }

    tracker.close(); // Больше задач добавлять не будут
    tracker.wait().await; // Ждём ВСЕ отслеживаемые задачи
    println!("Все задачи завершены");
}
}

Таймауты и повторы

#![allow(unused)]
fn main() {
use tokio::time::{timeout, sleep, Duration};

// Простой таймаут
async fn with_timeout() -> Result<Response, Error> {
    match timeout(Duration::from_secs(5), fetch_data()).await {
        Ok(Ok(response)) => Ok(response),
        Ok(Err(e)) => Err(Error::Fetch(e)),
        Err(_) => Err(Error::Timeout),
    }
}

// Повтор с экспоненциальной задержкой
async fn retry_with_backoff<F, Fut, T, E>(
    max_attempts: u32,
    base_delay_ms: u64,
    operation: F,
) -> Result<T, E>
where
    F: Fn() -> Fut,
    Fut: std::future::Future<Output = Result<T, E>>,
    E: std::fmt::Display,
{
    let mut delay = Duration::from_millis(base_delay_ms);

    for attempt in 1..=max_attempts {
        match operation().await {
            Ok(result) => return Ok(result),
            Err(e) => {
                if attempt == max_attempts {
                    eprintln!("Последняя попытка {attempt} не удалась: {e}");
                    return Err(e);
                }
                eprintln!("Попытка {attempt} не удалась: {e}, повтор через {delay:?}");
                sleep(delay).await;
                delay *= 2; // Экспоненциальная задержка
            }
        }
    }
    unreachable!()
}

// Использование:
// let result = retry_with_backoff(3, 100, || async {
//     reqwest::get("https://api.example.com/data").await
// }).await?;
}

Совет для продакшена — добавляйте jitter: функция выше использует чистую экспоненциальную задержку, но в продакшене много клиентов, падающих одновременно, будут повторять запросы в одни и те же моменты (эффект «стада», thundering herd). Добавьте случайный jitter — например, sleep(delay + rand_jitter), где rand_jitter — это значение из 0..delay/4, — чтобы повторы распределялись во времени.

Обработка ошибок в асинхронном коде

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

thiserror против anyhow — выбираем подходящий инструмент:

#![allow(unused)]
fn main() {
// thiserror: определяем типизированные ошибки для библиотек и публичных API
// Каждый вариант явный — вызывающий код может сопоставлять конкретные ошибки
use thiserror::Error;

#[derive(Error, Debug)]
enum DiagError {
    #[error("Команда IPMI не выполнена: {0}")]
    Ipmi(#[from] IpmiError),

    #[error("Датчик {sensor} вне диапазона: {value}°C (максимум {max}°C)")]
    OverTemp { sensor: String, value: f64, max: f64 },

    #[error("Операция завершилась по таймауту через {0:?}")]
    Timeout(std::time::Duration),

    #[error("Задача завершилась паникой: {0}")]
    TaskPanic(#[from] tokio::task::JoinError),
}

// anyhow: быстрая обработка ошибок для приложений и прототипов
// Оборачивает любую ошибку — не нужно определять типы для каждого случая
use anyhow::{Context, Result};

async fn run_diagnostics() -> Result<()> {
    let config = load_config()
        .await
        .context("Не удалось загрузить конфигурацию диагностики")?;  // Добавляет контекст

    let result = run_gpu_test(&config)
        .await
        .context("Диагностика GPU не выполнена")?;              // Строит цепочку контекста

    Ok(())
}
// anyhow выведет: "Диагностика GPU не выполнена: Команда IPMI не выполнена: timeout"
}

Паттерн двойного ? с tokio::spawn:

#![allow(unused)]
fn main() {
use thiserror::Error;
use tokio::task::JoinError;

#[derive(Error, Debug)]
enum AppError {
    #[error("Ошибка HTTP: {0}")]
    Http(#[from] reqwest::Error),

    #[error("Задача завершилась паникой: {0}")]
    TaskPanic(#[from] JoinError),
}

async fn spawn_with_errors() -> Result<String, AppError> {
    let handle = tokio::spawn(async {
        let resp = reqwest::get("https://example.com").await?;
        Ok::<_, reqwest::Error>(resp.text().await?)
    });

    // Двойной ?: первый ? разворачивает JoinError (паника задачи), второй ? — внутренний Result
    let result = handle.await??;
    Ok(result)
}
}

Проблема границы ошибок — tokio::spawn стирает контекст:

#![allow(unused)]
fn main() {
// ❌ Контекст ошибки теряется на границе spawn:
async fn bad_error_handling() -> Result<()> {
    let handle = tokio::spawn(async {
        some_fallible_work().await  // Возвращает Result<T, SomeError>
    });

    // handle.await возвращает Result<Result<T, SomeError>, JoinError>
    // У внутренней ошибки нет контекста о том, какая задача упала
    let result = handle.await??;
    Ok(())
}

// ✅ Добавляем контекст на границе spawn:
async fn good_error_handling() -> Result<()> {
    let handle = tokio::spawn(async {
        some_fallible_work()
            .await
            .context("ошибка рабочей задачи")  // Контекст до пересечения границы
    });

    let result = handle.await
        .context("рабочая задача завершилась паникой")??;  // Контекст и для JoinError
    Ok(())
}
}

Ошибки таймаута — оборачивание или замена:

#![allow(unused)]
fn main() {
use tokio::time::{timeout, Duration};

async fn with_timeout_context() -> Result<String, DiagError> {
    let dur = Duration::from_secs(30);
    match timeout(dur, fetch_sensor_data()).await {
        Ok(Ok(data)) => Ok(data),
        Ok(Err(e)) => Err(e),                      // Внутренняя ошибка сохраняется
        Err(_) => Err(DiagError::Timeout(dur)),     // Таймаут → типизированная ошибка
    }
}
}

Tower: паттерн middleware

Крейт Tower определяет компонуемый трейт Service — основу асинхронных middleware в Rust (его используют axum, tonic, hyper):

#![allow(unused)]
fn main() {
// Основной трейт Tower (упрощённо):
pub trait Service<Request> {
    type Response;
    type Error;
    type Future: Future<Output = Result<Self::Response, Self::Error>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
    fn call(&mut self, req: Request) -> Self::Future;
}
}

Middleware оборачивает Service, добавляя сквозное поведение — логирование, таймауты, ограничение частоты — без изменения внутренней логики:

#![allow(unused)]
fn main() {
use tower::{ServiceBuilder, timeout::TimeoutLayer, limit::RateLimitLayer};
use std::time::Duration;

let service = ServiceBuilder::new()
    .layer(TimeoutLayer::new(Duration::from_secs(10)))       // Самый внешний: таймаут
    .layer(RateLimitLayer::new(100, Duration::from_secs(1))) // Затем: ограничение частоты
    .service(my_handler);                                     // Самый внутренний: ваш код
}

Почему это важно: если вы использовали middleware в ASP.NET или Express.js, Tower — это его аналог в Rust. Так продакшен-сервисы на Rust добавляют сквозные задачи без дублирования кода.

Упражнение: graceful shutdown с пулом воркеров

use tokio::sync::{mpsc, watch};
use tokio::time::{sleep, Duration};

struct WorkItem { id: u64, payload: String }

#[tokio::main]
async fn main() {
    let (work_tx, work_rx) = mpsc::channel::<WorkItem>(100);
    let (shutdown_tx, shutdown_rx) = watch::channel(false);
    let work_rx = std::sync::Arc::new(tokio::sync::Mutex::new(work_rx));

    let mut handles = Vec::new();
    for id in 0..4 {
        let rx = work_rx.clone();
        let mut shutdown = shutdown_rx.clone();
        handles.push(tokio::spawn(async move {
            loop {
                let item = {
                    let mut rx = rx.lock().await;
                    tokio::select! {
                        item = rx.recv() => item,
                        _ = shutdown.changed() => {
                            if *shutdown.borrow() { None } else { continue }
                        }
                    }
                };
                match item {
                    Some(work) => {
                        println!("Воркер {id}: обрабатываю {}", work.id);
                        sleep(Duration::from_millis(200)).await;
                    }
                    None => break,
                }
            }
        }));
    }

    // Отправляем работу
    for i in 0..20 {
        let _ = work_tx.send(WorkItem { id: i, payload: format!("task-{i}") }).await;
        sleep(Duration::from_millis(50)).await;
    }

    // По Ctrl+C: сигнализируем об остановке и ждём воркеров
    // ПРИМЕЧАНИЕ: .unwrap() используется для краткости — в продакшене обрабатывайте ошибки.
    tokio::signal::ctrl_c().await.unwrap();
    shutdown_tx.send(true).unwrap();
    for h in handles { let _ = h.await; }
    println!("Корректно завершено.");
}

Ключевые выводы — продакшен-паттерны

• Используйте канал watch + select! для согласованного graceful shutdown
• Ограниченные каналы (mpsc::channel(N)) обеспечивают обратное давление — отправители блокируются, когда буфер полон
• JoinSet и TaskTracker обеспечивают структурированную конкурентность: отслеживание, прерывание и ожидание групп задач
• Для сетевых операций всегда задавайте таймауты — tokio::time::timeout(dur, fut)
• Трейт Service из Tower — стандартный паттерн middleware для продакшен-сервисов на Rust

См. также: Гл. 8 — Глубокое погружение в Tokio — каналы и примитивы синхронизации, Гл. 12 — Типичные ловушки — опасности отмены при остановке

14. Async — это оптимизация, а не архитектура 🔴

Что вы узнаете:

• Почему async обычно «заражает» всю кодовую базу — и почему это недостаток дизайна, а не возможность
• Паттерн «синхронное ядро, асинхронная оболочка», который делает большую часть кода тестируемой и понятной
• Как справляться со сложным случаем: логикой, которой тоже нужен ввод-вывод
• Когда spawn_blocking — это решение, а когда — симптом
• Когда async действительно уместен в основной логике
• Почему библиотеки, построенные сначала на синхронном коде, компонуемее библиотек, построенных сначала на async

Вы уже прошли 13 глав, изучая асинхронный Rust. Вот самое важное, о чём эта книга до сих пор не говорила: большая часть вашего кода не должна быть асинхронной.

Проблема раскраски функций

В статье Боба Ныстрома «What Color is Your Function?» описана основная проблема: асинхронные функции могут вызывать синхронные, а синхронные функции не могут вызывать асинхронные. Как только одна функция становится асинхронной, всё, что выше по цепочке вызовов, тоже должно стать асинхронным.

В Rust это хуже, чем в C# или JavaScript, потому что async заражает не только сигнатуры функций, но и типы:

Каждая строка — это решение, которое кто-то должен принять, сделать правильно и поддерживать, и ни одно из них не относится к бизнес-логике. Индустрия движется от этого: виртуальные потоки Project Loom в Java и горутины в Go позволяют писать код, который выглядит синхронным, а рантайм дёшево мультиплексирует его. Rust выбрал явный async ради контроля с нулевой стоимостью, но у этого контроля есть цена в сложности, которую стоит платить осознанно, а не по умолчанию.

«Но потоки дорогие»

Рефлекторный аргумент: «нам нужен async, потому что потоки дорогие». В масштабах, в которых работает большинство команд, это в основном неверно.

• Память стека: каждый поток ОС резервирует 8 МБ виртуального адресного пространства (по умолчанию в Linux), но ОС выделяет физическую память только под реально затронутые страницы — поток, который в основном простаивает, занимает 20–80 КБ физической памяти.
• Переключение контекста: ~1–5 мкс на современном железе. При 50 одновременных запросах это шум. При 100 тысячах переключений в секунду это уже заметно.
• Стоимость создания: ~10–30 мкс на поток в Linux. Пул потоков (rayon, std::thread::scope) амортизирует это до нуля.

Честный порог, при котором async окупает свою сложность, — примерно 1–10 тысяч одновременных, в основном простаивающих соединений: зона epoll/io_uring, где стеки на соединение становятся реальной стоимостью. Ниже этого порога пул потоков проще, легче отлаживается и достаточно быстр. Выше — побеждает async. Большинство сервисов находятся ниже этого порога.

Трудный пример: логика, которой тоже нужен ввод-вывод

Тривиальная чистая функция — fn add(a: i32, b: i32) -> i32 — очевидно не нуждается в async. Это неинтересный урок. Интересный случай — когда бизнес-логике кажется, что ей посреди работы нужен ввод-вывод: валидация, которая проверяет остатки на складе, ценообразование, которое запрашивает курс валют, конвейер заказов, который ищет клиента.

Рассмотрим сервис обработки заказов. Вариант «async везде» выглядит естественно:

Вариант A: async через всё ядро

#![allow(unused)]
fn main() {
// orders.rs — async до самого низа

pub async fn process_order(order: Order) -> Result<Receipt, OrderError> {
    // Шаг 1: валидация — чистые бизнес-правила, без ввода-вывода
    validate_items(&order)?;
    validate_quantities(&order)?;

    // Шаг 2: проверка остатков — нужен вызов к базе данных
    let stock = inventory_client.check(&order.items).await?;
    if !stock.all_available() {
        return Err(OrderError::OutOfStock(stock.missing()));
    }

    // Шаг 3: расчёт цены — чистая арифметика, но async, потому что мы уже здесь
    let pricing = calculate_pricing(&order, &stock);

    // Шаг 4: применение скидки — нужен вызов внешнего сервиса
    let discount = discount_service.lookup(order.customer_id).await?;
    let final_price = pricing.apply_discount(discount);

    // Шаг 5: формирование чека — чисто
    Ok(Receipt::new(order, final_price))
}
}

Это разумный асинхронный код. Никакого злоупотребления Arc<Mutex> — просто последовательные await. Большинство разработчиков написали бы именно так и пошли дальше. Но посмотрите, что произошло: validate_items, validate_quantities, calculate_pricing и Receipt::new — это чистые функции, которые затянуло в асинхронный контекст, потому что шагам 2 и 4 нужен ввод-вывод. Вся функция должна стать асинхронной, её тестам нужен рантайм, и каждый вызывающий выше по цепочке тоже оказывается «окрашенным».

Вариант B: синхронное ядро, асинхронная оболочка

Альтернатива: отделить что решить от как получить данные:

#![allow(unused)]
fn main() {
// core.rs — чистая бизнес-логика, ноль async, ноль зависимостей от tokio

pub fn validate_order(order: &Order) -> Result<ValidatedOrder, OrderError> {
    validate_items(order)?;
    validate_quantities(order)?;
    Ok(ValidatedOrder::from(order))
}

pub fn check_stock(
    order: &ValidatedOrder,
    stock: &StockResult,
) -> Result<StockedOrder, OrderError> {
    if !stock.all_available() {
        return Err(OrderError::OutOfStock(stock.missing()));
    }
    Ok(StockedOrder::from(order, stock))
}

pub fn finalize(
    order: &StockedOrder,
    discount: Discount,
) -> Receipt {
    let pricing = calculate_pricing(order);
    let final_price = pricing.apply_discount(discount);
    Receipt::new(order, final_price)
}
}

#![allow(unused)]
fn main() {
// shell.rs — тонкий асинхронный оркестратор
//
// Примечание: `?` на сетевых вызовах требует `impl From<reqwest::Error> for OrderError`
// (или единого перечисления ошибок). См. гл. 12 — паттерны обработки ошибок в async.

use crate::core;

pub async fn process_order(order: Order) -> Result<Receipt, OrderError> {
    // Синхронно: валидация
    let validated = core::validate_order(&order)?;

    // Асинхронно: получаем остатки (это задача оболочки)
    let stock = inventory_client.check(&validated.items).await?;

    // Синхронно: применяем бизнес-правило к полученным данным
    let stocked = core::check_stock(&validated, &stock)?;

    // Асинхронно: получаем скидку
    let discount = discount_service.lookup(order.customer_id).await?;

    // Синхронно: финализируем
    Ok(core::finalize(&stocked, discount))
}
}

Асинхронная оболочка — это конвейер «получить → решить → получить → решить». Каждый шаг «решить» — синхронная функция, которая получает результат ввода-вывода как входные данные, а не ходит за ним сама.

Тестирование разницы

Синхронное ядро проверяет каждое бизнес-правило без рантайма и моков:

#![allow(unused)]
fn main() {
#[test]
fn out_of_stock_rejects_order() {
    let order = validated_order(vec![item("widget", 10)]);
    let stock = stock_result(vec![("widget", 3)]); // доступно только 3

    let result = core::check_stock(&order, &stock);
    assert_eq!(result.unwrap_err(), OrderError::OutOfStock(vec!["widget"]));
}

#[test]
fn discount_applied_correctly() {
    let order = stocked_order(100_00); // цена в копейках
    let receipt = core::finalize(&order, Discount::Percent(15));
    assert_eq!(receipt.final_price, 85_00);
}
}

Асинхронная оболочка получает более тонкий интеграционный тест, который проверяет связку компонентов, а не логику:

#![allow(unused)]
fn main() {
#[tokio::test]
async fn process_order_integration() {
    let mock_inventory = mock_service(/* возвращает остатки */);
    let mock_discounts = mock_service(/* возвращает 10% */);
    let receipt = process_order(sample_order()).await.unwrap();
    assert!(receipt.final_price > 0);
    // Корректность логики уже доказана тестами ядра выше
}
}

Почему это важно

Ключевая мысль: вызовы I/O на шагах 2 и 4 не обязаны находиться внутри бизнес-логики. Они — её входные данные. Синхронное ядро принимает StockResult и Discount как аргументы. Откуда пришли эти значения — из HTTP, gRPC, тестовой фикстуры или кэша — это забота оболочки.

Запах spawn_blocking

В главе 12 spawn_blocking был представлен как средство от случайной блокировки исполнителя. Это правильное решение для разовых блокирующих вызовов — std::fs::read, библиотеки сжатия, устаревшей FFI-функции.

Но если вы обнаруживаете, что оборачиваете в spawn_blocking большие участки кода:

#![allow(unused)]
fn main() {
async fn handler(req: Request) -> Response {
    // Если это ваша кодовая база, граница проведена не там
    tokio::task::spawn_blocking(move || {
        let validated = validate(&req);       // синхронно
        let enriched = enrich(validated);      // синхронно
        let result = process(enriched);        // синхронно
        let output = format_response(result);  // синхронно
        output
    }).await.unwrap()
}
}

…то это кодовая база сообщает вам: эта логика никогда не была асинхронной. spawn_blocking здесь не нужен — нужен синхронный модуль, который асинхронный обработчик вызывает напрямую:

#![allow(unused)]
fn main() {
async fn handler(req: Request) -> Response {
    // validate → enrich → process → format — всё синхронное.
    // spawn_blocking не нужен: это быстро и почти не нагружает CPU.
    let response = my_core::handle(req);
    response
}
}

Оставляйте spawn_blocking для действительно тяжёлой работы с CPU (разбор больших полезных нагрузок, обработка изображений, сжатие), где затраты времени реально могут заморить исполнитель. Для обычной бизнес-логики, которая выполняется за микросекунды, прямой синхронный вызов проще и корректен.

Библиотеки: сначала синхронные, асинхронная обёртка — по желанию

Вопрос границы ещё важнее для авторов библиотек. Синхронная библиотека может использоваться как из синхронного, так и из асинхронного кода:

// Синхронная библиотека — пригодна везде
let report = my_lib::analyze(&data);

// Вызывающий A: синхронная CLI-утилита
fn main() {
    let report = my_lib::analyze(&data);
    println!("{report}");
}

// Вызывающий B: асинхронный обработчик, работает нормально
async fn handler() -> Json<Report> {
    let report = my_lib::analyze(&data); // синхронный вызов в async-контексте — нормально
    Json(report)
}

// Вызывающий C: тяжёлый анализ — вызывающий решает разгрузить его
async fn handler_heavy() -> Json<Report> {
    let data = data.clone();
    let report = tokio::task::spawn_blocking(move || {
        my_lib::analyze(&data) // граница async контролирует вызывающий
    }).await.unwrap();
    Json(report)
}

Асинхронная библиотека загоняет всех вызывающих в рантайм:

#![allow(unused)]
fn main() {
// Асинхронная библиотека — пригодна только из асинхронного контекста
let report = my_lib::analyze(&data).await; // вызывающий ОБЯЗАН быть асинхронным

// Синхронный вызывающий? Теперь нужен block_on — и надежда, что нет вложенного рантайма
let report = tokio::runtime::Runtime::new().unwrap().block_on(
    my_lib::analyze(&data)
); // хрупко, может паниковать, если мы уже внутри рантайма
}

По умолчанию проектируйте синхронные API. Если ваша библиотека выполняет чистые вычисления, преобразование данных или разбор, нет причин делать её асинхронной. Если она выполняет ввод-вывод, рассмотрите синхронное ядро с необязательным асинхронным удобным слоем за feature-флагом — пусть вызывающий сам решает, где граница.

Когда async принадлежит ядру

Не всё можно аккуратно разделить. Async уместен в основной логике, когда:

• Fan-out/fan-in и есть логика. Если бизнес-правило звучит как «одновременно запросить 5 сервисов ценообразования и вернуть самый дешёвый», конкурентность — это и есть логика, а не сантехника. Пытаться протащить это через синхронный код и потоки — значит заново изобретать худший async.

• Потоковая обработка и есть логика. Обработка непрерывного потока событий с обратным давлением — управление потоком нетривиальная бизнес-логика, а не просто обёртка над I/O.

• Долгоживущие состоятельные соединения. Обработчики WebSocket, двунаправленные потоки gRPC и конечные автоматы протоколов имеют переходы состояний, неразрывно связанные с событиями ввода-вывода. Итоговый проект в гл. 17 — асинхронный чат-сервер — как раз такой случай: одновременные соединения, рассылка по комнатам и корректное завершение по своей природе являются асинхронной работой.

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

Правило принятия решения

graph TD
    START["Должна ли эта функция быть async?"] --> IO{"Она выполняет I/O?"}
    IO -->|Нет| SYNC["sync fn — всегда"]
    IO -->|Да| BOUNDARY{"Она на границе?<br/>обработчик, главный цикл, accept()"}
    BOUNDARY -->|Да| ASYNC_SHELL["async fn — это оболочка"]
    BOUNDARY -->|Нет| CORE_IO{"Является ли I/O основной логикой?<br/>fan-out, потоковая обработка, состоятельное соединение"}
    CORE_IO -->|Да| ASYNC_CORE["async fn — оправдано"]
    CORE_IO -->|Нет| EXTRACT["Вынесите логику в sync fn.<br/>Передавайте результаты I/O как аргументы."]

    style SYNC fill:#d4efdf,stroke:#27ae60,color:#000
    style ASYNC_SHELL fill:#e8f4f8,stroke:#2980b9,color:#000
    style ASYNC_CORE fill:#e8f4f8,stroke:#2980b9,color:#000
    style EXTRACT fill:#d4efdf,stroke:#27ae60,color:#000

Эмпирическое правило: начинайте с синхронного кода. Добавляйте async только на самой внешней границе ввода-вывода. Перемещайте его внутрь, только когда можете назвать, какие конкурентные операции ввода-вывода оправдывают налог на сложность.

#![allow(unused)]
fn main() {
use axum::{Json, extract::Path};

async fn get_device_report(Path(device_id): Path<String>) -> Result<Json<Report>, AppError> {
    // Получаем сырую телеметрию с устройства по HTTP
    let raw = reqwest::get(format!("http://bmc-{device_id}/telemetry"))
        .await?
        .json::<RawTelemetry>()
        .await?;

    // Бизнес-логика: преобразуем сырые показания датчиков в калиброванные значения
    let mut readings = Vec::new();
    for sensor in &raw.sensors {
        let calibrated = (sensor.raw_value as f64) * sensor.scale + sensor.offset;
        if calibrated < sensor.min_valid || calibrated > sensor.max_valid {
            return Err(AppError::SensorOutOfRange {
                name: sensor.name.clone(),
                value: calibrated,
            });
        }
        readings.push(CalibratedReading {
            name: sensor.name.clone(),
            value: calibrated,
            unit: sensor.unit.clone(),
        });
    }

    // Бизнес-логика: классифицируем состояние устройства
    let critical_count = readings.iter()
        .filter(|r| r.value > 90.0)
        .count();
    let health = if critical_count > 2 { Health::Critical }
                 else if critical_count > 0 { Health::Warning }
                 else { Health::Ok };

    // Получаем метаданные устройства из сервиса инвентаризации
    let meta = reqwest::get(format!("http://inventory/devices/{device_id}"))
        .await?
        .json::<DeviceMetadata>()
        .await?;

    Ok(Json(Report {
        device_id,
        device_name: meta.name,
        health,
        readings,
        timestamp: chrono::Utc::now(),
    }))
}
}

#![allow(unused)]
fn main() {
// core.rs — ноль зависимостей от async

pub fn calibrate_sensors(raw: &RawTelemetry) -> Result<Vec<CalibratedReading>, AppError> {
    raw.sensors.iter().map(|sensor| {
        let calibrated = (sensor.raw_value as f64) * sensor.scale + sensor.offset;
        if calibrated < sensor.min_valid || calibrated > sensor.max_valid {
            return Err(AppError::SensorOutOfRange {
                name: sensor.name.clone(),
                value: calibrated,
            });
        }
        Ok(CalibratedReading {
            name: sensor.name.clone(),
            value: calibrated,
            unit: sensor.unit.clone(),
        })
    }).collect()
}

pub fn classify_health(readings: &[CalibratedReading]) -> Health {
    let critical_count = readings.iter()
        .filter(|r| r.value > 90.0)
        .count();
    if critical_count > 2 { Health::Critical }
    else if critical_count > 0 { Health::Warning }
    else { Health::Ok }
}

pub fn build_report(
    device_id: String,
    readings: Vec<CalibratedReading>,
    meta: &DeviceMetadata,
) -> Report {
    Report {
        device_id,
        device_name: meta.name.clone(),
        health: classify_health(&readings),
        readings,
        timestamp: chrono::Utc::now(),
    }
}
}

#![allow(unused)]
fn main() {
// shell.rs — только асинхронная граница

pub async fn get_device_report(
    Path(device_id): Path<String>,
) -> Result<Json<Report>, AppError> {
    let raw = reqwest::get(format!("http://bmc-{device_id}/telemetry"))
        .await?
        .json::<RawTelemetry>()
        .await?;

    let readings = core::calibrate_sensors(&raw)?;

    let meta = reqwest::get(format!("http://inventory/devices/{device_id}"))
        .await?
        .json::<DeviceMetadata>()
        .await?;

    Ok(Json(core::build_report(device_id, readings, &meta)))
}
}

#![allow(unused)]
fn main() {
// core_tests.rs — рантайм не нужен

// Вспомогательные функции для фикстур — создают данные без какого-либо ввода-вывода
fn sensor(name: &str, raw_value: f64, valid_range: std::ops::Range<f64>) -> RawSensor {
    RawSensor {
        name: name.into(),
        raw_value,
        scale: 1.0,
        offset: 0.0,
        min_valid: valid_range.start,
        max_valid: valid_range.end,
        unit: "unit".into(),
    }
}

fn raw_telemetry(sensors: Vec<RawSensor>) -> RawTelemetry {
    RawTelemetry { sensors }
}

fn reading(name: &str, value: f64) -> CalibratedReading {
    CalibratedReading { name: name.into(), value, unit: "unit".into() }
}

fn device_meta(name: &str) -> DeviceMetadata {
    DeviceMetadata { name: name.into() }
}

#[test]
fn sensor_out_of_range_rejected() {
    let raw = raw_telemetry(vec![sensor("gpu_temp", 105.0, 0.0..100.0)]);
    let result = core::calibrate_sensors(&raw);
    assert!(matches!(result, Err(AppError::SensorOutOfRange { .. })));
}

#[test]
fn health_classification() {
    let readings = vec![
        reading("a", 50.0),  // ok
        reading("b", 95.0),  // critical
        reading("c", 91.0),  // critical
        reading("d", 92.0),  // critical
    ];
    assert_eq!(core::classify_health(&readings), Health::Critical);
}

#[test]
fn normal_report() {
    let raw = raw_telemetry(vec![sensor("fan_rpm", 3000.0, 0.0..10000.0)]);
    let readings = core::calibrate_sensors(&raw).unwrap();
    let meta = device_meta("gpu-node-42");
    let report = core::build_report("dev-1".into(), readings, &meta);
    assert_eq!(report.health, Health::Ok);
    assert_eq!(report.readings.len(), 1);
}
}

Ключевые выводы:

1. Async — это оптимизация мультиплексирования I/O, а не архитектура приложения. Большая часть бизнес-логики синхронна.
2. Синхронное ядро, асинхронная оболочка: держите бизнес-правила в чистых синхронных функциях, которые принимают результаты I/O как аргументы. Асинхронная оболочка организует запросы и вызывает ядро.
3. Если вы оборачиваете большие блоки в spawn_blocking, граница проведена не там — вместо этого вынесите логику в синхронный модуль.
4. Библиотеки должны по умолчанию предлагать синхронные API. Асинхронная библиотека заставляет всех вызывающих работать с рантаймом; синхронная позволяет вызывающему самому определять асинхронную границу.
5. Async окупает себя в fan-out/fan-in, потоковой обработке и состоятельных соединениях — там, где конкурентность и есть бизнес-логика.

См. также: Гл. 12 — Типичные ловушки (spawn_blocking как тактическое решение) · Гл. 13 — Продакшен-паттерны (обратное давление, структурированная конкурентность) · Гл. 17 — Итоговый проект: асинхронный чат-сервер (случай, когда async — правильная архитектура)

15. Упражнения

Упражнения

Упражнение 1: асинхронный эхо-сервер

Напишите TCP-эхо-сервер, который обрабатывает нескольких клиентов одновременно.

Требования:

• Слушать 127.0.0.1:8080
• Принимать соединения и возвращать обратно каждую строку
• Корректно обрабатывать отключения клиентов
• Выводить лог при подключении и отключении клиентов

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpListener;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Эхо-сервер слушает :8080");

    loop {
        let (socket, addr) = listener.accept().await?;
        println!("[{addr}] Подключён");

        tokio::spawn(async move {
            let (reader, mut writer) = socket.into_split();
            let mut reader = BufReader::new(reader);
            let mut line = String::new();

            loop {
                line.clear();
                match reader.read_line(&mut line).await {
                    Ok(0) => {
                        println!("[{addr}] Отключён");
                        break;
                    }
                    Ok(_) => {
                        print!("[{addr}] Эхо: {line}");
                        if writer.write_all(line.as_bytes()).await.is_err() {
                            println!("[{addr}] Ошибка записи, отключаем");
                            break;
                        }
                    }
                    Err(e) => {
                        eprintln!("[{addr}] Ошибка чтения: {e}");
                        break;
                    }
                }
            }
        });
    }
}

Упражнение 2: конкурентный загрузчик URL с ограничением частоты

Загрузите список URL одновременно, но не более 5 запросов одновременно.

#![allow(unused)]
fn main() {
use futures::stream::{self, StreamExt};
use tokio::time::{sleep, Duration};

async fn fetch_urls(urls: Vec<String>) -> Vec<Result<String, String>> {
    // buffer_unordered(5) гарантирует, что опрашивается не более 5 future
    // одновременно — отдельный Semaphore здесь не нужен.
    let results: Vec<_> = stream::iter(urls)
        .map(|url| {
            async move {
                println!("Загружаем: {url}");

                match reqwest::get(&url).await {
                    Ok(resp) => match resp.text().await {
                        Ok(body) => Ok(body),
                        Err(e) => Err(format!("{url}: {e}")),
                    },
                    Err(e) => Err(format!("{url}: {e}")),
                }
            }
        })
        .buffer_unordered(5) // ← Этого достаточно, чтобы ограничить конкурентность 5
        .collect()
        .await;

    results
}

// ПРИМЕЧАНИЕ: используйте Semaphore, когда нужно ограничить конкурентность
// для независимо порождённых задач (tokio::spawn). Используйте buffer_unordered,
// когда обрабатываете поток. Не комбинируйте оба механизма для одного и того же лимита.
}

Упражнение 3: graceful shutdown с пулом воркеров

Постройте обработчик задач с:

• очередью работы на каналах
• N задачами-воркерами, которые читают из очереди
• graceful shutdown по Ctrl+C: прекратить приём, доделать текущую работу

use tokio::sync::{mpsc, watch};
use tokio::time::{sleep, Duration};

struct WorkItem {
    id: u64,
    payload: String,
}

#[tokio::main]
async fn main() {
    let (work_tx, work_rx) = mpsc::channel::<WorkItem>(100);
    let (shutdown_tx, shutdown_rx) = watch::channel(false);

    // Запускаем 4 воркера
    let mut worker_handles = Vec::new();
    let work_rx = std::sync::Arc::new(tokio::sync::Mutex::new(work_rx));

    for id in 0..4 {
        let rx = work_rx.clone();
        let mut shutdown = shutdown_rx.clone();
        let handle = tokio::spawn(async move {
            loop {
                let item = {
                    let mut rx = rx.lock().await;
                    tokio::select! {
                        item = rx.recv() => item,
                        _ = shutdown.changed() => {
                            if *shutdown.borrow() { None } else { continue }
                        }
                    }
                };

                match item {
                    Some(work) => {
                        println!("Воркер {id}: обрабатываю элемент {}", work.id);
                        sleep(Duration::from_millis(200)).await; // Имитация работы
                        println!("Воркер {id}: закончил элемент {}", work.id);
                    }
                    None => {
                        println!("Воркер {id}: канал закрыт, выходим");
                        break;
                    }
                }
            }
        });
        worker_handles.push(handle);
    }

    // Производитель: отправляет работу
    let producer = tokio::spawn(async move {
        for i in 0..20 {
            let _ = work_tx.send(WorkItem {
                id: i,
                payload: format!("task-{i}"),
            }).await;
            sleep(Duration::from_millis(50)).await;
        }
    });

    // Ждём Ctrl+C
    tokio::signal::ctrl_c().await.unwrap();
    println!("\nПолучен сигнал остановки!");
    shutdown_tx.send(true).unwrap();
    producer.abort(); // Отменяем производителя

    // Ждём, пока воркеры завершатся
    for handle in worker_handles {
        let _ = handle.await;
    }
    println!("Все воркеры остановлены. До свидания!");
}

Упражнение 4: простой асинхронный мьютекс с нуля

Реализуйте асинхронный мьютекс на основе каналов (без tokio::sync::Mutex).

Подсказка: используйте tokio::sync::Semaphore с одним разрешением, чтобы сериализовать доступ.

#![allow(unused)]
fn main() {
use std::cell::UnsafeCell;
use std::sync::Arc;
use tokio::sync::{OwnedSemaphorePermit, Semaphore};

pub struct SimpleAsyncMutex<T> {
    data: Arc<UnsafeCell<T>>,
    semaphore: Arc<Semaphore>,
}

// SAFETY: доступ к T сериализуется семафором (не более одного разрешения).
unsafe impl<T: Send> Send for SimpleAsyncMutex<T> {}
unsafe impl<T: Send> Sync for SimpleAsyncMutex<T> {}

pub struct SimpleGuard<T> {
    data: Arc<UnsafeCell<T>>,
    _permit: OwnedSemaphorePermit, // Уничтожается вместе с guard → освобождает блокировку
}

impl<T> SimpleAsyncMutex<T> {
    pub fn new(value: T) -> Self {
        SimpleAsyncMutex {
            data: Arc::new(UnsafeCell::new(value)),
            semaphore: Arc::new(Semaphore::new(1)),
        }
    }

    pub async fn lock(&self) -> SimpleGuard<T> {
        let permit = self.semaphore.clone().acquire_owned().await.unwrap();
        SimpleGuard {
            data: self.data.clone(),
            _permit: permit,
        }
    }
}

impl<T> std::ops::Deref for SimpleGuard<T> {
    type Target = T;
    fn deref(&self) -> &T {
        // SAFETY: у нас единственное разрешение семафора, поэтому других
        // SimpleGuard не существует → эксклюзивный доступ гарантирован.
        unsafe { &*self.data.get() }
    }
}

impl<T> std::ops::DerefMut for SimpleGuard<T> {
    fn deref_mut(&mut self) -> &mut T {
        // SAFETY: то же рассуждение — единственное разрешение гарантирует эксклюзивность.
        unsafe { &mut *self.data.get() }
    }
}

// Когда SimpleGuard уничтожается, уничтожается и _permit,
// что освобождает разрешение семафора — следующий lock() может продолжить.

// Использование:
// let mutex = SimpleAsyncMutex::new(vec![1, 2, 3]);
// {
//     let mut guard = mutex.lock().await;
//     guard.push(4);
// } // разрешение освобождено здесь
}

Упражнение 5: конвейер потоков

Постройте конвейер обработки данных на потоках:

1. Сгенерируйте числа 1..=100
2. Отфильтруйте чётные
3. Возведите каждое в квадрат
4. Обрабатывайте по 10 штук одновременно (имитация через sleep)
5. Соберите результаты

use futures::stream::{self, StreamExt};
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let results: Vec<u64> = stream::iter(1u64..=100)
        // Шаг 2: отфильтровать чётные
        .filter(|x| futures::future::ready(x % 2 == 0))
        // Шаг 3: возвести каждое в квадрат
        .map(|x| x * x)
        // Шаг 4: обрабатывать одновременно (имитация асинхронной работы)
        .map(|x| async move {
            sleep(Duration::from_millis(50)).await;
            println!("Обработано: {x}");
            x
        })
        .buffer_unordered(10) // 10 одновременно
        // Шаг 5: собрать
        .collect()
        .await;

    println!("Получено результатов: {}", results.len());
    println!("Сумма: {}", results.iter().sum::<u64>());
}

Упражнение 6: реализуйте select с таймаутом

Не используя tokio::select! и tokio::time::timeout, реализуйте функцию, которая состязает future с дедлайном и возвращает Either::Left(result) или Either::Right(()) при таймауте.

Подсказка: опирайтесь на комбинатор Select из главы 6 и TimerFuture из той же главы.

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

pub enum Either<A, B> {
    Left(A),
    Right(B),
}

pub struct Timeout<F> {
    future: F,
    timer: TimerFuture, // Из главы 6
}

impl<F: Future + Unpin> Timeout<F> {
    pub fn new(future: F, duration: Duration) -> Self {
        Timeout {
            future,
            timer: TimerFuture::new(duration),
        }
    }
}

impl<F: Future + Unpin> Future for Timeout<F> {
    type Output = Either<F::Output, ()>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // Проверяем, завершился ли основной future
        if let Poll::Ready(val) = Pin::new(&mut self.future).poll(cx) {
            return Poll::Ready(Either::Left(val));
        }

        // Проверяем, не истёк ли таймер
        if let Poll::Ready(()) = Pin::new(&mut self.timer).poll(cx) {
            return Poll::Ready(Either::Right(()));
        }

        Poll::Pending
    }
}

// Использование:
// match Timeout::new(fetch_data(), Duration::from_secs(5)).await {
//     Either::Left(data) => println!("Получены данные: {data}"),
//     Either::Right(()) => println!("Время вышло!"),
// }

Итоги и справочная карточка

Краткая справочная карточка

Ментальная модель async

┌─────────────────────────────────────────────────────┐
│  async fn → конечный автомат (enum) → impl Future   │
│  .await   → poll() вложенного future                │
│  исполнитель → loop { poll(); спать_до_пробуждения; }│
│  waker    → «эй, исполнитель, опроси меня снова»    │
│  Pin      → «обещаю, что не переместят в памяти»    │
└─────────────────────────────────────────────────────┘

Шпаргалка по частым паттернам

Краткая справка по закреплению (pinning)

Выбор канала

Выбор мьютекса

Краткая справка по решениям

Нужна конкурентность?
├── Ограничена I/O → async/await
├── Ограничена CPU → rayon / std::thread
└── Смешанная → spawn_blocking для CPU-части

Выбор рантайма?
├── Серверное приложение → tokio
├── Библиотека → не привязана к рантайму (крейт futures)
├── Встраиваемая система → embassy
└── Минимальный → smol

Нужны конкурентные future?
├── Могут быть 'static + Send → tokio::spawn
├── Могут быть 'static + !Send → LocalSet
├── Не могут быть 'static → FuturesUnordered
└── Нужно отслеживать/прерывать → JoinSet

Частые сообщения об ошибках и их исправление

Дополнительная литература

Конец руководства по асинхронному Rust

Итоговый проект: асинхронный чат-сервер

Этот проект объединяет паттерны из всей книги в одно приложение продакшен-уровня. Вы построите многокомнатный асинхронный чат-сервер с использованием tokio, каналов, потоков, graceful shutdown и корректной обработки ошибок.

Расчётное время: 4–6 часов | Сложность: ★★★

Что вы потренируете:

• tokio::spawn и требование 'static (гл. 8)
• Каналы: mpsc для сообщений, broadcast для комнат, watch для остановки (гл. 8)
• Потоки: чтение строк из TCP-соединений (гл. 11)
• Типичные ловушки: безопасность при отмене, MutexGuard через .await (гл. 12)
• Продакшен-паттерны: graceful shutdown, обратное давление (гл. 13)
• Async-трейты для подключаемых бэкендов (гл. 10)

Постановка задачи

Постройте TCP-чат-сервер, где:

1. Клиенты подключаются по TCP и входят в именованные комнаты
2. Сообщения рассылаются всем клиентам в той же комнате
3. Команды: /join <комната>, /nick <имя>, /rooms, /quit
4. Сервер корректно завершается по Ctrl+C — доставляя сообщения, которые уже в пути

graph LR
    C1["Клиент 1<br/>(Alice)"] -->|TCP| SERVER["Чат-сервер"]
    C2["Клиент 2<br/>(Bob)"] -->|TCP| SERVER
    C3["Клиент 3<br/>(Carol)"] -->|TCP| SERVER

    SERVER --> R1["#general<br/>канал broadcast"]
    SERVER --> R2["#rust<br/>канал broadcast"]

    R1 -->|сообщение| C1
    R1 -->|сообщение| C2
    R2 -->|сообщение| C3

    CTRL["Ctrl+C"] -->|watch| SERVER

    style SERVER fill:#e8f4f8,stroke:#2980b9,color:#000
    style R1 fill:#d4efdf,stroke:#27ae60,color:#000
    style R2 fill:#d4efdf,stroke:#27ae60,color:#000
    style CTRL fill:#fadbd8,stroke:#e74c3c,color:#000

Шаг 1: базовый цикл принятия TCP-соединений

Начните с сервера, который принимает соединения и возвращает строки обратно:

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpListener;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Чат-сервер слушает :8080");

    loop {
        let (socket, addr) = listener.accept().await?;
        println!("[{addr}] Подключён");

        tokio::spawn(async move {
            let (reader, mut writer) = socket.into_split();
            let mut reader = BufReader::new(reader);
            let mut line = String::new();

            loop {
                line.clear();
                match reader.read_line(&mut line).await {
                    Ok(0) | Err(_) => break,
                    Ok(_) => {
                        let _ = writer.write_all(line.as_bytes()).await;
                    }
                }
            }
            println!("[{addr}] Отключён");
        });
    }
}

Ваша задача: убедитесь, что это компилируется и работает через telnet localhost 8080.

Шаг 2: состояние комнат с broadcast-каналами

Каждая комната — это broadcast::Sender. Все клиенты в комнате подписываются, чтобы получать сообщения.

#![allow(unused)]
fn main() {
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{broadcast, RwLock};

type RoomMap = Arc<RwLock<HashMap<String, broadcast::Sender<String>>>>;

fn get_or_create_room(rooms: &mut HashMap<String, broadcast::Sender<String>>, name: &str) -> broadcast::Sender<String> {
    rooms.entry(name.to_string())
        .or_insert_with(|| {
            let (tx, _) = broadcast::channel(100); // Буфер на 100 сообщений
            tx
        })
        .clone()
}
}

Ваша задача: реализуйте состояние комнат так, чтобы:

• Клиенты начинают в #general
• /join <комната> переключает комнату (отписка от старой, подписка на новую)
• Сообщения рассылаются всем клиентам в текущей комнате отправителя

#![allow(unused)]
fn main() {
loop {
    tokio::select! {
        // Клиент прислал строку
        result = reader.read_line(&mut line) => {
            match result {
                Ok(0) | Err(_) => break,
                Ok(_) => {
                    // Разобрать команду или разослать сообщение
                }
            }
        }
        // Получено сообщение комнаты
        result = room_rx.recv() => {
            match result {
                Ok(msg) => {
                    let _ = writer.write_all(msg.as_bytes()).await;
                }
                Err(_) => break,
            }
        }
    }
}
}

Шаг 3: команды

Реализуйте протокол команд:

Ваша задача: разберите команды из входной строки. Для /rooms нужно читать из RoomMap — используйте RwLock::read(), чтобы не блокировать других клиентов.

Шаг 4: graceful shutdown

Добавьте обработку Ctrl+C, чтобы сервер:

1. Прекратил приём новых соединений
2. Отправил «Сервер останавливается…» во все комнаты
3. Дождался, пока сообщения в пути будут доставлены
4. Корректно завершился

#![allow(unused)]
fn main() {
use tokio::sync::watch;

let (shutdown_tx, shutdown_rx) = watch::channel(false);

// В цикле принятия соединений:
loop {
    tokio::select! {
        result = listener.accept() => {
            let (socket, addr) = result?;
            // запускаем клиентскую задачу с shutdown_rx.clone()
        }
        _ = tokio::signal::ctrl_c() => {
            println!("Получен сигнал остановки");
            shutdown_tx.send(true)?;
            break;
        }
    }
}
}

Ваша задача: добавьте shutdown_rx.changed() в select!-цикл каждого клиента, чтобы клиенты завершались при сигнале остановки.

Шаг 5: обработка ошибок и граничных случаев

Усильте сервер для продакшена:

1. Отстающие получатели: broadcast::recv() возвращает RecvError::Lagged(n), если медленный клиент пропустил сообщения. Обработайте это аккуратно (запишите в лог и продолжайте, не падайте).
2. Валидация имени: отклоняйте пустые и слишком длинные никнеймы.
3. Обратное давление: буфер broadcast-канала ограничен (100). Если клиент не успевает, он получает ошибку Lagged.
4. Таймаут: отключайте клиентов, которые простаивают больше 5 минут.

#![allow(unused)]
fn main() {
use tokio::time::{timeout, Duration};

// Оборачиваем чтение в таймаут:
match timeout(Duration::from_secs(300), reader.read_line(&mut line)).await {
    Ok(Ok(0)) | Ok(Err(_)) | Err(_) => break, // конец потока, ошибка или таймаут
    Ok(Ok(_)) => { /* обрабатываем строку */ }
}
}

Шаг 6: интеграционный тест

Напишите тест, который запускает сервер, подключает двух клиентов и проверяет доставку сообщений:

#![allow(unused)]
fn main() {
#[tokio::test]
async fn two_clients_can_chat() {
    // Запускаем сервер в фоне
    let server = tokio::spawn(run_server("127.0.0.1:0")); // Порт 0 = ОС выберет сама

    // Подключаем двух клиентов
    let mut client1 = TcpStream::connect(addr).await.unwrap();
    let mut client2 = TcpStream::connect(addr).await.unwrap();

    // Клиент 1 отправляет сообщение
    client1.write_all(b"Hello from client 1\n").await.unwrap();

    // Клиент 2 должен его получить
    let mut buf = vec![0u8; 1024];
    let n = client2.read(&mut buf).await.unwrap();
    let msg = String::from_utf8_lossy(&buf[..n]);
    assert!(msg.contains("Hello from client 1"));
}
}

Критерии оценки

Идеи для расширения

Когда базовый чат-сервер заработает, попробуйте такие улучшения:

1. Постоянная история: храните последние N сообщений каждой комнаты и проигрывайте их новым участникам
2. Поддержка WebSocket: принимайте и TCP-, и WebSocket-клиентов с помощью tokio-tungstenite
3. Ограничение частоты: используйте tokio::time::Interval, чтобы ограничить число сообщений от клиента в секунду
4. Метрики: отслеживайте число подключённых клиентов, сообщений в секунду и число комнат через крейт prometheus
5. TLS: добавьте tokio-rustls для зашифрованных соединений