Low-latency Rust: горячий путь без аллокаций

Бюджет из прошлого урока показал, что наш собственный код это меньше пятой части времени. Звучит как повод им не заниматься, но это ловушка: как только ядро обойдено, код становится главным участком, и вся экономия упирается в него. Этот урок про то, что значит «быстрый код» в торговом движке. Ответ короткий и неочевидный: не «мало инструкций», а ни одного обращения к тому, кто может заставить нас ждать.

Горячий путь и всё остальное

Раздели программу на две части и держи границу жёстко.

Горячий путь это цепочка от прихода пакета до ухода заявки. Всё, что в ней, выполняется на каждое событие рынка, то есть десятки тысяч раз в секунду.

Всё остальное это старт, разбор конфигурации, логирование, отчёты, метрики, переподключение. Оно выполняется редко, и там можно писать как угодно.

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

Аллокатор. Box::new, Vec::push с ростом, String, format!, collect. Аллокатор в общем случае берёт глобальную блокировку, а когда у него кончается память в пуле, идёт в ядро за страницами. Обычный вызов это десятки наносекунд, редкий вызов это десятки микросекунд, и никто не скажет заранее, какой из них выпадет.

Блокировка. Mutex, RwLock. Если блокировка занята, поток встаёт, планировщик снимает его с ядра, и вернуться он может через миллисекунду. Хуже того, снять его могут прямо посреди удержания блокировки, и тогда встанут все.

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

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

Линия кэша и почему соседние переменные дерутся

Процессор не читает байты, он читает линии кэша по 64 байта. Порядок задержек примерно такой:

Откуда читаем	Тактов	Наносекунд
L1	4	около 1
L2	12	около 4
L3	40	около 15
оперативная память	200 и больше	70 и больше

Разница между попаданием в L1 и походом в память это семьдесят раз. Никакая алгоритмическая оптимизация не даст такого множителя, а раскладка данных даёт.

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

Это ложное разделение, и лечится оно разведением по линиям:

/// Размер линии кэша на x86 и на современных ARM. На некоторых процессорах
/// линии подтягиваются парами, поэтому в промышленных библиотеках берут 128.
pub const CACHE_LINE: usize = 64;

/// Значение, занимающее целую линию кэша.
#[derive(Debug, Default)]
#[repr(align(64))]
pub struct CachePadded<T> {
    value: T,
}

impl<T> CachePadded<T> {
    pub fn new(value: T) -> Self {
        Self { value }
    }
}

impl<T> std::ops::Deref for CachePadded<T> {
    type Target = T;

    fn deref(&self) -> &T {
        &self.value
    }
}

#[repr(align(64))] заставляет компилятор выровнять структуру на границу линии и дополнить её до целой линии. Проверить можно тестом, и такой тест стоит написать: атрибут легко потерять при рефакторинге, а падение производительности от этого будет тихим.

#[test]
fn cache_padded_occupies_a_whole_line() {
    assert_eq!(std::mem::align_of::<CachePadded<u64>>(), CACHE_LINE);
    assert_eq!(std::mem::size_of::<CachePadded<u64>>(), CACHE_LINE);
}

Очередь на двух индексах

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

Нам нужен частный случай, и он проще общего: один писатель, один читатель. Такая очередь называется SPSC и строится на кольцевом буфере с двумя индексами. Писатель двигает только head, читатель только tail, и вся синхронизация умещается в два атомарных числа.

use std::cell::UnsafeCell;
use std::mem::MaybeUninit;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

/// Общее нутро очереди. Наружу оно не торчит.
struct Ring<T> {
    slots: Box<[UnsafeCell<MaybeUninit<T>>]>,
    mask: usize,
    /// Куда пишет производитель. Читается потребителем.
    head: CachePadded<AtomicUsize>,
    /// Откуда читает потребитель. Читается производителем.
    tail: CachePadded<AtomicUsize>,
}

// Слоты трогает ровно один поток за раз: писатель до публикации индекса,
// читатель после. Синхронизацию обеспечивают Release и Acquire на индексах.
unsafe impl<T: Send> Send for Ring<T> {}
unsafe impl<T: Send> Sync for Ring<T> {}

Обрати внимание на три решения, принятые ещё до первой строчки логики.

head и tail разведены по линиям кэша. Они лежат рядом и пишутся разными потоками, то есть это ровно тот случай ложного разделения, который мы только что разобрали. Без CachePadded очередь работает корректно и медленно.

Слоты это MaybeUninit, а не Option<T>. Option добавил бы к каждому слоту флаг и лишнюю проверку при чтении. Занятость слота и так известна из индексов, второй раз её хранить незачем.

Ёмкость это степень двойки. Тогда возврат индекса к началу буфера это index & mask, одна инструкция, а не деление. Одна ячейка при этом всегда остаётся пустой: иначе полная и пустая очереди выглядели бы одинаково.

Раздача концов:

/// Создать очередь на одного производителя и одного потребителя.
///
/// Концы раздаются отдельными типами, и это не украшение: `Producer` и
/// `Consumer` не `Clone`, поэтому «один писатель, один читатель» проверяет
/// компилятор, а не комментарий в заголовке файла.
pub fn channel<T>(capacity: usize) -> (Producer<T>, Consumer<T>) {
    let size = capacity.next_power_of_two().max(2);
    let mut slots = Vec::with_capacity(size);
    slots.resize_with(size, || UnsafeCell::new(MaybeUninit::uninit()));
    let ring = Arc::new(Ring {
        slots: slots.into_boxed_slice(),
        mask: size - 1,
        head: CachePadded::new(AtomicUsize::new(0)),
        tail: CachePadded::new(AtomicUsize::new(0)),
    });
    (
        Producer {
            ring: Arc::clone(&ring),
            cached_tail: 0,
        },
        Consumer {
            ring,
            cached_head: 0,
        },
    )
}

