UnitPartSortThread.pas

unit UnitPartSortThread;

interface

uses
  System.Classes, Winapi.Windows, UnitCommon;

type
  TPartSortThread = class(TCommonThread)
  private
    m_Offset      : UInt64;   //  ñìåùåíèå íàáîðà â èñõîäíîì ôàéëå
    m_Size        : UInt64;   //  ðàçìåð íàáîðà â èñõîäíîì ôàéëå
    m_pIDs        : Pointer;  //  áàçîâûå óêàçàòåëè âíóòðè áóôåðà
    m_pPtrs       : Pointer;  //  óêàçàòåëü íà ìàññèâ ñòðîê (áûñòðûé äîñòóï)
    m_pTail       : Pointer;  //  óêàçàòåëü íà êîíåö ðàáî÷åãî áóôåðà (íà÷àëî ðåâåðñèðîâàííîãî ìàññèâà äëèí ñòðîê)
    m_ItemsCount  : Word;     //  ÷èñëî çàïèñåé â áóôåðå
    m_PackedSize  : Cardinal; //  óïàêîâàííûé ðàçìåð áëîêà äàííûõ (ñòðîê)
    m_Results     : Integer;  //  ÷èñëî ÷àñòè÷íûõ ðåçóëüòàòîâ ñîðòèðîâêè òåêóùåãî ïîòîêà
    m_ResultsSize : UInt64;   //  îáúåì ÷àñòè÷íûõ ðåçóëüòàòîâ ñîðòèðîâêè òåêóùåãî ïîòîêà
    m_hFile       : THandle;  //  äåñêðèïòîð èñõîäíîãî ôàéëà
    m_hDstFile    : THandle;  //  äåñêðèïòîð öåëåâîãî ôàéëà
    m_hSrcLock    : THandle;  //  ìüþòåêñ, çàùèùàþùèé äîñòóï ê èñõîäíîìó ôàéëó
    m_hDstLock    : THandle;  //  ìüþòåêñ, çàùèùàþùèé äîñòóï ê öåëåâîìó ôàéëó
    m_FileSize    : UInt64;   //  ðàçìåð èñõîäíîãî ôàéëà
    m_Last        : Boolean;  //  ïðèçíàê õâîñòîâîãî ïîòîêà (ó÷åò ïîñëåäíåé ñòðîêè áåç CRLF)
    m_pRunning    : PInteger; //  îáùåå ÷èñëî âûïîëíÿþùèõñÿ ïîòîêîâ
    function Comp( a, b: Word ): Integer;
    procedure BuildList;
    procedure QuickSort( l, r: Word );
  protected
    procedure Execute; override;
    function GetID( Index: Word ): PWord; inline;
    function GetItem( Index: Word ): PAnsiChar; inline;
    function GetItemLength( Index: Word ): Word; inline;
    function GetItemOffset( Index: Word ): Cardinal; inline;
    procedure SortStrings;
    function WriteStrings( const FileOffset, FileSize: UInt64 ): Boolean;
  public
    procedure Init(
      hFile, hDstFile, hSrcLock, hDstLock: THandle;
      Offset, Size, FileSize: UInt64;
      pBuffer: Pointer;
      pRunning, pLastOutputTime: PInteger;
      Last: Boolean
    );
    procedure AddSize( Chunk: IntPtr );
    function AdjustOffset( hFile: THandle ): Word;
    property Results: Integer read m_Results;
    property ResultsSize: UInt64 read m_ResultsSize;
  end;

implementation

uses System.AnsiStrings, System.SysUtils, SDIMAIN;

