Средний

17 мин

141

Если попросить среднего Rust-разработчика объяснить, что делает .await, в ответ обычно звучит что-то про «приостанавливает выполнение, пока не придут данные». Это верно ровно настолько же, насколько «компьютер думает» объясняет работу процессора. На самом деле за .await не стоит ни потока, ни приостановки в привычном смысле. Стоит обычный enum и вызов функции по указателю.

Я полгода писал async-код на tokio, искренне считая рантайм чернм ящиком, в который лучше не лезть. Сломался этот настрой в тот день, когда у меня в проде намертво зависла одна задача: по логам она обязана была проснуться, но не просыпалась. Я потерял на ней вечер, а причина оказалась в одной строчке про Waker, к которой мы ещё вернёмся. Тогда я плюнул и за пару вечеров написал свой рантайм с нуля - и оказалось, что весь фундамент умещается в голове за один присест и примерно в 200 строк кода. После этого и тот баг стал очевидным, и исходники tokio начали читаться как книга, а не как заклинание.

В этой статье мы соберём работающий async-рантайм на голой std, без единой сторонней зависимости. Будет два полных компилируемых примера: исполнитель, который крутит сто тысяч задач в одном потоке, и async-эхо-сервер на epoll, держащий сотни соединений тем же одним потоком. По дороге станет понятно, откуда в сигнатурах берутся Pin, Send и 'static, почему Waker сделан так странно, и чем именно tokio отличается от нашей игрушки.

TL;DR. async fn компилируется в машину состояний (enum). Рантайм это три кубика: Future умеет делать шаг и говорить «готово» или «не сейчас»; исполнитель держит очередь готовых задач и опрашивает их; реактор спит на epoll_wait и будит задачи, когда ОС сообщает о готовности сокета. Связывает всё Waker - колбэк «верни мою задачу в очередь». Никакого опроса в цикле, никакого busy-wait: спящая задача стоит почти ноль.

Статья рассчитана на тех, кто уже пишет на Rust и пользуется async/.await, но хочет понять, что под капотом. Глубокого знания tokio не требуется, наоборот, к концу он перестанет быть чёрным ящиком.

Зачем вообще нужен рантайм

Классическая модель «поток на соединение» упирается в стоимость потока. Стек по умолчанию это 8 МБ виртуальной памяти на поток в Linux; даже если резидентно занято мало, на 100 тысячах потоков вы упираетесь в лимиты, а планировщик ядра захлёбывается на переключениях контекста. Async переносит многозадачность в пространство пользователя: несколько рабочих потоков кооперативно жонглируют десятками и сотнями тысяч задач, и переключение между задачами это не syscall, а обычный возврат из функции.

Цифры, ради которых всё затевается. Пустая async-задача в tokio занимает порядка сотни байт плюс размер её машины состояний. Поэтому миллион одновременных задач реально живёт в сотнях мегабайт, а не в терабайтах под стеки. Стоимость спящей задачи близка к нулю, потому что её никто не опрашивает, пока не произошло событие.

Грубое сравнение двух подходов на 100 000 одновременных «преимущественно ждущих» клиентов:

Цифры приблизительные и зависят от настроек, но порядок именно такой: разница не в проценты, а в тысячи раз по памяти и на порядок по стоимости переключения.

Чтобы такая схема работала, нужны три сущности, и дальше мы напишем каждую:

• Future - единица работы, которую можно опросить (poll) и которая отвечает «готово, вот результат» либо «не сейчас, разбужу позже».

• Исполнитель (executor) - планировщик с очередью готовых задач.

• Реактор (reactor) - мост к ОС, который через epoll/kqueue/IOCP ждёт готовности дескрипторов и будит задачи.

async fn это enum: доказательство на пальцах

Самое живучее заблуждение про async в Rust - что под капотом там зелёные потоки со своими стеками. Никаких потоков нет вообще. Возьмём функцию:

async fn handle(mut sock: TcpStream) -> usize {
    let mut buf = [0u8; 1024];
    let n = sock.read(&mut buf).await;   // точка приостановки №1
    process(&buf[..n]);
    n
}

