Concurrent ZIP Inflate ใน Delphi: Read Gate ของ HotXLS

HotXLS component Excel เนทีฟสำหรับ Delphi และ C++Builder คลายบีบอัดหลาย worksheet ของ XLSX พร้อมกันจาก ZIP package ที่เปิดอยู่ไฟล์เดียว กลไกคือ TZipReadGate คลาสเล็ก ๆ ใน lxZipArchive.pas ที่ถือ package stream บวก critical section หนึ่งตัว และเปิดให้ใช้เมธอดเดียวเท่านั้น มันทำให้คู่ seek-and-read เป็นแบบเรียงลำดับ ทุกอย่างเหนือคู่นั้นทำงานพร้อมกันได้

ปัญหาที่บังคับให้เกิดการออกแบบนี้เป็นสิ่งที่นักพัฒนา Delphi ทุกคนที่เคยเปิด workbook ขนาดใหญ่เคยเจอมาแล้ว xlsx ขนาด 80 MB คือ XML ที่บีบอัดแบบ deflate 80 MB และส่วน worksheet ภายในขยายออกประมาณห้าถึงสิบเท่า ถ้าเส้นทางการเปิดของคุณดึง worksheet แต่ละตัวเข้าไปใน memory stream ก่อน parse คุณจะจ่ายค่าไบต์ที่ขยายแล้วซ้อนบน workbook ที่คุณกำลังสร้าง และค่าสูงสุดจะมาถึงก่อนที่เซลล์แม้แต่ตัวเดียวจะถูกสร้างขึ้น บทความนี้พูดถึงการทำงานพร้อมกันระดับ package ที่กำจัดขั้นตอนการเตรียมข้อมูลนั้นออกไป เพดาน allocator ที่อยู่เหนือมันครอบคลุมใน บทความเรื่องการ parse XLSX แบบขนานกับ memory manager และ API แบบอ่านครั้งเดียวไม่เคยสร้างข้อมูลเลยครอบคลุมใน บทความ streaming direct reader

ทำไม open path เก่าถึงต้องเตรียม worksheet ทุกตัวไว้ใน RAM

Parallel open เดิมใน HotXLS เป็น pipeline สามระยะ และระยะกลางเป็นระยะเดียวที่รันบน worker Phase A เดินตามรายการ sheet แบบเรียงลำดับ สร้างแต่ละ worksheet อ่าน relationship part ของมัน และคัดลอก worksheet XML ที่คลายบีบอัดแล้วทั้งหมดเข้าไปใน TMemoryStream ส่วนตัว Phase B กระจาย ParseWorksheetXml ออกไปทั่ว pool Phase C กลับไปยัง archive บน thread ที่เรียกสำหรับ part เล็ก ๆ บริวาร คือ comment, threaded comment, drawing, chart, table รูปแบบนั้นถูกเลือกด้วยเหตุผลที่ระบุไว้ comment บนหัวไฟล์ lxParallelParse.pas เคยบอกไว้ตรง ๆ ว่า zip archive และสถานะ inflate ของมันไม่ thread-safe และบันทึกภายในไปไกลกว่านั้น อย่าเสียเวลาล็อก archive เพราะเมื่อสถานะ inflate ถูกทำเป็นเรียงลำดับต่อ entry แล้ว การล็อกไม่ได้ประโยชน์อะไรเลย Phase A มีไว้เพื่อให้การแตะ archive ทุกครั้งอยู่บน thread เดียว ต้นทุนคือ workbook ที่มีแปด sheet ที่ทำงานอยู่จะถือ worksheet XML buffer ที่คลายบีบอัดเต็มรูปแบบแปดตัวพร้อมกันในหน่วยความจำ และ buffer เหล่านั้นคือ object ชั่วคราวที่ใหญ่ที่สุดใน open path ทั้งหมด

สอง thread คลายบีบอัดจาก ZIP stream เดียวได้ไหม

ได้ และการตัดสินเดิมผิดในแบบที่เจาะจงและระบุตำแหน่งได้ มันยุบสถานะสองแบบที่ต่างกันเข้าเป็นประโยคเดียว สถานะ inflate นั้นแชร์กันจริง ๆ ไม่ได้ z_stream ของ zlib พก sliding window, Huffman table และตำแหน่ง bit สำหรับสมาชิกที่บีบอัดหนึ่งตัว และสอง thread ที่ดันไบต์ผ่านตัวเดียวกันจะสร้างขยะ แหล่งไบต์ที่อยู่ข้างใต้เป็นคำถามคนละเรื่องเลย และคำตอบตรงนั้นคือ file stream มีสถานะที่เปลี่ยนแปลงได้ที่ใช้ร่วมกันอยู่ตัวเดียวเท่านั้นที่คุ้มค่าจะป้องกัน นั่นคือ position cursor ของมัน

