Arbeiten mit TThreadlist

[norwegen60]

Hallo zusammen,

ich möchte im 0.1s Takt anfallende Daten in die Datenbank schreiben ohne dass eventuelle Verzögerungen beim Schreiben der Daten zu einer Verzögerungen beim Erfassen der Daten führt. Angedacht ist ein Thread der unabhängig vom Hauptthread die Daten in die DB schreibt. U.U. auch ein Thread, der die Datn erfasst. Dabei dachte ich an die Abarbeitung der Daten mit TThreadlist. Ich habe mir das Embacadero-Beispiel angeschaut.

Delphi-Quellcode:

procedure TListThread.Execute;
var
  I: Integer;
  Temp: TControl;
  myList: TList;
begin
  while(True) do
  begin
    if (Terminated) then
    begin
      listthreadRunning:= FALSE;
      exit;
    end;
    Form1.ListBox1.Clear;
    myList:= threadList1.LockList;
    try
      for I := 0 to myList.Count-1 do
      begin
        Temp:= myList.Items[I];
        Form1.ListBox1.Items.Add(Temp.Name);
      end;
    finally
      myList:= threadList1.UnLockList;;
    end;
    Sleep(1000);
  end;
  listthreadRunning:= FALSE;
end;

Meine Fragen:

1. Wozu dient die Anweisung myList:= threadList1.LockList; . Was spricht dagegen direkt auf ThreadList1 zuzugreifen
2. Wenn myList, kann ich die Anweisung myList:= threadList1.LockList; nicht gleich nach der Zuweisung wieder aufrufen

Problem dieses Beispiels: So lange die for-Schleife läuft ist die Liste gesperrt und mein Ziel, die Daten zu erfassen, verfehlt da ja threadList1.LockList; gesetzt ist.
Wie kann ich das verhindern?

Danke für eure Unterstützung
Gerd

[Zacherl]

Such mal nach dem Stichword "Consumer-Producer Problem". Wenn dein Consumer (Speichern in Datenbank) nicht hinterherkommt, hast du im Prinzip nur die Möglichkeit die Frequenz des Producers (Datenerfassung) entsprechend zu verringern. Am besten dynamisch. Alternativ - falls die Datenerfassung nur kurze Zeit läuft - könntest du erstmal alle Werte erfassen und danach erst in die Datenbank schreiben (dafür müsstest du einen genügend Großen Buffer erstellen). Ich meine mich zu erinnern, dass bei dieserart Problem meistens ein RingBuffer verwendet wird, der tatsächlich nicht per CriticalSection abgesichert wird. Stattdessen erfolgt das Lesen/Schreiben der Werte und das Inc/Decrementieren der aktuellen Positionen über (die sehr viel performanteren) atomaren Operationen aus (TInterlocked. ).

Zu deiner konkreten Frage:
LockList sichert intern die Liste mit einer CriticalSection gegen konkurrierende Threadzugriffe ab. Beim direkten Zugriff auf die Liste passiert dies nicht und du wirst früher oder später Probleme bekommen, sobald mehrere Threads gleichzeitig Aktionen auf der Liste ausführen.

[norwegen60]

Hallo Zacherl,

es ist klar, dass ich nicht auf Dauer mehr Daten erzeugen kann als ich wegschreiben kann. In der Praxis ist es aber so, dass ich in der Regel schneller Schreiben kann als ich erzeuge. Nur hin und wieder kommt es zu Engpässen und dann will ich puffern.
Ausserdem dauert der Prozess maximal 45min und falls dann noch Daten hängig sind, würden die eben verzögert geschrieben.

Ich weiß schon dass Locklist die Liste gegen andere Threads sichert. Wozu dient aber die Zuweisung zu myList und was, wenn ich die Liste so abarbeiten würde:

Delphi-Quellcode:

    myList:= threadList1.LockList;
    myList:= threadList1.UnLockList;      // Gleich nach Zuweisung wieder freigeben
    try
      for I := 0 to myList.Count-1 do
      begin
        Temp:= myList.Items[I];
        Form1.ListBox1.Items.Add(Temp.Name);
      end;
    finally
    end;

