Schnittmenge von mehreren Mengen ermitteln

[Laser]

Moin,

Zitat von BUG:

Aber mit den fortlaufenden Zahlen könntest du den Aufwand vermutlich auf die Größe der kleinsten Datei drücken, mehr Informationen (oder vielleicht ein Beispiel) wären schön.

Zitat von BUG:

Ich würde noch gerne mehr über die fortlaufenden Nummern erfahren.
Meint "fortlaufend" Zahlen in der Form: n, n+1, n+2, n+3, ..., n+m?

der 64-Bit Schlüssel ist eindeutig. Er besteht aus einem 32 Bit-Zeitstempel im oberen Teil und einer 32-Bit laufenden Nr. im unteren Teil.

In den Dateien n1...n12 kommt der Schlüssel jeweils 0 bis 1 mal vor. In den Dateien sind die Schlüssel aufsteigend sortiert.

Fortlaufend wäre also nur der untere 32-Bit Teil. Es kann aber Lücken zwischen 2 Schlüsseln geben. Es sind zwar alle Schlüssel lückenlos vergeben, aber es sind nicht unbedingt alle in den Dateien n1..n12 enthalten.

[Namenloser]

Zitat von Furtbichler:

Zitat von BUG:

Ich weiß nicht, was die Hashtable in diesem Fall für Verbesserungen bringen soll.

Suchen geht sehr schnell, der Aufbau ist aber grottenlahm.

Geht. Meine (String-)Hashmap addet eine Million Einträge in ca. einer halben Sekunde. Gut, deine 500ms kann man damit natürlich nicht mehr toppen, aber die Frage ist ja, wie schnell das Programm wirklich sein muss . Kommt es auf ein paar Sekunden überhaupt an?

Wie auch immer, ich habe noch mal kurz etwas drüber nachgedacht und jetzt eine deutlich einfachere Lösung – manchmal sieht man den Wald vor lauter Bäumen nicht. Kann sein, dass Patti in seinem ersten Post das gleiche meinte, allerdings bin ich mir bei seinem Pseudocode nicht ganz sicher

Mein Algorithmus geht so:

Code:

Zunächst haben wir für jede Datei quasi einen "Stream" mit einem Index. 
Der Index ist zu Beginn für alle Streams 0, also am Anfang der Datei. 
Jeder Stream hat eine Methode um das aktuelle Element zurückliefern, und eine Methode um einen Datensatz weiterzurücken.

Solange kein Stream das Dateiende erreicht hat:
  Bestimme PivotElement = höchstes (größter Wert) der aktuellen Elemente der Streams
  Für jeden Stream:
    Wenn das aktuelle Element des Streams < PivotElement:
      Aktuellen Stream eins weiterrücken lassen
  Wenn alle Elemente gleich sind:
    Element ausgeben
    Streams eins weiterrücken lassen (reicht an sich auch, einfach irgendeinen weiterrücken zu lassen, der Rest zieht automatisch nach)

Ich habe es nur an drei präparierten kleinen Testdateien ausprobiert, aber danach scheint es zu funktionieren. Hoffentlich ist kein Denkfehler drin (um diese Uhrzeit und bei 3 Stunden Schlaf kann viel passieren). Furtbichler, veröffentliche doch mal dein Programm zur Generierung der Testdaten. Ich würde mein Programm gerne mal damit testen (auch auf Geschwindigkeit, obwohl ich es jetzt nicht sonderlich optimiert habe).

[edit]
Alternativ kann man natürlich auch immer die Schnittmenge aus zwei Listen bilden, wobei bei den späteren Durchläufen eine der Listen selbst wieder je eine Schnittmenge ist. Das entspricht wohl Furtbichlers Ansatz. Dieses Verfahren ist im Best-Case schneller, da man bei einer leeren Zwischen-Schnittmenge gleich abbrechen kann, hat aber den Nachteil, dass es nicht In-Place arbeitet.

Man könnte den Ansatz auch parallelisieren und immer die Schnittmengen zwischen mehreren Listenpaare gleichzeitig bilden, und dann das gleiche wieder mit den sich daraus ergebenden Listen usw.... allerdings wächst dabei natürlich der Speicherbedarf noch weiter an.