/// Пишущий конец очереди.
pub struct Producer<T> {
    ring: Arc<Ring<T>>,
    /// Копия чужого индекса. Пока в очереди есть место по этой копии,
    /// чужую атомарную переменную читать не надо, и линия кэша остаётся нашей.
    cached_tail: usize,
}

/// Читающий конец очереди.
pub struct Consumer<T> {
    ring: Arc<Ring<T>>,
    cached_head: usize,
}

Кэш чужого индекса это главный приём этой конструкции, и он неочевиден. Писателю нужно знать, где стоит читатель, только чтобы понять, есть ли место. Но место обычно есть с большим запасом, и читать чужую атомарную переменную на каждой записи значит тянуть чужую линию кэша к себе на каждой записи. Поэтому мы запоминаем последнее известное значение и лезем за настоящим, только когда по запомненному места нет.

Теперь запись:

impl<T> Producer<T> {
    /// Положить значение. Возвращает его обратно, если места нет: очередь
    /// фиксированного размера, и «подожди, я сейчас выделю память» на горячем
    /// пути не бывает.
    pub fn push(&mut self, value: T) -> Result<(), T> {
        let head = self.ring.head.load(Ordering::Relaxed);
        let next = head.wrapping_add(1);

        if next.wrapping_sub(self.cached_tail) > self.ring.mask {
            // По кэшу места нет: только теперь идём читать чужой индекс.
            self.cached_tail = self.ring.tail.load(Ordering::Acquire);
            if next.wrapping_sub(self.cached_tail) > self.ring.mask {
                return Err(value);
            }
        }

        let slot = &self.ring.slots[head & self.ring.mask];
        // Безопасность: в этот слот пишет только производитель, и только пока
        // потребитель ещё не увидел новый head.
        unsafe { (*slot.get()).write(value) };
        // Release: запись в слот видна всем, кто увидел новый head.
        self.ring.head.store(next, Ordering::Release);
        Ok(())
    }
}

Разберём порядки памяти, они тут не украшение. Подробный разбор мы делали в уроке про атомики, здесь применяем.

Своё чтение head идёт с Relaxed. Кроме нас его никто не пишет, значит гарантий упорядочивания не нужно, нужно просто значение.

Чтение чужого tail идёт с Acquire. Мы хотим увидеть все записи, которые читатель сделал до того, как продвинул tail.

Публикация head идёт с Release. Это ключевой момент: запись в слот обязана стать видимой раньше, чем новое значение head. Иначе читатель увидит «есть новый элемент» и прочитает мусор. Компилятор и процессор оба имеют право переставлять записи местами, и Release это ровно запрет на такую перестановку.

Чтение симметрично:

impl<T> Consumer<T> {
    /// Забрать значение, если оно есть.
    pub fn pop(&mut self) -> Option<T> {
        let tail = self.ring.tail.load(Ordering::Relaxed);

        if tail == self.cached_head {
            self.cached_head = self.ring.head.load(Ordering::Acquire);
            if tail == self.cached_head {
                return None;
            }
        }

        let slot = &self.ring.slots[tail & self.ring.mask];
        // Безопасность: производитель опубликовал head больше tail, значит
        // слот проинициализирован и больше не трогается.
        let value = unsafe { (*slot.get()).assume_init_read() };
        self.ring.tail.store(tail.wrapping_add(1), Ordering::Release);
        Some(value)
    }

    /// Забрать всё, что накопилось, одним проходом. На всплеске рыночных
    /// данных это дешевле, чем дёргать `pop` по одному: индекс публикуется
    /// один раз на пачку.
    pub fn drain(&mut self, limit: usize, mut visit: impl FnMut(T)) -> usize {
        let tail = self.ring.tail.load(Ordering::Relaxed);
        self.cached_head = self.ring.head.load(Ordering::Acquire);
        let available = self.cached_head.wrapping_sub(tail).min(limit);
        for offset in 0..available {
            let slot = &self.ring.slots[tail.wrapping_add(offset) & self.ring.mask];
            let value = unsafe { (*slot.get()).assume_init_read() };
            visit(value);
        }
        if available > 0 {
            self.ring
                .tail
                .store(tail.wrapping_add(available), Ordering::Release);
        }
        available
    }
}

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

Один тест обязателен, и он не про API, а про корректность под настоящей конкуренцией:

#[test]
fn queue_survives_two_threads() {
    let (mut producer, mut consumer) = channel::<u64>(64);
    let total = 100_000u64;

    let writer = std::thread::spawn(move || {
        let mut value = 0;
        while value < total {
            if producer.push(value).is_ok() {
                value += 1;
            } else {
                std::hint::spin_loop();
            }
        }
    });

    let mut expected = 0;
    while expected < total {
        match consumer.pop() {
            Some(value) => {
                // Порядок обязан сохраниться: очередь FIFO.
                assert_eq!(value, expected);
                expected += 1;
            }
            None => std::hint::spin_loop(),
        }
    }
    writer.join().expect("писатель дожил до конца");
}

Тест на порядок ловит именно ошибку в Ordering: при неправильных порядках элементы не пропадают, они приезжают не тогда, когда должны, или приезжают недописанными.