Параллельная распаковка ZIP в Delphi: шлюз чтения HotXLS

Два потока делят один TFileStream над пакетом .xlsx. Поток A ставит Position на первый сжатый байт sheet1.xml, поток B мгновение спустя — на первый байт sheet2.xml, и Read потока A возвращает буфер, принадлежащий другому члену архива. Распаковщик потока A теперь декодирует середину чужого потока deflate; zlib обычно останавливается с Z_DATA_ERROR через несколько блоков, а если на выходе что-то и получилось, CRC32 члена не сойдётся. Deflate тут ни при чём. Гонка живёт целиком в курсоре позиции потока, и одной блокировки вокруг пары seek-и-read достаточно, чтобы много воркеров распаковывали члены одного пакета одновременно

Причина, по которой это вообще кому-то нужно, — память. Пусть книга весит 80 МБ на диске, а её XML листов при распаковке расширяется в пять-десять раз. Путь открытия, который извлекает каждый лист в буфер до разбора, платит этими распакованными байтами поверх книги, которую он строит, и пик наступает до того, как существует хоть одна ячейка. Эта статья — про конкурентность на уровне пакета, убирающую шаг извлечения. Потолок аллокатора над ней разбирает статья о параллельном разборе XLSX и менеджере памяти, а API чтения-один-раз, который вообще не строит книгу, — прогон потокового прямого читателя

Staging-дизайн и что он держит в памяти

Осторожный способ распараллелить открытие XLSX — трёхфазный конвейер, где на воркерах работает только средняя фаза. Фаза A на вызывающем потоке обходит список листов, создаёт каждый лист, читает его часть отношений и копирует весь распакованный XML листа в приватный TMemoryStream. Фаза B раскидывает разбор XML по пулу потоков. Фаза C возвращается в архив, по-прежнему на вызывающем потоке, за мелкими спутниками: комментариями, ветками комментариев, рисунками, диаграммами и таблицами

Аргумент за такую форму звучит непробиваемо. ZIP-архив и его состояние распаковки не потокобезопасны, так что держите каждое касание архива на одном потоке и вообще обходитесь без блокировок. Счёт за это — книга с восемью занятыми листами, которая в один и тот же момент держит восемь полностью распакованных буферов листов, и эти буферы — крупнейшие транзиентные объекты на всём пути открытия. При предполагаемых выше 80 МБ пакета это несколько сотен мегабайт XML-текста, существующего только ради того, чтобы парсер прочитал его один раз

HotXLS сохраняет фазы A и C, но выбрасывает копию. Каждый воркер фазы B открывает собственный поток записи на общем пакете и отдаёт его прямо XML-ридеру, так что распакованный лист никогда не существует одним куском. Это работает потому, что аргумент "не потокобезопасно" на поверку оказывается о двух разных вещах

Могут ли два потока распаковывать из одного потока ZIP?

Да, если каждый поток владеет своим распаковщиком, а общий источник байт сериализуется на чтении. Осторожный аргумент склеивает два вида состояния в одном предложении. Состоянием распаковки действительно нельзя делиться: zlib z_stream несёт скользящее окно на 32 КБ, которое определяет RFC 1951, текущие таблицы Хаффмана и битовую позицию внутри одного сжатого члена, и два потока, толкающие байты через один такой, производят мусор. Источник байт — другое дело. У файлового потока ровно один кусок изменяемого состояния, стоящий защиты, — его позиция

Контейнер ZIP делает это разделение законным. APPNOTE от PKWARE (раздел 4.3) раскладывает пакет на независимые члены, у каждого свой локальный заголовок файла, свои сжатые данные по собственному смещению и свои CRC32 и размеры, записанные в центральном каталоге. Ни один словарь не тянется через члены, как в сплошном блоке 7z, так что член N можно распаковать, не прочитав ни байта члена M. Дайте каждому воркеру приватный распаковщик над его диапазоном байт — и единственная оставшаяся коллизия это seek. В HotXLS этой коллизией занимается TZipReadGate в lxZipArchive.pas, и весь класс помещается на один экран

type
  TZipReadGate = class
  private
    FBaseStream: TStream;
    FLock: TXLSCriticalSection;   // обёртка lxPlatform над критической секцией
  public
    constructor Create(ABaseStream: TStream);
    destructor Destroy; override;
    function ReadAt(AOffset: Int64; var Buffer; Count: Longint): Longint;
  end;

function TZipReadGate.ReadAt(AOffset: Int64; var Buffer;
  Count: Longint): Longint;
begin
  if Count <= 0 then
  begin
    Result := 0;
    Exit;
  end;
  XLSEnterCriticalSection(FLock);
  try
    FBaseStream.Position := AOffset;
    Result := FBaseStream.Read(Buffer, Count);
  finally
    XLSLeaveCriticalSection(FLock);
  end;