Wie erreiche ich es dass die Liste nur kürzestmöglich gelockt ist damit sie wieder Daten empfangen kann

[Zacherl]

Zitat von norwegen60:

Wozu dient aber die Zuweisung zu myList

Die eigentliche TThreadList ist keine Liste, sondern verwaltet nur intern eine Solche. Konkret macht LockList folgendes:

Delphi-Quellcode:

function TThreadList.LockList: TList;
begin
  TMonitor.Enter(FLock);
  Result := FList;
end;

und UnlockList dies:

Delphi-Quellcode:

procedure TThreadList.UnlockList;
begin
  TMonitor.Exit(FLock);
end;

Zitat von norwegen60:

und was, wenn ich die Liste so abarbeiten würde:

Delphi-Quellcode:

    myList:= threadList1.LockList;
    myList:= threadList1.UnLockList;      // Gleich nach Zuweisung wieder freigeben

Naja, das wäre gleichbedeutend mit dem kompletten Weglassen von TThreadList ; sprich: Du operierst praktisch auf einer normalen Liste, welche entsprechend nicht Thread-safe ist.

Zum Verständnis:
Sobald TMonitor.Enter von einem deiner beiden Threads aufgerufen wurde, erhält dieser Thread exklusiven Zugriff, bis er selbst TMonitor.Exit erreicht. Hat ein Thread den exklusiven Zugriff, dann hält der zweite Thread in dem Moment, in dem er auch TMonitor.Enter erreicht (vereinfacht ausgedrückt) einfach an - und zwar so lange, bis der erste Thread den exklusiven Zugriff durch den Aufruf von TMonitor.Exit wieder abgibt. So ist garantiert, dass der abgesicherte Code-Block tatsächlich nicht konkurrierend abgearbeitet werden kann und keine Dateninkohärenzen entstehen.

Zitat von norwegen60:

In der Praxis ist es aber so, dass ich in der Regel schneller Schreiben kann als ich erzeuge. Nur hin und wieder kommt es zu Engpässen und dann will ich puffern.

Zitat von norwegen60:

Wie erreiche ich es dass die Liste nur kürzestmöglich gelockt ist damit sie wieder Daten empfangen kann

Das ist gut In diesem Falle würde ich dir zu so einem von mir beschriebenen RingBuffer raten. Mit TInterlocked.Read , TInterlocked.Exchange , TInterlocked.CompareExchange , TInterlocked.Increment und TInterlocked.Decrement stehen dir eine Reihe von atomaren Instruktionen zur Verfügung, welche alle nur eine Hand von CPU Zyklen verbrauchen (und demnach alle anderen Threads bei konkurrierenden Zugriffen auch nur für diese kurze Zeit blockieren).

Eine vielleicht einfachere Alternative wäre deine momentane Methode beizubehalten, allerdings darauf zu achten, dass du im Producer Thread die Liste tatsächlich NUR für den Augenblick lockst, in dem du das neue Element mit Add hinzufügst. Im Consumer könntest du die Liste locken, alle Daten in eine lokale (temporäre) zweite Liste kopieren, danach die gemeinsame Liste leeren und unlocken. Danach beginnst du auf der temporären Liste deinen Loop abzuarbeiten. Bei dieser Vorgehensweise sollten beide Threads nur kurze Zeit blockiert werden und der Producer muss vor allem nicht darauf warten, bis sämtliche Werte in die Datenbank geschrieben wurden.

[norwegen60]

Hallo Zacherl,

vielen Dank für die gute Erklärung. Auf die Idee, die Daten zu kopieren bin ich auch gekommen und habe es so umgesetzt.

Delphi-Quellcode:

type
  TValues = Class
    dtTime : TDateTime;
    rForce,
    rTemperature,
    rHumidity  : Real;
  End;

....

procedure TListThread.Execute;
var
  s:String;
  I, i1: Integer;
  Values: TValues;
  tmpList,
  my1List: TList;
  dtOld : TDateTime;
  rTdelta :REal;
