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

HotXLS, нативна бібліотека компонентів Excel для Delphi та C++Builder, розпаковує кілька аркушів XLSX одночасно з одного відкритого пакета ZIP. Механізм — TZipReadGate, невеликий клас у lxZipArchive.pas, що тримає потік пакета плюс одну критичну секцію й надає рівно один метод. Він серіалізує пару позиціонування та читання. Усе над цією парою виконується паралельно

Проблема, що змусила цей дизайн, — та, з якою стикався кожен розробник Delphi, що відкривав велику книгу. 80 МБ xlsx — це 80 МБ дефлейтованого XML, а частини аркушів усередині нього розширюються приблизно у п'ять-десять разів. Якщо ваш шлях відкриття витягує кожен аркуш у потік пам'яті перед розбором, ви платите за розпаковані байти на додачу до книги, яку будуєте, і пік настає до того, як створено хоч одну клітинку. Ця стаття про паралелізм на рівні пакета, що усуває цей проміжний крок. Стеля алокатора, що стоїть над цим, охоплена в статті про паралельний розбір XLSX і менеджер пам'яті, а API читання-один-раз, що ніколи не матеріалізує, охоплено в огляді потокового прямого читача

Чому старий шлях відкриття тримав кожен аркуш у RAM

Оригінальне паралельне відкриття в HotXLS було трифазним конвеєром, і середня фаза була єдиною, що виконувалась на воркерах. Фаза A послідовно обходила список аркушів, створювала кожен аркуш, читала його частину зв'язків і копіювала весь розпакований XML аркуша в приватний TMemoryStream. Фаза B розгортала ParseWorksheetXml по пулу. Фаза C поверталась до архіву в потоці виклику для малих супутніх частин: коментарів, потокових коментарів, малюнків, діаграм, таблиць. Ця форма була обрана з заявленої причини. Заголовковий коментар у lxParallelParse.pas раніше говорив прямим текстом, що zip-архів і його стан декомпресії не потокобезпечні, а внутрішні нотатки йшли далі: не турбуйтеся блокувати архів, бо коли стан декомпресії серіалізований по записах, замок нічого не купує. Фаза A існувала, щоб утримати кожен доторк до архіву в одному потоці. Ціна полягала в тому, що книга з вісьмома активними аркушами тримала вісім повністю розпакованих буферів XML аркушів у пам'яті одночасно, і ці буфери — найбільші перехідні об'єкти в усьому шляху відкриття

Чи можуть два потоки декомпресовувати з одного потоку ZIP?

Так, і старе судження було неправильним конкретним, локалізовним способом: воно згорнуло два різних шматки стану в одне речення. Стан декомпресії справді не можна ділити. z_stream zlib несе ковзне вікно, таблиці Хаффмана й бітову позицію для одного стисненого члена, і два потоки, що штовхають байти через один і той самий, видадуть сміття. Базове джерело байтів — зовсім інше питання, і відповідь там у тому, що потік файлу має рівно один шматок мутабельного спільного стану, вартого захисту, — свій курсор позиції

Контейнер ZIP робить розділення легальним. Кожен член у ZIP-архіві стискається незалежно: власний локальний заголовок файлу, власний бітовий потік deflate на власному DataOffset, власний CRC32 і розміри в центральному каталозі. Немає спільного словника, що охоплює члени, як у суцільному блоці 7z, тож запис N можна розпакувати, не торкаючись запису M. Дайте кожному воркеру власний z_stream над власним діапазоном байтів, і єдине, на чому вони зіштовхуються, — це позиціонування. Саме це зіткнення й усуває TZipReadGate, і весь клас достатньо короткий, щоб прочитати на одному екрані

type
  TZipReadGate = class
  private
    FBaseStream: TStream;
    FLock: TRTLCriticalSection;
  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;
  EnterCriticalSection(FLock);
  try
    FBaseStream.Position := AOffset;
    Result := FBaseStream.Read(Buffer, Count);
  finally
    LeaveCriticalSection(FLock);
  end;
end;

Що захищає TZipReadGate, а чого навмисно не захищає

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

Усе інше залишається поза замком, бо воно вже приватне чи вже незмінне. TZipSubStream тримає власну FPosition, тож кожен воркер відстежує власне місце у власному записі. TZLibStream, що TZipEntry.GetStream будує над цим підпотоком, — на кожен запис, створений з windowBits -15 для сирого deflate, і ніколи не спільний. Центральний каталог повністю розібраний до того, як почне будь-який воркер, включно з кожним локальним заголовком, тож GetEntryByName — це пошук у хеші лише для читання на момент початку паралелізму. Сама маршрутизація — три рядки в TZipSubStream.Read, і гілка без шлюзу — те, що утримує кожного наявного однопотокового викликача на старому шляху коду

function TZipSubStream.Read(var buffer; Count: longint): longint;
var
  rest: Int64;
  rc: longint;
begin
  rest := FSize - FPosition;
  if (Count > rest) then
    Count := rest;
  if FReadGate <> nil then
    rc := FReadGate.ReadAt(FOffset + FPosition, buffer, Count)
  else
  begin
    FBaseStream.Position := FOffset + FPosition;
    rc := FBaseStream.Read(buffer, Count);
  end;
  FPosition := FPosition + rc;
  Result := rc;
end;

Скільки коштує шлюз під конкуренцією?

