Współbieżny inflate ZIP w Delphi: bramka odczytu HotXLS

HotXLS, natywna biblioteka komponentów Excel dla Delphi i C++Builder, inflatuje kilka arkuszy XLSX jednocześnie z jednego otwartego pakietu ZIP. Mechanizmem jest TZipReadGate, niewielka klasa w lxZipArchive.pas, która trzyma strumień pakietu plus jedną sekcję krytyczną i udostępnia dokładnie jedną metodę. Serializuje ona parę seek-i-read. Wszystko ponad tą parą działa współbieżnie

Problem, który wymusił ten projekt, to problem, który spotkał każdego dewelopera Delphi, który otworzył duży skoroszyt. Plik xlsx o rozmiarze 80 MB to 80 MB zdeflatowanego XML, a części arkuszy w środku rozszerzają się mniej więcej pięcio- do dziesięciokrotnie. Jeśli Twoja ścieżka otwierania ekstrahuje każdy arkusz do strumienia pamięci przed parsowaniem, płacisz za rozdęte bajty na dodatek do budowanego skoroszytu, a szczyt przychodzi, zanim powstanie choćby jedna komórka. Ten artykuł dotyczy współbieżności na poziomie pakietu, która usuwa ten krok pośredni. Pułap alokatora siedzący nad tym omawia artykuł o równoległym parsowaniu XLSX i menedżerze pamięci, a API czytające raz, nigdy niematerializujące omawia przewodnik po strumieniowym czytniku bezpośrednim

Dlaczego stara ścieżka otwierania buforowała każdy arkusz w RAM

Pierwotne równoległe otwieranie w HotXLS było trójfazowym pipeline'em, a faza środkowa była jedyną, która działała na workerach. Faza A przechodziła listę arkuszy sekwencyjnie, tworzyła każdy arkusz, czytała jego część relacji i kopiowała cały zinflatowany XML arkusza do prywatnego TMemoryStream. Faza B rozdzielała ParseWorksheetXml na pulę. Faza C wracała do archiwum na wątku wywołującym po małe części satelitarne: komentarze, komentarze wątkowe, rysunki, wykresy, tabele. Ten kształt został wybrany z podanego powodu. Komentarz nagłówkowy w lxParallelParse.pas mówił dosłownie, że archiwum zip i jego stan inflate nie są bezpieczne wątkowo, a wewnętrzne notatki szły dalej: nie zawracać sobie głowy blokowaniem archiwum, ponieważ gdy stan inflate jest już serializowany per wpis, blokada niczego nie kupuje. Faza A istniała, by trzymać każdy dostęp do archiwum na jednym wątku. Kosztem było to, że skoroszyt z ośmioma zajętymi arkuszami trzymał jednocześnie w pamięci osiem w pełni zinflatowanych buforów XML arkuszy, a te bufory są największymi obiektami przejściowymi w całej ścieżce otwierania

Czy dwa wątki mogą inflatować z jednego strumienia ZIP?

Tak, a stary osąd był błędny w konkretny, dający się zlokalizować sposób: zwinął dwa różne kawałki stanu w jedno zdanie. Stan inflate faktycznie nie jest współdzielny. z_stream zlib niesie okno przesuwne, tabele Huffmana i pozycję bitową dla jednego skompresowanego elementu, a dwa wątki wpychające bajty przez ten sam produkują śmieci. Bazowe źródło bajtów to zupełnie inne pytanie, a odpowiedzią tam jest to, że strumień pliku ma dokładnie jeden kawałek mutowalnego stanu współdzielonego wartego ochrony, swój kursor pozycji

Kontener ZIP czyni ten rozdział legalnym. Każdy element w archiwum ZIP jest skompresowany niezależnie: własny lokalny nagłówek pliku, własny strumień bitów deflate pod własnym DataOffset, własne CRC32 i rozmiary w central directory. Nie ma współdzielonego słownika rozciągającego się przez elementy, tak jak ma to lity blok 7z, więc wpis N może być inflatowany bez dotykania wpisu M. Daj każdemu workerowi jego własny z_stream nad jego własnym zakresem bajtów, a jedyną rzeczą, na której się zderzają, jest seek. To zderzenie usuwa TZipReadGate, a cała klasa jest wystarczająco krótka, by przeczytać ją na jednym ekranie

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 chroni TZipReadGate, a czego celowo nie chroni

TZipReadGate.ReadAt strzeże jednej niepodzielnej operacji, pozycjonowania współdzielonego strumienia i czytania z niego, i niczego więcej. TZipArchive.OpenArchive konstruuje bramkę nad FInputStream, gdy tylko central directory zostanie pomyślnie sparsowany, a TZipArchive.Close ją zwalnia. Archiwa otwierane do zapisu nigdy jej nie dostają. Każdy odczyt, jaki worker wykonuje na pakiecie, przechodzi więc przez pojedynczą sekcję krytyczną trzymaną przez czas trwania jednego buforowanego odczytu

Wszystko inne pozostaje poza blokadą, ponieważ jest już prywatne albo już niemutowalne. TZipSubStream trzyma własne FPosition, więc każdy worker śledzi własne miejsce we własnym wpisie. TZLibStream, który TZipEntry.GetStream buduje nad tym podstrumieniem, jest per wpis, tworzony z windowBits równym -15 dla surowego deflate, i nigdy nie jest współdzielony. Central directory jest w pełni sparsowany, zanim jakikolwiek worker wystartuje, wraz z każdym lokalnym nagłówkiem, więc GetEntryByName jest wyszukiwaniem hashowym tylko-do-odczytu, zanim zacznie się współbieżność. Samo trasowanie to trzy linie w TZipSubStream.Read, a gałąź bez bramki jest tym, co trzyma każdego istniejącego jednowątkowego wywołującego na starej ścieżce kodu

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;