Компилятор не запускает её в отдельном стеке. Он генерирует тип-перечисление, где каждый .await это отдельное состояние, а всё, что должно пережить приостановку, складывается в поля:

enum HandleFuture {
    Start { sock: TcpStream },
    AwaitingRead { sock: TcpStream, buf: [u8; 1024], read: ReadFuture },
    Done,
}

poll это match по текущему состоянию: сделать шаг, при .await сохранить контекст в поле и вернуть Pending, при следующем вызове прыгнуть в нужную ветку. Размер future известен на этапе компиляции, аллокаций под стек нет вообще. Размер футуры это сумма самого «толстого» одновременно живого состояния, отсюда и знаменитая боль с гигантскими футурами, которые clippy советует боксировать. Это не абстракция: соберите проект с -Z print-type-sizes, и компилятор честно покажет размер вашей футуры в байтах. Конкретные числа зависят от версии rustc и оптимизаций, но картина будет примерно такая:

print-type-size type: `HandleFuture`: 1040 bytes, alignment: 8 bytes
print-type-size     variant `AwaitingRead`: 1040 bytes
print-type-size         field `.buf`: 1024 bytes
print-type-size         field `.sock`: 8 bytes
print-type-size         field `.read`: 8 bytes
print-type-size     variant `Start`: 8 bytes
print-type-size     variant `Done`: 0 bytes

Грубо говоря, вес футуры это в основном наш buf на 1024 байта, который обязан пережить .await, плюс мелочь под остальные поля. Каждый локальный массив, живущий через точку приостановки, целиком ложится в future. Отсюда практическое правило: не держите большие буферы на стеке через .await, иначе размер задачи раздувается, а вместе с ним и стоимость её перемещения между состояниями.

Из машины состояний напрямую вытекает Pin. В состоянии AwaitingRead вложенный read может держать ссылку на buf, который лежит в той же структуре. Это самоссылающийся тип: сдвинете его в памяти - внутренняя ссылка станет висячей. Pin это типобезопасное обещание компилятору «этот объект больше не переедет». Поэтому poll принимает Pin<&mut Self>, а не &mut Self. Не было бы самоссылок - не было бы и Pin.

Кубик 1: Future и его единственный метод

Весь трейт:

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

pub enum Poll<T> { Ready(T), Pending }

Контракт жёсткий и важный: если poll вернул Pending, future обязан был сохранить Waker из cx и позже дёрнуть его, когда появится шанс продвинуться. Нарушите контракт - задачу больше никто никогда не опросит, и она зависнет навсегда. Это причина доброй половины «подвисших» async-багов в ручном коде.

Кубик 2: Waker, и почему он выглядит как ассемблер

Waker это колбэк «положи мою задачу обратно в очередь исполнителя». Казалось бы, тут хватило бы Box<dyn Fn()>. Но стандартная библиотека делает Waker руками, как самодельный trait-объект: указатель на данные плюс таблица из четырёх функций.

pub struct RawWaker {
    data: *const (),
    vtable: &'static RawWakerVTable,
}

pub struct RawWakerVTable {
    clone: unsafe fn(*const ()) -> RawWaker,
    wake: unsafe fn(*const ()),
    wake_by_ref: unsafe fn(*const ()),
    drop: unsafe fn(*const ()),
}

Зачем так сложно? Затем, что один и тот же тип Waker должен обслуживать совершенно разные рантаймы: наш однопоточный исполнитель, многопоточный tokio с work stealing, block_on, который вообще паркует текущий поток. Каждый подставляет свою vtable и свои данные, а код future остаётся неизменным: он просто зовёт waker.wake() и не знает, что за этим стоит. Это ручная реализация полиморфизма без аллокаций там, где их можно избежать - ровно та работа, которую обычно прячет рантайм.

Наш Waker хранит идентификатор задачи и канал в очередь исполнителя. Дёрнули wake - идентификатор полетел в очередь:

fn make_waker(task_id: usize, queue: Sender<usize>) -> Waker {
    let data = Arc::new((task_id, queue));

    unsafe fn clone(ptr: *const ()) -> RawWaker {
        let arc = Arc::<(usize, Sender<usize>)>::from_raw(ptr as *const _);
        let cloned = arc.clone();
        std::mem::forget(arc);
        RawWaker::new(Arc::into_raw(cloned) as *const (), &VTABLE)
    }
    unsafe fn wake(ptr: *const ()) {
        let arc = Arc::<(usize, Sender<usize>)>::from_raw(ptr as *const _);
        arc.1.send(arc.0).ok();
    }
    unsafe fn wake_by_ref(ptr: *const ()) {
        let arc = Arc::<(usize, Sender<usize>)>::from_raw(ptr as *const _);
        arc.1.send(arc.0).ok();
        std::mem::forget(arc);
    }
    unsafe fn drop_fn(ptr: *const ()) {
        drop(Arc::<(usize, Sender<usize>)>::from_raw(ptr as *const _));
    }
    static VTABLE: RawWakerVTable =
        RawWakerVTable::new(clone, wake, wake_by_ref, drop_fn);

    let raw = RawWaker::new(Arc::into_raw(data) as *const (), &VTABLE);
    unsafe { Waker::from_raw(raw) }
}

Здесь много возни со счётчиком ссылок Arc и unsafe - это цена за то, что обычно делает за вас рантайм. Логика же тривиальна: разбудили - отправили id задачи в очередь. Один раз, ровно когда надо.

Кубик 3: исполнитель в десяток строк

Исполнитель хранит задачи и крутит цикл. Очередь готовых это канал с идентификаторами.

type BoxFuture = Pin<Box<dyn Future<Output = ()>>>;

struct Executor {
    tasks: HashMap<usize, BoxFuture>,
    ready_rx: Receiver<usize>,
    ready_tx: Sender<usize>,
    next_id: usize,
}

impl Executor {
    fn spawn(&mut self, fut: impl Future<Output = ()> + 'static) {
        let id = self.next_id;
        self.next_id += 1;
        self.tasks.insert(id, Box::pin(fut));
        self.ready_tx.send(id).unwrap();   // новая задача сразу готова
    }

    fn run(&mut self) {
        while let Ok(id) = self.ready_rx.recv() {
            let Some(task) = self.tasks.get_mut(&id) else { continue };
            let waker = make_waker(id, self.ready_tx.clone());
            let mut cx = Context::from_waker(&waker);
            if task.as_mut().poll(&mut cx).is_ready() {
                self.tasks.remove(&id);    // завершилась — выкидываем
            }
            // Pending: ничего не делаем, задача сама вернётся через waker
        }
    }
}

Вот и весь планировщик. Достали id из очереди, опросили задачу, при Ready выкинули. При Pending ничего не делаем: задача уже сохранила свой Waker и вернётся в очередь сама, когда её разбудят.

Здесь же прячется коварный баг, на котором горят все, кто пишет Future руками. Мы создаём свежий Waker на каждый poll. Контракт обязывает future при каждом опросе запоминать именно последний полученный Waker и забывать старый. Если закэшировать первый и звонить потом в него, то после миграции задачи между потоками (в многопоточном рантайме) пробуждение уйдёт в никуда. Дебажится это адски: задача по всем признакам должна была проснуться, но спит. Это ровно тот баг из начала статьи, на котором я потерял вечер: самописный Future кэшировал первый Waker, а tokio после work stealing перенёс задачу на другой поток и ждал звонка в новый. Правило простое - на каждом poll обновляй сохранённый waker через cx.waker().clone(), и никогда не предполагай, что Waker между опросами тот же самый.

Кубик 4: реактор и единственная настоящая блокировка

Пока что некому звонить в Waker, когда сокет готов. Этим занимается реактор - прослойка над epoll. Он регистрирует дескрипторы, спит на epoll_wait, а на событии будит соответствующие задачи.

struct Reactor {
    epoll_fd: RawFd,
    wakers: Mutex<HashMap<RawFd, Waker>>,   // fd -> кого будить
}

