Souběžné rozbalování ZIP v Delphi: HotXLS Read Gate

HotXLS, nativní knihovna komponent pro Excel v Delphi a C++Builder, rozbaluje několik listů XLSX zároveň z jediného otevřeného balíčku ZIP. Mechanismem je TZipReadGate, malá třída v lxZipArchive.pas, která drží stream balíčku plus jednu kritickou sekci a vystavuje přesně jednu metodu. Serializuje dvojici seek-and-read. Vše nad touto dvojicí běží souběžně

Problém, který si tento návrh vynutil, zná každý delphistický vývojář, který kdy otevřel velký sešit. 80MB xlsx je 80 MB deflatovaného XML a listové části uvnitř se rozbalí přibližně pětkrát až desetkrát. Pokud vaše otevírací cesta extrahuje každý list do paměťového streamu ještě před zpracováním, platíte za rozbalené bajty navrch k sešitu, který stavíte, a špička přichází dřív, než je vytvořena jediná buňka. Tento článek se zabývá souběžností na úrovni balíčku, která tento mezikrok odstraňuje. Strop alokátoru, který nad tím stojí, popisuje článek o paralelním parsování XLSX a správci paměti, a API pro čtení-jednou-nikdy-nematerializovat popisuje průvodce streamovacím přímým čtečem

Proč stará otevírací cesta ukládala každý list do RAM

Původní paralelní otevírání v HotXLS bylo trojfázové pipeline a prostřední fáze byla jediná, která běžela na workerech. Fáze A procházela seznam listů sériově, vytvářela každý list, četla jeho relační část a kopírovala celé rozbalené XML listu do soukromého TMemoryStream. Fáze B rozvětvila ParseWorksheetXml napříč poolem. Fáze C se vracela do archivu na volajícím vlákně kvůli malým satelitním částem: komentářům, vláknovým komentářům, kresbám, grafům, tabulkám. Tento tvar byl zvolen z uvedeného důvodu. Hlavičkový komentář v lxParallelParse.pas kdysi doslova říkal, že archiv zip a jeho stav rozbalování nejsou thread-safe, a interní poznámky šly dál: neobtěžujte se archiv zamykat, protože jakmile je stav rozbalování serializován na jeden vstup, zámek nic nepřinese. Fáze A existovala proto, aby udržela každý dotek archivu na jednom vlákně. Cenou bylo, že sešit s osmi rušnými listy držel v paměti současně osm plně rozbalených bufferů XML listu, a tyto buffery jsou největšími přechodnými objekty v celé otevírací cestě

Mohou dvě vlákna rozbalovat z jednoho streamu ZIP?

Ano, a starý úsudek byl chybný konkrétním, dohledatelným způsobem: sloučil dva různé kusy stavu do jedné věty. Stav rozbalování skutečně sdílet nelze. Struktura zlib z_stream nese posuvné okno, Huffmanovy tabulky a pozici v bitech pro jeden komprimovaný člen, a pokud dvě vlákna tlačí bajty přes tentýž stream, výsledkem je nesmysl. Podkladový zdroj bajtů je úplně jiná otázka a odpověď zní, že souborový stream má přesně jeden kus měnitelného sdíleného stavu, který stojí za ochranu — svůj kurzor pozice

Kontejner ZIP dělá toto oddělení legitimním. Každý člen v archivu ZIP je komprimován nezávisle: má vlastní lokální hlavičku souboru, vlastní bitový deflate stream na vlastním DataOffset, vlastní CRC32 a velikosti v centrálním adresáři. Neexistuje žádný sdílený slovník napříč členy tak, jak jej má souvislý blok 7z, takže vstup N lze rozbalit bez dotyku vstupu M. Dejte každému workeru vlastní z_stream nad vlastním rozsahem bajtů a jediné, na čem se srazí, je seek. Právě tuto srážku odstraňuje TZipReadGate, a celá třída je natolik krátká, že se dá přečíst 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;

Co TZipReadGate chrání a co záměrně ne

TZipReadGate.ReadAt hlídá jednu nedělitelnou operaci — nastavení pozice sdíleného streamu a čtení z něj — a nic jiného. TZipArchive.OpenArchive vytváří gate nad FInputStream jakmile se centrální adresář úspěšně zpracuje, a TZipArchive.Close jej uvolní. Archivy otevřené pro zápis jej nikdy nedostanou. Každé čtení, které worker provede nad balíčkem, tak protéká přes jedinou kritickou sekci držanou po dobu jednoho bufferovaného čtení

Vše ostatní zůstává mimo zámek, protože je to už soukromé nebo už neměnné. TZipSubStream si drží vlastní FPosition, takže každý worker sleduje vlastní místo ve vlastním vstupu. TZLibStream, který TZipEntry.GetStream staví nad tímto sub-streamem, je pro každý vstup vlastní, vytvořený s windowBits rovným -15 pro raw deflate, a nikdy se nesdílí. Centrální adresář je plně zpracován ještě před spuštěním jakéhokoli workeru, včetně každé lokální hlavičky, takže GetEntryByName je ve chvíli, kdy souběžnost začíná, jen vyhledávání v hash tabulce pouze pro čtení. Samotné směrování je tři řádky v TZipSubStream.Read, a větev bez gate je to, co udržuje každého existujícího jednovláknového volajícího na staré cestě 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;

