Ταυτόχρονο ZIP Inflate σε Delphi: Read Gate του HotXLS

Το HotXLS, η εγγενής βιβλιοθήκη component Excel για Delphi και C++Builder, αποσυμπιέζει πολλά φύλλα εργασίας XLSX ταυτόχρονα από ένα μόνο ανοιχτό πακέτο ZIP. Ο μηχανισμός είναι το TZipReadGate, μια μικρή κλάση στο lxZipArchive.pas που κρατά το stream πακέτου συν μία κρίσιμη ενότητα και εκθέτει ακριβώς μία μέθοδο. Σειριοποιεί το ζεύγος seek-και-read. Όλα πάνω από αυτό το ζεύγος τρέχουν ταυτόχρονα

Το πρόβλημα που ανάγκασε αυτόν τον σχεδιασμό είναι κάτι που έχει συναντήσει κάθε προγραμματιστής Delphi που έχει ανοίξει ένα μεγάλο βιβλίο εργασίας. Ένα xlsx 80 MB είναι 80 MB αποσυμπιεσμένου XML, και τα μέρη φύλλου εργασίας μέσα του διευρύνονται περίπου πέντε έως δέκα φορές. Αν η διαδρομή ανοίγματός σας εξάγει κάθε φύλλο εργασίας σε ένα memory stream πριν το αναλύσει, πληρώνετε για τα αποσυμπιεσμένα bytes πάνω από το βιβλίο εργασίας που χτίζετε, και η αιχμή φτάνει πριν δημιουργηθεί έστω ένα κελί. Αυτό το άρθρο αφορά την ταυτοχρονία επιπέδου πακέτου που αφαιρεί αυτό το βήμα σκηνοθεσίας. Το ανώτατο όριο allocator που κάθεται πάνω από αυτό καλύπτεται στο άρθρο για την παράλληλη ανάλυση XLSX και τον διαχειριστή μνήμης, και το API διάβασε-μία-φορά-ποτέ-μην-υλοποιείς καλύπτεται στην περιγραφή του streaming direct reader

Γιατί η παλιά διαδρομή ανοίγματος σκηνοθετούσε κάθε φύλλο εργασίας στη RAM

Το αρχικό παράλληλο άνοιγμα στο HotXLS ήταν μια pipeline τριών φάσεων, και η μεσαία φάση ήταν η μόνη που έτρεχε σε workers. Η Φάση Α περπατούσε τη λίστα φύλλων σειριακά, δημιουργούσε κάθε φύλλο εργασίας, διάβαζε το μέρος σχέσεών του, και αντέγραφε ολόκληρο το αποσυμπιεσμένο XML φύλλου εργασίας σε ένα ιδιωτικό TMemoryStream. Η Φάση Β διασκόρπιζε την ParseWorksheetXml στη δεξαμενή. Η Φάση Γ γυρνούσε πίσω στο αρχείο στο νήμα κλήσης για τα μικρά δορυφορικά μέρη: σχόλια, νηματοποιημένα σχόλια, σχέδια, γραφήματα, πίνακες. Αυτό το σχήμα επιλέχθηκε για δηλωμένο λόγο. Το σχόλιο κεφαλίδας στο lxParallelParse.pas συνήθιζε να λέει, με τόσα λόγια, ότι το αρχείο zip και η κατάσταση αποσυμπίεσής του δεν είναι thread-safe, και οι εσωτερικές σημειώσεις προχωρούσαν παραπέρα: μη μπείτε στον κόπο να κλειδώσετε το αρχείο, επειδή μόλις η κατάσταση αποσυμπίεσης σειριοποιηθεί ανά καταχώριση, το κλείδωμα δεν αγοράζει τίποτα. Η Φάση Α υπήρχε για να κρατά κάθε επαφή με το αρχείο σε ένα νήμα. Το κόστος ήταν ότι ένα βιβλίο εργασίας με οκτώ απασχολημένα φύλλα κρατούσε οκτώ πλήρως αποσυμπιεσμένα buffers XML φύλλου εργασίας στη μνήμη ταυτόχρονα, και αυτά τα buffers είναι τα μεγαλύτερα προσωρινά αντικείμενα σε ολόκληρη τη διαδρομή ανοίγματος

Μπορούν δύο νήματα να αποσυμπιέζουν από ένα stream ZIP;

Ναι, και η παλιά κρίση ήταν λάθος με συγκεκριμένο, εντοπίσιμο τρόπο: συνέπτυξε δύο διαφορετικά κομμάτια κατάστασης σε μία πρόταση. Η κατάσταση αποσυμπίεσης γνησίως δεν μοιράζεται. Ένα z_stream zlib φέρει το κυλιόμενο παράθυρο, τους πίνακες Huffman και τη θέση bit για ένα συμπιεσμένο μέλος, και δύο νήματα που σπρώχνουν bytes μέσα από το ίδιο παράγουν σκουπίδια. Η υποκείμενη πηγή byte είναι εντελώς διαφορετικό ερώτημα, και η απάντηση εκεί είναι ότι ένα file stream έχει ακριβώς ένα κομμάτι μεταβλητής κοινής κατάστασης που αξίζει προστασία, τον δείκτη θέσης του