impl Reactor {
    fn register(&self, fd: RawFd, waker: Waker) {
        self.wakers.lock().unwrap().insert(fd, waker);
        epoll_add(self.epoll_fd, fd);       // EPOLL_CTL_ADD, EPOLLONESHOT
    }

    fn run(&self) {                          // в отдельном потоке
        let mut events = [epoll_event::default(); 1024];
        loop {
            let n = epoll_wait(self.epoll_fd, &mut events, -1); // спим здесь
            let mut wakers = self.wakers.lock().unwrap();
            for ev in &events[..n] {
                if let Some(w) = wakers.remove(&ev.fd()) {
                    w.wake();                // будим задачу, ждавшую этот fd
                }
            }
        }
    }
}

epoll_wait с таймаутом -1 это единственная точка во всём рантайме, где поток реально блокируется и отдаёт управление ядру. Всё остальное время потоки либо выполняют poll, либо спят на пустой очереди. Вот почему миллион ждущих соединений почти не ест CPU: мы не опрашиваем их в цикле, мы спим на одном epoll_wait и просыпаемся строго на событиях.

IO-future, связывающий сокет с реактором, выглядит так:

impl Future for ReadReady<'_> {
    type Output = ();
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        match self.socket.try_read() {           // неблокирующее чтение
            Ok(_) => Poll::Ready(()),
            Err(e) if e.kind() == WouldBlock => {
                REACTOR.register(self.socket.fd(), cx.waker().clone());
                Poll::Pending
            }
            Err(e) => panic!("io error: {e}"),
        }
    }
}

Соберём весь путь в цепочку. Future читает сокет, получает WouldBlock, кладёт Waker в реактор, возвращает Pending. Исполнитель без готовых задач засыпает на recv. Реактор спит на epoll_wait. Пришли данные - ядро будит реактор, реактор зовёт waker.wake(), id задачи летит в очередь, исполнитель просыпается и опрашивает future снова, на этот раз чтение проходит. Круг замкнулся, и нигде не было ни одного лишнего потока.

Вся картина одной схемой

Прежде чем писать код целиком, зафиксируем цикл, который мы только что собрали из кусков. Стрелки это и есть весь async:

         spawn(future)
              │
              ▼
      ┌───────────────┐   poll()    ┌──────────────┐
      │  Исполнитель  │ ──────────▶ │    Future    │
      │ (очередь id)  │             │ (машина сост)│
      └───────────────┘ ◀────────── └──────────────┘
              ▲   send(id)   Poll::Pending │
              │              + сохранить    │ cx.waker().clone()
      wake()  │                Waker        ▼
      ┌───────────────┐  epoll_wait  ┌──────────────┐
      │    Реактор    │ ───спим───▶  │  ядро ОС     │
      │ fd -> Waker   │ ◀──событие── │   (epoll)    │
      └───────────────┘              └──────────────┘

Слева направо: исполнитель опрашивает future; та возвращает Pending и оставляет Waker в реакторе; реактор спит на epoll_wait; ядро сообщает о готовности; реактор зовёт wake(); id задачи возвращается в очередь исполнителя. Один поток исполнителя, один поток реактора, и между ними бегают только идентификаторы задач и пробуждения.

Проверьте себя на этой схеме: что произойдёт, если poll вернул Pending, но забыл сохранить Waker? Правильный ответ - ничего и никогда. Реактор не узнает, кого будить, исполнитель не получит id в очередь, и задача зависнет навсегда, не потребляя при этом ни процента CPU. Именно поэтому «сохрани Waker» это не рекомендация, а единственное, что вообще заставляет всю машину крутиться.

Собираем рабочий пример целиком

Хватит псевдокода. Вот полный файл, который компилируется на стабильном Rust без зависимостей и крутит 100 тысяч таймеров в одном потоке. Таймер реализован честно: future при первом опросе отдаёт Waker фоновому потоку-таймеру, который через нужное время звонит в него. Скопируйте в main.rs и запустите cargo run --release. Листинг длинный, но это весь рантайм целиком; читать удобно по секциям-комментариям внутри: сначала Waker, затем таймер, фоновый поток и в конце исполнитель.