Container ของ ZIP ทำให้การแยกนี้ถูกกฎหมาย แต่ละสมาชิกใน ZIP archive ถูกบีบอัดแยกอิสระจากกัน มี local file header ของตัวเอง มี deflate bit stream ของตัวเองที่ DataOffset ของตัวเอง มี CRC32 และขนาดของตัวเองใน central directory ไม่มี shared dictionary ที่ครอบคลุมข้ามสมาชิกแบบที่ solid block ของ 7z มี ดังนั้น entry N สามารถถูกคลายบีบอัดได้โดยไม่ต้องแตะ entry M เลย ให้ worker แต่ละตัวมี z_stream ของตัวเองเหนือช่วงไบต์ของตัวเอง และสิ่งเดียวที่พวกมันชนกันคือการ 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 เมื่อ central directory parse สำเร็จ และ TZipArchive.Close จะปล่อยมันทิ้ง archive ที่เปิดสำหรับการเขียนจะไม่มี gate เลย ทุกการอ่านที่ worker ทำบน package จะไหลผ่าน critical section ตัวเดียวที่ถูกถือไว้ตลอดระยะเวลาของการอ่านแบบ buffered หนึ่งครั้ง

ทุกอย่างอื่นอยู่นอกล็อกเพราะมันเป็นแบบส่วนตัวหรือไม่เปลี่ยนแปลงอยู่แล้ว TZipSubStream เก็บ FPosition ของตัวเอง ดังนั้น worker แต่ละตัวจึงติดตามตำแหน่งของตัวเองใน entry ของตัวเอง TZLibStream ที่ TZipEntry.GetStream สร้างเหนือ sub-stream นั้นเป็นแบบต่อ entry สร้างด้วย windowBits เป็น -15 สำหรับ raw deflate และไม่เคยถูกแชร์เลย central directory ถูก parse ครบสมบูรณ์ก่อนที่ worker ใด ๆ จะเริ่ม รวมถึง local header ทุกตัว ดังนั้น GetEntryByName จึงเป็น hash lookup แบบอ่านอย่างเดียวเมื่อการทำงานพร้อมกันเริ่มขึ้น การจัดเส้นทางเองเป็นแค่สามบรรทัดใน TZipSubStream.Read และสาขาที่ไม่มี gate คือสิ่งที่ทำให้ caller แบบ single-threaded เดิมทุกตัวยังอยู่บน code path เก่า

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 นี้มีต้นทุนเท่าไรภายใต้การแย่งชิง

น้อยกว่าที่วลี "global lock บน archive" บอกไว้มาก เพราะความละเอียดที่ TZLibStream ใช้อยู่แล้ว input buffer ของมันคือ BufferSize ที่กำหนดเป็น $4000 ดังนั้น ReadInputBuffer จึงดึงข้อมูลบีบอัด 16 KB ต่อการเติมหนึ่งครั้งและส่งให้ zng_inflate การได้ล็อกหนึ่งครั้งจึงครอบคลุมข้อมูลนำเข้า deflate 16 KB ซึ่งสำหรับ worksheet XML จะขยายเป็น markup ประมาณ 100 KB ที่ worker จะถอดรหัสและ parse ต่อไปโดยไม่ต้องถืออะไรไว้เลย ล็อกถูกถือไว้กับการอ่านที่กำหนดตำแหน่งแล้วเทียบกับ cache ของระบบปฏิบัติการ งานที่มันกันไว้วัดเป็นมิลลิวินาที

ขอบเขตที่ตรงไปตรงมาคือจุดที่อัตราส่วนนั้นกลับด้าน entry ที่เก็บแบบ stored แทนที่จะ deflate จะอ่านผ่าน gate แบบหนึ่งต่อหนึ่งโดยไม่มีงาน inflate มาซ่อน latency ดังนั้น package ที่เต็มไปด้วยสมาชิกแบบ stored จะทำให้เกิดการเรียงลำดับหนักกว่ามาก ไฟล์เย็นบนสื่อที่ช้าจะขยาย critical section เพราะการอ่านข้างในตอนนี้คือการโอนดิสก์จริง ไม่ใช่ cache hit และเกินกว่า worker ไม่กี่ตัว gate ไม่ใช่สิ่งที่คุณเจอก่อนอยู่ดี การ parse worksheet หนักเรื่องการ allocation และ Delphi memory manager เรียงลำดับการ allocate ข้าม thread ก่อนที่ read gate จะกลายเป็นข้อจำกัดเสียอีก นี่คือเหตุผลที่ TXLSXWorkbook.ParallelParseThreads ค่าเริ่มต้นเป็นเพดานอัตโนมัติแทนที่จะเป็นหนึ่ง thread ต่อหนึ่งคอร์

ตัว worker และลูป drain ที่ลืมง่าย

