Cross-Cleaner: стриминг JSON частями вместо загрузки всей базы очистки в память

Стриминговый разбор JSON в Cross-Cleaner на serde/serde_json: практическая реализация

Что уже есть в коде

Cross-Cleaner уже проделал большую часть работы по снижению памяти — но не на уровне парсинга, а на уровне представления данных:

• Arc<str> вместо String для category, program, class, sub_category, files_to_remove, directories_to_remove — с дедупликацией через string_interner (intern_arc);

• SharedPath — путь хранится как Arc<[Arc<str>]> из интернированных сегментов, что даёт разделение общих префиксов (AppData\Local\...) между десятками тысяч записей;

• CleanerIndex и RegistryIndex — лёгкие проекции для сканирования категорий, где «неизвестные поля игнорируются serde и не аллоцируются».

Это уже сильно снижает установившийся размер базы в памяти. Но пик потребления при десериализации всё ещё определяется тем, как именно вызывается serde_json. Если база грузится через serde_json::from_reader::<_, Vec<CleanerData>>, то в момент разбора в памяти одновременно живут: буфер чтения, строящийся Vec, и промежуточные String внутри кастомного Deserialize for CleanerData (структура Helper содержит String, а не Arc<str>). Именно Helper — главный источник временных аллокаций.

Где именно теряется память при текущем Deserialize

Кастомный impl<'de> Deserialize<'de> for CleanerData делает:

rust

#[derive(Deserialize)]struct Helper {    path: String,    category: String,    program: String,    class: String,    sub_category: String,    files_to_remove: Vec<String>,    directories_to_remove: Vec<String>,    // ...}let h = Helper::deserialize(deserializer)?;// затем всё это переинтернивается в Arc<str>

То есть на каждую запись сначала аллоцируются String, а уже потом они конвертируются в Arc<str> через intern_arc. Пиковая память на запись ≈ размер всех её строк в виде String + размер интернированных Arc<str> + накладные расходы Vec. Для одной записи это немного, но при потоковом разборе важно, что Helper освобождается сразу после конвертации — а при from_reader в Vec<CleanerData> это происходит для каждой записи последовательно, так что пик по временным строкам не суммируется. Хорошая новость: сам по себе этот Deserialize уже не «загружает всё в строки разом».

Проблема остаётся в итоговом Vec<CleanerData>: все записи живут одновременно. Для базы очистки это, вероятно, приемлемо, если она нужна целиком для сканирования. Но если задача — например, построить только CleanerIndex или прогнать категорию, держать полный Vec<CleanerData> не нужно.

Что даёт StreamDeserializer в этой кодовой базе

serde_json::StreamDeserializer уместен в двух сценариях.

Сценарий 1: построение индекса без полной базы

Если база — JSON-массив CleanerData, а нужен только CleanerIndex (для сканирования категорий), можно не десериализовать CleanerData вовсе. CleanerIndex уже игнорирует лишние поля, и serde_json не будет их аллоцировать. При потоковом обходе элементов массива в памяти живёт один CleanerIndex за раз:

rust