use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::mpsc::{channel, Receiver, Sender};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker};
use std::thread;
use std::time::{Duration, Instant};

// --- Waker через RawWakerVTable ---
fn make_waker(id: usize, q: Sender<usize>) -> Waker {
    type Data = (usize, Sender<usize>);
    unsafe fn clone(p: *const ()) -> RawWaker {
        let a = Arc::<Data>::from_raw(p as *const Data);
        let c = a.clone();
        std::mem::forget(a);
        RawWaker::new(Arc::into_raw(c) as *const (), &VT)
    }
    unsafe fn wake(p: *const ()) {
        let a = Arc::<Data>::from_raw(p as *const Data);
        let _ = a.1.send(a.0);
    }
    unsafe fn wake_ref(p: *const ()) {
        let a = Arc::<Data>::from_raw(p as *const Data);
        let _ = a.1.send(a.0);
        std::mem::forget(a);
    }
    unsafe fn drop_d(p: *const ()) { drop(Arc::<Data>::from_raw(p as *const Data)); }
    static VT: RawWakerVTable = RawWakerVTable::new(clone, wake, wake_ref, drop_d);
    let arc: Arc<Data> = Arc::new((id, q));
    let raw = RawWaker::new(Arc::into_raw(arc) as *const (), &VT);
    unsafe { Waker::from_raw(raw) }
}

// --- Future-таймер: будит себя через фоновый поток ---
struct Timer { when: Instant, registered: bool, shared: Arc<TimerThread> }

impl Future for Timer {
    type Output = ();
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        if Instant::now() >= self.when { return Poll::Ready(()); }
        if !self.registered {
            self.registered = true;
            self.shared.add(self.when, cx.waker().clone());
        }
        Poll::Pending
    }
}

// Один фоновый поток обслуживает все таймеры
struct TimerThread { tx: Sender<(Instant, Waker)> }
impl TimerThread {
    fn start() -> Arc<Self> {
        let (tx, rx) = channel::<(Instant, Waker)>();
        thread::spawn(move || {
            let mut pending: Vec<(Instant, Waker)> = Vec::new();
            loop {
                while let Ok(item) = rx.try_recv() { pending.push(item); }
                let now = Instant::now();
                pending.retain(|(when, w)| {
                    if now >= *when { w.wake_by_ref(); false } else { true }
                });
                thread::sleep(Duration::from_millis(1));
            }
        });
        Arc::new(TimerThread { tx })
    }
    fn add(&self, when: Instant, w: Waker) { let _ = self.tx.send((when, w)); }
}

// --- Исполнитель ---
struct Executor {
    tasks: HashMap<usize, Pin<Box<dyn Future<Output = ()>>>>,
    rx: Receiver<usize>,
    tx: Sender<usize>,
    next: usize,
}
impl Executor {
    fn new() -> Self {
        let (tx, rx) = channel();
        Executor { tasks: HashMap::new(), rx, tx, next: 0 }
    }
    fn spawn(&mut self, f: impl Future<Output = ()> + 'static) {
        let id = self.next; self.next += 1;
        self.tasks.insert(id, Box::pin(f));
        let _ = self.tx.send(id);
    }
    fn run(&mut self) {
        while !self.tasks.is_empty() {
            let id = self.rx.recv().unwrap();
            let Some(t) = self.tasks.get_mut(&id) else { continue };
            let w = make_waker(id, self.tx.clone());
            let mut cx = Context::from_waker(&w);
            if t.as_mut().poll(&mut cx).is_ready() { self.tasks.remove(&id); }
        }
    }
}

