Отмена и select!

Ты собрал рантайм и заглянул внутрь Tokio. Осталась семантика, та, на которой спотыкаются даже опытные. Что значит «отменить» async-задачу? В Rust ответ неожиданно простой и неожиданно опасный: отмена это drop. Никакого сигнала внутрь не приходит, задачу перестают опрашивать и роняют. Отсюда растёт понятие cancellation safety и большинство ловушек select!. Разберёмся, собрав комбинатор гонки своими руками.

Отмена это drop

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

У этого две стороны. Отмена бесплатна и сама освобождает все ресурсы футуры: деструкторы полей отработают. Но тут же кроется ловушка. Футуру можно уронить посреди работы, между двумя await, и весь прогресс, накопленный до этой точки и ещё не зафиксированный наружу, исчезает.

select! и гонка футур

Главный способ нарваться на отмену это select!: запустить несколько футур и дождаться первой готовой. Аналог Promise.race из инструментов конкурентности в JS, но с важной разницей: проигравшие футуры дропаются. Соберём минимальный select на две футуры, чтобы увидеть это в коде:

pub enum Either<L, R> { Left(L), Right(R) }

pub struct Select2<A, B> {
    a: Option<A>, // Option, чтобы забрать и уронить проигравшую
    b: Option<B>,
}

impl<A: Future, B: Future> Future for Select2<A, B> {
    type Output = Either<A::Output, B::Output>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        // SAFETY: проекция Pin на поля; futures a и b двигаем только через
        // Pin::new_unchecked и роняем лишь когда другая победила (drop футуры
        // всегда допустим).
        let this = unsafe { self.get_unchecked_mut() };

        if let Some(a) = this.a.as_mut() {
            let a_pin = unsafe { Pin::new_unchecked(a) };
            if let Poll::Ready(out) = a_pin.poll(cx) {
                this.b = None; // левая победила: роняем правую (отмена через drop)
                return Poll::Ready(Either::Left(out));
            }
        }

        if let Some(b) = this.b.as_mut() {
            let b_pin = unsafe { Pin::new_unchecked(b) };
            if let Poll::Ready(out) = b_pin.poll(cx) {
                this.a = None; // правая победила: роняем левую
                return Poll::Ready(Either::Right(out));
            }
        }

        Poll::Pending
    }
}

Строки this.b = None и this.a = None это и есть отмена. Победила одна, проигравшую немедленно роняем, на любом этапе её работы. Настоящий tokio::select! делает то же самое со всеми ветками, кроме выигравшей.

Cancellation safety на пальцах

Теперь главное понятие. Cancellation safety это свойство футуры: безопасно ли её ронять на точке await. Футура безопасна к отмене, если её drop не теряет уже сделанную работу. Опасна, если успела что-то взять из мира, но ещё не отдала наружу, и при drop это «что-то» пропадает.

Сделаем потерю осязаемой. Возьмём футуру, которая копит прогресс по шагу за poll, и стравим её в гонке с быстрой:

struct Progress {
    progress: Rc<Cell<u32>>, // прогресс виден снаружи даже после drop
    target: u32,
}

impl Future for Progress {
    type Output = u32;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<u32> {
        let current = self.progress.get();
        if current >= self.target {
            Poll::Ready(current)
        } else {
            self.progress.set(current + 1); // продвинулись на шаг
            // Waker не трогаем: пусть гонку двигает соперник, а мы честно «застрянем».
            let _ = cx;
            Poll::Pending
        }
    }
}

// Гонка: slow (target=100) против мгновенной. slow успеет продвинуться,
// но проиграет и будет дропнута на середине.
let reached = slow_progress.get();
assert!(reached > 0);    // успела поработать
assert!(reached < 100);  // но до конца не дошла, прогресс потерян

Прогресс застрял где-то между 0 и 100. Никто его не продолжит: футуры больше нет. Если бы Progress вынимала элементы из канала и теряла их при drop, это были бы потерянные сообщения. Это и есть цена отмены через drop, и за неё отвечаешь ты, а не компилятор.

Классическая ловушка select! в цикле

Самый частый баг в реальном коде выглядит так:

loop {
    tokio::select! {
        msg = socket.read_message() => handle(msg),
        _ = shutdown.recv() => break,
    }
}

Каждая итерация цикла создаёт read_message() заново. Если на этой итерации сработала ветка shutdown, то футура read_message дропается. И если read_message была не cancellation safe (успела прочитать из сокета половину сообщения в свой внутренний буфер, но ещё не вернула целое), эта половина теряется вместе с футурой. На следующей итерации читается уже хвост следующего сообщения, и протокол рассинхронизируется.

Поэтому документация Tokio честно помечает каждый async-метод: cancellation safe он или нет. tokio::sync::mpsc::Receiver::recv безопасен (если его дропнуть, сообщение остаётся в канале). А AsyncReadExt::read в общем случае нет. Правило простое: в ветках select! зови только cancellation safe операции, а если нужна небезопасная, выноси её состояние наружу цикла, чтобы оно пережило drop футуры.

Структурная конкурентность и таймауты

Раз отмена это drop, из неё естественно вырастают полезные инструменты.

Таймаут это select! между твоей работой и таймером: кто первый, тот и победил, проигравший дропается. tokio::time::timeout(dur, fut) это ровно гонка fut против sleep(dur); если победил таймер, fut отменяется через drop.

Структурная конкурентность это принцип «дочерние задачи не переживают родителя». Если функция запустила несколько подзадач, при её завершении (или отмене) они должны отмениться, а не утечь в фон осиротевшими. tokio::task::JoinSet поддерживает это: дроп набора отменяет все его задачи. Сравни с голым tokio::spawn, который, наоборот, отвязывает задачу: она живёт сама по себе, и чтобы её прервать снаружи, нужен AbortHandle, его abort() роняет футуру задачи на ближайшей точке приостановки.

Заметь сквозную мысль: и select!, и timeout, и JoinSet::drop, и AbortHandle::abort это одна и та же операция, drop футуры. Поняв drop как отмену, ты понял их все.

Что унести из урока

Отмена async-задачи в Rust это её drop: футуру перестают опрашивать и роняют, никакого сигнала внутрь не приходит. Это бесплатно и освобождает ресурсы, но роняет работу, накопленную до точки приостановки и не зафиксированную наружу. Cancellation safety это свойство «безопасно ли ронять футуру на await»; в ветках select! зови только безопасные операции, иначе словишь потерю частичного прогресса (классика: read в select!-цикле теряет полусчитанное сообщение). Таймауты, структурная конкурентность через JoinSet и AbortHandle::abort это всё надстройки над одной идеей drop как отмены.

Дальше последний урок блока: Stream как асинхронный итератор и каналы (mpsc, oneshot, broadcast, watch) с backpressure.

Домашка