Async в Rust подробно: select!, тайм-ауты, стримы и Pin

На странице Async/Await ты разобрался с основами: async fn возвращает ленивую футуру, .await её выполняет, #[tokio::main] запускает рантайм, а tokio::join! и tokio::spawn выполняют несколько дел одновременно. Если что-то из этого звучит незнакомо, сначала прочитай ту страницу. Эта строится на ней.

Настоящим асинхронным программам нужно ещё несколько инструментов. Снова представь оживлённую кухню. Иногда ты ждёшь все блюда (join!). Иногда тебе важно только первое, что произойдёт: еда готова или гость ушёл (select!). Иногда ты сдаёшься через пять минут (timeout). А иногда заказы весь вечер поступают один за другим (стримы). На этой странице разобрано всё это, а ещё Pin — странное слово, которое ты рано или поздно увидишь в сообщениях об ошибках, — простыми словами.

Во всех примерах используется tokio с features = ["full"], как и на странице основ. Кнопки «Попробуй сам» работают: в Rust Playground tokio уже есть.

Запускаем много дел одновременно

Ты уже знаешь tokio::join!: он ждёт все свои футуры и возвращает их результаты кортежем. Его родственник tokio::try_join! предназначен для футур, которые возвращают Result. Он даёт Ok со всеми значениями или первую Err и перестаёт ждать остальных, как только одна из футур завершилась ошибкой:

use std::time::Duration;
use tokio::time::sleep;

async fn load_profile(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(20)).await;
    Ok(format!("профиль #{id}"))
}

async fn load_settings(id: u32) -> Result<String, String> {
    sleep(Duration::from_millis(10)).await;
    if id == 2 {
        return Err(format!("нет настроек для пользователя {id}"));
    }
    Ok(String::from("тёмная тема"))
}

#[tokio::main]
async fn main() {
    for id in [1, 2] {
        match tokio::try_join!(load_profile(id), load_settings(id)) {
            Ok((profile, settings)) => println!("пользователь {id}: {profile}, {settings}"),
            Err(e) => println!("пользователь {id}, ошибка: {e}"),
        }
    }
}

Вывод:

пользователь 1: профиль #1, тёмная тема
пользователь 2, ошибка: нет настроек для пользователя 2

▶ Попробуй сам

join! и try_join! выполняют футуры внутри текущей задачи, по очереди. Это идеально для нескольких футур, известных заранее. Когда у тебя список заданий, размер которого известен только во время работы, запускай их как отдельные задачи. Можно складывать каждый JoinHandle в Vec, как на странице основ. Но у tokio для этого есть инструмент удобнее: JoinSet — мешок задач, который отдаёт каждый результат, как только задача закончилась:

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::sleep;

async fn check_site(site: &'static str, ms: u64) -> (&'static str, u64) {
    sleep(Duration::from_millis(ms)).await; // представим, что это сетевой запрос
    (site, ms)
}

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();
    for (site, ms) in [("rust-lang.org", 30), ("docs.rs", 10), ("crates.io", 20)] {
        set.spawn(check_site(site, ms));
    }

    let mut results = Vec::new();
    while let Some(res) = set.join_next().await {
        results.push(res.unwrap()); // Err, только если задача запаниковала
    }

    results.sort(); // порядок завершения может меняться от запуска к запуску
    for (site, ms) in results {
        println!("{site} ответил за {ms} мс");
    }
}

Вывод:

crates.io ответил за 20 мс
docs.rs ответил за 10 мс
rust-lang.org ответил за 30 мс

▶ Попробуй сам

join_next() возвращает None, когда набор пуст, поэтому цикл while let заканчивается сам. Результаты приходят в порядке завершения, а не запуска. Когда задачи занимают примерно одинаковое время, этот порядок может меняться между запусками, поэтому перед выводом мы сортируем. И ещё одно приятное свойство: если удалить JoinSet, все задачи, которые ещё в нём лежат, отменяются. Ничего не остаётся работать случайно.