fn main() {
    let timer = TimerThread::start();
    let mut ex = Executor::new();
    let counter = Arc::new(Mutex::new(0usize));

    let n = 100_000;
    for i in 0..n {
        let timer = timer.clone();
        let counter = counter.clone();
        ex.spawn(async move {
            let ms = 50 + (i % 100) as u64;     // 50..150 мс
            Timer { when: Instant::now() + Duration::from_millis(ms),
                    registered: false, shared: timer }.await;
            *counter.lock().unwrap() += 1;
        });
    }

    let start = Instant::now();
    ex.run();
    println!("{} задач завершено за {:?} в ОДНОМ потоке исполнителя",
             *counter.lock().unwrap(), start.elapsed());
}

Запустите и посмотрите на вывод: сто тысяч таймеров с разбросом 50…150 мс завершаются за время порядка верхней границы их сна (то есть доли секунды, а не часы), и всё это в одном потоке исполнителя плюс один поток-таймер. Память держится в единицах десятков мегабайт. Для сравнения попробуйте честно завести 100 тысяч thread::spawn со sleep - на типичной Linux-машине вы упрётесь в лимит потоков или в гигабайты под стеки задолго до этого. Точные цифры замерьте на своём железе, но соотношение будет именно таким, и это и есть весь смысл async, потроганный руками.

От таймера к настоящему серверу: async-эхо на epoll

Таймер это хорошо, но настоящий async начинается с сокетов. Заменим источник пробуждений: вместо потока-таймера поставим реактор на epoll, а задачами сделаем обслуживание TCP-соединений. Ниже минимальный эхо-сервер: каждое соединение это отдельная задача, все они крутятся в одном потоке исполнителя, и ни один read/write не блокирует поток. Для краткости опущены use и хелперы epoll_add/read_nb/accept_nb - это тонкие обёртки над libc-вызовами (epoll_ctl, read, accept4), и пишутся они в десяток строк каждая. Здесь же оставляю только async-логику, чтобы скелет был виден целиком.

// Реактор: один epoll на всех, fd -> Waker ждущей задачи
static REACTOR: OnceLock<Reactor> = OnceLock::new();

struct Reactor {
    epoll: RawFd,
    wakers: Mutex<HashMap<RawFd, Waker>>,
}

impl Reactor {
    fn get() -> &'static Reactor { REACTOR.get().unwrap() }

    // future зовёт это, когда получил WouldBlock
    fn wait(&self, fd: RawFd, interest: u32, waker: Waker) {
        self.wakers.lock().unwrap().insert(fd, waker);
        epoll_mod(self.epoll, fd, interest | EPOLLONESHOT); // ADD или MOD
    }

    fn run(&self) { // крутится в фоновом потоке
        let mut evs = [epoll_event::zeroed(); 1024];
        loop {
            let n = epoll_wait(self.epoll, &mut evs, -1);   // спим тут
            let mut map = self.wakers.lock().unwrap();
            for ev in &evs[..n] {
                if let Some(w) = map.remove(&(ev.u64 as RawFd)) {
                    w.wake();                                // будим задачу
                }
            }
        }
    }
}

// Future готовности сокета: poll возвращает Ready, когда fd готов
struct Ready { fd: RawFd, interest: u32, armed: bool }

impl Future for Ready {
    type Output = ();
    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        if self.armed {            // нас уже разбудил реактор — fd готов
            self.armed = false;
            return Poll::Ready(());
        }
        self.armed = true;
        Reactor::get().wait(self.fd, self.interest, cx.waker().clone());
        Poll::Pending
    }
}

// Обслуживание одного соединения как async-задача
async fn serve(fd: RawFd) {
    let mut buf = [0u8; 4096];
    loop {
        // ждём, пока в сокете появятся данные
        Ready { fd, interest: EPOLLIN, armed: false }.await;
        let n = match read_nb(fd, &mut buf) {
            Ok(0) => break,                 // клиент закрыл соединение
            Ok(n) => n,
            Err(WouldBlock) => continue,    // ложное пробуждение — ждём ещё
            Err(_) => break,
        };
        // эхо обратно (упрощённо, без досыла частичных записей)
        Ready { fd, interest: EPOLLOUT, armed: false }.await;
        let _ = write_nb(fd, &buf[..n]);
    }
    close(fd);
}

