Multi Pattern Suche

[wido]

Hallo,

ich geb zu nur oberflächliches Wissen über das Thema zu haben, daher hoffe ich das mir evtl. jemand der mehr Kenntnisse hat evtl. weiterhelfen kann ... .

Folgende Aufgabenstellung:
Ich benötige eine Klasse um eine große Anzahl an Pattern mit unterschiedlicher Länge innerhalb eines Buffers zu suchen. Zuerst hab ich mich an Brute Force probiert - unglaublich langsam.

Meine nächste Überlegung war die Anzahl der Vergleiche zu minimieren. Entsprechend hab ich Pattern sortiert nach den ersten 2 Bytes in separate Arrays einsortiert:

Delphi-Quellcode:

type  TSignature      = record
       Signature      : array[0..63] of byte;
       Size           : byte;
       ID             : integer;
       end;
      TSignatureArray = array of TSignature;
      TSignatureGrid  = array[0..65536 - 1] of TSignatureArray;

Das hat die Vergleichsoperationen natürlich drastisch reduziert und das Ganze wurde um einiges schneller - aber nicht schnell genug :).

Ich hab mir daraufhin überlegt ob man den Boyer Moore Algo nicht so anpassen könnte, das er mit mehr als einem Pattern funktioniert.

Folgendes ist das Resultat:

Delphi-Quellcode:

unit MultiPatternBMSearch;

interface

uses Classes, SysUtils, Windows;

const
  MAXLONGWORD = 4294967295;
  MAXPATTERNS = 1048576;             
  MAXLEN = 32768;
  HASHSHIFT = 5;
  HASHMASK = 31;
  ASIZE: array [1..3] of longword = (32, 1024, 32768);

type
  PPattern = ^TPattern;
  TPattern = record
    ID: longword;                    
    Length: longword;                
    Ptr: PByteArray;                 
    Hash: longword;                  
    Prefix: longword;                
  end;
  TMultiPatternBMEventHandler = 
    procedure (Sender : TObject; dwID : DWORD; var bContinue : boolean) of object;
  TMultiPatternBM = class
  private
    BlockSize: byte;
    Buffer: PByteArray;
    BufferLength: longword;
    Index: longword;
    MinPatSize: longword;
    PrefixSize: byte;
    PATTERNS: TList;
    HASH: array [0..MAXLEN] of word;
    SHIFT: array [0..MAXLEN-1] of longword;
    FOnPatternFound: TMultiPatternBMEventHandler;
    function HashFunc(Ptr: PByteArray; Count: byte): longword;
    function Min(Item1, Item2: longword): longword;
    procedure FindMinimalPatternSize;
    procedure HashTableEvaluation;
    procedure ShiftTableEvaluation;
    procedure Preprocessing;
    procedure Search;
    procedure SetBlockAndPrefixSizes;
  public
    property OnPatternFound:
      TMultiPatternBMEventHandler read FOnPatternFound write FOnPatternFound;
    constructor Create(MultiPatternBMEventHandler: TMultiPatternBMEventHandler);
    destructor Destroy; override;
    procedure AddPattern(dwID : DWORD; Pattern : pointer; dwPatternLength : DWORD);
    procedure SearchBuffer(pBuffer : pointer; dwBufferLength : DWORD);
  end;

implementation


function ComparePatterns(Item1, Item2: pointer): integer;
begin
  if PPattern(Item1).Hash > PPattern(Item2).Hash then
    Result := 1
  else
    if PPattern(Item1).Hash = PPattern(Item2).Hash then
      Result := 0
    else
      Result := -1;
end;

procedure TMultiPatternBM.AddPattern(dwID : DWORD; Pattern : pointer;
  dwPatternLength : DWORD);
var
  Pat: PPattern;
begin
  if (dwPatternLength > 0) and (PATTERNS.Count < MAXPATTERNS) then begin
    New(Pat);
    Pat.ID := dwID;
    Pat.Length := dwPatternLength;
    Pat.Ptr := Pattern;
    PATTERNS.Add(Pat);
  end;
end;

constructor TMultiPatternBM.Create(MultiPatternBMEventHandler: TMultiPatternBMEventHandler);
begin
  PATTERNS := TList.Create;
  FOnPatternFound := MultiPatternBMEventHandler;
end;

destructor TMultiPatternBM.Destroy;
var
  i: longword;
begin
  for i := 0 to PATTERNS.Count-1 do
    Dispose(PATTERNS[i]);
  PATTERNS.Free;
end;

procedure TMultiPatternBM.FindMinimalPatternSize;
var
  i: longword;
  Pat: PPattern;
begin
  MinPatSize := MAXLONGWORD;
  for i := 0 to PATTERNS.Count-1 do begin
    Pat := PATTERNS[i];
    if Pat.Length < MinPatSize then
      MinPatSize := Pat.Length;
  end;
end;

function TMultiPatternBM.HashFunc(Ptr: PByteArray; Count: byte): longword;
var
  i: byte;
begin
  Result := Ptr[0] and HASHMASK;
  if Count > 1 then
    for i := 1 to Count-1 do
      Result := (Result shl HASHSHIFT)+(Ptr[i] and HASHMASK);
end;

procedure TMultiPatternBM.Preprocessing;
begin
  FindMinimalPatternSize;
  SetBlockAndPrefixSizes;
  ShiftTableEvaluation;
  PATTERNS.Sort(ComparePatterns);
  HashTableEvaluation;
  Index := BlockSize-1;
end;

procedure TMultiPatternBM.Search;
var
  bContinue: boolean;
  i, Hash1, Hash2, PrefixValue, ShiftValue, HashValue: longword;
  Pat: PPattern;
  PrefixPtr: PByteArray;