Zusätzlich wäre ein Hybridansatz denkbar, dass man z.B. wenn nur (noch) sehr wenig Elemente in einer Liste (bzw. Zwischen-Schnittmenge) sind, die Strategie wechselt, und eben doch eine binäre Suche durchführt, da es ineffizient wäre, 5 Millionen Datensätze einzulesen, nur um zu prüfen, ob ein oder zwei Datensätze existieren.
[/edit]

[Horst_]

Hallo,

im Prinzip bleibt es bei Patti's Vorschlag aus PostNr 2
Aber in 500 ms aus 480 Mbyte die Schnittmenge zu bilden heißt nur, dass alle Dateien im Cache waren.

Gruß Horst

[Furtbichler]

Zitat von Laser:

@Furtbichler: Wie hast Du bei Deinem Test sichergestellt, dass Du die Dateien nicht aus dem Puffer des Betriebssystems liest?

Gar nicht. Aber andere Verfahren lesen ja auch aus dem Cache, sodaß es eigentlich egal ist.

Ich habe zudem eine Abbruchbedingung: Wenn die Schnittmenge leer ist, werden keine weiteren Dateien mehr angefasst, wozu auch. Das verfälscht natürlich das Ergebnis.

Also: Durchscannen aller 5 Mio Werte einer Datei geht ohne Optimierung im Worstcase in 140ms, wenn nämlich alle Dateien die gleichen 5 Mio Zahlen enthalten.

Zitat von Laser:

Memory Mapped Files scheint mir auch eher ungünstig.

Beim Einlesen habe ich eben mit Mapped Files getestet: Das scheint doppelt so schnell zu sein, wie ein TMemorystream.LoadFromFile, zumindest, wenn die Daten im RAM vorliegen.

Das Mapped File habe ich von hier: http://landman-code.blogspot.com/200...ce-i-last.html

Zitat von Horst_:

im Prinzip bleibt es bei Patti's Vorschlag aus PostNr 2: Aber in 500 ms aus 480 Mbyte die Schnittmenge zu bilden heißt nur, dass alle Dateien im Cache waren.

Korrekt. und nochmal korrekt.

Das, zusammen mit memory mapped files, dürfte die ideale Lösung sein bzw. die, gegen die eine Alternative bestehen müsste.

[Amateurprofi]

Hallo Furtbichler,
mich hat das auch interessiert und ich habe das mit binärer Suche versucht – war aber, die Performance betreffend, ein Flop.
Also hab ich mir mal deine Lösung angeschaut.
Sehr interessanter Ansatz, leider aber fehlerhaft.

for I := 0 to High(data) - 1 do begin Das " – 1 " gehört da m.E. nicht hin. Es verursacht, dass das letzte Element von data nicht in Intersect übernommen wrid.
Ich nehme an da stand ursprünglich Length(data) – 1


while (j < n) and (Intersect[j] < data[i]) do inc(j); Wenn das letzte Element in data größer ist als das letzte Element in Intersect dann wird j=n=Length(Intersect) und bei
if data[i] = Intersect[j] then begin wird auf Intersect[Length(Intersect)] zugegriffen, was einen Laufzeitfehler verursachen müßte.


Du schriebst 140 ms bei 12 Files je 5 Mio Werten, wobei die Files bereits im RAM liegen und alle Files die gleichen Daten enthalten.

Ich hab mal deine Prozedur auf meinem Rechner laufen lassen.
Das "Alle Files identisch und bereits im RAM" habe ich so realisiert, dass das Array "data", das bei dir lokal definiert ist, als Parameter mitgegeben wird.

Bei mir dauert das ganze bei unten stehendem Ablauf 300 ms.
Hab ich da vielleicht irgendwas falsch verstanden? Oder hab ich nur 'nen lahmen Rechner? Auf was für einer Maschine erreichst du 140 ms?

Delphi-Quellcode:

procedure IntersectFileWithHashmap(var Intersect,Data:TSampleArray);
var
   newIntersect{, data}: TSampleArray;
   n, i, j, k: Integer;
