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 на каждый элемент отдельно.