Problem: Mehrere Threads auf ein Dictionary

[Cubysoft]

Hallo,

ich habe derzeit ein Problem, was in meinem nun sehr umfangreichen Projekt aufgetreten ist.

Delphi-Quellcode:

procedure TTeItemInfo.AddSmallDataFromListAsync(list: TList<TeItemDB.TTeItemDBData>; lang: TTeLanguage);
begin
  FEs := TTask.Create(procedure()
  var
    i: Integer;
    s: String;
  begin
    for i := 0 to list.Count -1 do
    begin
      if smalldata.ContainsKey(list[i].id) = false then
      begin
        s := DownloadInfo(list[i].id, lang.LoadedLanguage);
        AddSmallInfo(s,list[i].id,lang);
      end;
    end;
    if Assigned(FOnUpdate) then FOnUpdate(self);
  end);
  FEs.Start;
end;

Generell ersteinmal eine Erklärung: Da es mir unmöglich ist, den gesamten Code zu posten, poste ich mal den, der in meinen Augen zu dem Problem führt. Im Prinzip habe ich ein Edit auf einem Formular liegen. Gibt der Benutzer dort einen Begriff ein, der in der Datenbank zu finden ist, wird die obengenannte Funktion aufgerufen und eine Liste mit allen gefundenen Ergebnissen an diese übergeben. Die Funktion startet dann einen neuen Thread und geht die Liste durch. Dabei läd sie zu jedem Eintrag einen String aus dem Internet herunter. Die Funktion "AddSmallInfo" bereitet diesen String dann auf und erzeugt schlussendlich einen Eintrag in dem TDictionary "smalldata".

Nun generell Informationen, damit ihr mein Können einschätzen könnt: Dies ist mein erstes Projekt, bei dem ich so intensiv mit Multithreading arbeite, deshalb habe ich auch keine Ahnung, wie der unten genannte Fehler zustande kommen kann.

So nun zum Problem:
Da die Funktion jedes mal aufgerufen wird, wenn im Edit (im Hauptthread) eine Änderung passiert und der entstandene Text in einer Datenbank vorhanden ist, kommt es zwangsläufig zu Überschneidungen. Ich bekomme bei übermäßigem Ändern des Textes in immer wieder den selben Text (zB 20x von "Beispiel" (in Datenbank) zu "Beispiel1" (nicht in Datenbank) und wieder zurück) dann folgenden Fehler:

Code:

---------------------------
Benachrichtigung über Debugger-Exception
---------------------------
Im Projekt tep.exe ist eine Exception der Klasse EListError mit der Meldung 'Duplikate nicht zulässig' aufgetreten.
---------------------------
Anhalten   Fortsetzen   Hilfe   
---------------------------

Der Fehler tritt auch manchmal komplett unerwartet auf. Die Quelle ist wohl genau die o.g. Funktion und das damit verbundene Dictionary. Das Problem ist hierbei, dass ich in keinem Falle den Hauptthread warten lassen kann, bis ein zuvor gestarteter Thread beendet ist, auch wenn das das Problem warscheinlich lösen würde, denn es könnte Folgendes auftreten:

Der Benutzer möchte "Beispiel12345" eintragen. In der Datenbank befinden sich aber Einträge zu "Beispiel", "Beispiel1", "Beispiel12", Beispiel12345". Dadurch würde der Benutzer viel zu lange brauchen, um den gewünschten Datensatz zu erhaten. Der Gedanke hinter dem Ganzen ist folgender:

Das Dictionary smalldata wird nur während dem Programmlauf gefüllt, dadurch muss man nicht jedes Mal die Informationen erneut herunterladen, obwohl man den selben Datensatz eintragen möchte. Deshalb wird zu Beginn genau diese Liste heruntergeladen und eingetragen.


Ich hoffe ihr könnt mir helfen Herr über diesen Fehler zu werden, den leider habe ich davon nicht genug Ahnung. Was mir auch aufgefallen ist, ist dass das Event "OnUpdate" meiner Klasse nicht immer so oft aufgerufen wird, wie die Asynchrone Funktion, also habert es wohl irgendwo..

[Bambini]

Das Suchen und hinzufügen in TDictionay muß bei Threads gegen parallele Zugriffe geschützt werden.
D.h. sollte das ContainsKey false zurück liefern und den Eintrag hinzugefügt werden,
müss man diesen Bereich in ein TCriticalSection Bereich durchgeführt werden.

Wenn zur gleichen Zeit eine weitere Abfrage kommt, wird diese diesen Eintrag
auch in das Dictinary einstellen wollen. Was dann zu den genannten Fehlermeldung führt.

Delphi-Quellcode:

var
  FLock : TCriticalSection;

