Große Textdatei - einzelne Zeile löschen

[himitsu]

Es kommt auch nicht nur auf die Rechenleistung an, denn "gerechnet" wird hier ja fast nichts.
Speicherzugriffe, Caching und Co. sind bei diesem Thema die Hauptaufgabe.

Und auf dem Pi werden bestimmt auch weniger andere Prozesse parallel arbeiten, als wie im Windows.

[Rollo62]

Ich bin immer noch der Meinung das ein "Buffering" sich lohnen würde.
Vieleicht hilft ja schon das TBufferedFileStream ?

[himitsu]

Ein Buffering kann es aber auch verschlimmern.

siehe das zu "winzige" Buffering der alten Textdatei-APIs in Delphi (AssignFile),
welches mehr gegen, als für Flashspeicher und das Buffering des OS, kämpft.

[Rollo62]

Ja aber in dem Fall hier, zeilenweises Einlesen von 4GB ...
Da würde ich 64K Buffer einlesen und Zeilenweises Lesen im Specher vorziehen.

[himitsu]

jo, midestens 16 bis 64 KB wäre schon gut.

AssignFile hat standardmäßig 128 Byte, was ja zu überhaupt nichst passt.

512 Byte ist ja die kleinste Einheit (Sektor), für Dateizugriffe, und Cluster sind auch mindestens 4 KB groß ... bis 32 KB oder gar 64 KB bei broßen Festplatten.
Und 64 KB (eigentlich 8 KB) ist die Verwaltungsgröße im Arbeitsspeicher.

Die Verwaltungsgröße des WindowsFileCache wird vermutlich auch in mehereren Cluster-Größen arbeiten.
In Delphi ist der nahezu immer mit dazwischen, da der FileCache praktisch nirgendwo deaktiviert wird. (Parameter ans CreateFile)

[mytbo]

Vermutlich habe ich die Aufgabenstellung nicht richtig verstanden. Die hier berichteten Laufzeiten lassen mich daran zweifeln. Bei meiner Lösung bin ich für 1GB Rohdaten auf eine Laufzeit von 2 Sekunden gekommen. Der gesamte Scan für ca. 100GB sollte in weniger als 4 Minuten zu schaffen sein. Die erzielte Laufzeit entspricht damit meiner Erwartung für einen ersten Entwurf. Der Quelltext ist ein Proof of Concept, weder getestet noch optimiert.

Delphi-Quellcode:

uses
  Windows, Messages, SysUtils, Variants, Classes, Contnrs,
  mormot.core.base,
  mormot.core.text,
  mormot.core.test,
  mormot.core.perf,
  mormot.core.os;

type
  TOnScanProgressEvent = procedure(pmFileSize, pmDoneSize: Int64) of Object;

  TPartFile = class(TObject)
  private
    FPartFile: TFileName;
    FTextWriter: TTextWriter;
  public
    constructor Create(const pmcPartFile: TFileName); reintroduce;
    destructor Destroy; override;
    procedure AddLine(const pmcLine: RawUtf8);
  end;

  TMapIndex = array[Byte] of Integer;

  TFileParts = class(TObject)
  private
    FFileIndex: TMapIndex;
    FFileParts: TObjectList;
    FSourceFile: TFileName;
    FBasePartFileExt: String;
    FBasePartFileName: TFileName;
    FBasePartFilePath: TFileName;
    FOnScanProgress: TOnScanProgressEvent;
    function FindFileIndex(const pmcLine: RawUtf8): Integer;
    function AddNewPartFileItem(const pmcLine: RawUtf8): Integer;
    procedure ScannedLine(const pmcLine: RawUtf8);
  public
    constructor Create(const pmcSourceFile: TFileName); reintroduce;
    destructor Destroy; override;
    procedure ProcessFile;
    property OnScanProgress: TOnScanProgressEvent
      read FOnScanProgress write FOnScanProgress;
  end;


//==============================================================================
// TPartFile
//==============================================================================

constructor TPartFile.Create(const pmcPartFile: TFileName);
begin
  inherited Create;
  FPartFile := pmcPartFile;
  FTextWriter := TTextWriter.CreateOwnedFileStream(pmcPartFile);
end;

destructor TPartFile.Destroy;
begin
  FTextWriter.FlushFinal;
  FTextWriter.Free;
  inherited Destroy;
end;

procedure TPartFile.AddLine(const pmcLine: RawUtf8);
begin
  FTextWriter.AddString(pmcLine);
  FTextWriter.AddCR;
end;

//==============================================================================
// TFileParts
//==============================================================================

constructor TFileParts.Create(const pmcSourceFile: TFileName);
begin
  inherited Create;
  FFileParts := TObjectList.Create(True);
  FillChar(FFileIndex, SizeOf(FFileIndex), -1);
  FSourceFile := pmcSourceFile;
  FBasePartFileExt := ExtractFileExt(pmcSourceFile);
  FBasePartFilePath := ExtractFilePath(pmcSourceFile);
  FBasePartFileName := Utf8ToString(GetFileNameWithoutExtOrPath(pmcSourceFile));
end;

destructor TFileParts.Destroy;
begin
  FFileParts.Free;
  inherited Destroy;
end;

