Súbežný ZIP inflate v Delphi: HotXLS read gate

HotXLS, natívna knižnica Excel komponenty pre Delphi a C++Builder, súčasne rozbaľuje niekoľko XLSX hárkov z jedného otvoreného ZIP balíka. Mechanizmom je TZipReadGate, malá trieda v lxZipArchive.pas, ktorá drží stream balíka plus jednu kritickú sekciu a sprístupňuje presne jednu metódu. Sériovo vykonáva dvojicu seek-a-read. Všetko nad touto dvojicou beží súbežne

Problém, ktorý si tento dizajn vynútil, pozná každý Delphi vývojár, ktorý niekedy otvoril veľký zošit. 80 MB xlsx je 80 MB deflate-ovaného XML, a časti hárkov vnútri sa rozbaľujú zhruba päť až desaťnásobne. Ak vaša cesta otvárania extrahuje každý hárok do pamäťového streamu ešte pred jeho parsovaním, platíte za rozbalené bajty popri zošite, ktorý staviate, a vrchol nastane skôr, než je vytvorená čo i len jedna bunka. Tento článok sa venuje súbežnosti na úrovni balíka, ktorá tento medzikrok odstraňuje. Strop alokátora, ktorý sedí nad ním, je pokrytý v článku o paralelnom parsovaní XLSX a správcovi pamäte, a API čítania raz, nikdy nematerializuj je pokryté v prehľade streamovacieho priameho čítača

Prečo stará cesta otvárania staggovala každý hárok v RAM

Pôvodné paralelné otváranie v HotXLS bolo trojfázovou pipeline, a stredná fáza bola jediná, ktorá bežala na workeroch. Fáza A sériovo prešla zoznam hárkov, vytvorila každý hárok, prečítala jeho časť vzťahov, a skopírovala celé rozbalené XML hárku do súkromného TMemoryStream. Fáza B rozdelila ParseWorksheetXml naprieč poolom. Fáza C sa vrátila k archívu na volajúcom vlákne pre malé satelitné časti: komentáre, vláknové komentáre, kresby, grafy, tabuľky. Tento tvar bol zvolený zo stanoveného dôvodu. Hlavičkový komentár na lxParallelParse.pas kedysi hovoril, doslova, že ZIP archív a jeho stav inflate nie sú thread-safe, a interné poznámky išli ešte ďalej: neobťažujte sa zamykaním archívu, pretože akonáhle je stav inflate sériovaný na jednu položku, zámok nekúpi nič. Fáza A existovala na to, aby udržala každý dotyk archívu na jednom vlákne. Cena bola, že zošit s ôsmimi zaneprázdnenými hárkami držal osem plne rozbalených XML bufferov hárkov v pamäti súčasne, a tieto buffery sú najväčšie prechodné objekty v celej ceste otvárania

Môžu dve vlákna rozbaľovať z jedného ZIP streamu?

Áno, a starý úsudok bol nesprávny konkrétnym, lokalizovateľným spôsobom: zlúčil dva odlišné kusy stavu do jednej vety. Stav inflate skutočne nie je zdieľateľný. zlib z_stream nesie posuvné okno, Huffmanove tabuľky a bitovú pozíciu pre jedného komprimovaného člena, a dve vlákna tlačiace bajty cez ten istý produkujú nezmysel. Podkladový bajtový zdroj je úplne odlišná otázka, a odpoveď tam je, že súborový stream má presne jeden kus meniteľného zdieľaného stavu, ktorý stojí za ochranu, svoj kurzor pozície

ZIP kontajner robí toto oddelenie legálnym. Každý člen v ZIP archíve je komprimovaný nezávisle: vlastnú lokálnu hlavičku súboru, vlastný bitový prúd deflate na vlastnom DataOffset, vlastné CRC32 a veľkosti v centrálnom adresári. Neexistuje žiadny zdieľaný slovník naprieč členmi, tak ako to má solídny blok 7z, takže položku N možno rozbaliť bez dotyku položky M. Dajte každému workeru vlastný z_stream nad vlastným rozsahom bajtov a jediné, na čom kolidujú, je seek. Práve túto kolíziu TZipReadGate odstraňuje, a celá trieda je dosť krátka na to, aby sa dala prečítať na jednu obrazovku

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;

Čo TZipReadGate chráni a čo zámerne nechráni

TZipReadGate.ReadAt stráži jednu nedeliteľnú operáciu, umiestnenie zdieľaného streamu a čítanie z neho, a nič iné. TZipArchive.OpenArchive zostaví gate nad FInputStream potom, čo sa centrálny adresár úspešne spracoval, a TZipArchive.Close ho uvoľní. Archívy otvorené na zápis ho nikdy nedostanú. Každé čítanie, ktoré worker vykoná na balíku, sa preto zlieva cez jednu kritickú sekciu, drženú počas trvania jedného bufferovaného čítania

Všetko ostatné zostáva mimo zámku, pretože je to už súkromné alebo už nemenné. TZipSubStream si drží vlastnú FPosition, takže každý worker sleduje svoje vlastné miesto vo svojej vlastnej položke. TZLibStream, ktorú TZipEntry.GetStream stavia nad týmto substreamom, je na jednu položku, vytvorená s windowBits -15 pre surový deflate, a nikdy nie je zdieľaná. Centrálny adresár je úplne spracovaný skôr, než sa spustí čo i len jeden worker, vrátane každej lokálnej hlavičky, takže GetEntryByName je do momentu, keď súbežnosť začne, iba na čítanie hashové vyhľadávanie. Samotné smerovanie sú tri riadky v TZipSubStream.Read, a vetva bez gate je to, čo udržiava každého existujúceho jednovláknového volajúceho na starej ceste kódu

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;

