Это продолжение статьи «Сортировка слиянием — не так просто, как кажется». Там я довёл тестовое задание — отсортировать 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 для культурно-чувствительного сравнения считается один раз на уникальный текст :
Вместе с заполнением массивов смещений для CSR ключи из байтов превращаются в char в отдельном буфере на весь чанк.
Параллельно
GetSortKeyLengthна каждый ключ, потом подсчет точного размера буфера для sort keys и префикс-сумма для смещений в буфере.Параллельно
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вместо ручного протаскивания завершения по ChannelBoundedCapacityдает backpressureEnsureOrdered = 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 ГБ, разные размеры чанка — влияет ли число источников на мерж:
Прогон | Чанков | Split | Merge |
|---|---|---|---|
Настройки по умолчанию | 4 | 1:05 | 1:08 |
Форсированно мелкий чанк | 282 | 1:19 | 1:01 |
Merge почти не зависит от числа источников: 282 пайпа и канала работают так же, как четыре. Для сравнения, в версии с синхронной записью 206 источников стоили 1:28:46.
100 ГБ, холодный HDD:
Split | Merge | Всего | |
|---|---|---|---|
net7 (Война и мир, Brotli, ключи на диске) | 11:35 | 20:16 | ~32 мин |
первая версия нового мержа, случайные строки | 22:59 | 1:28:46 | ~1:52 |
финал, случайные строки | 23:15 | 29:08 | 52:23 |
Важно: . net7 мерил на «Войне и мире», где Brotli жал временные файлы в 20 раз; финал — на случайных строках, где zstd даёт 1.3×: энтропийному кодировщику нечего находить. На 100 ГБ это разница между ~10 ГБ и ~77 ГБ промежуточных файлов через один шпиндель. На “Войне и мире” финальный код тоже быстрее старого — там получается 2-3 чанка и 12 минут.
сплит — чтение и запись вперемешку до 210 МБ/с (эффект кэша файловой системы), CPU до 100% на сортировке чанков.
мерж — ровная запись 100–130 МБ/с с непрерывным фоновым чтением, disk time 100–400% (очередь к шпинделю — это нормально), CPU 20–100% пилой параллельных парсеров.
Итог: что каждое решение дало
Словарь-дедупликация чанка (предложение из комментариев). Эффективность: sort key считается на уникальный текст, на данных с повторами чанк сжимается кратно — вплоть до вырождения в in-memory. Структура: словарь
текст → (номер, count)стал ядром чанка, формат временных файлов — по уникальным текстам.Alternate lookup (.NET 9). Эффективность: ноль аллокаций на ключи словаря — текст копируется в буфер чанка один раз. Структура: хэш, сравнение и аллокация ключа собраны в одном типе-компарере; словарь — над структурами-адресами.
CSR вместо списков значений (раскладку подсказал ИИ). Эффективность: 8 байт на значение подряд вместо 32+ в куче; сортировка значений — независимыми диапазонами в
Parallel.For. Структура: накопление отделено от раскладки (префикс-суммы, scatter) — стадииChunk → SortedChunkявные.varint + дельты в формате чанка. Эффективность: значение 1–5 байт вместо восьми, числа уже упорядочены. Структура: один формат на обе фазы, мерж сливает отсортированные потоки без досортировки.
Pipe для чтения. Эффективность: по бенчмаркам — не медленнее ручных буферов при меньшей памяти, backpressure 8/16 МБ бесплатно. Структура: парсинг через
SequenceReader— два десятка строк вместо страницы арифметики с остатками.TPL Dataflow в сплите. Эффективность: backpressure и параллелизм декларацией. Структура: три блока и
LinkToвместо массивов Task и ручногоComplete().Пайп + две задачи + канал на каждый чанк мержа. Эффективность: чтение перестало ждать CPU и записи. Структура: поток мержа делает только сравнение ключей; IO и разбор — в задачах.
MergeSortedкак операторIAsyncEnumerable. Эффективность: «полёт» MoveNext-задач держит сотни источников читаемыми одновременно. Структура: тестируемый переиспользуемый оператор; каналы превращаются в потоки однимReadAllAsync.MergeGroup: zero-copy для одиночной записи, материализация для групп, пул. Эффективность: большинство групп — одна запись, ноль копий и аллокаций. Структура: раскладка «данные с головы, мета с хвоста» — один буфер на группу.Пулы массивов и памяти на всех уровнях. Эффективность: GC-паузы 17 с на 52 минуты, стабильный рабочий набор. Структура: владение явное (
Return/Reset/Dispose), но требует дисциплины — компилятор за ним не следит.One-shot zstd, адаптивное окно, преаллокация файлов точным размером. Эффективность: один вызов компрессора на чанк, ни одного расширения файла по кусочкам. Структура: сериализация и сжатие — одна стадия, точный размер известен до записи.
Честный итог по вкладам: самое большое ускорение дала не алгоритмика, а развязка IO и CPU — семьдесят минут мержа исчезли от того, что чтение перестало стоять в очереди за записью. Словарь из комментариев и CSR дали структуру, без которой развязывать было бы нечего.
Репозиторий — gandjustas/HugeFileSort: ветка net7 — финал первой части, master — этот код.
PS. Runtime async в .NET 11 Preview тоже тестил - получилось медленнее, повторю после релиза.