function SearchCRLF( p: Pointer; Length: Cardinal ): Cardinal;
begin
  Result  := 0;
  while Length > 0 do
    begin
      if PAnsiChar( p )^ = #13 then
        begin
          Inc( PAnsiChar( p ) );
          Dec( Length );
          if ( Length > 0 ) and ( PAnsiChar( p )^ = #10 ) then
            Exit
          else
            Inc( Result );
          if Length = 0 then Break;
        end;
      Inc( PAnsiChar( p ) );
      Dec( Length );
      Inc( Result );
    end;
end;

{ TPartSortThread }

procedure TPartSortThread.AddSize( Chunk: IntPtr );
begin
  Inc( m_Size, Chunk );
end;

function TPartSortThread.AdjustOffset( hFile: THandle ): Word;
var Offset: UInt64; ToRead, Readed: Cardinal; CRLF: Word;
begin
  //  çàõâàòûâàåì ïîñëåäíèå 2 ñèìâîëà ïðåäûäóùåãî áóôåðà
  //  (ó÷åò ñèòóàöèè, êîãäà ãðàíèöà áóôåðîâ ïåðåðåçàåò ñòðîêó )
  Offset  := Self.m_Offset - 2;
  //  ÷èòàåì áóôåð
  SetFilePointer(
    hFile, Int64Rec( Offset ).Lo, @Int64Rec( Offset ).Hi, FILE_BEGIN
  );
  ToRead  := Min( CoreMemSize, m_Size + 2 );
  if False = ReadFile( hFile, m_pBuffer^, ToRead, Readed, nil ) then
    begin
      DebugBreak;
      Result  := 0;
      Exit;
    end;
  if ToRead <> Readed then
    begin
      DebugBreak;
      Result  := 0;
      Exit;
    end;
  //
  CRLF    := SearchCRLF( m_pBuffer, Readed );
  Result  := CRLF;
  if CRLF < Readed then
    Inc( Result, 2 );
  Dec( Result, 2 );
  //  êîððåêòèðîâêà ñìåùåíèé
  Inc( Self.m_Offset, Result );
  Dec( Self.m_Size, Result );
end;

procedure TPartSortThread.BuildList;
var
  ps: Cardinal;
  i, Length: Word;
  pString: PAnsiChar;
  pSize, pID: PWord;
  pPtr: PPointer;
begin
  m_pIDs    := m_pBuffer;
  m_pPtrs   := m_pBuffer;
  m_pTail   := m_pBuffer;
  //
  Inc( PAnsiChar( m_pPtrs ), m_PackedSize );
  Inc( PWord( m_pPtrs ), m_ItemsCount );
  Inc( PAnsiChar( m_pIDs ), m_PackedSize );
  Inc( PAnsiChar( m_pTail ), CoreMemSize );
  //
  pPtr    := m_pPtrs;
  pID     := m_pIDs;
  pSize   := m_pTail;
  pString := m_pBuffer;
  ps      := 0;
  for i := 0 to m_ItemsCount - 1 do
    begin
      Dec( pSize );
      //
      pPtr^   := pString;
      pID^    := i;
      Length  := Min( pSize^, 50 );
      //
      Inc( pString, Length );
      Inc( pPtr );
      Inc( pID );
      Inc( ps, Length );
    end;
  Assert( ps = m_PackedSize );
end;

function TPartSortThread.Comp(a, b: Word): Integer;
var al, bl: Word;
begin
  al      := GetItemLength( a );
  bl      := GetItemLength( b );
  if al < bl then
    Result  := -1
  else if al > bl then
    Result  := +1
  else
    Result  := System.AnsiStrings.AnsiStrLComp(
      GetItem( a ), GetItem( b ), Min( 50, al )
    );
end;

procedure TPartSortThread.Execute;
var
  BufferOffset, RelOffset, Unreaded: UInt64;
  ToRead, Readed, CRLF, _CRLF, RecSize, UnpackedSize, SysSize: Cardinal;
  p, pSizes: Pointer;
  Done: Boolean;
label loop;
begin
  while ( WAIT_OBJECT_0 = WaitForSingleObject( m_hWakeup, INFINITE ) ) and not m_fAbort do
    begin
      BufferOffset  := m_Offset;
      RelOffset     := m_Offset;
      Unreaded      := m_Size;
      UnpackedSize  := 0;
      m_Results     := 0;
      m_ResultsSize := 0;
      while Unreaded > 0 do
        begin
          p           := m_pBuffer;
          pSizes      := m_pBuffer;
          Done        := False;
          m_PackedSize  := 0;
          m_ItemsCount  := 0;
          Inc( PAnsiChar( pSizes ), CoreMemSize );
          while not Done and ( Unreaded > 0 ) do
            begin
              RelOffset := BufferOffset;
              //  ÷èòàåì áóôåð
              WaitForSingleObject( m_hSrcLock, INFINITE );
              SetFilePointer(
                m_hFile, Int64Rec( BufferOffset ).Lo, @Int64Rec( BufferOffset ).Hi,
                FILE_BEGIN
              );
              ToRead    := Min(
                CoreMemSize - m_PackedSize - m_ItemsCount * sizeof( Word ), Unreaded
              );
              Assert( IntPtr(p) < ( IntPtr( m_pBuffer ) + CoreMemSize ) );
              Assert( ( UIntPtr(p) + ToRead ) <= ( UIntPtr( m_pBuffer ) + CoreMemSize ) );
              if False = ReadFile( m_hFile, p^, ToRead, Readed, nil ) then
                begin
                  DebugBreak;
                  ReleaseMutex( m_hSrcLock );
                  Exit;
                end;
              if ToRead <> Readed then
                begin
                  DebugBreak;
                  ReleaseMutex( m_hSrcLock );
                  Exit;
                end;
              ReleaseMutex( m_hSrcLock );
              //  ó÷èòûâàåì îñòàòîê ñ ïðåäûäóùåé èòåðàöèè
              Dec( PAnsiChar( p ), UnpackedSize );
              Inc( ToRead, UnpackedSize );
              //  óïàêîâûâàåì ñòðîêè
              UnpackedSize  := ToRead;
loop:
              CRLF  := SearchCRLF( p, UnpackedSize );
              _CRLF := CRLF;
              if CRLF < UnpackedSize then
                Inc( _CRLF, 2 );
              if ( CRLF < UnpackedSize ) or ( m_Last and ( UnpackedSize > 0 ) and ( UnpackedSize = Unreaded ) ) then
                begin
                  Assert( CRLF <= 500 );
                  RecSize := Min( CRLF, 50 );
                  //  îñòàòîê áóôåðà äîëæåí âìåùàòü âñå èíäåêñû è óêàçàòåëè äëÿ óïàêîâàííûõ ñòðîê
                  SysSize  := 2 * ( m_ItemsCount + 1 ) * ( 2 * sizeof( Word ) + sizeof( Pointer ) );
                  if ( CoreMemSize - m_PackedSize - RecSize ) >= SysSize then
                    begin
                      Inc( m_PackedSize, RecSize );
                      Dec( UnpackedSize, _CRLF );
                      Inc( BufferOffset, _CRLF );
                      Dec( Unreaded, _CRLF );
                      MoveMemory(
                        PAnsiChar( p ) + RecSize, PAnsiChar( p ) + _CRLF, UnpackedSize
                      );
                      Inc( m_ItemsCount );
                      //  ïèøåì ôàêòè÷åñêèé ðàçìåð ñòðîêè â êîíåö áóôåðà
                      Assert( CRLF <= $0000FFFF );
                      Dec( PWord( pSizes ) );
                      PWord( pSizes )^  := Word( CRLF );
                      //
                      Inc( PAnsiChar( p ), RecSize );
                      goto loop;
                    end
                  else
                    begin
                      UnpackedSize  := 0;
                      Done          := True;
                    end;
                end;
              //  â UnpackedSize îñòàòîê, êîòîðûé íóæíî ïåðåíåñòè íà ñëåä. èòåðàöèþ
              Inc( PAnsiChar( p ), UnpackedSize );
            end;
          //  çàïîëíÿåì ìàññèâ èíäåêñîâ
          BuildList;
          //  ñîðòèðóåì èíäåêñû ñòðîê
          SortStrings;
          //  çàïèñûâàåì ÷àñòè÷íûå ðåçóëüòàòû â ôàéë
          WaitForSingleObject( m_hDstLock, INFINITE );
          if not WriteStrings( RelOffset, m_FileSize ) then
            begin
              DebugBreak;
              ReleaseMutex( m_hDstLock );
              Exit;
            end;
          ReleaseMutex( m_hDstLock );
          //
          Inc( m_Results );
          //  ðàçìåð äàííûõ
          Inc( m_ResultsSize, m_PackedSize );
          //  ðàçìåð äëèí è ñìåùåíèé ñòðîê
          Inc( m_ResultsSize, m_ItemsCount * PartSortStringHeaderSize );
          //  ðàçìåð çàãîëîâêà (ðàçìåð è ÷èñëî ñòðîê)
          Inc( m_ResultsSize, PartSortHeaderSize );
          //  îáíîâëÿåì ñòàòóñ
          UpdateProgress( 100 - ( Unreaded * 100 ) / m_Size );
        end;
      if ( InterlockedDecrement( m_pRunning^ ) = 0 ) and not m_fAbort then
        Synchronize( SDIAppForm.NextState );
    end;
end;

function TPartSortThread.GetID(Index: Word): PWord;
begin
  Result  := m_pIDs;
  Inc( Result, Index );
end;

function TPartSortThread.GetItem(Index: Word): PAnsiChar;
begin
  Result  := PPointer( IntPtr( m_pPtrs ) + Index * sizeof( IntPtr ) )^;
end;

function TPartSortThread.GetItemLength(Index: Word): Word;
begin
  Result  := PWord( IntPtr( m_pTail ) - ( Index + 1 ) * sizeof( Word ) )^;
end;

function TPartSortThread.GetItemOffset(Index: Word): Cardinal;
var pSize: PWord;
begin
  Result  := 0;
  pSize   := m_pTail;
  while Index > 0 do
    begin
      Dec( pSize );
      Dec( Index );
      Inc( Result, 2 + pSize^ );
    end;
end;

procedure TPartSortThread.Init(
  hFile, hDstFile, hSrcLock, hDstLock: THandle;
  Offset, Size, FileSize: UInt64;
  pBuffer: Pointer;
  pRunning, pLastOutputTime: PInteger;
  Last: Boolean
);
begin
  m_hFile            := hFile;
  m_hDstFile         := hDstFile;
  m_hSrcLock         := hSrcLock;
  m_hDstLock         := hDstLock;
  m_hWakeup          := m_hWakeup;
  m_FileSize         := FileSize;
  m_Offset           := Offset;
  m_Size             := Size;
  m_pRunning         := pRunning;
  m_Last             := Last;
  m_ItemsCount       := 0;
  inherited Init( pBuffer, pLastOutputTime);
end;

procedure TPartSortThread.QuickSort(l, r: Word);
var i, j, q, sw: Integer;
begin
  if ( r - l ) < 1 then Exit;
  repeat
    i := l;
    j := r;
    q := GetID( l + ( r - l ) shr 1 )^;
    repeat
      while Comp( GetID( i )^, q ) < 0 do Inc(I);
      while Comp( GetID( j )^, q ) > 0 do Dec(J);
      if I <= J then
        begin
          if i <> j then
            begin
              sw          := GetID( i )^;
              GetID( i )^ := GetID( j )^;
              GetID( j )^ := sw;
            end;
          Inc( i );
          Dec( j );
        end;
    until i > j;
    if l < j then
      QuickSort( l, j );
    l := i;
  until i >= r;
end;

procedure TPartSortThread.SortStrings;
begin
  if m_ItemsCount > 0 then
    QuickSort( 0, m_ItemsCount - 1 );
end;

function TPartSortThread.WriteStrings( const FileOffset, FileSize: UInt64): Boolean;
var i, ID, Size, Length: Word; Written: Cardinal; ui64: UInt64;
begin
  Result  := False;
  //  ðàçìåð ïîñëåäîâàòåëüíîñòè
  ui64  := PartSortHeaderSize + m_ItemsCount * PartSortStringHeaderSize + m_PackedSize;
  if False = WriteFile( m_hDstFile, ui64, sizeof( ui64 ), Written, nil ) then
    begin
      DebugBreak;
      Exit;
    end;
  if Written <> sizeof( ui64 ) then
    begin
      DebugBreak;
      Exit;
    end;
  //  ÷èñëî ñòðîê
  ui64  := m_ItemsCount;
  if False = WriteFile( m_hDstFile, ui64, sizeof( ui64 ), Written, nil ) then
    begin
      DebugBreak;
      Exit;
    end;
  if Written <> sizeof( ui64 ) then
    begin
      DebugBreak;
      Exit;
    end;
  //  ñòðîêè
  for i := 0 to m_ItemsCount - 1 do
    begin
      ID    := GetID( i )^;
      Size  := GetItemLength( ID );
      //  ñìåùåíèå ñòðîêè
      ui64  := FileOffset + GetItemOffset( ID );
      if ( ui64 + Size ) > FileSize then
        begin
          //Assert( ( ui64 + Size ) <= FileSize );
          DebugBreak;
        end;
      if False = WriteFile( m_hDstFile, ui64, sizeof( UInt64 ), Written, nil ) then
        begin
          DebugBreak;
          Exit;
        end;
      if Written <> sizeof( UInt64 ) then
        begin
          DebugBreak;
          Exit;
        end;
      //  äëèíà ñòðîêè
      if False = WriteFile( m_hDstFile, Size, sizeof( Word ), Written, nil ) then
        begin
          DebugBreak;
          Exit;
        end;
      if Written <> sizeof( Word ) then
        begin
          DebugBreak;
          Exit;
        end;
      //  ñòðîêà
      Length  := Min( Size, 50 );
      if False = WriteFile( m_hDstFile, GetItem( ID )^, Length, Written, nil ) then
        begin
          DebugBreak;
          Exit;
        end;
      if Written <> Length then
        begin
          DebugBreak;
          Exit;
        end;
    end;
  //
  Result  := True;
end;

end.