Сортировка слиянием, часть 2

Это продолжение статьи «Сортировка слиянием — не так просто, как кажется». Там я довёл тестовое задание — отсортировать 100 ГБ строк вида число. текст за час — до 32 минут: пайплайн из пяти стадий на Channels, Brotli, sort keys во временных файлах. Сортировка, конечно же с учетом культуры.

Код той версии остался в ветке net7.

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

Год назад, когда появилось время, я начал переделывать решение. Даже сделал пост, но переработка затянулась, а пост по таймеру опубликовался сам, когда я забыл переставить дату.

Итог: 100 ГБ случайных строк (худший случай для сжатия) на обычном HDD — за 52 минуты.

Рассказываю, как к этому пришёл: тут и предложение из комментариев, которое сработало слишком хорошо, и CSR-раскладка, которую подсказал ИИ, и малоизвестные части .NET.

С чего всё началось: комментарий, который сработал слишком хорошо

В комментариях к первой части @gchebanov написал:

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

(gchebanov в комментариях к части 1)

Он прав: тестовые данные генерировались из «Войны и мира», и уникальных строк там около шестнадцати тысяч. Записывать в чанк все строки нет никакой необходимости, можно записать каждую строку один раз и список чисел рядом с ней. Тогда и brotli-сжатие не даст такого заметного эффекта, как в первом варианте.

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

varint(textLen) text varint(count) varint(δ₁) varint(δ₂) ...

Причем числа хранятся дельтами с предыдущим по возрастанию, поэтому varint в среднем 3–5 байт на значение вместо восьми.

На данных исходной статьи она сработала слишком хорошо: чанк растёт только на уникальных текстах, а их шестнадцать тысяч, поэтому все 100 ГБ сложились в один чанк, целиком живущий в памяти. Внешняя сортировка выродилась в обычную: час, конечно, «сошёлся», но задача перестала быть задачей.

Тогда я написал генератор случайных строк (команда generate в репозитории) и прогнал старый алгоритм на 100 ГБ честных данных — где почти каждый текст уникален, дедупликации нет, а сжатие не находит повторов. Он не уложился в час: промежуточные чанки выросли в разы — с ~10 ГБ (Война и мир, Brotli жал в 20 раз) до ~77 ГБ, и всё это пришлось прокрутить через один медленный шпиндель.

Почему эта задача интересная: CPU-bound и IO-bound одновременно

Большинство задач связаны либо процессором (сжатие, хэширование), либо диском (копирование). Внешняя сортировка слиянием — редкий зверь, связанный и тем, и другим одновременно: чтение исходника, запись чанков, потом чтение чанков и запись результата — это IO; разбор строк, словари, sort keys, сортировка, распаковка — это CPU. На SSD узким местом почти всегда окажется CPU, и половина усилий по развязке пропадёт зря. На HDD — наоборот: если читающий поток хоть на секунду встал в очередь за процессорным слотом, шпиндель простаивает, и эти секунды уже не вернуть.

Дедупликация работает как честная оптимизация: на данных с повторами она ускоряет всё, на данных без повторов не мешает.

Карта решения

Пайплайн из двух фаз. Сплит читает исходник и выдаёт сжатые отсортированные чанки:

Мерж сливает чанки в результат:

Формат чанка один на обе фазы — varint(textLen) text varint(count) δ-varint значения на каждый уникальный текст, sort key в файл не пишется (почему — ниже). Дальше по порядку.

Чтение: Pipe вместо жонглирования буферами

В net7 чтение выглядело так:

var bytesRead = stream.ReadBlock(readBuffer, remainingBytes, readBuffer.Length - remainingBytes, out eof);
int chunkSize = remainingBytes + bytesRead;
var lastNewLine = readBuffer.AsSpan(0, bytesRead).LastIndexOf(NewLine);
if (lastNewLine >= 0) chunkSize = lastNewLine + NewLine.Length;
remainingBytes = remainingBytes + bytesRead - chunkSize;
...
if (remainingBytes > 0) oldBuffer.AsSpan(chunkSize,remainingBytes).CopyTo(readBuffer.AsSpan());

Rent, ReadBlock, поиск последнего перевода строки, перенос хвоста в начало следующего буфера. Тридцать строк, каждая — место для off-by-one.