begin
  while(True) do
  begin
    if (Terminated) then
    begin
      listthreadRunning:= FALSE;
      exit;
    end;

    my1List := TList.Create;
    // Noch nicht bearbeitete Daten in Temporäre Liste kopieren
    try
      tmpList := threadList1.LockList;
      for I := i1 to tmpList.Count-1 do
      begin
        Values:= TValues.Create;
        Values:= tmpList.Items[I];
        my1List.Add(Value);
      end;
      i1 := i;
    finally
      threadList1.UnlockList;
    end;

    // Kopierte Daten abhandeln (z.B. in DB speichern) und Liste wieder löschen
    try
      for I := 0 to my1List.Count-1 do
      begin
        Values:= my1List.Items[I];
        rTdelta := (Values.dtTime - dtOld)*(24*60*60);
        s:=formatDateTime('yyyy-mm-dd HH:MM:SS.zzz',Values.dtTime)+ format(': %.7d, %5.3f s', [i1, rTdelta]);
        Form1.ListBox1.Items.Add(s);
        dtOld := Temp.dtTime;
//          Values.Free;      // Speicher wieder freigeben    NOTWENDIG ??  Führt zu Fehler
      end;
      Form1.Listbox1.ItemIndex :=  Form1.Listbox1.Count-1;  // Auf letzte Zeile in Listbox springen
    finally
      while my1List.Count > 0 do
        my1List.Delete(0);                           // NOTWENDIG ??
      my1List.Free;
    end;

    Sleep(100);
  end;
  listthreadRunning:= FALSE;
end;

Da ich bisher nie mit Listen gearbeitet habe wäre ich froh wenn jemannd drauf schaut um mir zu sagen wo ich Problemstellen kreiert habe. Ich bin z.B. nicht sicher, was ich wieder freigeben muss. Viele Beispiele machen einfach my1List.Free .
Wenn ich allerdings Value.Free; aufrufe stimmt meine Liste nicht mehr

Vielen Dank
Gerd

[HolgerX]

Hmm..

Wieso umkopieren?
Während des Umkopieren ist die Liste gelockt, ein Hinzufügen würde nicht möglich sein...


Alternative:

Delphi-Quellcode:

procedure TListThread.Add(AItem : TValues);
begin
  tmpList := threadList1.LockList;   // Liste Locken
  try
    tmpList.Add(AItem);              // Item anhängen
  finally
    threadList1.UnlockList;          // Liste Unlock
  end;
end;



procedure TListThread.Execute;
var
  Temp: TValues ;
  tmpList : TList;
begin
  ListthreadRunning:= true;

  while(not Terminated) do             // Dauerschleife
  begin
    Temp := nil;

    tmpList := threadList1.LockList;   // Liste Locken
    try
      if tmpList.Count > 0 then        // Sind EIntrage vorhanden
      begin
        Temp:= tmpList.Items[0];       // Nur den ersten nehmen
        tmpList.Delete(0);             // den Ersten aus der Liste entfernen
      end;
    finally
      threadList1.UnlockList;          // Liste Unlock
    end;

    if Assigned(Temp) then
      DoItemWork(Temp);                // -> Verarbeiten (Datenbank), während Liste wieder frei zum Füllen

    Sleep(1);                          // anderem Thread Zeit geben

  end; // While..
  listthreadRunning:= false;
end;

Da Du ja eh in einer (While) Dauerschleife bist, kannst du ja nach dem FIFO-Prinzip immer nur den ersten Eintrag aus der Liste nehmen, wenn er vorhanden ist.

Nur während des Überprüfen, ob eine Eintrag vorhanden ist und der Entnahme ist die Liste gelockt (sehr kurz).

Die Verarbeitung des Eintrages erfolgt außerhalb des Locks und somit kann währenddessen weiter neue Einträge angehängt werden.

Neue Einträge können also ohne große Verzögerung angehängt werden, ungestört von der Datenbank-Aktion.

[norwegen60]

Hallo,