use std::fs::File;use std::io::BufReader;fn build_index(path: &str) -> Result<Vec<CleanerIndex>, Box<dyn std::error::Error>> {    let file = File::open(path)?;    let reader = BufReader::new(file);    let mut index = Vec::new();    for item in iter_json_array::<CleanerIndex, _>(reader) {        index.push(item?); // в памяти — только CleanerIndex, без Helper и без полных данных    }    Ok(index)}

Ключевой момент: CleanerIndex использует deserialize_shared_opt/deserialize_shared_str_default, то есть сразу интернирует в Arc<str>. Временные String живут только внутри одного вызова Deserialize и освобождаются до перехода к следующему элементу. По сравнению с Vec<CleanerData>, где каждая запись тащит files_to_remove, directories_to_remove и flags, это радикально меньше.

Сценарий 2: обработка базы по одной записи без накопления

Если CleanerData нужны, но не все сразу (например, фильтрация по ОС, по категории, немедленная регистрация в планировщике), потоковый обход позволяет не держать Vec:

rust

fn for_each_rule<F>(path: &str, mut f: F) -> Result<(), Box<dyn std::error::Error>>where    F: FnMut(CleanerData) -> Result<(), Box<dyn std::error::Error>>,{    let file = File::open(path)?;    let reader = BufReader::new(file);    for item in iter_json_array::<CleanerData, _>(reader) {        f(item?)?; // item освобождается после вызова    }    Ok(())}

Здесь Helper внутри Deserialize for CleanerData создаётся и уничтожается на каждой записи, а итоговый CleanerData живёт ровно один вызов f.

Обёртка iter_json_array под топ-левел массив

StreamDeserializer не умеет итерировать элементы массива напрямую, поэтому нужна тонкая обёртка. Она не держит данные — только читает байты и пропускает [, ,, ]. Минимальная версия:

rust

use serde::de::DeserializeOwned;use serde_json::Deserializer;use std::io::{self, Read};fn skip_ws<R: Read>(reader: &mut R) -> io::Result<Option<u8>> {    loop {        let mut b = [0u8; 1];        match reader.read(&mut b)? {            0 => return Ok(None),            _ if b[0].is_ascii_whitespace() => continue,            _ => return Ok(Some(b[0])),        }    }}fn deserialize_one<T: DeserializeOwned, R: Read>(reader: &mut R) -> io::Result<T> {    Deserializer::from_reader(reader)        .into_iter::<T>()        .next()        .ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "premature EOF"))?        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))}pub fn iter_json_array<T, R>(mut reader: R) -> impl Iterator<Item = io::Result<T>>where    T: DeserializeOwned,    R: Read,{    let mut started = false;    let mut done = false;    std::iter::from_fn(move || {        if done {            return None;        }        if !started {            started = true;            match skip_ws(&mut reader) {                Ok(Some(b'[')) => {}                Ok(_) => {                    done = true;                    return Some(Err(io::Error::new(                        io::ErrorKind::InvalidData,                        "expected `[`",                    )));                }                Err(e) => {                    done = true;                    return Some(Err(e));                }            }            match skip_ws(&mut reader) {                Ok(Some(b']')) => {                    done = true;                    return None;                }                Ok(Some(first)) => {                    // «вернуть» первый байт в поток невозможно, поэтому                    // используем Cursor с цепочкой                    let chained = std::io::Cursor::new([first]).chain(reader);                    let mut chained = chained;                    return Some(deserialize_one(&mut chained));                }                Ok(None) => {                    done = true;                    return Some(Err(io::Error::new(                        io::ErrorKind::UnexpectedEof,                        "unterminated array",                    )));                }                Err(e) => {                    done = true;                    return Some(Err(e));                }            }        }        match skip_ws(&mut reader) {            Ok(Some(b',')) => Some(deserialize_one(&mut reader)),            Ok(Some(b']')) => {                done = true;                None            }            Ok(_) => {                done = true;                Some(Err(io::Error::new(                    io::ErrorKind::InvalidData,                    "expected `,` or `]`",                )))            }            Err(e) => {                done = true;                Some(Err(e))            }        }    })}

Важная деталь: deserialize_one создаёт StreamDeserializer на одном элементе и сразу его отбрасывает, поэтому внутренние буферы serde_json не накапливаются между элементами. Обёртка Cursor::new([first]).chain(reader) нужна только для первого элемента, чтобы не потерять уже прочитанный байт.

Почему это сочетается с текущим дизайном

Текущий код уже готов к потоковому использованию:

1. CleanerIndex и RegistryIndex — готовые лёгкие типы. Их Deserialize использует deserialize_shared_opt, то есть сразу интернирует. Временные String живут в пределах одного элемента.

2. deserialize_vec_arc — принимает Vec<String>, интернирует и возвращает Vec<Arc<str>>. Временные String для files_to_remove/directories_to_remove освобождаются сразу после интернирования, не дожидаясь конца базы.

3. SharedPath::deserialize — читает String, разбивает на сегменты и интернирует каждый. Временная строка пути живёт только на время одного Deserialize.

4. CleanerData::Deserialize через Helper — единственное место, где временные String живут чуть дольше, но всё равно в пределах одной записи.

То есть переход от from_reader::<_, Vec<CleanerData>> к iter_json_array::<CleanerData, _> не требует менять ни один Deserialize — он уже написан так, что временные аллокации локальны для записи.

Ограничения и подводные камни

• StreamDeserializer и BufReader. serde_json не буферизует ввод. Без BufReader каждый read — это syscall, и потоковый разбор станет медленнее полного. BufReader обязателен.

• Формат базы. Если база — не топ-левел массив, а, например, объект {"rules": [...]}, обёртку нужно адаптировать под префикс. Альтернатива — перевести базу на JSON Lines, тогда iter_json_array не нужен вовсе: BufReader::lines() + serde_json::from_str на строку.

• CustomCleaner не сериализуется. Он определён в коде, а не в JSON, поэтому к нему потоковый разбор неприменим — там Vec<CustomCleaner> собирается из зарегистрированных функций.

• CleanerDataRegistry (Windows) — обычный derive(Deserialize), без кастомного Helper. Его тоже можно гонять через iter_json_array, если база реестра — массив.

• Порядок и агрегация. Если после разбора нужна сортировка или глобальная дедупликация по всей базе, потоковый обход сам по себе не поможет — придётся либо накапливать индекс (что и делает CleanerIndex), либо использовать внешнюю сортировку.

Ожидаемый эффект

Для базы очистки Cross-Cleaner переход на потоковый разбор даёт:

• пик памяти при загрузке определяется одной записью (CleanerData + временный Helper), а не всей базой;

• при построении индекса (CleanerIndex) в памяти остаются только SharedPath + три Arc<str> на запись — без files_to_remove, directories_to_remove, flags и без Helper;

• возможность обрабатывать базы, превышающие доступную RAM, если обработка не требует держать все записи одновременно;

• сохранение всех существующих оптимизаций (string_interner, SharedPath, Arc<str>) без изменений в Deserialize.

Заключение

В Cross-Cleaner снижение памяти уже достигнуто на уровне представления: интернирование строк и сегментированные пути. Следующий шаг — на уровне способа чтения: заменить serde_json::from_reader::<_, Vec<CleanerData>> на iter_json_array поверх StreamDeserializer (или на JSON Lines). Существующие Deserialize-реализации уже устроены так, что временные String локальны для одной записи, поэтому потоковый обход не требует их переписывать — достаточно обёртки, которая пропускает разделители массива и создаёт StreamDeserializer на каждый элемент отдельно.