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
   Delphi-Quellcode:

   myList:= threadList1.LockList;

   . Was spricht dagegen direkt auf ThreadList1 zuzugreifen
2. Wenn myList, kann ich die Anweisung
   Delphi-Quellcode:

   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

Delphi-Quellcode:

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 (

Delphi-Quellcode:

TInterlocked.

).

Zu deiner konkreten Frage:

Delphi-Quellcode:

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:

Zitat von norwegen60 (Beitrag 1359317)

Wozu dient aber die Zuweisung zu myList

Die eigentliche

Delphi-Quellcode:

TThreadList

ist keine Liste, sondern verwaltet nur intern eine Solche. Konkret macht

Delphi-Quellcode:

LockList

folgendes:

Delphi-Quellcode:

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

und

Delphi-Quellcode:

UnlockList

dies:

Delphi-Quellcode:

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

Zitat:

Zitat von norwegen60 (Beitrag 1359317)

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

Delphi-Quellcode:

TThreadList

; sprich: Du operierst praktisch auf einer normalen Liste, welche entsprechend nicht Thread-safe ist.

Zum Verständnis:
Sobald

Delphi-Quellcode:

TMonitor.Enter

von einem deiner beiden Threads aufgerufen wurde, erhält dieser Thread exklusiven Zugriff, bis er selbst

Delphi-Quellcode:

TMonitor.Exit

erreicht. Hat ein Thread den exklusiven Zugriff, dann hält der zweite Thread in dem Moment, in dem er auch

Delphi-Quellcode:

TMonitor.Enter

erreicht (vereinfacht ausgedrückt) einfach an - und zwar so lange, bis der erste Thread den exklusiven Zugriff durch den Aufruf von

Delphi-Quellcode:

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:

Zitat von norwegen60 (Beitrag 1359317)

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:

Zitat von norwegen60 (Beitrag 1359317)

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

Delphi-Quellcode:

TInterlocked.Read

,

Delphi-Quellcode:

TInterlocked.Exchange

,

Delphi-Quellcode:

TInterlocked.CompareExchange

,

Delphi-Quellcode:

TInterlocked.Increment

und

Delphi-Quellcode:

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

Delphi-Quellcode:

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

Delphi-Quellcode:

my1List.Free

.
Wenn ich allerdings

Delphi-Quellcode:

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:

Zitat von HolgerX (Beitrag 1359321)

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

Delphi-Quellcode:

TQueue<>

nehmen und den Zugriff manuell mit einem

Delphi-Quellcode:

TMonitor

btw. einer

Delphi-Quellcode:

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.

Delphi-Quellcode:

TMonitor

implementiert sogar schon eine Form des Spin-Lockings, washalb man nichtmal manuell mit

Delphi-Quellcode:

TInterlocked

arbeiten müsste.

Statt

Delphi-Quellcode:

Sleep(1)

geht btw. auch

Delphi-Quellcode:

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:

Zitat von norwegen60 (Beitrag 1359327)

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.