fn main() {
    REACTOR.set(Reactor::new()).ok();
    thread::spawn(|| Reactor::get().run());

    let listener = listen_nonblocking("127.0.0.1:8080");
    let mut ex = Executor::new();

    // задача-акцептор: принимает соединения и спавнит на каждое serve()
    ex.spawn(async move {
        loop {
            Ready { fd: listener, interest: EPOLLIN, armed: false }.await;
            while let Ok(conn) = accept_nb(listener) {
                set_nonblocking(conn);
                SPAWNER.spawn(serve(conn));  // новая задача на соединение
            }
        }
    });

    ex.run();
}

Один нюанс честности: чтобы акцептор мог спавнить задачи прямо во время работы исполнителя, очередь spawn выносится в общий Spawner (по сути тот же Sender плюс хранилище задач за Mutex), а ex.run() тянет из него. Это пара строк поверх нашего Executor и единственное отличие от листинга с таймерами. Проверяется одной строкой: cargo run --release, затем в другом терминале nc 127.0.0.1 8080 и печатаете что угодно - сервер возвращает эхо. Откройте сто параллельных nc - все обслуживаются одним потоком исполнителя плюс один поток реактора. Это уже не игрушка-таймер, а полноценная архитектура tokio в миниатюре: акцептор, задача на соединение, реактор на epoll, кооперативное переключение на каждом .await.

Обратите внимание на EPOLLONESHOT и повторную регистрацию wait на каждом круге цикла. Без ONESHOT уровневый epoll будет будить нас в цикле, пока в сокете есть данные, и мы получим busy-loop. С ONESHOT дескриптор после события «снимается» с наблюдения, и задача сама перевзводит его, когда снова готова ждать. Это ровно та грань, об которую спотыкаются почти все, кто впервые пишет реактор руками.

Чем tokio отличается от нашей игрушки

Скелет готов, и в нём видны все идеи tokio: машина состояний, Waker на vtable, очередь исполнителя, реактор на epoll. Дальше tokio навешивает слои, каждый из которых это ответ на конкретную боль, которую в игрушке не успеваешь почувствовать.

Многопоточность и work stealing. Наш исполнитель однопоточный. Tokio поднимает пул потоков по числу ядер, у каждого своя локальная lock-free очередь плюс глобальная. Простаивающий поток ворует задачи из хвоста очереди соседа. Отсюда требование Send на спавнящиеся future: задача может проснуться не на том потоке, где заснула, поэтому всё, что она тащит через .await, обязано безопасно переезжать между потоками. Половина строгости компилятора вокруг async это плата именно за это.

Блокирующие операции. Если задача внутри .await-цепочки сделает блокирующий syscall или тяжёлый счёт, она застопорит весь рабочий поток вместе со всеми его задачами. Tokio выносит такое в отдельный пул через spawn_blocking. У нас этой защиты нет вовсе.

Таймеры. Наш таймер на Vec с линейным сканом - это O(n) на каждый тик, на больших объёмах смерть. Tokio использует иерархическое колесо таймеров (hierarchical timing wheel) с почти константной вставкой.

IO-драйвер. У нас реактор это HashMap под мьютексом и отдельный поток. Tokio интегрирует драйвер с парковкой воркеров: поток, которому нечего делать, сам идёт крутить epoll_wait, экономя лишний поток и переключения. Плюс на свежих ядрах есть бэкенд на io_uring.

Суть в том, что концепции те же. Сложность tokio не в идеях, а в том, чтобы сделать их быстрыми, корректными в многопоточке и устойчивыми под нагрузкой.

Что обычно спрашивают в комментариях

Чтобы не разводить холивар, отвечу заранее на предсказуемое.

«Это же не настоящий рантайм, у тебя busy-loop в таймере». В демо-таймере да, фоновый поток крутится с sleep(1ms) ради простоты примера. В нормальной реализации таймеры регистрируются как таймерфд в том же epoll, и никакого опроса нет. Сам исполнитель уже честный: он спит на recv и просыпается только на пробуждениях.