Ещё в первой части я гонял бенчмарки чтения — и Pipelines показывали скорость не хуже при меньшем потреблении памяти. Теперь чтение везде работает через Pipe: выделенный поток-производитель пишет в пайп синхронным FileStream без собственного буфера, потребитель режет ReadOnlySequence итератором на SequenceReader. Backpressure получился автоматически: пороги resume/pause 8/16 МБ — производитель физически не может убежать от потребителя дальше шестнадцати мегабайт, никакой ручной арифметики с остатками.

Неожиданный бонус — парсинг упростился. SequenceReader поверх ReadOnlySequence умеет TryReadTo по байтовому разделителю и отдаёт срезы без копий: строка — это два среза (число и текст), живущие в буфере пайпа. Итератор строк — два десятка строк вместо страницы ручной арифметики:

SequenceReader<byte> sequenceReader = new(search);
if (!sequenceReader.TryReadTo(out ReadOnlySpan<byte> number, dotSpace))
    throw new FormatException(...);
var x = long.Parse(number, null);          // utf-8 парсится прямо из байтов
var text = reader.Sequence.Slice(reader.Position);   // текст — срез до конца строки

Ключ без аллокаций: alternate lookup (.NET 9)

Pipe решает проблему чтения и тут же создаёт новую: строка теперь — Span<byte> из буфера пайпа, а Span — ref struct, в Dictionary его не положишь. Копировать в string — вернуть все аллокации, от которых уходили. До выхода .NET 9 я пытался сам собрать dictionary, который поможет искать ключ по Span, который приходит от SequenceReader .

В .NET 9 появился: Dictionary<TKey, TValue>.GetAlternateLookup<TAlternate>() и IAlternateEqualityComparer<TAlternate, TKey>. Реализуя интерфейс в компараторе для Dictionary вы получаете возможность искать и создавать элементы с ключом типа TAlternate.

Я сделал словарь чанка, который типизирован структурой (смещение, длина) в отдельном буфере чанка, а компарер, при добавлении нового ключа, сам копирует байты Span’а в буфер чанка и возвращает адрес:

public BufferString Create(ReadOnlySpan<byte> alternate)
{
    if (alternate.Length > Free)
        throw new OverflowException(...);
    alternate.CopyTo(buffer.AsSpan(Offset));
    Offset += alternate.Length;
    return new BufferString(Offset - alternate.Length, alternate.Length);
}

Хэш для ключа — XxHash32 по спану, сравнение — SequenceEqual для спанов, ноль аллокаций в куче. Текст живёт в буфере ровно один раз — и как ключ словаря, и как содержимое будущего чанка.

Когда буфер переполняется я кидаю OverflowException, который ловит цикл разбора и превращает в «выдать чанк, начать новый». Исключение как сигнал — я сам морщился, но это один throw на сотни тысяч добавлений, а взамен из всех слоёв исчезает протаскивание статуса «места нет».

CSR: почему не List на каждый ключ

Куда класть числа, которые относятся к ключу? Первая наивная версия — List<long> при каждом ключе — и была тем местом, где словарный подход умирал: на миллион ключей это миллион объектов списка по 24+ байта каждый плюс отдельные массивы по 8 байт на элемент, размазанные по куче. GC-давление, плохая локальность, и всё это ещё надо как-то сериализовать.

Решение — CSR (compressed sparse row), классическая раскладка разреженных матриц. Его подсказал мне ИИ.

Как работает CSR:

На фазе накопления чанка значения пишутся в буфер парами: (номер ключа, значение). В значениях словаря лежат пары (номер ключа, количество значений).

comparer.ReservePair();                    // 16 байт с хвоста
ref var info = ref CollectionsMarshal.GetValueRefOrAddDefault(alt, text, out var exists);
if (!exists) info = new KeyInfo(lookup.Count, 1);
else info = info with { Count = info.Count + 1 };
slots[^valueCount++] = (info.Number, value);   // пара — в хвост буфера

CollectionsMarshal.GetValueRefOrAddDefault позволяет за одну операцию взять ссылку на существующее значение в словаре или добавить новое. Пары пишутся в тот же буфер, что и ключи, но с конца.

Перед сортировкой пары (номер ключа, значение) рассыпаются в один массив значений, где значения каждого ключа лежат подряд, плюс массив индексов, с которых начинаются значения данного ключа, и массив количеств значений для данного ключа.

Итого 3 массива независимо от количества ключей. Что это даёт по сравнению со списками:

• 8 байт на значение — против 32+ байт в куче на каждый ключ + давление на GC из-за увеличения списков.

• Локальность, которую чувствует и сортировка, и сериализация.