Ile kosztuje bramka pod rywalizacją?

Mniej niż sugeruje fraza „globalna blokada na archiwum”, dzięki granularności, jakiej akurat używa TZLibStream. Jego bufor wejściowy to BufferSize, zdefiniowany jako $4000, więc ReadInputBuffer pobiera 16 KB skompresowanych bajtów na dopełnienie i oddaje je zng_inflate. Jedno przejęcie blokady pokrywa więc 16 KB danych wejściowych deflate, co dla XML arkusza rozszerza się do czegoś rzędu 100 KB znaczników, które worker dekoduje i parsuje, nie trzymając niczego. Blokada jest trzymana na czas pozycjonowanego odczytu wobec cache systemu operacyjnego; praca, którą bramkuje, mierzona jest w milisekundach

Uczciwą granicą jest miejsce, gdzie ta proporcja się odwraca. Wpisy przechowywane, a nie deflatowane, czytają przez bramkę jeden do jednego bez żadnej pracy inflate, która ukrywałaby opóźnienie, więc pakiet pełen przechowywanych elementów serializowałby się dużo mocniej. Zimny plik na wolnym nośniku poszerza sekcję krytyczną, ponieważ odczyt w jej wnętrzu jest teraz prawdziwym transferem dyskowym, a nie trafieniem w cache. A poza garstką workerów bramka i tak nie jest tym, na co trafisz najpierw: parsowanie arkusza jest ciężkie alokacyjnie, a menedżer pamięci Delphi serializuje alokacje między wątkami na długo, zanim bramka odczytu stanie się ograniczeniem. Dlatego TXLSXWorkbook.ParallelParseThreads domyślnie ma automatyczny pułap zamiast jednego wątku na rdzeń

Ciało workera i pętla dopróżniająca, o której łatwo zapomnieć

Z bramką na miejscu HotXLS całkowicie usunął buforowanie fazy A. Worker otwiera teraz własny strumień wpisu i podaje go prosto do parsera. Dwa pola przejściowe niosą dane wejściowe: FParZip trzyma archiwum na czas trwania fazy równoległej, FParSheetPartNames trzyma nazwy części, a oba są czyszczone w bloku finally, tak by żaden nieaktualny wskaźnik nie przetrwał nieudanego otwarcia. Strumień, który wraca z TZipArchive.OpenFile, to TZipVerifiedStream opakowujący TZLibStream opakowujący TZipSubStream, a zwolnienie zewnętrznego zwalnia cały łańcuch

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;

Pętla dopróżniająca to szczegół, który prosty port starego kodu by pominął, a pominięcie go po cichu wyłącza sprawdzanie integralności. TZipVerifiedStream gromadzi bieżące CRC32, gdy bajty przez niego przepływają, i wywołuje VerifyComplete tylko wtedy, gdy jego pozycja dociera do rozmiaru nieskompresowanego zapisanego w central directory; stąd biorą się wyjątki niezgodności rozmiaru i niezgodności CRC32, plus jednobajtowy odczyt sondujący, który wychwytuje wpis dłuższy niż zadeklarowany. Czytnik XML zatrzymuje się na elemencie zamykającym i zwykle zostawia nieprzeczytany znak nowej linii albo kilka bajtów końcowej białej spacji, więc bez dopróżnienia pozycja nigdy nie dociera do zadeklarowanego rozmiaru, a sprawdzenia nigdy się nie uruchamiają. Odczytanie reszty do bufora roboczego nic nie kosztuje i je przywraca. Gdy istniały strumienie buforujące, robił to przypadkiem XlsxCopyStreamAll

Co wciąż działa sekwencyjnie i flaga, która to wszystko wyłącza

Faza A przetrwała, minus ekstrakcja. Wciąż tworzy każdy arkusz i czyta jego relacje na wątku wywołującym, co jest tym, co pozostawia każdą współdzieloną mapę niemutowalną, gdy workery startują. Faza C wciąż przechodzi arkusze sekwencyjnie potem po komentarze, rysunki, wykresy i tabele, a jej strażnica zmieniła się z sprawdzenia null na starej tablicy buforującej na zip.Exists wobec nazwy części. Współdzielone dane wejściowe tylko-do-odczytu, których dotykają workery, współdzielona tablica ciągów znaków i mapy cellXf, są kompletne, zanim rozpocznie się faza B, i nigdy nie są zapisywane w jej trakcie

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;

Ustawienie ParallelParse na False przed Open wysyła tę samą procedurę zadania z liczbą wątków równą jeden, a RunParallelJobs degeneruje się do zwykłej pętli na wątku wywołującym. Warto to wiedzieć z dwóch powodów: to jednolinijkowa odpowiedź, gdyby kiedykolwiek w terenie wypłynął problem wątkowania, i oznacza, że ścieżka sekwencyjna i równoległa dzielą jedno ciało kodu parsującego zamiast się rozjeżdżać. Wyjątki workerów są przechwytywane, wygrywa najniższy indeks zadania, a błąd jest ponownie zgłaszany na wątku wywołującym, gdy każdy worker dołączy, więc uszkodzony arkusz wciąż ujawnia się jako jeden wyjątek w oczekiwanym miejscu. Ogólne strojenie otaczającej ścieżki otwierania omawia przewodnik po wydajności dużych skoroszytów w Delphi

Bramka odczytu, równoległa faza otwierania i strumieniowy dostęp do wpisów opisane tutaj są dostarczane jako część standardowego komponentu HotXLS Excel dla Delphi i C++Builder, z pełnym źródłem; strona produktu niesie pełną dokumentację TXLSXWorkbook, wraz z właściwościami równoległego otwierania