Kolik gate stojí pod zátěží?

Méně, než naznačuje fráze „globální zámek na archivu", a to díky granularitě, kterou TZLibStream náhodou používá. Jeho vstupní buffer má velikost BufferSize, definovanou jako $4000, takže ReadInputBuffer při každém doplnění stáhne 16 KB komprimovaných bajtů a předá je zng_inflate. Jedno zamknutí tedy pokrývá 16 KB deflate vstupu, což se u XML listu rozbalí do řádově 100 KB značkovaného obsahu, který worker následně dekóduje a zpracuje bez toho, aby cokoli držel. Zámek je držen po dobu čtení na dané pozici proti cache operačního systému; práce, kterou hlídá, se měří v milisekundách

Poctivá hranice je tam, kde se tento poměr obrátí. Vstupy uložené bez komprese místo deflatovaných procházejí gate jedna ku jedné bez jakékoli práce rozbalování, která by latenci skryla, takže balíček plný uložených členů by se serializoval mnohem tvrději. Studený soubor na pomalém médiu rozšiřuje kritickou sekci, protože čtení uvnitř je nyní skutečný přenos z disku, ne zásah do cache. A za hrstkou workerů stejně narazíte na jinou hranici dřív: parsování listu je náročné na alokace, a správce paměti Delphi serializuje alokace napříč vlákny dávno předtím, než se stane limitem read gate. Proto má TXLSXWorkbook.ParallelParseThreads ve výchozím stavu automatický strop místo jednoho vlákna na jádro

Tělo workeru a smyčka vyprázdnění, na kterou se snadno zapomíná

S gate na místě HotXLS rovnou smazal staging Fáze A. Worker nyní otevře vlastní stream vstupu a předá jej přímo parseru. Dvě přechodná pole nesou vstupy: FParZip drží archiv po dobu paralelní fáze, FParSheetPartNames drží názvy částí, a obě jsou vyprázdněna v bloku finally, aby žádný zastaralý ukazatel nepřežil neúspěšné otevření. Stream, který se vrátí z TZipArchive.OpenFile, je TZipVerifiedStream obalující TZLibStream obalující TZipSubStream, a uvolnění vnějšího uvolní celý řetěz

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;

Smyčka vyprázdnění je detail, který by přímý port starého kódu vynechal, a jeho vynechání tiše vypne kontrolu integrity. TZipVerifiedStream průběžně akumuluje CRC32 při průchodu bajtů a volá VerifyComplete teprve tehdy, když jeho pozice dosáhne nekomprimované velikosti zaznamenané v centrálním adresáři; odtud pocházejí výjimky pro neshodu velikosti a neshodu CRC32, plus jednobajtové sondovací čtení, které zachytí vstup delší, než bylo deklarováno. Čtečka XML se zastaví na uzavíracím elementu a obvykle ponechá nepřečtený nový řádek nebo pár bajtů koncových mezer, takže bez vyprázdnění pozice nikdy nedosáhne deklarované velikosti a kontroly se nikdy nespustí. Přečtení zbytku do pomocného bufferu nic nestojí a kontroly obnoví. Pokud existovaly staging streamy, XlsxCopyStreamAll to dělal jaksi mimochodem

Co pořád běží sériově a příznak, který vše vypne

Fáze A přežívá, jen bez extrakce. Stále vytváří každý list a čte jeho vztahy na volajícím vlákně, což je to, co ponechává každou sdílenou mapu neměnnou v okamžiku, kdy workeři začnou. Fáze C stále poté sériově prochází listy kvůli komentářům, kresbám, grafům a tabulkám, a její strážce se změnil z kontroly na null nad starým staging polem na zip.Exists vůči názvu části. Sdílené vstupy jen pro čtení, kterých se workeři dotýkají, sdílená tabulka řetězců a mapy cellXf, jsou kompletní ještě před začátkem Fáze B a během ní se nikdy 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;

Nastavení ParallelParse na False před Open spustí stejnou proceduru úlohy s počtem vláken rovným jedné, a RunParallelJobs degeneruje na obyčejnou smyčku na volajícím vlákně. To stojí za znalost ze dvou důvodů: je to jednořádková odpověď, pokud se v terénu kdy objeví problém s vlákny, a znamená to, že sériová a paralelní cesta sdílejí jedno tělo parsovacího kódu místo toho, aby se rozcházely. Výjimky workerů jsou zachyceny, vyhrává nejnižší index úlohy, a chyba je znovu vyvolána na volajícím vlákně poté, co se ke všem workerům připojí, takže poškozený list se stále projeví jako jedna výjimka na očekávaném místě. Obecné ladění okolní otevírací cesty popisuje průvodce výkonem velkých sešitů v Delphi

Read gate, paralelní fáze otevírání a streamovaný přístup ke vstupům popsané zde jsou součástí standardní komponenty HotXLS Excel pro Delphi a C++Builder, s plným zdrojovým kódem; stránka produktu obsahuje kompletní referenci TXLSXWorkbook včetně vlastností paralelního otevírání