procedure TTeItemInfo.AddSmallDataFromListAsync(list: TList<TeItemDB.TTeItemDBData>; lang: TTeLanguage);
begin
  FEs := TTask.Create(procedure()
  var
    i: Integer;
    s: String;
  begin
    if not Assigned(FLock) then
      FLock := TCriticalSection.Create

    for i := 0 to list.Count -1 do
    begin
      FLock.Enter;
      if smalldata.ContainsKey(list[i].id) = false then
      begin
        s := DownloadInfo(list[i].id, lang.LoadedLanguage);
        AddSmallInfo(s,list[i].id,lang);
      end;
      FLock.Leave;
    end;
    if Assigned(FOnUpdate) then FOnUpdate(self);
  end);
  FEs.Start;
end;

[HolgerX]

Damit nicht bei jeder Änderung im Edit die gesamte Suche erneut aufgerufen wird, warte doch, bis er mit der Eingabe fertig ist.

Sprich gib ihm Zeit und starte die Suche erst, wenn der User nichts mehr eingibt.

Hierfür könnte ein Timer verwendet werden, der bei jedem OnChange des Edits neu gestartet wird und erst, wenn ca. 300 ms (Timerdauer) kein weiteres OnChange erfolgt, dann beginne mit der Suche.

Somit kann der User erst seinen Suchstring eingeben, bevor überhaupt gearbeitet wird.

[Blup]

Bambini hat uns in seinem Beispiel leider gezeigt, wie man eine CriticalSection auf keinen Fall verwenden sollte.
1. Diese muss schon erstellt sein, bevor mehrere Threads mit den Daten arbeiten, sonst ist auch der schreibende Zugriff auf FLock unzulässig. Am besten im Constructor von TTeItemInfo.
2. Auch der Zugriff auf "list" muss geschützt werden, diese Variable ist nur eine Referenz auf ein Listenobjekt, das alle Threads verwenden.
3. Auch der Aufruf von DownloadInfo() liegt im geschützten Bereich, da dies sehr viel Zeit benötigt ist die Sperre fast permanent aktiv.
4. FOnUpdate wird nicht geschützt.

Ich versuch mich mal daran, hab aber das neue Delphi noch nicht:

Delphi-Quellcode:

procedure TTeItemInfo.AddSmallDataFromListAsync(list: TList<TeItemDB.TTeItemDBData>; lang: TTeLanguage);
var
  DownloadList: array of record
    id: string; // anderer Typ? 
    Value: string;
  end;
  i1, i2, id: Integer;