Το container ZIP κάνει τον διαχωρισμό νόμιμο. Κάθε μέλος σε ένα αρχείο ZIP συμπιέζεται ανεξάρτητα: η δική του τοπική κεφαλίδα αρχείου, το δικό του deflate bit stream στη δική του θέση DataOffset, το δικό του CRC32 και μεγέθη στον κεντρικό κατάλογο. Δεν υπάρχει κοινό λεξικό που εκτείνεται σε μέλη όπως έχει ένα συμπαγές μπλοκ 7z, οπότε η καταχώριση N μπορεί να αποσυμπιεστεί χωρίς να αγγίξει την καταχώριση M. Δώστε σε κάθε worker το δικό του z_stream πάνω στο δικό του εύρος byte και το μόνο πράγμα στο οποίο συγκρούονται είναι το seek. Αυτή τη σύγκρουση αφαιρεί το 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 φρουρεί μία αδιαίρετη λειτουργία, την τοποθέτηση του κοινού stream και την ανάγνωση από αυτό, και τίποτα άλλο. Η TZipArchive.OpenArchive κατασκευάζει το gate πάνω στο FInputStream μόλις ο κεντρικός κατάλογος αναλυθεί επιτυχώς, και η TZipArchive.Close το ελευθερώνει. Αρχεία ανοιγμένα για εγγραφή ποτέ δεν παίρνουν ένα. Κάθε ανάγνωση που εκτελεί ένας worker στο πακέτο επομένως διοχετεύεται μέσα από μία μόνο κρίσιμη ενότητα κρατημένη για τη διάρκεια μιας buffered ανάγνωσης

Όλα τα υπόλοιπα παραμένουν έξω από το κλείδωμα επειδή είναι ήδη ιδιωτικά ή ήδη αμετάβλητα. Το TZipSubStream κρατά τη δική του FPosition, οπότε κάθε worker παρακολουθεί τη δική του θέση στη δική του καταχώριση. Το TZLibStream που χτίζει η TZipEntry.GetStream πάνω σε εκείνο το sub-stream είναι ανά καταχώριση, δημιουργημένο με windowBits -15 για raw deflate, και ποτέ δεν μοιράζεται. Ο κεντρικός κατάλογος αναλύεται πλήρως πριν αρχίσει οποιοσδήποτε worker, συμπεριλαμβανομένης κάθε τοπικής κεφαλίδας, οπότε η GetEntryByName είναι μια αναζήτηση hash μόνο-για-ανάγνωση μέχρι να αρχίσει η ταυτοχρονία. Η ίδια η δρομολόγηση είναι τρεις γραμμές στην TZipSubStream.Read, και ο κλάδος χωρίς gate είναι αυτό που κρατά κάθε υπάρχοντα καλούντα ενός νήματος στην παλιά διαδρομή κώδικα

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;

Πόσο κοστίζει το gate κάτω από ανταγωνισμό;

Λιγότερο απ' όσο υποδηλώνει η φράση «καθολικό κλείδωμα στο αρχείο», λόγω της κοκκομέτρησης που τυχαίνει να χρησιμοποιεί το TZLibStream. Το buffer εισόδου του είναι BufferSize, ορισμένο ως $4000, οπότε η ReadInputBuffer τραβά 16 KB συμπιεσμένων bytes ανά ανανέωση και τα δίνει στην zng_inflate. Μία απόκτηση κλειδώματος επομένως καλύπτει 16 KB εισόδου deflate, που για XML φύλλου εργασίας διευρύνεται σε κάτι της τάξης των 100 KB markup που ο worker μετά αποκωδικοποιεί και αναλύει χωρίς να κρατά τίποτα. Το κλείδωμα κρατιέται για μια θετημένη ανάγνωση έναντι της cache του λειτουργικού συστήματος· η δουλειά που κλειδώνει μετριέται σε χιλιοστά του δευτερολέπτου

Το ειλικρινές όριο είναι εκεί όπου αυτός ο λόγος αντιστρέφεται. Καταχωρίσεις αποθηκευμένες αντί για αποσυμπιεσμένες διαβάζονται μέσα από το gate ένα προς ένα χωρίς δουλειά αποσυμπίεσης να κρύψει την καθυστέρηση, οπότε ένα πακέτο γεμάτο αποθηκευμένα μέλη θα σειριοποιούνταν πολύ πιο σκληρά. Ένα κρύο αρχείο σε αργά μέσα διευρύνει την κρίσιμη ενότητα, επειδή η ανάγνωση μέσα της είναι τώρα πραγματική μεταφορά δίσκου αντί για hit στην cache. Και πέρα από μια χούφτα workers το gate δεν είναι έτσι κι αλλιώς αυτό που χτυπάτε πρώτα: η ανάλυση φύλλου εργασίας είναι βαριά σε δεσμεύσεις, και ο διαχειριστής μνήμης Delphi σειριοποιεί δεσμεύσεις κατά μήκος νημάτων πολύ πριν το read gate γίνει ο περιορισμός. Γι' αυτό η TXLSXWorkbook.ParallelParseThreads έχει προεπιλογή ένα αυτόματο ανώτατο όριο αντί για ένα νήμα ανά πυρήνα