• Сортировка значений распадается на независимые диапазоны — выполняется в Parallel.For по ключам без единой блокировки.

Sort keys для культурно-чувствительного сравнения считается один раз на уникальный текст :

1. Вместе с заполнением массивов смещений для CSR ключи из байтов превращаются в char в отдельном буфере на весь чанк.

2. Параллельно GetSortKeyLength на каждый ключ, потом подсчет точного размера буфера для sort keys и префикс-сумма для смещений в буфере.

3. Параллельно GetSortKey параллельно для всех ключей.

Первая версия делала это последовательно, и профилирование показало яму в несколько секунд на чанке.

Вся эта конструкция оформлена как явные стадии: Chunk (накопление) → SortedChunk (CSR, порядок, сериализация) → CompressedChunk (сжатые байты). Каждая стадия отдаёт буферы, которые больше не нужны, — владение видно по коду, а не по комментариям.

Оркестрация: Dataflow вместо Channels

В net7 оркестрация — это четыре массива Task’ов, ёмкость каналов 1 и ручное протаскивание завершения:

readToParse.Writer.Complete();
await Task.WhenAll(parserThreads);
parseToSort.Writer.Complete();
await Task.WhenAll(sorterThreads);
...

TPL Dataflow (который давно входит в .net) делает то же самое декларативно:

var sorter = new TransformBlock<Chunk, SortedChunk>(
    chunk => chunk.Sort(encoding, compareInfo, compareOptions, isOrdinal, ct),
    new() { 
            MaxDegreeOfParallelism = stages,
            BoundedCapacity = stages + 2,
            EnsureOrdered = false, 
            CancellationToken = ct });
var compressor = new TransformBlock<SortedChunk, CompressedChunk>(...);
var saver = new ActionBlock<CompressedChunk>(...);
sorter.LinkTo(compressor, new DataflowLinkOptions { PropagateCompletion = true });

• PropagateCompletion вместо ручного протаскивания завершения по Channel

• BoundedCapacity дает backpressure

• EnsureOrdered = false обязателен — иначе степень параллелизма больше единицы бессмысленна.

Число блоков сортировки и сжатия я подбирал на своем компьютере, получилось 1/4 от количества ядер.

Формат чанка и сжатие

Словарь изменил формат временных файлов: вместо lineLen, line, keyLen, key на каждую строку в net7, я сделал одд=ну запись на каждый уникальный текст: varint(textLen) text varint(count) δ-varint значения. Числа идут дельтами по возрастанию внутри ключа — значит, они уже отсортированы, и мерж-фазе остаётся сливать упорядоченные потоки.

И одна сделка: sort key в файл больше не пишется — мерж пересчитывает его при чтении. Ключ в ICU — до 3.5 байта на символ; хранение ключей раздувает чанки в 2–4 раза, а мерж на HDD-bound данных упирается в объём чтения. Лучше еще раз получать ключи мерже — там, где ядра всё равно простаивали.

Сжатие — one-shot: размер сериализованного чанка считается проходом, под него выделяется ровно столько памяти, весь чанк жмётся одним вызовом ZstandardEncoder.Compress (окно адаптивное: бюджет ~2 ГБ на все параллельные декодеры мержа, делённый на число чанков) — и пишется одним Write в преаллоцированный файл точного размера. Уровень zstd по умолчанию — 9: подъём до 15 на случайных строках дал 7–8 минут сплита против 2:30 — высшие уровни жгут CPU на поиск повторов, которых нет.

Мерж: первая попытка провалилась

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

На 100 ГБ (206 чанков) получилось 1:28:46 — и мониторинг показал почему: диск простаивал 70–90% времени. Синхронная запись на потоке мержа мешала чтению: пока шла запись буфера, никто не читал чанки; пока читались чанки, запись стояла. Шпиндель метался между двумя файлами короткими рывками, агрегатная скорость падала в разы. Классическая ошибка для задачи, где IO — главный ресурс: читатель должен читать всегда.

Обновленный мерж: пайп на чанк, канал между задачами, оператор для слияния

Переписал по принципу «каждому занятию — свой исполнитель». На каждый чанк — три сущности:

Pipe chunkPipe = new(new PipeOptions(
    pool: memoryPool, 
    minimumSegmentSize: 1 << 20, // 1MB
    resumeWriterThreshold: 8 << 20, // 8MB 
    pauseWriterThreshold: 16 << 20, // 16MB
    useSynchronizationContext: false));