«Зачем Box::pin, это же аллокация на задачу». Для гетерогенного HashMap из разных футур - да, проще боксировать. tokio избегает этого, выделяя задачу одним блоком и работая с ней через сырой указатель, но это усложнение, которое в обучающем коде только мешает.

«Arc::from_raw в каждой функции vtable выглядит жутко». Так и есть, это ручное управление счётчиком ссылок. Именно поэтому в реальной жизни вы берёте Waker из futures::task::waker или из рантайма и не пишете vtable руками. Но один раз написать полезно, чтобы перестать бояться.

«У тебя в реакторе один Waker на дескриптор - а если два таска ждут один и тот же fd?». Справедливо: HashMap<RawFd, Waker> затирает предыдущего ждущего, и в общем случае нужен список вокеров на дескриптор плюс разделение интересов на чтение и запись. В реальных рантаймах для каждого fd хранится отдельная пара слотов под read/write ожидание. В обучающем коде я намеренно держу по одному, чтобы не утонуть в деталях, но знать про это ограничение полезно.

«Откуда требование 'static на спавне». Задача живёт в рантайме сколько угодно долго и не привязана к стек-фрейму того, кто её заспавнил, поэтому она не может одалживать ссылки с коротким временем жизни. Всё, что нужно задаче, она должна владеть - отсюда move в async move и 'static в сигнатуре spawn. Это не каприз, а прямое следствие того, что future переживает функцию, которая его создала.

«Почему не async-std / smol / glommio». Smol устроен идейно очень близко к тому, что мы написали, и его исходники отлично читаются после этой статьи. Glommio это thread-per-core поверх io_uring - другая архитектура, заслуживающая отдельного разбора.

Итог

Свой рантайм не нужен в проде: для этого есть tokio, и обогнать его за выходные нельзя. Но как способ выбить из головы слово «магия» он бесценен. Собрав цепочку Future → Waker → реактор → исполнитель руками, вы начинаете видеть за .await машину состояний и понимать, где задача засыпает, кто её будит и почему компилятор требует Pin, Send и 'static ровно там, где требует.

Эхо-сервер выше это уже рабочий каркас: добавьте к нему буферизацию частичных записей, таймауты через таймерфд в том же epoll и парсер протокола - и получится основа реального сетевого сервиса на голой std. А дальше слоями ложатся многопоточность и work stealing, и внутренности tokio открываются не по верхам, а по существу.

Если хочется копнуть глубже самостоятельно, вот проверенный временем минимум. Async Book с главой, где Waker собирают руками примерно как у нас. Серия статей «Async/await» Уилла Крайтона и разбор «Tasks» в блоге tokio про устройство планировщика. Исходники крейта smol - он маленький и читается за вечер, а идеи ровно те же. И, конечно, документация модуля std::task, где RawWakerVTable описан без посредников. На русском есть отличная серия «Асинхронный Rust в трёх частях» (перевод Jack O’Connor в блоге Beget) - там тот же путь разбирается ещё подробнее, с пошаговой эволюцией кода и фирменным багом «sleep forever?». Если после моей статьи захочется медленнее и с большим числом промежуточных шагов - вам туда.

И ещё одно практическое следствие из всего сказанного, на сладкое. Раз async fn это машина состояний, которую двигает poll, то блокирующий вызов внутри .await-цепочки останавливает не одну задачу, а весь поток исполнителя со всеми его задачами. Это объясняет classic-ловушку «положил std::fs::read в async-функцию и весь сервер встал». Теперь понятно, почему: пока poll не вернул управление, исполнитель физически не может опросить никого другого.

Я копаю Rust и async на этом же уровне в телеграме - короткими заметками, которые в статью не попадают: @rust_code. Если зашло, заглядывайте, там продолжение этой темы.

Если зайдёт, в следующей части разберу work stealing вплотную: lock-free деки локальных очередей, протокол воровства из хвоста, проблема ABA и почему наивная реализация ломается под нагрузкой. Напишите в комментариях, копать ли в эту сторону, и накидайте, что ещё во внутренностях async осталось непонятным.