begin
   n := Length(Intersect);
   if n = 0 then exit;
   //ReadSamples(aFilename, data);
   j := 0;
   k := 0;
   SetLength(newIntersect, n);
   for I := 0 to High(data) {- 1} do begin
      while (j < n) and (Intersect[j] < data[i]) do inc(j);
      if data[i] = Intersect[j] then begin
         newIntersect[k] := data[i];
         inc(k);
      end;
   end;
   setLength(newIntersect, k);
   Intersect := newIntersect;
end;

procedure TMain.Button1Click(Sender: TObject);
const count=5000000;
var intersect,data:TSampleArray; i:integer; t:cardinal;
begin
   SetLength(data,count);
   for i:=0 to High(data) do data[i]:=i+1;
   intersect:=Copy(data);
   t:=GetTickCount;
   for i:=1 to 11 do IntersectFileWithHashmap(intersect,data);
   t:=GetTickCount-t;
   ShowMessage('Anzahl='+IntToStr(Length(intersect))+#13+
               'Zeit='+IntToStr(t)+' ms');
end;

Du wolltest gern bessere Lösungen sehen.

Ich habe da einfach mal deine Prozedur genommen und an ein paar Stellen etwas entfernt.

Du erstellst ein Array newIntersect, schreibst die gefundenen Werte hinein und stellst am Schluss newIntersect in Intersect.
Das ist überflüssig.
Anstatt kann man die gefundenen Werte direkt in das Array Intersect stellen.

Mit diesen Änderungen braucht das Ganze (auf meinem Rechner) nur noch 250 ms, also eine Verbesserung um ca 15 %.
Auf deinem Rechner müsste das dann 140*250/300 = 117 ms brauchen.

Delphi-Quellcode:

procedure xIntersectFileWithHashmap(var Intersect,Data: TSampleArray);
var n, i, j, k: Integer;
begin
   n := Length(Intersect);
   if n = 0 then exit;
   j := 0;
   k := 0;
   for I := 0 to High(data) do begin
     while (j < n) and (Intersect[j] < data[i]) do inc(j);
     if (j < n ) and (data[i] = Intersect[j]) then begin
       Intersect[k] := data[i];
       inc(k);
     end;
   end;
   setLength(Intersect, k);
end;

procedure TMain.Button2Click(Sender: TObject);
const count=5000000;
var intersect,data:TSampleArray; i:integer; t:cardinal;
begin
   SetLength(data,count);
   for i:=0 to High(data) do data[i]:=i+1;
   intersect:=Copy(data);
   t:=GetTickCount;
   for i:=1 to 11 do xIntersectFileWithHashmap(intersect,data);
   t:=GetTickCount-t;
   ShowMessage('Anzahl='+IntToStr(Length(intersect))+#13+
               'Zeit='+IntToStr(t)+' ms');
end;

Aber damit war ich auch nicht zufrieden, denn ich wollte ja auch auf meiner lahmen Krücke deine 140 ms toppen.

Die unten stehende Version braucht bei identischen Daten (auf meinem Rechner) nur noch 110 ms, was, auf deinen Rechner umgerechnet 140*110/300 = 51 ms heißen sollte.
Verglichen mit den 300 ms, die deine Prozedur auf meinem Rechner brauchte, ist das eine Verbesserung um ca. 65 %.

Jedoch möchte ich mich nicht mit fremden (deinen) Federn schmücken.
Auch meine Asm-Version baut im Prinzip auf deiner Lösung auf - und die ist einfach nur gut, auch wenn da ein paar Flüchtigkeitsfehler drin waren.
Jetzt hoffe ich nur, daß ich bei meiner Asm-Version nichts übersehen habe......

Delphi-Quellcode:

FUNCTION IntersectData(var Intersect,Data:TSampleArray; length:integer):integer;
asm
// IN  : EAX=@Intersect, EDX=@Data, ECX=Anzahl der Elemente der bisherigen Schnittmenge
// Out : Neue Anzahl der Elemente der Schnittmenge
               pushad  // Temp:=ESP; Push EAX,ECX,EDX,EBX,Temp,EBP,ESI,EDI
               mov   ebp,ecx          // n := Length(intersect)
               test  ebp,ebp
               je    @ReturnZero      // Schnittmenge ist leer
               mov   esi,[edx]        // @data[0]
               test  esi,esi
               je    @ReturnZero      // Data ist leer
               mov   edi,[eax]        // @Intersect[0]
               test  edi,edi          // nur zur Sicherheit
               je    @ReturnZero      // Intersect leer
               xor   ecx,ecx           // j := 0
               xor   edx,edx           // k := 0;
               xor   ebx,ebx           // for i := 0
               jmp   @CheckFor