end;

Что блокирует шлюз чтения, а что остаётся за его пределами

TZipReadGate.ReadAt охраняет одну неделимую операцию — позиционирование общего потока и чтение из него. TZipArchive.OpenArchive строит шлюз, как только разобран центральный каталог, для перегрузок и по имени файла, и по TStream, а освобождение архива выпускает его. Архивам, созданным для записи, шлюз не достаётся никогда, потому что читать из них конкурентно нечего

Всё остальное остаётся вне блокировки, потому что уже приватно или уже заморожено. Каждый поток записи, который возвращает TZipArchive.OpenFile, хранит собственную позицию внутри своего члена и переводит её в абсолютное смещение пакета только в момент вызова ReadAt. Обёрнутый вокруг него распаковщик создаётся на каждый член с размером окна -15, что zlib читает как сырой deflate без zlib-заголовка, и никогда не передаётся другому потоку. Центральный каталог, включая локальные заголовки, полностью разобран до старта любого воркера, так что к моменту, когда начинается конкурентность, поиск члена по имени — это read-only поиск по хешу. Поиск не учитывает регистр и отвергает имена, содержащие сегменты .. в любом месте

Одно следствие важно, когда вы вызываете TXLSXWorkbook.Open(Stream) со своим потоком. Шлюз оборачивается вокруг этого потока, и его блокировка координирует воркеров библиотеки между собой. О вашем коде он ничего не знает. Если другой поток вашего приложения читает или переставляет тот же TStream, пока идёт Open, гонка курсора из первого абзаца возвращается, так что отдайте поток и не трогайте его до возврата Open

Сколько стоит одна блокировка под конкуренцией?

Меньше, чем подсказывает "глобальная блокировка архива", — из-за того, сколько работы покрывает каждое приобретение. Распаковщик пополняет вход чанками сжатых байт по 64 КБ, так что один проход через шлюз подаёт воркеру 64 КБ данных deflate, которые он затем декодирует и разбирает, ничего не удерживая. При предполагавшихся выше пяти-десяти кратках расширения это пополнение превращается примерно в 320–640 КБ разметки листа. Сама блокировка покрывает позиционированное чтение, которое обычно обслуживается из файлового кэша операционной системы

Соотношение переворачивается в трёх местах, и все они про вход или машину. Члены, хранящиеся без сжатия, проходят через шлюз один раз на вызов чтения без распаковочной работы между приобретениями; большинство писателей сжимают deflate части листов, но пакет от инструмента, который всё хранит, сериализуется жёстче. Холодный файл на медленном или сетевом хранилище расширяет критическую секцию, поскольку чтение внутри неё становится настоящей передачей. А за пределами горстки воркеров шлюз — не первая стена: разбор листов жирно аллоцирует, и менеджер памяти Delphi сериализует аллокации между потоками задолго до того, как насытится шлюз чтения

Последний пункт и определяет число потоков. TXLSXWorkbook.ParallelParseThreads по умолчанию 0 — по воркеру на логический процессор, но никогда больше числа листов. На ноутбуке с четырьмя-восемью логическими процессорами это значение в порядке. На 32-поточном сборочном сервере, открывающем книги с десятками листов, поставьте явный небольшой предел вроде 4: лишние воркеры в основном стоят в очереди к аллокатору, и каждый держит свой распаковщик и буферы. Чтобы найти правильное число для своих файлов, откройте одну показательную книгу в цикле со значением 1, 2, 4 и 8 и сравните время по стенке с пиковым working set в Task Manager или Process Explorer

Дочитывание записи до конца — то, что запускает проверку CRC

Каждый поток записи, который возвращает архив, проверяет сам себя, но только в конце. Он копит бегущий CRC32, пока байты проходят через него, а когда позиция достигает распакованного размера из центрального каталога, сравнивает значения и щупает один лишний байт. Отказы всплывают исключениями с литеральными сообщениями: 'ZIP entry uncompressed size mismatch', 'ZIP entry CRC32 mismatch', 'ZIP entry exceeds its declared size' и 'ZIP entry ended before its declared size' для члена, закончившегося раньше времени

XML-ридер останавливается на закрывающем теге </worksheet>. Всё, что после него — часто просто хвостовой перевод строки, — остаётся непрочитанным, позиция никогда не доходит до заявленного размера, и сравнение не выполняется никогда. Staging-дизайн получал эту проверку бесплатно, поскольку копирование записи в буфер читает каждый байт. Парсер, читающий поток записи напрямую, — нет, поэтому каждый воркер HotXLS дочитывает то, что оставил парсер, буфером на 32 КБ, прежде чем освободить поток, а освобождение внешнего потока выпускает распаковщик и подпоток под ним

