Inflate ZIP Concorrente em Delphi

Duas threads compartilham um TFileStream sobre um pacote .xlsx. A thread A define o Position para o primeiro byte comprimido de sheet1.xml, a thread B o define para o primeiro byte de sheet2.xml um momento depois, e o Read da A retorna um buffer que pertence ao outro membro. O inflater da A agora decodifica o meio do stream deflate de outra pessoa; o zlib normalmente para com Z_DATA_ERROR alguns blocos adiante, e se por acaso produzir saída, o CRC32 do membro não vai bater. O Deflate não é o problema aqui. A corrida vive inteiramente no cursor de posição do stream, e um lock em torno do par seek-e-leitura basta para deixar muitos workers inflando membros do mesmo pacote ao mesmo tempo

A razão de alguém querer isso é memória. Suponha uma pasta de trabalho de 80 MB em disco cujo XML de planilha expande cinco a dez vezes quando inflado. Um caminho de abertura que extrai cada planilha para um buffer antes de interpretá-la paga esses bytes inflados por cima da pasta de trabalho que está construindo, e o pico chega antes de uma única célula existir. Este artigo cobre a concorrência de nível de pacote que remove a etapa de extração. O teto do alocador acima dela está coberto em o artigo sobre parsing XLSX paralelo e o gerenciador de memória, e a API de leitura única que nunca constrói uma pasta de trabalho está em o passo a passo do streaming direct reader

O design de staging, e o que ele guarda em memória

O jeito cauteloso de paralelizar uma abertura de XLSX é um pipeline de três fases em que apenas a fase do meio roda em workers. A fase A caminha pela lista de planilhas na thread chamadora, cria cada planilha, lê a part de relationships dela, e copia o XML inflado da planilha inteira para um TMemoryStream privado. A fase B espalha o parsing do XML por um pool de threads. A fase C volta ao arquivo, ainda na thread chamadora, para as pequenas parts satélites: comentários, threaded comments, drawings, charts e tables

O argumento para essa forma soa incontestável. Um arquivo ZIP e o estado de inflate dele não são thread-safe, então mantenha todo toque no arquivo em uma thread e dispense locks por completo. A conta é uma pasta de trabalho com oito planilhas ocupadas segurando oito buffers de planilha totalmente inflados no mesmo momento, e esses buffers são os maiores objetos transitórios em qualquer lugar do caminho de abertura. Com o pacote de 80 MB presumido acima, são várias centenas de megabytes de texto XML que existem apenas para um parser ler uma vez

O HotXLS mantém as fases A e C mas descarta a cópia. Cada worker na fase B abre o próprio entry stream dele no pacote compartilhado e o entrega direto ao leitor de XML, então a planilha inflada nunca existe como um bloco único. Isso só funciona porque o argumento de "não thread-safe" se revela sobre duas coisas diferentes

Duas threads podem inflar a partir de um stream ZIP?

Sim, contanto que cada thread possua o inflater dela e a fonte de bytes compartilhada seja serializada na leitura. O argumento cauteloso dobra dois tipos de estado em uma frase. O estado de inflate realmente não pode ser compartilhado: um z_stream do zlib carrega a janela deslizante de 32 KB que a RFC 1951 define, as tabelas de Huffman atuais e a posição de bit dentro de um membro comprimido, e duas threads empurrando bytes pelo mesmo produzem lixo. A fonte de bytes é outra história. Um file stream tem exatamente uma peça de estado mutável que vale proteger, a posição dele

O container ZIP torna a divisão legal. O APPNOTE da PKWARE (seção 4.3) dispõe um pacote como membros independentes, cada um com o próprio local file header dele, os próprios dados comprimidos no próprio offset dele, e o próprio CRC32 e tamanhos registrados no central directory. Nenhum dicionário atravessa membros do jeito que um bloco solid do 7z faz, então a entrada N pode ser inflada sem ler um byte da entrada M. Dê a cada worker um inflater privado sobre a própria faixa de bytes dele e a única colisão restante é o seek. No HotXLS essa colisão é tratada pelo TZipReadGate em lxZipArchive.pas, e a classe inteira cabe em uma tela

type
  TZipReadGate = class
  private
    FBaseStream: TStream;
    FLock: TXLSCriticalSection;   // wrapper lxPlatform sobre uma critical section
  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;
  XLSEnterCriticalSection(FLock);
  try
    FBaseStream.Position := AOffset;
    Result := FBaseStream.Read(Buffer, Count);
  finally
    XLSLeaveCriticalSection(FLock);
  end;
end;

O que o read gate tranca, e o que fica fora dele