begin
  {Daten für den Task vorbereiten}
  i2 := 0;
  for i1 := 0 to list.Count -1 do
  begin
    id := list[i1].id;
    if not smalldata.ContainsKey(id) then
    begin
      SetLength(DownloadList, i2 + 1);
      DownloadList[i2].id := id;
      Inc(i2);
    end;
  end;
  {Task ausführen}
  if i2 > 0 then
  begin
    TTask.Run(
      procedure
      var
        i: Integer;
      begin
        for i := 0 to Length(DownloadList) - 1 do
          DownloadList[i].Value := DownloadInfo(DownloadList[i].id, lang.LoadedLanguage);
        {Ergebnis an den Hauptthread übergeben}
        TThread.Synchronize(nil,
          procedure
          var
            i: Integer;
          begin
            for i := 0 to Length(DownloadList) - 1 do
              if not smalldata.ContainsKey(DownloadList[i].id) then
                AddSmallInfo(DownloadList[i].Value, DownloadList[i].id, lang);
            if Assigned(FOnUpdate) then FOnUpdate(self);
          end;
      end;
  end;
end;

Das einzig zeitintensive Teil ist "DownloadInfo()", nicht weil die Datenmenge so groß ist, sondern wegen der Latenz zwischen Anfrage und Antwort.
Besteht die Möglichkeit dem Dienst im Internet mehrere Anfragen in einem Block zu übertragen und die Antworten als Block zurück zu erhalten?
Das würde die Netzlast senken und die Geschwindigkeit insgesamt erhöhen.

[Cubysoft]

Hallo,

ersteinmal danke für deine Antwort. Ich werde sie gleich testen und mich dann nochmal melden. Das mit dem Download muss leider so bleiben, da eine Block-Anfrage ausgeschlossen ist.

[Sir Rufo]

Warum solltest du den Task nicht abbrechen können? Du solltest den auf jeden Fall abbrechen und den gesamten Ablauf etwas anders aufbauen:

Delphi-Quellcode:

procedure TTeItemInfo.AddSmallDataFromListAsync(list: TList<TeItemDB.TTeItemDBData>; lang: TTeLanguage);
var
  LList : TArray<TeItemDB.TTeItemDBData>;
  LIdx : Integer;
begin
  // Wenn es einen Task gibt
  if Assigned( FEs ) then
    // dann brechen wir den mal ab
    FEs.Cancel;

  // Wir prüfen schon vorher, welche Items abgearbeitet werden müssen
  for LIdx := 0 to list.Count - 1 do
    if not smalldata.ContainsKey(list[LIdx].id) then
      LList := LList + [ list[LIdx] ];

  FEs := TTask.Create(procedure()
  var
    i: Integer;
    s: String;
  begin
    for i := Low( LList ) to High( LList ) do
    begin
      // Prüfen ob dieser Task abgebrochen wurde
      TTask.CurrentTask.CheckCanceled; // wirft eine Exception, wenn abgebrochen

      s := DownloadInfo(LList[i].id, lang.LoadedLanguage);
      
      TThread.Synchronize( nil, procedure
        begin
          // wir müssen nochmal prüfen, 
          // denn es könnten parallel Einträge hinzugekommen sein
          if not smalldata.ContainsKey(LList[i].id) then
            AddSmallInfo(s,list[i].id,lang);
        end );
    end;
    
    // Synchronisierter Zugriff
    TThread.Synchronize( nil, procedure 
      begin
        if Assigned(FOnUpdate) then FOnUpdate(self);
      end );
  end);
  FEs.Start;
end;

[Cubysoft]

Hallo,

also ich habe jetzt den Code von Blup etwas angepasst. Dieser läuft schonmal perfekt und tut, was er tun soll. Wichtig ist noch Folgendes, weswegen ich nicht weiß, wie ich Sir Rufos Code verwenden soll:

Das Event OnUpdate muss IMMER aufgerufen werden. Es sagt nämlich nicht aus, dass etwas neues hinzugekommen ist, sondern teilt meinem Fenster mit, dass es die Ansicht aktualisieren muss. Zu deinem Code Sir Rufos habe ich allerdings doch noch eine Frage:

1. Wie kann ich mein OnUpdate Event aufrufen, wenn der Thread abgebrochen wurde?
2. Wenn das CheckCanceled ne Exception wirft, bedeutet das, dass ich ne Fehlermeldung bekomme und abfangen muss?

[Sir Rufo]

Wie einsetzen?

Einfach meinen Code über deinen kopieren, fertig!

Wenn FOnUpdate immer aufgerufen werden soll muss, dann einfach so:

Delphi-Quellcode:

procedure TTeItemInfo.AddSmallDataFromListAsync(list: TList<TeItemDB.TTeItemDBData>; lang: TTeLanguage);
var
  LList : TArray<TeItemDB.TTeItemDBData>;
  LIdx : Integer;
begin
  // Wenn es einen Task gibt
  if Assigned( FEs ) then
    // dann brechen wir den mal ab
    FEs.Cancel;

  // Wir prüfen schon vorher, welche Items abgearbeitet werden müssen
  for LIdx := 0 to list.Count - 1 do
    if not smalldata.ContainsKey(list[LIdx].id) then
      LList := LList + [ list[LIdx] ];

  FEs := TTask.Create(procedure()
  var
    i: Integer;
    s: String;
  begin
    try

      for i := Low( LList ) to High( LList ) do
      begin
        // Prüfen ob dieser Task abgebrochen wurde
        TTask.CurrentTask.CheckCanceled; // wirft eine Exception, wenn abgebrochen

        s := DownloadInfo(LList[i].id, lang.LoadedLanguage);
      
        TThread.Synchronize( nil, procedure
          begin
            // wir müssen nochmal prüfen,
            // denn es könnten parallel Einträge hinzugekommen sein
            if not smalldata.ContainsKey(LList[i].id) then
              AddSmallInfo(s,list[i].id,lang);
          end );
      end;

    finally
      // Synchronisierter Zugriff
      TThread.Synchronize( nil, procedure
        begin
          if Assigned(FOnUpdate) then FOnUpdate(self);
        end );
    end;
  end);
  FEs.Start;
end;

Nein, die Exception brauchst du nicht fangen (nur wenn du möchtest) und es wird dir auch nicht um die Ohren gehauen, da die Exception im Thread aufschlägt und diese nicht an den Haupt-Thread weitergereicht wird (das müsste man dann selber regeln, wenn erwünscht).

[Cubysoft]

Okay, da dein Code das ganze warscheinlich schneller macht, wollte ich ihn ausprobieren. Leider bekomme ich gesagt:

Code:

[dcc32 Fehler] TeItemInfo.pas(153): E2003 Undeklarierter Bezeichner: 'CheckCancelled'

EDIT: Hat sich nur ein Schreibfehler eingeschlichen. Funktioniert perfekt. Vielen Dank!

[Sir Rufo]

Zitat:

Zitat von Cubysoft (Beitrag 1301073)

EDIT: Hat sich nur ein Schreibfehler eingeschlichen. Funktioniert perfekt. Vielen Dank!

Jo, den habe ich auch in meinem Code gerade korrigiert ;)