function TFileParts.FindFileIndex(const pmcLine: RawUtf8): Integer;
begin
  if pmcLine <> '' then
    Result := FFileIndex[Ord(pmcLine[1])]
  else
    Result := -1;
end;

function TFileParts.AddNewPartFileItem(const pmcLine: RawUtf8): Integer;
const
  FORMAT_PARTFILE = '%s-%.3d%s';
var
  fileOrd: Integer;
  fileName: TFileName;
begin
  Result := -1;
  if pmcLine <> '' then
  begin
    fileOrd := Ord(pmcLine[1]);
    fileName := MakePath([FBasePartFilePath, Format(FORMAT_PARTFILE, [FBasePartFileName, fileOrd, FBasePartFileExt])]);
    Result := FFileParts.Add(TPartFile.Create(fileName));
    FFileIndex[fileOrd] := Result;
  end;
end;

procedure TFileParts.ScannedLine(const pmcLine: RawUtf8);
var
  idx: Integer;
begin
  if pmcLine <> '' then
  begin
    idx := FindFileIndex(pmcLine);
    if idx < 0 then
      idx := AddNewPartFileItem(pmcLine);

    if idx >= 0 then
      TPartFile(FFileParts.Items[idx]).AddLine(pmcLine);
  end;
end;

procedure TFileParts.ProcessFile;
const
  BUFFER_SIZE = 1 shl 20;
  BUFFER_ENDCHUNK = 1 shl 10;
var
  fileHnd: THandle;
  filePos: Int64;
  fileSize: Int64;
  readLine: RawUtf8;
  readCount: Integer;
  readBuffer: RawByteString;
  p, pStart: PUtf8Char;
begin
  fileHnd := FileOpenSequentialRead(FSourceFile);
  if ValidHandle(fileHnd) then
  try
    fileSize := mormot.core.os.FileSize(fileHnd);

    filePos := 0;
    readCount := 0;
    FastSetRawByteString(readBuffer, Nil, BUFFER_SIZE);
    repeat
      Inc(filePos, readCount);
      if Assigned(FOnScanProgress) then
        FOnScanProgress(fileSize, filePos);

      FileSeek64(fileHnd, filePos, soFromBeginning);
      readCount := FileRead(fileHnd, Pointer(readBuffer)^, Length(readBuffer));
      if readCount <= 0 then
        Exit; //=>

      if readCount < BUFFER_SIZE then
        FakeLength(readBuffer, readCount);

      readCount := 0;
      p := PUtf8Char(Pointer(readBuffer));
      while p <> Nil do
      begin
        pStart := p;
        readLine := GetNextLine(p, p);
        if readLine <> '' then
          ScannedLine(readLine);

        Inc(readCount, (p - pStart));
        if (BUFFER_SIZE - readCount) < BUFFER_ENDCHUNK then
          Break; //->
      end;
    until False;
  finally
    FileClose(fileHnd);
  end;
end;

Testdaten erzeugt wie folgt:

Delphi-Quellcode:

const
  ITEM_COUNT = 50000000;
var
  i: Integer;
  value: RawUtf8;
  textWriter: TTextWriter;
begin
  i := 0;
  textWriter := TTextWriter.CreateOwnedFileStream(MakePath([Executable.ProgramFilePath, 'random.data']));
  try
    while i < ITEM_COUNT do
    begin
      value := TSynTestCase.RandomIdentifier(12 + Random(20));
      if value[1] <> '_' then
      begin
        textWriter.AddString(value);
        textWriter.AddCR;
        Inc(i);
      end;
    end;

    textWriter.FlushFinal;
  finally
    textWriter.Free;
  end;
end;

Die Anwendung wie folgt:

Delphi-Quellcode:

var
  test: TFileParts;
  timer: TPrecisionTimer;
begin
  test := TFileParts.Create(MakePath([Executable.ProgramFilePath, 'random.data']));
  try
    timer.Start;
    test.ProcessFile;
    ShowMessage(Format('Total time: %s', [timer.Stop]))
  finally
    test.Free;
  end;

Der Quelltext sollte mit Delphi 7 kompatibel sein. mORMot ist bei mir immer mit dabei. Und jetzt ab ins Schwimmbad.

Nachtrag: Das Benchmark-Test-Programm ist auf einem Rechner mit SATA SSD gelaufen. Die theoretischen Transferraten sind: Lesegeschwindigkeit 550 MB/s, Schreibgeschwindigkeit 520 MB/s. Am Ergebnis sieht man, dass der limitierende Faktor für die Verarbeitung die Geschwindigkeit der SSD ist. Sie erreicht praktisch das maximal Mögliche. Die verarbeitende CPU wird mit ca. 50% belastet. Der benötigte Arbeitsspeicher ist vernachlässigbar. Vermutlich wird sich die Verteilung auch bei einer superschnellen SSD mit NVMe Anschluss kaum ändern. Die Geschwindigkeit des Datenspeichers bleibt der bestimmende Faktor. Eine weitere Optimierung macht in diesem Fall keinen Sinn. Ein kleines Schmankerl am Rande: Im Testszenario sieht man auch, dass eine SSD beides (nicht immer) kann, annähernd maximal Lesen und Schreiben gleichzeitig.

Bis bald...
Thomas