O TZipReadGate.ReadAt protege uma única operação indivisível, posicionar o stream compartilhado e ler dele. O TZipArchive.OpenArchive constrói o gate uma vez que o central directory foi interpretado, tanto para o overload de nome de arquivo quanto para o de TStream, e liberar o arquivo o libera. Arquivos criados para gravação nunca recebem um, porque nada lê deles concorrentemente

Todo o resto fica fora do lock porque já é privado ou já está congelado. Cada entry stream que o TZipArchive.OpenFile retorna mantém a própria posição dele dentro do próprio membro dele e a converte para um offset absoluto do pacote apenas no momento em que chama o ReadAt. O inflater embrulhado ao redor dele é criado por entrada com um tamanho de janela de -15, que o zlib lê como deflate cru sem header de zlib, e nunca é entregue a outra thread. O central directory, headers locais incluídos, é totalmente interpretado antes de qualquer worker começar, então consultar uma entrada pelo nome é uma consulta de hash somente leitura quando a concorrência começa. Consultas são case-insensitive e rejeitam nomes com segmentos .. no caminho

Uma consequência importa quando você chama TXLSXWorkbook.Open(Stream) com o seu próprio stream. O gate é embrulhado em torno desse stream, e o lock dele coordena os workers da biblioteca entre si. Ele não sabe nada do seu código. Se outra thread na sua aplicação ler ou reposicionar o mesmo TStream enquanto o Open roda, a corrida de cursor do primeiro parágrafo volta, então entregue o stream e deixe-o quieto até o Open retornar

Quanto custa um lock sob contenção?

Menos do que "um lock global no arquivo" sugere, por causa de quanto trabalho cada aquisição cobre. O inflater recarrega a entrada dele em blocos de 64 KB de bytes comprimidos, então uma viagem pelo gate alimenta 64 KB de dados deflate a um worker que então decodifica e interpreta sem segurar nada. À expansão de cinco a dez vezes presumida antes, essa recarga vira aproximadamente 320 a 640 KB de marcação de planilha. O lock em si cobre uma leitura posicionada que normalmente é servida do cache de arquivos do sistema operacional

A razão se inverte em três lugares, todos sobre a entrada ou a máquina. Membros guardados sem compressão passam pelo gate uma vez por chamada de leitura sem trabalho de inflate entre aquisições; a maioria dos gravadores comprime parts de planilha com deflate, mas um pacote produzido por uma ferramenta que guarda tudo vai serializar mais. Um arquivo frio em armazenamento lento ou de rede alarga a critical section, já que a leitura dentro dela vira uma transferência real. E além de um punhado de workers o gate não é a primeira parede de qualquer jeito: parsing de planilha aloca pesado, e o gerenciador de memória do Delphi serializa alocações entre threads bem antes de o read gate saturar

Esse último ponto decide a contagem de threads. O TXLSXWorkbook.ParallelParseThreads tem padrão 0, que inicia um worker por processador lógico, nunca mais que o número de planilhas. Em um laptop com quatro a oito processadores lógicos esse padrão está bom. Em um servidor de build de 32 threads abrindo pastas de trabalho com dezenas de planilhas, defina um teto pequeno explícito como 4: os workers extras na maior parte esperam na fila do alocador e cada um segura o próprio inflater e buffers dele. Para achar o número certo para os seus arquivos, abra uma pasta de trabalho representativa em um loop com a propriedade definida como 1, 2, 4 e 8, e compare o tempo de parede contra o pico de working set no Task Manager ou no Process Explorer

Ler uma entrada até o fim é o que dispara a checagem de CRC

Todo entry stream que o arquivo retorna se checa, mas apenas no fim. Ele acumula um CRC32 acumulado enquanto os bytes passam, e quando a posição dele alcança o tamanho não comprimido registrado no central directory ele compara os dois valores e sonda por um byte extra. As falhas aparecem como exceções com mensagens literais: 'ZIP entry uncompressed size mismatch', 'ZIP entry CRC32 mismatch', 'ZIP entry exceeds its declared size', e 'ZIP entry ended before its declared size' para um membro que acaba cedo

Um leitor de XML para na tag de fechamento </worksheet>. Qualquer coisa depois dela, frequentemente apenas uma quebra de linha final, fica não lida, a posição nunca alcança o tamanho declarado, e a comparação nunca roda. O design de staging recebia a checagem de graça, já que copiar uma entrada para um buffer lê cada byte. Um parser lendo o entry stream direto não recebe, então cada worker do HotXLS drena o que o parser deixou com um buffer de rascunho de 32 KB antes de liberar o stream, e liberar o stream mais externo libera o inflater e o sub-stream embaixo dele

A mesma regra vale quando o seu próprio código usa o TZipArchive do lxZipArchive, por exemplo para fazer um sanity check de um pacote que chega antes de entregá-lo ao loader de pasta de trabalho. Uma função que lê apenas os primeiros poucos kilobytes de cada part para farejar o elemento raiz dela não prova nada sobre integridade. Leia até o fim:

uses
  Classes, SysUtils, lxZipArchive;

// Lê uma part do pacote até o fim declarado dela para que as checagens de tamanho e CRC32 rodem.
// PartIsIntact é um helper que você escreve; não faz parte da biblioteca
function PartIsIntact(Zip: TZipArchive; const PartName: WideString;
  out Problem: string): Boolean;
var
  Stream: TStream;
  Buffer: array [0..32767] of Byte;
begin
  Result := False;
  Problem := '';
  Stream := Zip.OpenFile(PartName);   // nil quando a part não existe
  if Stream = nil then
  begin
    Problem := 'part not found';
    Exit;
  end;
  try
    try
      while Stream.Read(Buffer, SizeOf(Buffer)) > 0 do
        ;
      Result := True;
    except
      on E: Exception do
        Problem := E.Message;          // p.ex. 'ZIP entry CRC32 mismatch'
    end;
  finally
    Stream.Free;
  end;
end;

var
  Zip: TZipArchive;
  Problem: string;
begin
  Zip := TZipArchive.Create;
  try
    Zip.OpenArchive('quarterly-consolidation.xlsx');
    if not PartIsIntact(Zip, 'xl/worksheets/sheet1.xml', Problem) then
      Writeln('sheet1.xml: ', Problem);
  finally
    Zip.Free;
  end;
end;

A segunda armadilha naquele código é rebobinar. Entry streams são somente para frente: ler o Size funciona, porque o tamanho declarado é conhecido de antemão, mas definir Position := 0 depois de uma espiada levanta 'ZIP entry stream is forward-only'. Se um componente insiste em seek, abra a entrada uma segunda vez com OpenFile, o que é barato e lhe dá um stream fresco no offset zero. Várias threads podem cada uma chamar OpenFile em um arquivo aberto para leitura, já que é exatamente o que o loader de pasta de trabalho faz, contanto que nada libere o arquivo enquanto os streams delas estão vivos

O que ainda roda serialmente, e o switch que desliga

A fase A ainda roda, menos a extração. Ela cria toda planilha e lê as relationships delas na thread chamadora, que é o que deixa todo mapa compartilhado imutável quando os workers começam. A tabela de shared strings e as tabelas de formato de célula estão completas antes de a fase B começar e nunca são gravadas durante ela. A fase C ainda caminha pelas planilhas em ordem depois, para comentários, drawings, charts e tables, pulando qualquer planilha cuja part falta no pacote

uses
  SysUtils, lxHandleX;

var
  Wb: TXLSXWorkbook;
begin
  Wb := TXLSXWorkbook.Create;
  try
    Wb.ParallelParse := True;        // padrão; False interpreta uma planilha por vez
    Wb.ParallelParseThreads := 4;    // 0 = um worker por processador lógico
    Wb.SelectedSheets.Add('Q3 Detail');  // opcional: interpreta células apenas desta planilha
    if Wb.Open('quarterly-consolidation.xlsx') <> 1 then
      raise Exception.Create('Workbook open failed');
    // A pasta de trabalho carregada é idêntica qualquer que seja o caminho que a interpretou
  finally
    Wb.Free;
  end;
end;

Definir o ParallelParse como False antes do Open roda o mesmo job por planilha com uma contagem de threads de um, e o pool degrada para um loop simples na thread chamadora. Uma pasta de trabalho com uma única planilha segue esse caminho de qualquer jeito. Mantenha o switch em mente ao diagnosticar um problema de abertura: ele tira threading da equação em uma linha, e os dois caminhos rodam o mesmo código de parsing, então a pasta de trabalho que você recebe de volta não depende da configuração

Erros mantêm a forma serial também. Quando um worker levanta exceção, nenhuma planilha adicional é despachada, todo worker se junta, e a exceção do menor índice de planilha é relançada na thread chamadora, então um sheet3.xml corrompido ainda aparece como uma única exceção do Open. Com o RecoveryMode definido como True, uma planilha que falha ao ser interpretada é pulada em vez disso e registrada como aviso nos diagnósticos da pasta de trabalho. Planilhas deixadas fora do SelectedSheets nunca abrem o entry stream delas de forma alguma, e o MetadataOnly pula a fase paralela por inteiro. Ajustes mais amplos do caminho de abertura estão cobertos em o guia de desempenho com pastas de trabalho grandes no Delphi

O read gate, a fase de abertura paralela e o acesso ZIP de nível de entrada mostrados aqui fazem parte do componente Excel HotXLS Delphi para Delphi e C++Builder, que vem com código-fonte completo, então você pode ler o lxZipArchive.pas ao lado das propriedades de abertura do TXLSXWorkbook descritas na página de produto