Koľko stojí gate pod záťažou?

Menej, než naznačuje fráza „globálny zámok na archíve“, vďaka granularite, ktorú TZLibStream náhodou používa. Jej vstupný buffer je BufferSize, definovaný ako $4000, takže ReadInputBuffer ťahá 16 KB komprimovaných bajtov na doplnenie a odovzdáva ich zng_inflate. Jedno získanie zámku teda pokrýva 16 KB deflate vstupu, ktorý sa pri XML hárku rozbalí na niečo rádovo 100 KB markupu, ktorý worker potom dekóduje a parsuje bez toho, aby čokoľvek držal. Zámok je držaný počas polohovaného čítania voči cache operačného systému; práca, ktorú stráži, sa meria v milisekundách

Úprimná hranica je tam, kde sa tento pomer obráti. Položky uložené namiesto deflate-ovaných čítajú cez gate jedna k jednej bez akejkoľvek inflate práce, ktorá by skryla latenciu, takže balík plný uložených členov by sa sérioval oveľa tvrdšie. Studený súbor na pomalom médiu rozširuje kritickú sekciu, pretože čítanie vnútri neho je teraz skutočný diskový prenos namiesto zásahu do cache. A za hŕstkou workerov gate rovnako nie je to, na čo narazíte ako prvé: parsovanie hárku je náročné na alokácie, a Delphi správca pamäte sérioval alokácie naprieč vláknami dávno predtým, než sa read gate stane obmedzením. Preto TXLSXWorkbook.ParallelParseThreads predvolene automaticky stropuje namiesto jedného vlákna na jadro

Telo workera, a drenážna slučka, na ktorú sa ľahko zabúda

S gate na mieste HotXLS úplne zmazal staging Fázy A. Worker teraz otvorí vlastný stream položky a nakŕmi ho priamo parserom. Dva prechodné polia nesú vstupy: FParZip drží archív počas trvania paralelnej fázy, FParSheetPartNames drží názvy častí, a oba sa vyčistia v bloku finally, takže žiadny zastaraný ukazovateľ neprežije neúspešné otvorenie. Stream, ktorý sa vráti z TZipArchive.OpenFile, je TZipVerifiedStream obaľujúci TZLibStream obaľujúci TZipSubStream, a uvoľnenie vonkajšieho uvoľní celý reťazec

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;

Drenážna slučka je detail, ktorý by priamy port starého kódu vynechal, a jej vynechanie ticho vypne kontrolu integrity. TZipVerifiedStream akumuluje bežiace CRC32, ako bajty prechádzajú, a zavolá VerifyComplete iba vtedy, keď jej pozícia dosiahne nekomprimovanú veľkosť zaznamenanú v centrálnom adresári; odtiaľ pochádzajú výnimky nezhody veľkosti a nezhody CRC32, plus jednobajtové skúšobné čítanie, ktoré zachytí položku dlhšiu, než je deklarované. XML čítač sa zastaví na uzatváracom elemente a zvyčajne necháva neprečítaný nový riadok alebo pár bajtov koncového bieleho priestoru, takže bez drenáže pozícia nikdy nedosiahne deklarovanú veľkosť a kontroly sa nikdy nespustia. Prečítanie zvyšku do pomocného buffera nič nestojí a obnovuje ich. Keď existovali staging streamy, XlsxCopyStreamAll to robil náhodou

Čo stále beží sériovo, a príznak, ktorý to všetko vypne

Fáza A prežíva, mínus extrakcia. Stále vytvára každý hárok a číta jeho vzťahy na volajúcom vlákne, čo je to, čo necháva každú zdieľanú mapu nemennú, akonáhle workeri začnú. Fáza C stále prechádza hárky sériovo potom, kvôli komentárom, kresbám, grafom a tabuľkám, a jej poistka sa zmenila z null kontroly na starom staging poli na zip.Exists voči názvu časti. Zdieľané iba na čítanie vstupy, ktorých sa workeri dotýkajú, zdieľaná tabuľka reťazcov a mapy cellXf, sú kompletné skôr, než začne Fáza B, a nikdy sa počas nej nezapisujú

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;

Nastavenie ParallelParse na False pred Open odošle tú istú job procedúru s počtom vlákien jedna, a RunParallelJobs degeneruje na obyčajnú slučku na volajúcom vlákne. To stojí za to vedieť z dvoch dôvodov: je to jednoriadková odpoveď, ak sa niekedy v teréne objaví problém s vláknami, a znamená to, že sériová a paralelná cesta zdieľajú jedno telo parsovacieho kódu namiesto toho, aby sa rozchádzali. Výnimky workerov sa zachytávajú, vyhráva najnižší index úlohy, a chyba sa znovu vyvolá na volajúcom vlákne potom, čo sa každý worker pripojí, takže poškodený hárok sa stále objaví ako jedna výnimka na očakávanom mieste. Všeobecné ladenie okolitej cesty otvárania je pokryté v sprievodcovi výkonom veľkých zošitov v Delphi

Read gate, paralelná fáza otvárania a streamovací prístup k položkám opísané tu sú súčasťou štandardnej HotXLS Excel komponenty pre Delphi a C++Builder, s úplným zdrojovým kódom; stránka produktu nesie úplnú referenciu TXLSXWorkbook vrátane vlastností paralelného otvárania