cih werde mir auch das mal genauer anschauen. Ich möchte allerdings die Ursprungsliste nicht löschen, da ich die Daten auch anzeigen oder ausgeben möchte. Und dann geht es mir auch ums Verständnis ob mein Ansatz korrekt ist.

Gerd

[Zacherl]

Zitat von HolgerX:

Wieso umkopieren?
Während des Umkopieren ist die Liste gelockt, ein Hinzufügen würde nicht möglich sein...

Dein Ansatz wird vermutlich tatsächlich noch performanter sein. Dann würde ich aber direkt eine TQueue<> nehmen und den Zugriff manuell mit einem TMonitor btw. einer TCriticalSection absichern. Tatsächlich bevorzuge ich nach wie vor den RingBuffer Ansatz, da dort die Memory Operationen eingespart werden können und man mit lediglich zwei Pointern/Indizes sehr einfach das FIFI Prinzip umsetzen kann. TMonitor implementiert sogar schon eine Form des Spin-Lockings, washalb man nichtmal manuell mit TInterlocked arbeiten müsste.

Statt Sleep(1) geht btw. auch TThread.Yield() . Das ist etwas intuitiver zu verstehen, falls mal jemand anderes den Code lesen sollte.

Edit:
Meine Überlegung bezüglich des Kopierens und Verarbeiten der Liste in einem Thread war im Grunde genommen die folgende Datenbank-Operation. Hier kann ich mir gut vorstellen, dass viele Inserts mit einer einzelnen Value deutlich langsamer sind, als das einmalige Inserten mehrerer Werte.

Zitat von norwegen60:

ich möchte allerdings die Ursprungsliste nicht löschen, da ich die Daten auch anzeigen oder ausgeben möchte. Und dann geht es mir auch ums Verständnis ob mein Ansatz korrekt ist.

Puh, also da wird dir bei einer 45 Minütigen Datensammlung aber ziemlich der Speicher volllaufen. Ich würde das Anzeigen der Daten direkt im Producer Thread erledigen, oder notfalls noch einen dritten Thread erstellen, der das übernimmt.

Dein Code sieht noch nicht korrekt aus. Ich schaue im Laufe des Tages nochmal genauer drüber, sobald ich Zeit habe.

[Zacherl]

Hier mal noch die Alternative mit einer Queue (erscheint mir mit Boardmitteln und ohne großen Aufwand die eleganteste Lösung zu sein):

Delphi-Quellcode:

type
  TDataStruct = record
  public
    Timestamp: Cardinal;
    Value: Integer;
  end;
var
  Queue: TQueue<TDataStruct>;
begin

  Queue := TQueue<TDataStruct>.Create;
  try
    // Producer
    TThread.CreateAnonymousThread(
      procedure
      var
        Item: TDataStruct;
        Counter: Integer;
      begin
        Counter := 0;
        while True do
        begin
          // Collect data
          Item.Timestamp := GetTickCount;
          Item.Value := Counter;
          Inc(Counter);
          // Enqueue
          System.TMonitor.Enter(Queue);
          try
            Queue.Enqueue(Item);
          finally
            System.TMonitor.Exit(Queue);
          end;
          Yield;
        end;
      end).Start;

    // Consumer
    TThread.CreateAnonymousThread(
      procedure
      var
        Item: TDataStruct;
      begin
        while True do
        begin
          // Dequeue
          System.TMonitor.Enter(Queue);
          try
            if (Queue.Count > 0) then
            begin
              Item := Queue.Dequeue;
            end;
          finally
            System.TMonitor.Exit(Queue);
          end;
          // Work with local copy
          // ...
          WriteLn(Item.Value);
        end;
      end).Start;

  finally
    Queue.Free;
  end;
end;

[norwegen60]

Oje, mir raucht der Kopf. Aber ich werde auch das anschauen.

Ich wüsste aber auch gerne, was in meinem Ansatz korrigiert werden müsste. Von der Performance konte ich nichts nachteiliges feststellen. Selbst bei 1ms Intervallen gab es keine Verzögerungen bei der Erfassung der Daten. Nur bei der Freigabe der Speicher bin ich nicht sicher.