То же правило действует, когда ваш собственный код использует TZipArchive из lxZipArchive, скажем для проверки входящего пакета перед передачей его загрузчику книги. Функция, читающая лишь первые килобайты каждой части, чтобы разглядеть корневой элемент, не доказывает ничего о целостности. Дочитывайте до конца:

uses
  Classes, SysUtils, lxZipArchive;

// Дочитывает одну часть пакета до заявленного конца, чтобы выполнились проверки размера и CRC32.
// PartIsIntact — хелпер, который вы пишете сами; он не входит в библиотеку
function PartIsIntact(Zip: TZipArchive; const PartName: WideString;
  out Problem: string): Boolean;
var
  Stream: TStream;
  Buffer: array [0..32767] of Byte;
begin
  Result := False;
  Problem := '';
  Stream := Zip.OpenFile(PartName);   // nil, когда части нет
  if Stream = nil then
  begin
    Problem := 'part not found';
    Exit;
  end;
  try
    try
      while Stream.Read(Buffer, SizeOf(Buffer)) > 0 do
        ;
      Result := True;
    except
      on E: Exception do
        Problem := E.Message;          // например 'ZIP entry CRC32 mismatch'
    end;
  finally
    Stream.Free;
  end;
end;

var
  Zip: TZipArchive;
  Problem: string;
begin
  Zip := TZipArchive.Create;
  try
    Zip.OpenArchive('quarterly-consolidation.xlsx');
    if not PartIsIntact(Zip, 'xl/worksheets/sheet1.xml', Problem) then
      Writeln('sheet1.xml: ', Problem);
  finally
    Zip.Free;
  end;
end;

Вторая ловушка в этом коде — перемотка. Потоки записи идут только вперёд: чтение Size работает, потому что заявленный размер известен заранее, но присвоение Position := 0 после подглядывания возбуждает 'ZIP entry stream is forward-only'. Если компонент настаивает на seek, откройте запись вторично через OpenFile — это дёшево и даёт свежий поток с нулевым смещением. Несколько потоков могут звать OpenFile на одном архиве, открытом для чтения, — ровно это делает загрузчик книги, — пока ничто не освобождает архив, пока живы их потоки

Что всё ещё выполняется последовательно и выключатель, который это отключает

Фаза A всё ещё выполняется, минус извлечение. Она создаёт каждый лист и читает его отношения на вызывающем потоке, что и оставляет все разделяемые карты неизменяемыми после старта воркеров. Таблица общих строк и таблицы форматов ячеек полны до начала фазы B и никогда не пишутся во время неё. Фаза C после всё так же обходит листы по порядку за комментариями, рисунками, диаграммами и таблицами, пропуская лист, чьей части нет в пакете

uses
  SysUtils, lxHandleX;

var
  Wb: TXLSXWorkbook;
begin
  Wb := TXLSXWorkbook.Create;
  try
    Wb.ParallelParse := True;        // по умолчанию; False разбирает по одному листу за раз
    Wb.ParallelParseThreads := 4;    // 0 = по воркеру на логический процессор
    Wb.SelectedSheets.Add('Q3 Detail');  // опционально: разобрать ячейки только этого листа
    if Wb.Open('quarterly-consolidation.xlsx') <> 1 then
      raise Exception.Create('Workbook open failed');
    // Загруженная книга идентична, каким бы путём она ни разбиралась
  finally
    Wb.Free;
  end;
end;

Установка ParallelParse в False перед Open выполняет ту же работу по листам с числом потоков один, и пул вырождается в простой цикл на вызывающем потоке. Книга с единственным листом идёт этим путём в любом случае. Держите выключатель в уме при диагностике проблем открытия: он убирает потоки из картины одной строкой, и оба пути гоняют один и тот же код разбора, так что полученная книга не зависит от настройки

Ошибки тоже сохраняют серийную форму. Когда воркер возбуждает исключение, новые листы не диспетчеризуются, все воркеры соединяются, и исключение с наименьшим индексом листа перевозбуждается на вызывающем потоке, так что битый sheet3.xml по-прежнему всплывает одним исключением из Open. При RecoveryMode равном True лист, не подавшийся разбору, пропускается и записывается предупреждением в диагностику книги. Листы, не попавшие в SelectedSheets, вообще не открывают свой поток записи, а MetadataOnly пропускает параллельную фазу целиком. Более широкая настройка пути открытия разобрана в руководстве по производительности больших книг в Delphi

Шлюз чтения, параллельная фаза открытия и доступ к записям ZIP уровня записи, показанные здесь, входят в HotXLS Delphi Excel component для Delphi и C++Builder, поставляемый с полным исходным кодом, так что можно читать lxZipArchive.pas рядом со свойствами открытия TXLSXWorkbook, описанными на странице продукта