Менше, ніж підказує фраза "глобальний замок на архів", завдяки гранулярності, яку випадково використовує TZLibStream. Його вхідний буфер — BufferSize, визначений як $4000, тож ReadInputBuffer витягує 16 КБ стиснених байтів за поповнення й передає їх zng_inflate. Тому одне отримання замка покриває 16 КБ вхідних даних deflate, що для XML аркуша розширюється у щось порядку 100 КБ розмітки, яку воркер потім декодує й розбирає, нічого не тримаючи. Замок утримується на позиційне читання проти кешу операційної системи; робота, яку він гейтує, вимірюється мілісекундами

Чесна межа там, де це співвідношення перевертається. Записи, збережені без стискання, а не дефлейтовані, читаються через шлюз один до одного без роботи декомпресії, що приховала б затримку, тож пакет, повний нестиснутих членів, серіалізувався б набагато сильніше. Холодний файл на повільному носії розширює критичну секцію, бо читання всередині неї тепер — реальна передача з диска, а не влучання в кеш. І понад жменькою воркерів шлюз — не те, у що ви впираєтесь першим у будь-якому разі: розбір аркуша насичений виділеннями, і менеджер пам'яті Delphi серіалізує виділення по потоках задовго до того, як шлюз читання стане обмеженням. Саме тому TXLSXWorkbook.ParallelParseThreads за замовчуванням має автоматичну стелю замість одного потоку на ядро

Тіло воркера та цикл вичерпання, який легко забути

З шлюзом на місці HotXLS взагалі видалила стадіювання фази A. Воркер тепер відкриває власний потік запису й подає його прямо в парсер. Два перехідні поля несуть входи: FParZip тримає архів на час паралельної фази, FParSheetPartNames тримає імена частин, і обидва очищаються в блоці finally, тож жоден застарілий покажчик не переживає невдале відкриття. Потік, що повертається з TZipArchive.OpenFile, — це TZipVerifiedStream, що обгортає TZLibStream, що обгортає TZipSubStream, і звільнення зовнішнього звільняє весь ланцюжок

procedure TXLSXWorkbook.ParseSheetJob(AIndex: Integer);
var
  Stream: TStream;
  DrainBuffer: array [0..32767] of Byte;
  PartName: WideString;
begin
  PartName := WideString(FParSheetPartNames[AIndex]);
  Stream := FParZip.OpenFile(PartName);
  if Stream = nil then
    Exit;
  try
    ParseWorksheetXml(Stream, FSheets.ByPos[AIndex], FParSst,
      FParRels[AIndex], FParFontMap, FParFillMap, FParBorderMap,
      FParNumFmtMap, FParAlignMap, FParProtMap, FParDateMap);
    // Consume any trailing bytes so the ZIP entry size and CRC are verified.
    while Stream.Read(DrainBuffer, SizeOf(DrainBuffer)) > 0 do
      ;
  finally
    Stream.Free;
  end;
end;

Цикл вичерпання — це деталь, яку прямий перенос старого коду упустив би, і його втрата тихо вимикає перевірку цілісності. TZipVerifiedStream накопичує біжучий CRC32 у міру проходження байтів і викликає VerifyComplete лише коли його позиція досягає нестисненого розміру, записаного в центральному каталозі; саме звідти й беруться винятки невідповідності розміру та CRC32, плюс однобайтове зондувальне читання, що ловить запис, довший за заявлений. Читач XML зупиняється на закриваючому елементі й зазвичай залишає непрочитаним новий рядок чи кілька байтів кінцевого пробілу, тож без вичерпання позиція ніколи не досягає заявленого розміру, і перевірки ніколи не спрацьовують. Читання решти в чорновий буфер нічого не коштує й відновлює їх. Коли існували стадіювальні потоки, XlsxCopyStreamAll робила це випадково

Що все ще виконується послідовно, і прапорець, що вимикає все

Фаза A виживає, мінус витяг. Вона все ще створює кожен аркуш і читає його зв'язки в потоці виклику, що й залишає кожну спільну мапу незмінною, щойно воркери почнуть. Фаза C все ще послідовно обходить аркуші потім для коментарів, малюнків, діаграм і таблиць, і її захист змінився з перевірки на null у старому стадіювальному масиві на zip.Exists проти імені частини. Спільні входи лише для читання, яких торкаються воркери, спільна таблиця рядків та мапи cellXf, завершені до початку Фази B і ніколи не записуються під час неї

var
  Wb: TXLSXWorkbook;
begin
  Wb := TXLSXWorkbook.Create;
  try
    Wb.ParallelParse := True;      // default; False forces one sheet at a time
    Wb.ParallelParseThreads := 4;  // 0 selects the automatic cap
    Wb.Open('quarterly-consolidation.xlsx');
    // ... workbook is identical either way ...
  finally
    Wb.Free;
  end;
end;

Встановлення ParallelParse у False перед Open надсилає ту саму процедуру завдання з кількістю потоків один, і RunParallelJobs вироджується у звичайний цикл у потоці виклику. Це варто знати з двох причин: це відповідь в один рядок, якщо занепокоєння щодо потоків коли-небудь виникне на практиці, і це означає, що послідовний і паралельний шляхи ділять одне тіло коду розбору, а не розходяться. Винятки воркерів захоплюються, найнижчий індекс завдання перемагає, і помилка повторно піднімається в потоці виклику після того, як кожен воркер приєднується, тож пошкоджений аркуш все одно проявляється як один виняток в очікуваному місці. Загальне налаштування навколишнього шляху відкриття охоплено в керівництві з продуктивності великих книг у Delphi

Шлюз читання, паралельна фаза відкриття та потоковий доступ до записів, описані тут, постачаються як частина стандартного компонента HotXLS Excel для Delphi та C++Builder, з повним вихідним кодом; сторінка продукту несе повну довідку TXLSXWorkbook, включно з властивостями паралельного відкриття