Дополнение · X-алгоритм · страница 2 из 11

Пайплайн: как лента собирается из типовых стадий

Лента «Для вас» описана как последовательность из 75 стадий шести типов. Разбираем фреймворк, на котором это держится: что выполняется параллельно, а что обязано идти по порядку, что происходит при падении источника, и почему гидратация запроса разделена на два круга.

1. Шесть типов стадий

Весь фреймворк — это 1234 строки на семь файлов. Каждый файл описывает один тип стадии, и все они устроены одинаково.

pub trait Filter<Q, C>: Any + Send + Sync {
    fn enable(&self, _query: &Q) -> bool { true }
    fn filter(&self, query: &Q, candidates: Vec<C>) -> FilterResult<C>;
    fn name(&self) -> &'static str { ... }
}

Фрагмент из filter.rs · код X, Apache 2.0, коммит 28e414f

Три метода: включён ли, что делает, как называется. Имя нужно для метрик — по нему считается, сколько эта стадия отбросила и сколько заняла. Метод run, который вызывает пайплайн, — это обёртка вокруг filter, добавляющая замеры.

2. Порядок выполнения

Главная функция пайплайна читается как оглавление всей ленты:

let hydrated_query = self.hydrate_query(query).await;
let hydrated_query = self.hydrate_dependent_query(hydrated_query).await;
let candidates = self.fetch_candidates(&hydrated_query).await;
let hydrated_candidates = self.hydrate(&hydrated_query, candidates).await;
let (kept, mut filtered) = self.filter(&hydrated_query, hydrated_candidates.clone());
let scored = self.score(&hydrated_query, kept).await;
let SelectResult { selected, non_selected } = self.select(&hydrated_query, scored);
let post_selection = self.hydrate_post_selection(&hydrated_query, selected).await;
let (mut final_candidates, post_filtered) = self.filter_post_selection(&hydrated_query, post_selection);

Фрагмент из candidate_pipeline.rs · код X, Apache 2.0, коммит 28e414f

Что параллельно, а что нет

Разница видна прямо в коде и она не косметическая.

Источники — параллельно:

let source_futures = sources.iter().map(|s| s.run(query));
let results = join_all(source_futures).await;

Фильтры — последовательно:

for filter in enabled {
    let result = filter.run(query, candidates);
    ...
}

Фрагмент из candidate_pipeline.rs · код X, Apache 2.0, коммит 28e414f

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

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

3. Что происходит, когда источник падает

Строка, которую легко пропустить, но которая определяет поведение системы под нагрузкой:

let results = join_all(source_futures).await;

let mut collected = Vec::new();
for mut candidates in results.into_iter().flatten() {
    collected.append(&mut candidates);
}

Фрагмент из candidate_pipeline.rs · код X, Apache 2.0, коммит 28e414f

Каждый источник возвращает Result<Vec<C>, String> — либо кандидатов, либо ошибку. А flatten по итератору результатов молча выбрасывает все ошибки и оставляет только успешные.

Значит, если сервис свежих постов подписок упал, запрос не падает. Лента соберётся из того, что вернули остальные источники: она станет хуже — не будет постов от тех, на кого вы подписаны, — но она будет.

4. Гидратация запроса: два круга

Обратите внимание, что гидратация запроса вызывается дважды:

let hydrated_query = self.hydrate_query(query).await;
let hydrated_query = self.hydrate_dependent_query(hydrated_query).await;

Фрагмент из candidate_pipeline.rs · код X, Apache 2.0, коммит 28e414f

Первый круг — гидраторы, которым нужен только исходный запрос: список подписок, блокировок, скрытых аккаунтов, демографические данные. Все они параллельны друг другу.

Второй круг — те, которым нужны результаты первого. Например, чтобы посчитать взаимные подписки, надо сначала получить список подписок. Или чтобы построить последовательность действий для модели, надо знать, какие посты уже показывались.

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

Что именно дотягивается к запросу

Гидраторов запроса двадцать. Вот наиболее показательные.

5. Две трубы: посты и всё остальное

Пайплайнов на самом деле два, и они вложены друг в друга.

Post Pipeline — то, что мы разбирали выше: 75 стадий, находящих и упорядочивающих посты.

Blending Pipeline — обёртка вокруг него. Для неё отранжированные посты — всего лишь один источник среди прочих. Другие источники приносят то, что моделью не ранжируется: рекламу, блок «кого читать», промо-подсказки.

6. Побочные эффекты: после ответа

fn run_side_effects(&self, input: Arc<SideEffectInput<Q, C>>) {
    tokio::spawn(... async move {
        let futures = side_effects.iter()
            .filter(|se| se.enable(input.query.clone()))
            .map(|se| se.run(input.clone()));
        let _ = join_all(futures).await;
    });
}

Фрагмент из candidate_pipeline.rs · код X, Apache 2.0, коммит 28e414f

Ключевое здесь — tokio::spawn и let _ =. Задача запускается в фоне, её результат не ожидается и не проверяется. Ответ пользователю уже отправлен.

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

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

Частые ошибки и подводные камни

Вопросы с собеседований

Шпаргалка одним экраном

Первоисточники

• candidate_pipeline.rs — порядок стадий, параллелизм, обработка ошибок.
• filter.rs, source.rs и соседние — определения типов стадий.
• phoenix_candidate_pipeline.rs — все 75 стадий ленты по порядку.
• Курс: глава «Архитектура рантайма» об архитектуре рантайма и отказоустойчивости, глава «Многостадийная воронка» о многостадийности, глава «Как проектировать систему с нуля» о слоях с разными целями.