Το σώμα του worker, και ο βρόχος αποστράγγισης που είναι εύκολο να ξεχαστεί

Με το gate στη θέση του, το HotXLS διέγραψε τη σκηνοθεσία της Φάσης Α εντελώς. Ο worker τώρα ανοίγει το δικό του stream καταχώρισης και το τροφοδοτεί απευθείας στον parser. Δύο προσωρινά πεδία μεταφέρουν τις εισόδους: το FParZip κρατά το αρχείο για τη διάρκεια της παράλληλης φάσης, το FParSheetPartNames κρατά τα ονόματα μερών, και και τα δύο καθαρίζονται στο block finally ώστε κανένας μπαγιάτικος δείκτης να μην επιβιώνει ενός αποτυχημένου ανοίγματος. Το stream που επιστρέφει από την 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 καθώς περνούν bytes και καλεί το VerifyComplete μόνο όταν η θέση του φτάσει στο ασυμπίεστο μέγεθος καταγεγραμμένο στον κεντρικό κατάλογο· εκεί προέρχονται η εξαίρεση ασυμφωνίας μεγέθους και η εξαίρεση ασυμφωνίας CRC32, συν μια δοκιμαστική ανάγνωση ενός byte που πιάνει μια καταχώριση μακρύτερη από τη δηλωμένη. Ένας XML reader σταματά στο κλείσιμο στοιχείο και συνήθως αφήνει μια νέα γραμμή ή λίγα bytes κενού διαστήματος αδιάβαστα, οπότε χωρίς την αποστράγγιση η θέση ποτέ δεν φτάνει στο δηλωμένο μέγεθος και οι έλεγχοι ποτέ δεν ενεργοποιούνται. Η ανάγνωση του υπολοίπου σε ένα προσωρινό buffer δεν κοστίζει τίποτα και τους επαναφέρει. Όταν υπήρχαν τα streams σκηνοθεσίας, η XlsxCopyStreamAll το έκανε αυτό κατά τύχη

Τι εξακολουθεί να τρέχει σειριακά, και η σημαία που τα απενεργοποιεί όλα

Η Φάση Α επιβιώνει, μείον την εξαγωγή. Εξακολουθεί να δημιουργεί κάθε φύλλο εργασίας και να διαβάζει τις σχέσεις του στο νήμα κλήσης, που είναι αυτό που αφήνει κάθε κοινό map αμετάβλητο μόλις αρχίσουν οι workers. Η Φάση Γ εξακολουθεί να περπατά τα φύλλα σειριακά μετά για σχόλια, σχέδια, γραφήματα και πίνακες, και η φρουρά της άλλαξε από έναν έλεγχο null στον παλιό πίνακα σκηνοθεσίας σε zip.Exists έναντι του ονόματος μέρους. Οι κοινές εισόδοι μόνο-για-ανάγνωση που αγγίζουν οι workers, ο κοινός πίνακας strings και τα maps cellXf, είναι πλήρεις πριν αρχίσει η Φάση Β και ποτέ δεν γράφονται κατά τη διάρκειά της

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 εκφυλίζεται σε έναν απλό βρόχο στο νήμα κλήσης. Αυτό αξίζει να το γνωρίζετε για δύο λόγους: είναι η απάντηση μίας γραμμής αν ποτέ εμφανιστεί ένα ζήτημα νηματοποίησης στο πεδίο, και σημαίνει ότι η σειριακή και η παράλληλη διαδρομή μοιράζονται ένα ενιαίο σώμα κώδικα ανάλυσης αντί να αποκλίνουν. Οι εξαιρέσεις worker συλλαμβάνονται, ο χαμηλότερος δείκτης εργασίας κερδίζει, και το σφάλμα επαναπροκαλείται στο νήμα κλήσης αφού κάθε worker ενωθεί, οπότε ένα κατεστραμμένο φύλλο εργασίας εξακολουθεί να εμφανίζεται ως μία εξαίρεση στο αναμενόμενο σημείο. Γενική ρύθμιση της γύρω διαδρομής ανοίγματος καλύπτεται στον οδηγό για την απόδοση μεγάλων βιβλίων εργασίας σε Delphi

Το read gate, η φάση παράλληλου ανοίγματος και η streaming πρόσβαση καταχωρίσεων που περιγράφονται εδώ διατίθενται ως μέρος του τυπικού HotXLS Excel component για Delphi και C++Builder, με πλήρη πηγαίο κώδικα· η σελίδα προϊόντος φέρει την πλήρη αναφορά TXLSXWorkbook συμπεριλαμβανομένων των ιδιοτήτων παράλληλου ανοίγματος