producers.Add(ProduceChunk(file, chunkPipe.Writer, ct));        // FileStream → zstd → пайп

Channel<ChunkEntry> channel = Channel.CreateBounded<ChunkEntry>(16);
materializers.Add(MaterializeChunk(chunkPipe.Reader, channel.Writer, ...)); // парсинг + sort key

sources.Add(channel.Reader.ReadAllAsync(ct));                    // IAsyncEnumerable записей

• Продюсер читает файл через FileStream с буфером 4 МБ и SequentialScan и перекачивает ZstandardStream в пайп. Большой буфер чтения — не прихоть: ОС получает последовательные запросы по 4 МБ независимо от чередования чтений сотни чанков, мелкие запросы роняют полосу HDD в разы.

• Парсер разбирает пайп, материализует каждую запись в один арендованный из пула массив — [текст][sort key][значения] — и кладёт в ограниченный канал на 16 записей.

• Получение sort key идёт параллельно по всем чанкам

• Channel даёт backpressure — останавливает читателей, которые читают строки из чанка быстрее, чем эти строки мержатся.

• Поток мержа только сравнивает ключи и собирает равные в группы, группы попадают в ActionBlock-писатель.

• ActionBlock-писатель пишет группы в выходной FileStream (буфер 4 МБ, преаллокация точного размера) синхронно в своём потоке — запись не останавливает мерж.

Каналы здесь работают как мост от PipeReader к IAsyncEnumerable: channel.Reader.ReadAllAsync(ct) превращает N каналов в N потоков записей, для которых написан отдельный оператор:

await foreach (var group in sources.MergeSorted(ChunkEntryComparer.Instance, groupPool, ct))
    await writer.SendAsync(group, ct);

MergeSorted — классический k-way merge на PriorityQueue с одной нетривиальной деталью. MoveNextAsync всех источников нужно делать параллельно, и желательно асинхронно, не дожидаясь результата. Синхронно завершённые MoveNextAsync разбираются сразу, отложенные ждут раундом Task.WhenEach:

void Advance(IAsyncEnumerator<ChunkEntry> enumerator)
{
    ValueTask<bool> vt = enumerator.MoveNextAsync();
    if (vt.IsCompletedSuccessfully)
    {
        if (vt.Result) pq.Enqueue(enumerator, enumerator.Current);
    }
    else flight.Add((enumerator, vt.AsTask()));
}

Грабля, на которую я потратил вечер отладки: синхронно завершённый MoveNextAsync().AsTask() у async-итераторов возвращает один общий кэшированный Task — положишь такие в словарь, получишь дубликаты ключей и тихую порчу данных. Поэтому в полёт попадают только действительно отложенные задачи, а синхронные результаты разбираются немедленно.

Равные ключи собираются в MergeGroup и на него было потрачего много моего времени и токенов ИИ. Группа из одной записи — а их большинство для случайных строк — не копируется вовсе: группа держит ChunkEntry и пишет прямо из его сегментов. Когда добавляется второй ChunkEntry группа материализует групповую раскладку в одном буфере: [count][ключ][текст₁][значения₁][текст₂][значения₂]…, данные растут от головы, мета информация (смещение/длина текста, смещение/число значений) — от хвоста, при встрече буфер удваивается. Значения записей уже отсортированы — при добавлении они просто memcpy’ятся, а при выдаче сливаются k-way кучей по возрастанию. После записи группа уходит в ObjectPool: Reset() возвращает все арендованные массивы.

Результат переписывания: мерж 100 ГБ с 1:28:46 до 29:08, простоев диска — ноль, запись держит ровные 100–130 МБ/с на фоне непрерывного чтения чанков — это максимум для моего HDD.

Пулы везде

Оставшиеся аллокации добили пулами, трёх уровней:

• ArrayPool<byte>.Shared — массивы записей мержа и буферы групп/

• Отдельный ArrayPool для чанков — текстовые буферы, sort keys, буферы сериализации и сжатия, сегменты пайпов (через обёртку MemoryPool<byte> над ArrayPool), с максимальным размером более 1 МБ.

• ObjectPool — словари чанков и группы мержа.

Итог по 100-гигабайтному прогону: суммарные GC-паузы 17 секунд на 52 минуты (0,5%), рабочий набор стабилен (плато ~8 ГБ; скачок до 10,4 ГБ на старте мержа — это 206 пайпов по 16 МБ плюс 206 zstd-декодеров, дальше константа). Цена — владение памятью становится ручным: компилятор не следит, вернул ли ты массив в пул, и один пропавший Dispose в ранней версии съедал буферы молча.

