Едновременно ZIP inflate в Delphi: HotXLS read gate

HotXLS, нативната библиотека за компонент за Excel за Delphi и C++Builder, декомпресира няколко XLSX работни листа едновременно от един отворен ZIP пакет. Механизмът е TZipReadGate, малък клас в lxZipArchive.pas, който държи потока на пакета плюс една критична секция и излага точно един метод. Той серализира двойката seek и read. Всичко над тази двойка се изпълнява едновременно

Проблемът, наложил този дизайн, е такъв, който всеки Delphi разработчик, отварял голяма работна книга, е срещал. 80 МБ xlsx е 80 МБ deflated XML, а частите на работния лист вътре в него се разширяват приблизително пет до десет пъти. Ако пътят ви за отваряне извлича всеки работен лист в поток в паметта преди да го разбере, плащате за декомпресираните байтове върху работната книга, която изграждате, а пикът пристига преди дори една клетка да е създадена. Тази статия разглежда едновременността на ниво пакет, която премахва тази стъпка на подготовка. Таванът на разпределителя на памет, който седи над него, е разгледан в статията за паралелен разбор на XLSX и мениджъра на паметта, а API-то за четене веднъж, никога не материализирай, е разгледано в разбора на директния поточен четец

Защо старият път за отваряне подготвяше всеки работен лист в RAM

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

Могат ли две нишки да декомпресират от един ZIP поток?

Да, а старата преценка беше грешна по конкретен, локализируем начин: тя сгъна две различни части на състоянието в едно изречение. Състоянието на декомпресия наистина не е споделимо. zlib z_stream носи плъзгащия прозорец, таблиците на Хъфман и позицията на бита за един компресиран член, а две нишки, бутащи байтове през същия, произвеждат боклук. Основният байтов източник е съвсем различен въпрос, а отговорът там е, че файлов поток има точно едно парче мутируемо споделено състояние, което си струва да се защити — своя курсор за позиция

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 е read-only търсене по хеш, докато едновременността започне. Самото маршрутизиране е три реда в 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 КБ маркиране, което работникът после декодира и разбира без да държи каквото и да е. Заключването се държи за позиционирано четене спрямо кеша на операционната система; работата, която то пази, се измерва в милисекунди

Честната граница е там, където това съотношение се обръща. Записи, съхранени, а не deflated, се четат през портата едно към едно без работа по декомпресия, която да скрие латентността, така че пакет, пълен със съхранени членове, би серализирал много по-трудно. Студен файл на бавен носител разширява критичната секция, защото четенето вътре в нея вече е реален трансфер от диска, а не попадение в кеша. И отвъд шепа работници портата така или иначе не е първото, което удряте: разборът на работен лист е тежък откъм разпределяне на памет, а мениджърът на паметта на Delphi серализира разпределянията през нишки много преди read gate-ът да стане ограничението. Затова 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 спрямо името на частта. Споделените read-only входове, които работниците докосват, споделената таблица от низове и картите 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 се изражда до обикновен цикъл на извикващата нишка. Това си струва да се знае по две причини: това е едноредовият отговор, ако притеснение за нишки някога изникне на терен, и означава, че последователният и паралелният път споделят единствено тяло от код за разбор, вместо да се разминават. Изключенията на работниците се улавят, най-ниският индекс на job печели, а грешката се хвърля повторно на извикващата нишка, след като всеки работник се присъедини, така че повреден работен лист все пак се показва като едно изключение на очакваното място. Общото настройване на обкръжаващия път за отваряне е разгледано в ръководството за производителност на големи работни книги в Delphi

Read gate-ът, паралелната фаза на отваряне и достъпът чрез поточно четене до записите, описани тук, се доставят като част от стандартния HotXLS Excel компонент за Delphi и C++Builder, с пълен изходен код; страницата на продукта носи пълната справка за TXLSXWorkbook, включително свойствата за паралелно отваряне