เมื่อมี gate แล้ว HotPDF ลบ Phase A staging ออกทั้งหมด ตอนนี้ worker เปิด entry stream ของตัวเองและป้อนมันตรงเข้าไปยัง parser เลย มีฟิลด์ชั่วคราวสองตัวที่พาข้อมูลนำเข้า FParZip ถือ archive ไว้ตลอดระยะเวลาของ parallel phase FParSheetPartNames ถือชื่อ part และทั้งสองถูก clear ใน block finally เพื่อไม่ให้ pointer ที่ค้างรอดจากการเปิดที่ล้มเหลว stream ที่ได้กลับมาจาก TZipArchive.OpenFile คือ TZipVerifiedStream ที่ห่อ TZLibStream ที่ห่อ TZipSubStream และการปล่อยตัวนอกสุดจะปล่อยทั้ง chain

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;

ลูป drain คือรายละเอียดที่การพอร์ตโค้ดเดิมตรง ๆ จะพลาด และการพลาดมันจะปิดการตรวจสอบความสมบูรณ์อย่างเงียบ ๆ TZipVerifiedStream สะสม CRC32 ที่กำลังคำนวณอยู่ขณะที่ไบต์วิ่งผ่าน และเรียก VerifyComplete ก็ต่อเมื่อตำแหน่งของมันไปถึงขนาดไม่บีบอัดที่บันทึกไว้ใน central directory เท่านั้น นั่นคือจุดที่ exception ของขนาดไม่ตรงกันและ CRC32 ไม่ตรงกันมาจาก บวกกับการอ่าน probe หนึ่งไบต์ที่จับ entry ที่ยาวกว่าที่ประกาศไว้ ตัวอ่าน XML จะหยุดที่ element ปิดและมักจะเหลือ newline หรือ whitespace ท้ายไม่กี่ไบต์ที่ยังไม่ได้อ่าน ดังนั้นถ้าไม่มี drain ตำแหน่งจะไม่มีวันไปถึงขนาดที่ประกาศไว้ และการตรวจสอบจะไม่มีวันทำงาน การอ่านส่วนที่เหลือเข้าไปใน scratch buffer ไม่มีต้นทุนอะไรเลยและคืนการตรวจสอบเหล่านั้นกลับมา เมื่อ staging stream ยังมีอยู่ XlsxCopyStreamAll ทำสิ่งนี้อยู่แล้วโดยบังเอิญ

สิ่งที่ยังรันแบบเรียงลำดับ และ flag ที่ปิดทุกอย่าง

Phase A ยังคงอยู่ ลบแค่ส่วนการดึงข้อมูลออก มันยังคงสร้าง worksheet แต่ละตัวและอ่าน relationship ของมันบน thread ที่เรียก ซึ่งเป็นสิ่งที่ทำให้ shared map ทุกตัวไม่เปลี่ยนแปลงเมื่อ worker เริ่มทำงาน Phase C ยังคงเดินผ่าน sheet แบบเรียงลำดับหลังจากนั้นสำหรับ comment, drawing, chart และ table และการป้องกันของมันเปลี่ยนจาก null check บน staging array เดิมเป็น zip.Exists เทียบกับชื่อ part ข้อมูลนำเข้าที่ใช้ร่วมกันแบบอ่านอย่างเดียวที่ worker แตะต้อง คือ shared string table และ map cellXf จะสมบูรณ์ก่อน Phase B เริ่มต้น และไม่เคยถูกเขียนระหว่างนั้นเลย

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 จะส่ง job procedure เดียวกันด้วยจำนวน thread เป็นหนึ่ง และ RunParallelJobs จะเสื่อมกลายเป็นลูปธรรมดาบน thread ที่เรียก นี่ควรค่าแก่การรู้ไว้ด้วยเหตุผลสองข้อ มันคือคำตอบบรรทัดเดียวถ้าความกังวลเรื่อง threading เกิดขึ้นในภาคสนาม และหมายความว่า path แบบเรียงลำดับกับแบบขนานใช้ parsing code ชุดเดียวกันแทนที่จะแยกออกจากกัน ข้อผิดพลาดของ worker ถูกดักไว้ job index ที่ต่ำที่สุดชนะ และ error จะถูก re-raise บน thread ที่เรียกหลังจาก worker ทุกตัว join แล้ว ดังนั้น worksheet ที่เสียหายจะยังคงปรากฏเป็น exception เดียวในที่ที่คาดหวังไว้ การปรับแต่งทั่วไปของ open path โดยรอบครอบคลุมใน คู่มือประสิทธิภาพ workbook ขนาดใหญ่ใน Delphi

Read gate, parallel open phase และ streaming entry access ที่กล่าวถึงในบทความนี้มาพร้อมกับ HotXLS Excel component มาตรฐานสำหรับ Delphi และ C++Builder พร้อม source เต็ม หน้าผลิตภัณฑ์มีเอกสารอ้างอิง TXLSXWorkbook ฉบับเต็ม รวมถึง property การเปิดแบบขนาน