TL;DR. Каждый async fn в Rust компилируется в enum-стейт-машину. Размер этой стейт-машины равен размеру самого толстого варианта, поэтому забытая через .await переменная на пару мегабайт превращается в утечку памяти, помноженную на число задач. Pin существует, чтобы запретить перемещать такие стейт-машины после первого poll, потому что внутри них живут указатели на собственные поля. select! молча теряет данные, если использовать в нём future без cancellation safety. И executor в Tokio, при всей его магии, концептуально умещается в сотню строк.
Содержание
Представь сервис на Tokio, который держит десять тысяч соединений и неожиданно начинает съедать восемь гигабайт памяти на ровном месте. В коде нет утечек, Arc всё считает корректно, valgrind молчит. Виновник — один безобидный async fn, в котором между двумя .await лежит массив на пару мегабайт. Компилятор честно положил его в стейт-машину, и теперь каждое из десяти тысяч соединений таскает за собой эту память. Чтобы такие истории перестали быть мистикой, надо перестать думать об async Rust как о чёрном ящике и разобрать три темы, которые в обычной жизни не встречаются: трансформацию async fn в стейт-машину, самоссылающиеся структуры и зачем Pin, и как на самом деле работает executor.
Под капотом async Rust устроен неожиданно просто и неожиданно жестоко: каждый твой async fn это сгенерированный компилятором enum, каждый .await это match по этому enum, а легендарный Pin, на который ругается половина туториалов, существует ровно потому, что без него вся эта конструкция разваливается на первом же mem::swap. В конце соберём свой собственный executor за двести строк, чтобы окончательно стало ясно, что Tokio это не магия.
async fn это синтаксический сахар над enum
Возьмём абсолютно ничего не делающую функцию:
async fn foo() -> u32 {
let a = bar().await;
let b = baz().await;
a + b
}
Если поставить cargo install cargo-expand и запустить cargo expand, можно увидеть реальный enum, в который компилятор разворачивает эту функцию. В упрощённом виде он выглядит так:
enum FooState {
Start,
WaitingOnBar { bar_fut: BarFuture },
WaitingOnBaz { a: u32, baz_fut: BazFuture },
Done,
}
Каждый .await это точка приостановки, и для каждой такой точки в enum появляется свой вариант с локальными переменными, которые должны пережить приостановку. poll у этого Future это огромный match, который смотрит текущее состояние, вызывает вложенный future, и если тот вернул Pending, сохраняет состояние и возвращает Pending наверх. Если Ready — переходит в следующий вариант enum.
Размер future и почему твой сервис ест память
Размер future равен размеру самого толстого варианта enum. Если у тебя в async fn живёт [u8; 1_000_000] через одну .await-точку, твой future весит мегабайт, и Box::pin его аллоцирует на каждом вызове. Переменные, которые не пересекают .await, в enum не попадают, поэтому иногда переписать код так, чтобы временные значения дропались до await, реально уменьшает память.
Проверить размер собственного future можно прямо в рантайме:
println!("{}", std::mem::size_of_val(&my_async_fn()));
А на nightly есть полезный флаг, который покажет все типы вместе с их размерами:
cargo +nightly rustc -- -Zprint-type-sizes
В выводе ищи строчки про async-блоки. Реальный случай из практики: у меня была корутина, которая держала в памяти String с подробным сообщением об ошибке через границу .await, и весила 8 КБ на задачу. Замена String на Box<str> срезала размер до 16 байт. На десяти тысячах задач это разница в восемьдесят мегабайт.
Самоссылающиеся структуры и зачем Pin
Теперь самое интересное. Что если внутри async fn есть такой код:
async fn weird() {
let s = String::from("hello");
let r = &s;
other().await;
println!("{}", r);
}
Ссылка r указывает на s, и обе живут в одном и том же варианте enum-а. То есть в сгенерированной структуре есть поле, которое указывает на другое поле той же структуры. Это и называется самоссылающейся структурой. Пока структура лежит на месте, всё работает. Но если её переместить в памяти (а в Rust перемещение это просто memcpy байтов), указатель внутри останется указывать на старый адрес, и ты получишь dangling pointer без единого unsafe в коде пользователя.
Решение могло быть таким: запретить перемещать future после первого poll. Именно это и делает Pin<P>. Важно понимать, что Pin это не магия и не часть компилятора — это обычный библиотечный тип. Вся гарантия держится на одном unsafe fn Pin::new_unchecked плюс автотрейте Unpin. Если ты как-то получил Pin<&mut T> безопасным путём, значит кто-то на стороне (компилятор для async-блоков, Box::pin, pin! макрос) заплатил за это unsafe-обязательство. Все методы Future::poll принимают Pin<&mut Self> именно для того, чтобы вынудить вызывающую сторону пройти через эту дверь.
Тут есть популярное недоразумение. Pin<&mut T> не запрещает мутировать T. Он запрещает только перемещать его. Если T: Unpin (а это автотрейт, который реализован почти для всего, кроме самоссылающихся future и нескольких специальных типов), Pin вообще ничего не даёт и сквозит насквозь. Поэтому Pin<&mut i32> это просто странно написанный &mut i32. А вот Pin<&mut SomeAsyncBlock> это уже настоящее ограничение.
Waker, или как future вообще узнаёт, что пора просыпаться
Если future вернул Pending, executor должен как-то понять, когда его снова poll-ить. Тупой вариант — поллить в цикле — съест CPU. Поэтому в Context, который передаётся в poll, лежит Waker. Это типизированный Arc-подобный объект с одним методом wake(), который говорит «эй, тот future, которому я принадлежу, готов двигаться дальше».
Кто вызывает wake()? Тот, кто знает, что данные появились. Если ты ждёшь сетевого пакета, wake() вызовет реактор поверх epoll/kqueue/io_uring, когда сокет станет читаемым. Если ты ждёшь таймера, его вызовет таймер-тред. Если ты ждёшь канала, его вызовет тот, кто отправил сообщение. Executor получает сигнал и кладёт future обратно в очередь готовых к опросу.
Cancellation safety, или почему select! иногда теряет данные
Прежде чем писать executor, упомяну ещё одну штуку, на которую регулярно наступают даже опытные. tokio::select! опрашивает несколько future параллельно и при первой готовности отбрасывает остальные. Отбрасывает буквально — дропает. Если ты внутри отброшенной ветки уже считал часть данных из сокета, эти данные исчезли вместе с future.
Каноничный пример, на котором горят люди:
let mut buf = [0u8; 1024];
tokio::select! {
result = socket.read_exact(&mut buf) => {
// обработать buf
}
_ = timeout => {
// таймаут
}
}
Если таймаут сработал, когда read_exact уже прочитал из сокета 500 байт, эти 500 байт улетели в небытие вместе с дропнутым future. Следующая попытка чтения вычитает «обрезанный» поток, протокол сломается, и понять, что произошло, по логам почти невозможно. Это называется отсутствием cancellation safety, и в документации Tokio для каждого метода честно написано, безопасен ли он в select!. read_exact — нет, recv у канала — да. Эта мелочь стоит людям недель отладки, и из синтаксиса select! она вообще никак не видна.
Свой executor за двести строк
Соберём минимальный однопоточный executor. Без I/O, только чтобы стало ясно, как крутится механизм.
use std::collections::VecDeque;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Wake, Waker};
struct Task {
future: Mutex<Pin<Box<dyn Future<Output = ()> + Send>>>,
executor: Arc<Executor>,
}
impl Wake for Task {
fn wake(self: Arc<Self>) {
self.executor.queue.lock().unwrap().push_back(self.clone());
}
}
struct Executor {
queue: Mutex<VecDeque<Arc<Task>>>,
}
impl Executor {
fn new() -> Arc<Self> {
Arc::new(Self { queue: Mutex::new(VecDeque::new()) })
}
fn spawn(self: &Arc<Self>, fut: impl Future<Output = ()> + Send + 'static) {
let task = Arc::new(Task {
future: Mutex::new(Box::pin(fut)),
executor: self.clone(),
});
self.queue.lock().unwrap().push_back(task);
}
fn run(&self) {
loop {
let task = match self.queue.lock().unwrap().pop_front() {
Some(t) => t,
None => break,
};
let waker = Waker::from(task.clone());
let mut cx = Context::from_waker(&waker);
let mut fut = task.future.lock().unwrap();
let _ = fut.as_mut().poll(&mut cx);
}
}
}
Это умышленно упрощённая модель. В реальном Tokio очередь lock-free, future внутри задачи защищён хитрой схемой состояний, чтобы wake() изнутри poll не привёл к дедлоку, плюс отдельный механизм для I/O-реактора и пула потоков с work-stealing. Но концептуальное ядро именно такое: spawn кладёт future в очередь, run берёт по одному, делает poll, и если future вернул Pending, он просто выпадает из очереди до следующего wake(). wake() кладёт его обратно. Всё.
Если добавить сюда mio для epoll и таймер-колесо, получится примитивный аналог async-std. Если добавить ещё несколько потоков и стащить идею work-stealing у Go, получится почти Tokio. Если добавить io_uring, получится glommio или monoio.
Бонус: async fn в трейтах, и почему это было больно десять лет
С 2015 года в Rust ждали возможность писать async fn в трейтах. В стабильной версии это появилось только в конце 2023-го (AFIT, async fn in trait), и причина задержки прямо вытекает из всего написанного выше. Когда у тебя в трейте есть async fn foo(&self) -> u32, компилятор должен вернуть наружу какой-то конкретный тип future. Но этот тип у каждой реализации трейта свой, размер у него свой, и для dyn Trait это превращается в кошмар: vtable не может заранее знать размер возвращаемого future, потому что разные реализации генерируют разные стейт-машины.
Решение в стабильной версии умышленно неполное: async fn в трейтах работает для impl Trait, но не для dyn Trait. Для dyn приходится явно писать fn foo(&self) -> Pin<Box<dyn Future<Output = u32> + '_>> и платить за бокс. Это не косметическая придирка — на этом сейчас стоит весь дизайн трейт-объектов в async-экосистеме, и крейт async-trait десять лет существовал именно потому, что он автоматизировал это боксирование.
Что с этим делать сегодня
Если ты пишешь обычный сервис на Tokio, можно расслабиться. Но как только начинаются нестандартные вещи — свой Stream, ручной Future, прокидывание данных между задачами через select! — стоит держать в голове четыре правила:
Размер future имеет значение. Прогоняй
-Zprint-type-sizesхотя бы раз перед релизом, иBox::pinдля большого future это не лень, а оптимизация..awaitэто граница дляSend. Если планируешьtokio::spawn, всё, что живёт через.await, должно бытьSend, и если компилятор ругается на!Sendfuture, виноват, скорее всего,RcилиRefCell, который зацепился через границуawait.select!без cancellation safety это мина. Перед использованием чего-либо вselect!смотри в доке cancellation safety, и если её там нет — считай, что небезопасно.cargo expandтвой друг. Если не понимаешь, что компилятор сделал с твоим async-блоком, не гадай, посмотри глазами.
Финальная мысль
Когда в очередной раз увидишь в стеке трейса Pin<&mut dyn Future> и Poll::Pending, не закрывай терминал. Это компилятор показывает, что между твоим представлением о «функции, которая просто ждёт» и его представлением — сгенерированной стейт-машиной с самоссылающимися полями — никакого зазора нет, всё сходится по битам. И ровно поэтому async Rust способен держать миллионы соединений в одном процессе без потери производительности, чего другие языки добиваются только ценой отдельного рантайма со своим сборщиком мусора и собственными зелёными потоками.
Люблю Rust, пишу на нём и разбираю код так, чтобы сложные вещи становились понятнее — подписывайтесь: t.me/rust_code. А здесь я собрал репо с роадмэпом по Rust, для входа в мой любимый язык, надеюсь, будет полезно.
Спасибо за внимание, пишите в комментах ваши замечания!