@WhileLoop:    add   ecx,1             // inc(j)
               cmp   ecx,ebp           // While (j < n)
               jae   @SetRes
@ForLoop:      cmp   [edi+ecx*4],eax   // and Intersect[j] < data[i]
               jb    @WhileLoop        // do
               jne   @NextFor
               mov   [edi+edx*4],eax   // Intersect[k] := data[i];
@NoStore:      add   edx,1             // inc(k);
@NextFor:      add   ebx,1             // next i
@CheckFor:     cmp   ebx,[esi-4]       // i > High(data)
               jae   @SetRes           // ja, fertig
               mov   eax,[esi+ebx*4]   // data[i]
               jmp   @ForLoop          // Prüfung j<n nicht erforderlich

@ReturnZero:   xor   edx,edx           // k := 0
@SetRes:       mov   [esp+28],edx      // popad stellt [esp-28] in EAX
               popad
end;

procedure TMain.Button3Click(Sender: TObject);
const count=5000000;
var intersect,data:TSampleArray; i,len:integer; t:cardinal;
begin
   SetLength(data,count);
   for i:=0 to High(data) do data[i]:=i+1;
   intersect:=Copy(data);
   len:=Length(intersect);
   t:=GetTickCount;
   for i:=1 to 11 do len:=IntersectData(intersect,data,len);
   SetLength(intersect,len);
   t:=GetTickCount-t;
   ShowMessage('Anzahl='+IntToStr(Length(intersect))+#13+
               'Zeit='+IntToStr(t)+' ms');
end;

[Namenloser]

Also das in ASM umzusetzen halte ich nicht für sinnvoll. Da geht nur die Wartbarkeit flöten, und die Performance wird sich dadurch kein bisschen okay, kaum verbessern, da die Festplatte der Flaschenhals ist, nicht die CPU.

[Panthrax]

Zitat von Furtbichler:

Ich würde allen vorschlagen, weniger über die Theorie zu sinieren, als einfach ein Proof-Of-Concept zu präsentieren.

+1

Code:

11 Messungen:
function Intersect(var Left: TSampleArray; const Right: TSampleArray);
* mit Length(Left) = Length(Right) = N = 10000000 // 10 Mio.
* mit denselben Daten für jede Routine
* mit zufällig generierten Daten für jede Messung

              Messung  #19, Pascal  #37, Pascal  #35, Assembler
                    1      254          221              68
                    2      276          218              68
                    3      256          220              62
                    4      250          226              65
                    5      266          214              64
                    6      258          201              62
                    7      258          234              64
                    8      248          222              64
                    9      262          225              63
                   10      253          226              66
                   11      250          225              63

           Mittelwert      257,364      221,091          64,455
   Standardabweichung        8,201        8,432           2,115

[Amateurprofi]

Zitat von NamenLozer:

Also das in ASM umzusetzen halte ich nicht für sinnvoll. Da geht nur die Wartbarkeit flöten, und die Performance wird sich dadurch kein bisschen okay, kaum verbessern, da die Festplatte der Flaschenhals ist, nicht die CPU.

Hallo NamenLozer,
ich sehe das ganz anders.
Auch wenn mein Flug nach Neuseeland 36 Stunden dauert, laufe ich nicht zu Fuß zum Flughafen sondern versuche eine schnellere Lösung zu finden. Zudem ging es hier um den speziellen Fall, daß sich die Daten bereits im RAM befinden. Den Flaschenhals Festplatte gab es hier also nicht.
Auch Wartbarkeit ist hier kein Problem, denn es handelt sich um einen kurzen und überschaubaren Code. Der Teil, der die Arbeit macht, umfaßt gerade mal 13 Asm-Anweisungen.

[Furtbichler]

Hi Leute,