Гонка футур с select!

join! ждёт всё. tokio::select! ждёт первое. У каждой ветки есть образец, футура и блок кода: pattern = future => code. Какая футура закончится первой, того код и выполнится, а остальные ветки отменяются.

Что значит «отменяется» в Rust? Проигравшая футура просто удаляется (drop), как любое другое значение, вышедшее из области видимости. Она больше никогда не выполнится. Её локальные переменные тоже удаляются, так что их код очистки срабатывает. В этом примере небольшой тип с Drop делает это видимым:

use std::time::Duration;
use tokio::time::sleep;

struct Download(&'static str);

impl Drop for Download {
    fn drop(&mut self) {
        println!("зеркало {}: убираем загрузку", self.0);
    }
}

async fn download(mirror: &'static str, ms: u64) -> &'static str {
    let _download = Download(mirror);
    sleep(Duration::from_millis(ms)).await;
    println!("зеркало {mirror}: загрузка завершена");
    mirror
}

#[tokio::main]
async fn main() {
    tokio::select! {
        winner = download("Европа", 50) => println!("победитель: {winner}"),
        winner = download("Азия", 1000) => println!("победитель: {winner}"),
    }
    println!("main продолжает работу");
}

Вывод:

зеркало Европа: загрузка завершена
зеркало Европа: убираем загрузку
зеркало Азия: убираем загрузку
победитель: Европа
main продолжает работу

▶ Попробуй сам

Посмотри, чего не хватает: строка «зеркало Азия: загрузка завершена» так и не напечаталась. Футура «Азии» остановилась на своём .await и была удалена, а её значение Download сразу же очистилось. Так отмена работает во всём асинхронном Rust: перестань опрашивать футуру и удали её. Никакого специального вызова «отменить» не нужно.

Классическое применение select! — тайм-аут: устраиваем гонку между работой и таймером.

tokio::select! {
    report = build_report() => println!("{report}"),
    _ = tokio::time::sleep(Duration::from_secs(5)) => println!("слишком долго, сдаёмся"),
}

Это прекрасно работает, но для такого случая есть сокращение, tokio::time::timeout. О нём — в следующем разделе.

Работаем до сигнала остановки

select! по-настоящему раскрывается внутри loop. Здесь воркер берёт задания из канала, но заодно слушает второй канал, по которому может прийти просьба «пожалуйста, остановись». Канал oneshot переносит ровно одно сообщение, поэтому он идеально подходит для сигнала остановки:

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

#[tokio::main]
async fn main() {
    let (job_tx, mut job_rx) = mpsc::channel::<String>(10);
    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();

    let worker = tokio::spawn(async move {
        let mut done = 0;
        loop {
            tokio::select! {
                Some(job) = job_rx.recv() => {
                    println!("в работе: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("получен сигнал остановки");
                    break;
                }
            }
        }
        done
    });

    for job in ["уменьшить фото", "отправить письмо", "обновить поисковый индекс"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    sleep(Duration::from_millis(100)).await; // даём воркеру время закончить
    stop_tx.send(()).unwrap();

    let done = worker.await.unwrap();
    println!("воркер выполнил заданий: {done}");
}

Вывод:

в работе: уменьшить фото
в работе: отправить письмо
в работе: обновить поисковый индекс
получен сигнал остановки
воркер выполнил заданий: 3

▶ Попробуй сам

Несколько деталей, которые полезно знать:

• Мы пишем &mut stop_rx, а не stop_rx. На каждом круге цикла select! нужен тот же самый получатель, поэтому мы его одалживаем, а не перемещаем в первый же круг.
• На каждом круге job_rx.recv() создаёт новую футуру, а проигравшая удаляется. Это нормально, потому что recv() безопасен для отмены (cancel-safe): если его удалить, сообщение не потеряется.
• Если образец не совпал (здесь — recv() вернул None, потому что все отправители исчезли), select! отключает эту ветку и продолжает ждать остальные.
• Если несколько веток готовы одновременно, select! выбирает одну случайно, чтобы ни одна ветка не захватила цикл. Напиши biased; первой строкой внутри select!, и ветки будут проверяться сверху вниз.

Безопасность отмены — в одном абзаце. Раз проигравшая ветка удаляется, всё, что она успела сделать наполовину, выбрасывается. Для recv() у канала или sleep() это безвредно: ничего ещё не забрано. Но представь футуру, которая читает 100 байт из сетевого сокета и уже прочитала 60, когда её удалили. Эти 60 байт потеряны. Такая футура не безопасна для отмены. В документации tokio для каждого метода написано, безопасен ли он для отмены. Когда используешь select! в цикле, проверяй методы, которые вызываешь, или создай футуру один раз вне цикла и продолжай опрашивать её же (как это сделать, увидишь в разделе про Pin).

Тайм-ауты и таймеры

tokio::time::timeout(duration, future) оборачивает любую футуру. Если она успела закончиться, получишь Ok(value). Если время вышло раньше, футура удаляется (отменяется, как проигравшая ветка select!), и ты получаешь Err(Elapsed):

use std::time::Duration;
use tokio::time::{sleep, timeout};

async fn fetch_price(server_delay_ms: u64) -> u32 {
    sleep(Duration::from_millis(server_delay_ms)).await;
    42
}

#[tokio::main]
async fn main() {
    let limit = Duration::from_millis(200);

    for (server, delay) in [("быстрый сервер", 20), ("медленный сервер", 2000)] {
        match timeout(limit, fetch_price(delay)).await {
            Ok(price) => println!("{server}: цена {price}"),
            Err(e) => println!("{server}: {e}"),
        }
    }
}

Вывод:

быстрый сервер: цена 42
медленный сервер: deadline has elapsed

▶ Попробуй сам

Текст ошибки deadline has elapsed («срок истёк») печатает сам tokio, поэтому он на английском. Ставь тайм-аут на всё, что общается с внешним миром: HTTP-запросы, запросы к базе данных, чтение из сокета. Сервер, который никогда не ответит, не должен навсегда замораживать твою программу.

sleep или interval. sleep ждёт один раз. interval задаёт ровный ритм, как метроном. Ты вызываешь .tick().await в цикле, и каждый тик приходит по расписанию. Первый тик срабатывает сразу:

use std::time::Duration;
use tokio::time::interval;

#[tokio::main]
async fn main() {
    let mut heartbeat = interval(Duration::from_millis(50));

    for beat in 1..=3 {
        heartbeat.tick().await; // первый тик завершается сразу
        println!("пульс {beat}");
    }
}

Вывод:

пульс 1
пульс 2
пульс 3

▶ Попробуй сам

Почему бы просто не делать sleep на 50 мс в цикле? Потому что тогда каждый круг занимает «твоя работа + 50 мс», и расписание постепенно уплывает. interval отсчитывает от запланированного начала каждого тика, так что если работа заняла 10 мс, он подождёт только оставшиеся 40 мс. Используй interval для сигналов «я жив» (heartbeat), периодического опроса и задач вроде «сохранять каждую минуту». Используй sleep для одной паузы.

Стримы: асинхронные итераторы

Итератор отдаёт значения по одному через next(). Стрим (stream) — его асинхронная версия: next().await отдаёт следующее значение, когда оно готово, и None, когда значений больше нет. Представь бегущую строку с ценами акций, строки, приходящие из базы данных, или сообщения от чат-сервера.

На самом деле ты уже пользовался чем-то похожим на стрим: получателем канала tokio. rx.recv().await ждёт следующее сообщение и возвращает None, когда все отправители исчезли:

use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(8);

    tokio::spawn(async move {
        for temp in [21, 22, 24, 23] {
            tx.send(temp).await.unwrap();
            sleep(Duration::from_millis(10)).await; // измерение каждые 10 мс
        }
        // здесь tx удаляется, и канал закрывается
    });

    while let Some(temp) = rx.recv().await {
        println!("температура: {temp}°C");
    }
    println!("датчик остановился");
}

Вывод:

температура: 21°C
температура: 22°C
температура: 24°C
температура: 23°C
датчик остановился

▶ Попробуй сам

В стандартной библиотеке пока нет трейта Stream (он ещё проектируется). Экосистема использует трейт из крейта futures, а вспомогательный крейт tokio tokio-stream добавляет трейт StreamExt с методами, знакомыми по итераторам: next(), map(), filter(), take() и другими. Добавь его в проект:

[dependencies]
tokio = { version = "1", features = ["full"] }
tokio-stream = "0.1"

use tokio_stream::{self as stream, StreamExt};

#[tokio::main]
async fn main() {
    let prices = stream::iter(vec![100, 102, 98, 105, 110]);

    let mut doubled = prices.map(|p| p * 2).take(3);

    while let Some(p) = doubled.next().await {
        println!("{p}");
    }
}

Программа печатает 200, 204 и 196. Читается так же, как код с итераторами, только с .await после next(). Чтобы превратить получатель канала tokio в настоящий стрим, оберни его: tokio_stream::wrappers::ReceiverStream::new(rx). (У крейта futures есть свой StreamExt с той же идеей. Выбирай один на файл: у обоих трейтов есть метод next(), и они конфликтуют.)

Pin простыми словами

Рано или поздно ты увидишь Pin в типе или в сообщении об ошибке. Вот идея без глубокой теории.

Когда Rust компилирует async fn, он превращает её в скрытую структуру — конечный автомат (state machine). Каждая локальная переменная, которая нужна после .await, становится полем этой структуры. А теперь представь, что одна локальная переменная заимствует другую, например let name = String::from("Аня"); let r = &name;, а между ними есть .await. Тогда структура хранит указатель на собственное поле. Если переместить структуру в другое место памяти, этот указатель будет указывать на старое, пустое место.

Pin решает эту проблему. Закреплённое (pinned) значение — это обещание: «это значение больше не будет перемещаться в памяти». Как только футура закреплена, её безопасно опрашивать. Чаще всего ты этого даже не замечаешь, потому что .await закрепляет футуру за тебя.

С Pin ты встретишься в трёх ситуациях:

1. Многократный опрос одной и той же футуры, например дедлайна в цикле с select!.
2. Хранение футур в Vec или поле структуры, обычно как Pin<Box<dyn Future<Output = T>>>.
3. Рекурсия в async fn.

Для первого случая вот что будет, если попытаться переиспользовать sleep в цикле, не закрепив его:

let deadline = sleep(Duration::from_millis(300));

loop {
    tokio::select! {
        _ = &mut deadline => break,
        Some(n) = rx.recv() => println!("{n}"),
    }
}

error[E0277]: `PhantomPinned` cannot be unpinned
   |
   = note: consider using the `pin!` macro
           consider using `Box::pin` if you need to access the pinned value outside of the current scope
   = note: required for `Sleep` to implement `Unpin`

Компилятор (по-английски) сам подсказывает исправление: воспользоваться макросом pin! или Box::pin. std::pin::pin! (или tokio::pin!) закрепляет футуру на месте, в стеке. Теперь один и тот же дедлайн общий для всех кругов цикла:

use std::pin::pin;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    let (tx, mut rx) = mpsc::channel(8);
    tokio::spawn(async move {
        for order in 1..=3 {
            tx.send(order).await.unwrap();
            sleep(Duration::from_millis(10)).await;
        }
        sleep(Duration::from_secs(5)).await; // потом в магазине становится тихо
    });

    let mut deadline = pin!(sleep(Duration::from_millis(300)));

    loop {
        tokio::select! {
            _ = &mut deadline => {
                println!("магазин закрывается");
                break;
            }
            Some(order) = rx.recv() => println!("получен заказ №{order}"),
        }
    }
}

Вывод:

получен заказ №1
получен заказ №2
получен заказ №3
магазин закрывается

▶ Попробуй сам

Если бы мы написали sleep(...) внутри select!, каждый новый заказ запускал бы свежий таймер на 300 мс, и загруженный магазин никогда бы не закрылся. Один дедлайн, закреплённый вне цикла, это исправляет.

С рекурсией дело в том, что компилятору нужно знать размер футуры, а футура, которая содержит саму себя, была бы бесконечно большой. Вызов countdown(n - 1).await внутри async fn countdown даёт:

error[E0733]: recursion in an async fn requires boxing
  |
  = note: a recursive `async fn` call must introduce indirection such as `Box::pin` to avoid an infinitely sized future

Ошибка говорит, что рекурсия в async fn требует упаковки в Box. Box::pin перемещает внутреннюю футуру в кучу и закрепляет её там. Внешняя футура теперь хранит только указатель, а у него фиксированный размер:

async fn countdown(n: u32) {
    if n == 0 {
        println!("пуск!");
        return;
    }
    println!("{n}...");
    Box::pin(countdown(n - 1)).await;
}

#[tokio::main]
async fn main() {
    countdown(3).await;
}

Вывод:

3...
2...
1...
пуск!

▶ Попробуй сам

Честно говоря, большинству людей этого достаточно: pin! — для футуры, которую опрашиваешь много раз, Box::pin — для хранения футур и рекурсии. Если хочешь полную историю, глава книги Rust из списка «Что почитать» объясняет её шаг за шагом.

async fn в трейтах и асинхронные замыкания

Начиная с Rust 1.75, async fn можно писать прямо в трейте. А с Rust 1.85 появились ещё и асинхронные замыкания, async |x| ..., с соответствующими трейтами AsyncFn, AsyncFnMut и AsyncFnOnce (асинхронные версии трейтов замыканий). Вот и то и другое в одном примере:

use std::time::Duration;
use tokio::time::sleep;

trait Notifier {
    async fn notify(&self, user: &str) -> String;
}

struct Email;

impl Notifier for Email {
    async fn notify(&self, user: &str) -> String {
        sleep(Duration::from_millis(10)).await; // как будто общаемся с почтовым сервером
        format!("письмо отправлено: {user}")
    }
}

async fn for_each_user(users: &[&str], job: impl AsyncFn(&str) -> String) {
    for user in users {
        println!("{}", job(user).await);
    }
}

#[tokio::main]
async fn main() {
    let email = Email;
    let users = ["Аня", "Бен"];

    for_each_user(&users, async |user: &str| email.notify(user).await).await;
}

Вывод:

письмо отправлено: Аня
письмо отправлено: Бен

▶ Попробуй сам

С async fn в трейтах есть одна загвоздка. Трейт не обещает, что возвращаемая футура реализует Send, а tokio::spawn нужны футуры с Send (они могут переходить между потоками). Поэтому запуск вызова через обобщённый N: Notifier не компилируется:

fn send_in_background<N: Notifier + Send + Sync + 'static>(notifier: N) {
    tokio::spawn(async move {
        let msg = notifier.notify("ana").await;
        println!("{msg}");
    });
}

error: future cannot be sent between threads safely
  |
7 |         let msg = notifier.notify("ana").await;
  |                   ^^^^^^^^^^^^^^^^^^^^^^ await occurs here on type `impl Future<Output = String>`, which is not `Send`
  |
help: `Send` can be made part of the associated future's guarantees for all implementations of `Notifier::notify`
  |
2 -     async fn notify(&self, user: &str) -> String;
2 +     fn notify(&self, user: &str) -> impl std::future::Future<Output = String> + Send;

Компилятор сообщает, что футуру нельзя безопасно передать между потоками, и предлагает добавить Send в сигнатуру метода трейта. Это и есть обычное исправление: записать метод трейта как fn ... -> impl Future<Output = String> + Send. Реализации при этом могут по-прежнему писать обычную async fn. И учти: трейты с async fn пока нельзя использовать как трейт-объекты dyn Notifier. Если это нужно, популярный крейт async-trait упакует футуры в Box за тебя.

Частые ошибки

Блокировать рантайм. std::thread::sleep, тяжёлые вычисления или большие чтения через std::fs внутри асинхронной задачи замораживают весь рабочий поток, и все остальные задачи на нём ждут. На странице основ был показан вариант со sleep. Тяжёлую для процессора или блокирующую работу отдавай в tokio::task::spawn_blocking: он выполняет её в отдельном пуле потоков, созданном как раз для этого:

fn count_primes(limit: u64) -> usize {
    (2..limit)
        .filter(|&n| (2..).take_while(|d| d * d <= n).all(|d| n % d != 0))
        .count()
}

#[tokio::main]
async fn main() {
    let handle = tokio::task::spawn_blocking(|| count_primes(50_000));

    println!("пока считаем, продолжаем отвечать на запросы...");

    let primes = handle.await.unwrap();
    println!("простых чисел меньше 50000: {primes}");
}

Вывод:

пока считаем, продолжаем отвечать на запросы...
простых чисел меньше 50000: 5133

▶ Попробуй сам

Держать охрану std::sync::Mutex через .await. Выглядит невинно:

let handle = tokio::spawn(async move {
    let mut count = visits.lock().unwrap();
    *count += 1;
    save_to_disk(*count).await;
});

error: future cannot be sent between threads safely
   |
   = help: within `{async block@src/main.rs:13:31: 13:41}`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, u32>`
note: future is not `Send` as this value is used across an await
   |
14 |         let mut count = visits.lock().unwrap();
   |             --------- has type `std::sync::MutexGuard<'_, u32>` which is not `Send`
15 |         *count += 1;
16 |         save_to_disk(*count).await;
   |                              ^^^^^ await occurs here, with `mut count` maybe used later

Компилятор объясняет, что MutexGuard не реализует Send, а используется через .await. Так он тебя защищает. Пока эта задача спит на .await, она всё ещё держала бы блокировку, и любая другая задача, которой нужна эта блокировка, заблокировала бы весь свой поток. Исправление — удалить охрану до .await, вынеся блокировку в отдельный блок:

let count = {
    let mut count = visits.lock().unwrap();
    *count += 1;
    *count
}; // охрана удаляется здесь
save_to_disk(count).await;

Если тебе действительно нужно держать блокировку во время ожидания, используй tokio::sync::Mutex: его lock().await ждёт, не блокируя поток. (Подробнее о блокировках — на странице Общее состояние.)

Забыть .await. Футуры ленивые, поэтому save_file(); без .await вообще ничего не делает. Компилятор предупредит тебя сообщением «futures do nothing unless you .await or poll them». Подробности — на странице Async/Await. И не игнорируй это предупреждение.

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

use std::time::Duration;
use tokio::time::sleep;

#[tokio::main]
async fn main() {
    tokio::spawn(async {
        sleep(Duration::from_millis(100)).await;
        println!("отчёт сохранён"); // никогда не напечатается!
    });

    println!("main закончила");
}

Вывод:

main закончила

▶ Попробуй сам

Отчёт так и не сохраняется. Если работа важна, сохрани JoinHandle и сделай для него .await (или положи задачи в JoinSet и дождись их все) до того, как main закончится.

Практика

1. Проверка цен с дедлайном

Четыре магазина отвечают на запрос цены через 20, 40, 300 и 60 миллисекунд. Спроси все четыре одновременно через JoinSet, но дай каждому магазину только 150 мс. Выведи пришедшие цены (отсортированные по названию магазина), а потом — сколько магазинов не успели. Подсказка: запускай в наборе timeout(limit, ask_shop(name, delay)). Каждая законченная задача тогда даёт Ok или Err(Elapsed), так что возвращай название магазина вместе с этим результатом.

Показать решение

use std::time::Duration;
use tokio::task::JoinSet;
use tokio::time::{sleep, timeout};

async fn ask_shop(delay_ms: u64) -> u32 {
    sleep(Duration::from_millis(delay_ms)).await;
    100 + delay_ms as u32 // выдуманная цена
}

#[tokio::main]
async fn main() {
    let shops = [("Corner Shop", 20), ("MegaMart", 40), ("SlowCo", 300), ("Bits & Bobs", 60)];
    let limit = Duration::from_millis(150);

    let mut set = JoinSet::new();
    for (name, delay) in shops {
        set.spawn(async move { (name, timeout(limit, ask_shop(delay)).await) });
    }

    let mut prices = Vec::new();
    let mut timed_out = 0;
    while let Some(res) = set.join_next().await {
        match res.unwrap() {
            (name, Ok(price)) => prices.push((name, price)),
            (_, Err(_)) => timed_out += 1,
        }
    }

    prices.sort();
    for (name, price) in prices {
        println!("{name}: {price}");
    }
    println!("не успели ответить: {timed_out}");
}

Вывод:

Bits & Bobs: 160
Corner Shop: 120
MegaMart: 140
не успели ответить: 1

▶ Попробуй сам

Названия магазинов остались английскими: список сортируется по буквам. Все магазины ждут одновременно, поэтому вся проверка занимает около 150 мс, как бы медленно ни работал SlowCo. Когда срабатывает его тайм-аут, его футура удаляется и не может продолжать работать в фоне.

2. Размеры папок с асинхронной рекурсией

У Folder есть имя, список размеров файлов в КБ и список подпапок. Напиши async fn total_size(folder: &Folder) -> u64, которая складывает размеры собственных файлов и размер каждой подпапки, вызывая саму себя для каждой из них. Представь, что каждой папке сначала нужно короткое асинхронное «чтение с диска», через tokio::task::yield_now().await. Подсказка: рекурсивный вызов нужно обернуть в Box::pin(...), иначе получишь ошибку E0733.

Показать решение

struct Folder {
    name: &'static str,
    files_kb: Vec<u64>,
    subfolders: Vec<Folder>,
}

async fn total_size(folder: &Folder) -> u64 {
    tokio::task::yield_now().await; // как будто читаем диск

    let mut total: u64 = folder.files_kb.iter().sum();
    for sub in &folder.subfolders {
        total += Box::pin(total_size(sub)).await;
    }
    println!("{}: {} КБ", folder.name, total);
    total
}

#[tokio::main]
async fn main() {
    let home = Folder {
        name: "home",
        files_kb: vec![4],
        subfolders: vec![
            Folder { name: "photos", files_kb: vec![2048, 1024], subfolders: vec![] },
            Folder {
                name: "code",
                files_kb: vec![12, 8],
                subfolders: vec![Folder { name: "target", files_kb: vec![500], subfolders: vec![] }],
            },
        ],
    };

    let total = total_size(&home).await;
    println!("итого: {total} КБ");
}

Вывод:

photos: 3072 КБ
target: 500 КБ
code: 520 КБ
home: 3596 КБ
итого: 3596 КБ

▶ Попробуй сам

Каждая папка печатается после всех своих подпапок, потому что сначала ей нужно дождаться их размеров. Box::pin кладёт футуру каждого вложенного вызова в кучу, так что внешняя футура сохраняет фиксированный размер.

3. Разобрать очередь, потом остановиться

Измени пример с остановкой так, чтобы перед отправкой сигнала остановки не нужен был sleep(100ms). Отправь три задания и сигнал остановки сразу, но сделай так, чтобы воркер всё равно выполнил все задания из очереди, прежде чем остановиться. Подсказка: без помощи select! выбирает случайную ветку, когда готовы обе. Добавь biased; и поставь ветку с заданиями первой.

Показать решение

use tokio::sync::{mpsc, oneshot};

#[tokio::main]
async fn main() {
    let (job_tx, mut job_rx) = mpsc::channel::<String>(10);
    let (stop_tx, mut stop_rx) = oneshot::channel::<()>();

    for job in ["уменьшить фото", "отправить письмо", "обновить поисковый индекс"] {
        job_tx.send(job.to_string()).await.unwrap();
    }
    stop_tx.send(()).unwrap();

    let worker = tokio::spawn(async move {
        let mut done = 0;
        loop {
            tokio::select! {
                biased;
                Some(job) = job_rx.recv() => {
                    println!("в работе: {job}");
                    done += 1;
                }
                _ = &mut stop_rx => {
                    println!("очередь пуста, останавливаемся");
                    break;
                }
            }
        }
        done
    });

    let done = worker.await.unwrap();
    println!("воркер выполнил заданий: {done}");
}

Вывод:

в работе: уменьшить фото
в работе: отправить письмо
в работе: обновить поисковый индекс
очередь пуста, останавливаемся
воркер выполнил заданий: 3

▶ Попробуй сам

С biased; макрос select! всегда сначала пробует ветку с заданиями, поэтому на сигнал остановки он смотрит, только когда у recv() ничего нет наготове. Без этого воркер мог бы остановиться после одного-двух заданий — а такая ошибка проявлялась бы лишь время от времени.

Итоги

• join! ждёт все футуры, try_join! останавливается на первой Err, а JoinSet запускает много задач и отдаёт результаты по мере завершения (сортируй их, если важен порядок).
• select! ждёт первую закончившуюся футуру. Остальные удаляются — так в асинхронном Rust работает отмена. Внутри цикла используй футуры, безопасные для отмены, и biased;, когда важен порядок.
• timeout сдаётся, если футура слишком медленная. interval тикает по ровному расписанию, sleep ждёт один раз.
• Стрим — это асинхронный итератор. Получатель канала работает похоже, а tokio-stream добавляет map, take и их друзей.
• Pin обещает, что футура не будет перемещаться в памяти. pin! нужен, чтобы опрашивать одну футуру много раз, а Box::pin — для хранения футур и асинхронной рекурсии.
• Не блокируй рантайм (spawn_blocking), не держи блокировку из std через .await и дожидайся задач, которые запускаешь.

Проверь себя

Проверь себя

1. Что происходит в tokio::select! с ветками, которые закончили не первыми?

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

2. Что возвращает tokio::time::timeout, когда время вышло?

Паникует Ok со значением по умолчанию Err(Elapsed), а внутренняя футура удаляется

3. Зачем нужен Box::pin, когда async fn вызывает саму себя?

Футура, которая содержит саму себя, была бы бесконечного размера Рекурсия в асинхронном коде вообще запрещена Чтобы функция выполнялась в другом потоке

4. В программе на tokio нужно выполнить 2 секунды тяжёлой работы процессора. Какой инструмент лучше всего?

Обернуть её в std::thread::sleep tokio::select! JoinSet с tokio::spawn tokio::task::spawn_blocking

Результат: 0 / 4

Что почитать

• Select — учебник Tokio (на английском): select! подробно — как работают ветки, что значит отмена и как использовать его в цикле.
• Streams — учебник Tokio (на английском): как использовать, адаптировать и создавать стримы с tokio-stream.
• A Closer Look at the Traits for Async — The Rust Book (на английском): спокойный пошаговый разбор Future, Pin, Unpin и Stream. В русском переводе книги этой главы пока нет.
• std::pin — стандартная библиотека Rust (на английском): официальное объяснение закрепления, Pin, Unpin и макроса pin!.
• tokio::time на docs.rs (на английском): sleep, timeout, interval и что происходит с пропущенными тиками.