Цифры

Один и тот же вход 10 ГБ, разные размеры чанка — влияет ли число источников на мерж:

Merge почти не зависит от числа источников: 282 пайпа и канала работают так же, как четыре. Для сравнения, в версии с синхронной записью 206 источников стоили 1:28:46.

100 ГБ, холодный HDD:

Важно: . net7 мерил на «Войне и мире», где Brotli жал временные файлы в 20 раз; финал — на случайных строках, где zstd даёт 1.3×: энтропийному кодировщику нечего находить. На 100 ГБ это разница между ~10 ГБ и ~77 ГБ промежуточных файлов через один шпиндель. На “Войне и мире” финальный код тоже быстрее старого — там получается 2-3 чанка и 12 минут.

• сплит — чтение и запись вперемешку до 210 МБ/с (эффект кэша файловой системы), CPU до 100% на сортировке чанков.

• мерж — ровная запись 100–130 МБ/с с непрерывным фоновым чтением, disk time 100–400% (очередь к шпинделю — это нормально), CPU 20–100% пилой параллельных парсеров.

Итог: что каждое решение дало

1. Словарь-дедупликация чанка (предложение из комментариев). Эффективность: sort key считается на уникальный текст, на данных с повторами чанк сжимается кратно — вплоть до вырождения в in-memory. Структура: словарь текст → (номер, count) стал ядром чанка, формат временных файлов — по уникальным текстам.

2. Alternate lookup (.NET 9). Эффективность: ноль аллокаций на ключи словаря — текст копируется в буфер чанка один раз. Структура: хэш, сравнение и аллокация ключа собраны в одном типе-компарере; словарь — над структурами-адресами.

3. CSR вместо списков значений (раскладку подсказал ИИ). Эффективность: 8 байт на значение подряд вместо 32+ в куче; сортировка значений — независимыми диапазонами в Parallel.For. Структура: накопление отделено от раскладки (префикс-суммы, scatter) — стадии Chunk → SortedChunk явные.

4. varint + дельты в формате чанка. Эффективность: значение 1–5 байт вместо восьми, числа уже упорядочены. Структура: один формат на обе фазы, мерж сливает отсортированные потоки без досортировки.

5. Pipe для чтения. Эффективность: по бенчмаркам — не медленнее ручных буферов при меньшей памяти, backpressure 8/16 МБ бесплатно. Структура: парсинг через SequenceReader — два десятка строк вместо страницы арифметики с остатками.

6. TPL Dataflow в сплите. Эффективность: backpressure и параллелизм декларацией. Структура: три блока и LinkTo вместо массивов Task и ручного Complete().

7. Пайп + две задачи + канал на каждый чанк мержа. Эффективность: чтение перестало ждать CPU и записи. Структура: поток мержа делает только сравнение ключей; IO и разбор — в задачах.

8. MergeSorted как оператор IAsyncEnumerable. Эффективность: «полёт» MoveNext-задач держит сотни источников читаемыми одновременно. Структура: тестируемый переиспользуемый оператор; каналы превращаются в потоки одним ReadAllAsync.

9. MergeGroup: zero-copy для одиночной записи, материализация для групп, пул. Эффективность: большинство групп — одна запись, ноль копий и аллокаций. Структура: раскладка «данные с головы, мета с хвоста» — один буфер на группу.

10. Пулы массивов и памяти на всех уровнях. Эффективность: GC-паузы 17 с на 52 минуты, стабильный рабочий набор. Структура: владение явное (Return/Reset/Dispose), но требует дисциплины — компилятор за ним не следит.

11. One-shot zstd, адаптивное окно, преаллокация файлов точным размером. Эффективность: один вызов компрессора на чанк, ни одного расширения файла по кусочкам. Структура: сериализация и сжатие — одна стадия, точный размер известен до записи.

Честный итог по вкладам: самое большое ускорение дала не алгоритмика, а развязка IO и CPU — семьдесят минут мержа исчезли от того, что чтение перестало стоять в очереди за записью. Словарь из комментариев и CSR дали структуру, без которой развязывать было бы нечего.

Репозиторий — gandjustas/HugeFileSort: ветка net7 — финал первой части, master — этот код.

PS. Runtime async в .NET 11 Preview тоже тестил - получилось медленнее, повторю после релиза.