Endlich geht es mal los. Ich lese gerade 'Clean Coder' von Robert C. Martin, und da geht es darum, wie sich ein 'professioneller Programmierer' verhalten soll. U.a. empfiehlt er 'Programmierwettbewerbe' im Team zu veranstalten, z.B. wer schreibt den schnellsten (oder elegantesten, kompaktesten etc.) Code. Einerseits geht es um Verfahren, andererseits auch um Implementierung. Und wenn man gerade kein Team bei der Hand hat, soll man immer mal wieder aus Spaß in seiner Freizeit seinen Stil vervollkommnen. Dazu ist so ein Forum echt gut.

Zum, Thema;
Bezüglich der Performance: Nun ja. Ich teste mit 5 Mio Werten, Du mit 10 Mio.

@Amateurprofi: Danke für das Review. Die Fehler, die ich da gemacht habe, sind mir teilweise hinterher auch aufgefallen. (aber die -1 nicht, weil ich -na ja- zu blöd bin ). Genau das ist der Sinn von Codereview. Die Optimierungen sind natürlich sehr gut bemerkt.

In einer Produktivumgebung würde ich den Code vermutlich sauber ausprogrammieren. Wenn es um jede 10tel Sekunde geht, dagegen Assembler nehmen, aber den Code im Klartext (d.h.) Delphi oder Pseudocode darüber schreiben.

Wir sollten jedoch nicht nur den worst case betrachten, sondern in etwa mit den Zahlen hantieren, die in der Produktivumgebung zu erwarten sind. Dann wird die zu untersuchende Schnittmenge sehr schnell klein werden und es wäre noch eine Optimierung drin: In dem Augenblick, wo data[i] größer als das letzte Element von Intersect ist, kann die For-Schleife beendet werden.

Desweiteren ist der Name der Routine unglücklich: "xIntersectFileWithHashmap". Da ist erstens keine Hashmap und zweitens keine Datei. Entweder wird das eine Methode des TSampleArray oder man gibt dem Kind einen anständigen Namen, z.B. so:

Delphi-Quellcode:

procedure IntersectSampleArray (var Intersect,Data: TSampleArray);
var n, i, j, k: Integer;
begin
  n := Length(Intersect);
  if n = 0 then exit;
  j := 0;
  k := 0;
  for I := 0 to High(data) do begin
    while (j < n) and (Intersect[j] < data[i]) do inc(j);
    if j = n then  // Intersect-Array ist durch, alle weiteren Daten in data können ignoriert werden
      break
    else if data[i] = Intersect[j] then begin
      Intersect[k] := data[i];
      inc(k);
    end
  end;
  setLength(Intersect, k);
end;

Weiteres Optimierungspotential wäre:
1. 'SetLength' am Ende entfernen und stattdessen die Arraygröße im TSampleArray selbst mitzuspeichern.
2. In der inneren While-Schleife kann man auf das (j<n) verzichten, wenn das letzte Element von Intersect immer MaxInt64 ist und dieses Element selbst nicht in dem Wertebereich vorkommt.
3. Zumindest bei Strings ist die Verwendung von inkrementierenden Zeigern (PChar) schneller als der Array-Zugriff. Hier könnte man noch etwas herausholen. Dann ist das in etwa so schnell wie Assembler. Aber auch so unübersichtlich.

So, muss zur Arbeit

[Panthrax]

Code:

11 Messungen:
function Intersect(var Left: TSampleArray; const Right: TSampleArray);
* mit Length(Left) = Length(Right) = N = 10000000 // 10 Mio.
* mit denselben Daten für jede Routine
* mit zufällig generierten Daten für jede Messung

           Messung  #19, Pascal  #39, Pascal  #37, Pascal #35, Assembler
                 1      260          231          221             65
                 2      259          238          209             65
                 3      264          238          205             68
                 4      267          233          204             67
                 5      264          232          225             72
                 6      266          239          221             64
                 7      262          253          217             66
                 8      263          243          210             66
                 9      280          232          210             66
                10      265          230          212             65
                11      264          231          209             65

        Mittelwert      264,909      236,364      213,000         66,273
Standardabweichung        5,540        6,932        6,957          2,195