begin
  while Index < BufferLength do begin
    HashValue := HashFunc(@Buffer[Index-BlockSize+1], BlockSize);
    ShiftValue := SHIFT[HashValue];
    if ShiftValue = 0 then begin
      Hash1 := HASH[HashValue];
      Hash2 := HASH[HashValue+1];
      while Hash1 < Hash2 do begin
        Pat := PATTERNS[Hash1];
        PrefixPtr := @Buffer[Index-Pat.Length+1];
        PrefixValue := HashFunc(PrefixPtr, PrefixSize);
        if PrefixValue = Pat.Prefix then begin
          i := 0;
          while (Pat.Ptr[i] = PrefixPtr[i]) and (i < Pat.Length) do
            i := i+1;
          if i = Pat.Length then begin
            FOnPatternFound(Self, Pat.ID, bContinue);
            if not bContinue then exit;
          end;
        end;
        Hash1 := Hash1+1;
      end;
      ShiftValue := 1;
    end;
    Index := Index+ShiftValue;
  end;
end;

procedure TMultiPatternBM.SearchBuffer(pBuffer : pointer; dwBufferLength : DWORD);
begin
  if (dwBufferLength > 0) and (PATTERNS.Count > 0) then begin
    Buffer := pBuffer;
    BufferLength := dwBufferLength;
    Preprocessing;
    Search;
  end;
end;

procedure TMultiPatternBM.SetBlockAndPrefixSizes;
begin
  case MinPatSize of
    1: BlockSize := 1;
    2: BlockSize := 2;
    else
       BlockSize := 3;
  end;
  if BufferLength < BlockSize then
    BlockSize := BufferLength;
  if BlockSize = 1 then
    PrefixSize := 1
  else
    PrefixSize := 2;
end;

procedure TMultiPatternBM.ShiftTableEvaluation;
var                                  
  i, j, InitSHIFTValue, HashValue, BlockCount: longword;
  Pat: PPattern;
begin
  InitSHIFTValue := MinPatSize-BlockSize+1;
  for i := 0 to ASIZE[BlockSize]-1 do
    SHIFT[i] := InitSHIFTValue;
  for i := 0 to PATTERNS.Count-1 do begin
    Pat := PATTERNS[i];
    BlockCount := Pat.Length-BlockSize;
    for j := 0 to BlockCount do begin
      HashValue := HashFunc(@Pat.Ptr[j], BlockSize);
      SHIFT[HashValue] := Min(SHIFT[HashValue], BlockCount-j);
      if j = BlockCount then
        Pat.Hash := HashValue;
    end;
    Pat.Prefix := HashFunc(Pat.Ptr, PrefixSize);
  end;
end;

function TMultiPatternBM.Min(Item1, Item2: longword): longword;
begin                                
  if Item1 <= Item2 then
    Result := Item1
  else
    Result := Item2;
end;

procedure TMultiPatternBM.HashTableEvaluation;
var
  i, HashValue: longword;
begin
  for i := 0 to ASIZE[BlockSize]-1 do
    HASH[i] := MAXWORD;
  for i := 0 to PATTERNS.Count-1 do begin
    HashValue := PPattern(PATTERNS[i]).Hash;
    if HASH[HashValue] = MAXWORD then
      HASH[HashValue] := i;
  end;
  HashValue := PATTERNS.Count;       
  for i := ASIZE[BlockSize]-1 downto 0 do
    if HASH[i] = MAXWORD then HASH[i] := HashValue
    else HashValue := HASH[i];
end;

end.

Bei Bedarf kann ich das Ganze nochmal kommentieren, sollte mein Gedankengang nicht klar sein. Allerdings ist das Ganze immer noch nicht schnell genug - um ehrlich zu sein ist es relativ langsam. Es ist sogar langsamer als meine Idee mit den vorsortierten Pattern.

Hat einer evtl. ne Idee wie man das Ganze weiter optimieren kann? Hab ich evtl. etwas übersehen oder komplett falsch implementiert? Gibts evtl. bereits Implementierungen dieses Problems in Delphi die ich mir zu Gemüte führen könnte?

Vielen Dank im Vorraus :).

[stoxx]

Hi ..
darf ich interessenhalber mal fragen, was Du für eine Patternsuche machst ?
Deine Idee der Vorsortierung ist schonmal der richtige Ansatz.
Boyer Moore hilft Dir dabei aber nicht mehr.
Du solltest aber nicht nur nach den ersten 2 Byte sortieren, sondern die kompletten Suffixe. Suche dann über binäres Suchen.

Suffix Arrays bieten die Lösung für Dein Problem.
http://www.informatik.hu-berlin.de/w...LinearTime.ppt

wenn es noch schneller gehen soll, solltest Du Dir Suffix Trees angucken. Deren Platzbedarf ist aber recht groß, wenn nicht sogar enorm :-)

[wido]

Ich bin kein Student - daher versteh ich bei der Präsentation nur Bahnhof. Hast Du evtl. ne Beispiel Implementation der dort vorgestellten Algorithmen? Kann durchaus auch in C oder Assembler sein. Bin bei sowas eher der "Learning by Reading Code" Typ und weniger der Theoretiker.

[stoxx]

Hier eine Implementation vom Skew Algorithmus ( suffix array) , welcher das Erstellen (sortieren) der Suffixe in linearer Zeit bewältigt ( wurde erst 2003 "erfunden" )
Wenn Du es mal in Delphi umgesetzt haben solltest, dann kannst Dich ja mal melden ;-)


http://www.informatik.hu-berlin.de/~weese/files/skew/

Gruß stoxx