[gelöst] Designproblem Threads

[Michl]

Hallo wertes Forum,

angeregt durch diesen Beitrag:

- Der GUI Thread darf natürlich nicht warten, sondern muss ein Callback-Event ausführen
- Andere Threads dürfen auf die Events warten.

"Message" an den GUI-Thread kannst Du mit QueueAsyncCall schicken.

Ein Worker-Thread kann mit TEvent.Waitfor auf eine "Message" warten und sie dann bearbeiten wenn ein anderer Thread das Event auslöst.l

will ich ein bestehendes Projekt umbauen und bin nach verschiedenen Versuchen immer noch nicht schlauer:

Ich habe ein Standalone-Programm, was lange Zeit eigenständig läuft. Zum mitloggen von Fehlern oder falls ich mal wissen, will, was es gerade macht, habe ich ein "Status-Programm" laufen (ähnlich WriteLn oder DebugLn). Es soll threadsicher sein und alle Ereignisse chronologisch schreiben.

Z.Zt. mache ich folgendes (alle Threads inkl. Mainthread greifen auf die Klasse TStatus mit Methode StatLn zu):

Code: Alles auswählen

procedure TStatus.StatLn(const aMethName, aStr: String);    //Sendet einen Status
begin
  EnterCriticalsection(StatusThread.ACriticalSection);
  StatusThread.AktText:='['+IntToStr(FLastView)+']['+aMethName+'] '+aStr;
  LeaveCriticalsection(StatusThread.ACriticalSection);
  RtlEventSetEvent(StatusThread.StartWrite);
  RtlEventWaitFor(StatusThread.EndWrite);
end; 

im Thread, der dann den String (AktText) verarbeitet:

Code: Alles auswählen

procedure TStatusThread.Schreib;
var
  s: String;
begin
  EnterCriticalsection(ACriticalSection);
  s:=AktText;
  LeaveCriticalsection(ACriticalSection); 
  RtlEventSetEvent(EndWrite);
  ... //hier wird mit SimpleIPC der Text an ein externes Log-Programm übergeben 
end;
 
procedure TStatusThread.Execute;
begin
  while not Terminated do begin
    RtlEventWaitFor(StartWrite);
    Schreib;
  end;
end; 

Ich habe mal einen HärteFall getestet und mit 400Threads und dem Mainthread hintereinanderweg darauf zugriffen und es läuft einwandfrei. Allerdings erfüllt es die Anforderung "Der GUI Thread darf natürlich nicht warten" nicht (auch wenn die Unterbrechung genau eine Stringkopierung lang dauert).

Mit Criticalsection, ohne ein Waitfor läuft das Programm in etwa 4 mal so schnell, dafür werden nur ca. 10% aller Meldungen mitgeschrieben und diese nicht chronologisch. Ich habe auch überlegt eine "SammelListe" der Meldungen zu machen. Das führt mMn aber zu höheren String-Kopierungen, die dann ebenfalls Zeit fressen.

Wie könnte ein sauberer Aufbau des Programms aussehen?

[Socke]

Ich habe mal einen HärteFall getestet und mit 400Threads und dem Mainthread hintereinanderweg darauf zugriffen und es läuft einwandfrei. Allerdings erfüllt es die Anforderung "Der GUI Thread darf natürlich nicht warten" nicht (auch wenn die Unterbrechung genau eine Stringkopierung lang dauert).

Bei 400 Threads kann es natürlich sein, dass ein anderer Thread gerade in der CriticalSection ist und der Mainthread diese noch nicht betreten darf. In dieser Zeit bleibt er dann stehen.

Abmildern lässt sich das mit einem Application.QueueAsyncCall(); dann wird der Status außerhalb der echten GUI-Operationen abgearbeitet. Andernfalls schaltest du einen eigenen Puffer-Thread dazwischen, der den Status vom Mainthread kurzfristig annehmen kann und bei Gelegenheit weitergibt.

[Michl]

Die 400 Threads sind nur mal zum Test gewesen. So konnte ich über 50.000 Meldungen / Sekunde schreiben, in dem eigentlichen Programm hoffe ich auf weniger als 1 Meldung/Sekunde (wobei ich meine Statusklasse als Package installiert habe und auch für andere Tests nutze).

Generell will ich eigentlich nicht unter dem Aufruf MainThread -> StatLn(...) oder IrgendeinThread -> StatLn(...) unterscheiden. D.h. würde ich einen Puffer-Thread für den MainThread verwenden wollen, würde ich für alle Threads jeweils einen erstellen (das habe ich noch gar nicht getestet, wie das von der Performance aussieht). Genauso würde ich Application.QueueAsyncCall() für den Mainthread und alle anderen Threads nutzen. Weiss gar nicht, ob man das kann/darf.

Ok, werde nochmals umbauen und schauen, in welche Richtung es geht. Danke für die schnelle Antwort und Anregung.

[mschnell]

Abmildern lässt sich das mit einem Application.QueueAsyncCall(); dann wird der Status außerhalb der echten GUI-Operationen abgearbeitet.

Ich weiß nicht ob das so richtig bztw verständlich ausgedrückt ist.
Wenn in einem Worker-Thread Application.QueueAsyncCall() aufgerufen wird macht das einen Eintrag in der Event-Queue des Main (GUI) Threads. Dieser Wird abgearbeitet, wenn der GUI-Thread alle vorher eingestellten Einträge (ausgelöst durch Tastatur, Maus, Timer, ...) abgearbeitet hat. Es kann also sehr lange dauern, bis die Aktion tatsächlich gestartet wird, aber der Worker-Thread kann (im Gegensatz zu TThread.Synchronize()) während dessen weiterlaufen.

-Michael

[Michl]

Wenn in einem Worker-Thread Application.QueueAsyncCall() aufgerufen wird macht das einen Eintrag in der Event-Queue des Main (GUI) Threads. Dieser Wird abgearbeitet, wenn der GUI-Thread alle vorher eingestellten Einträge (ausgelöst durch Tastatur, Maus, Timer, ...) abgearbeitet hat.

Uha, das klingt nach genau dem, was ich nicht will. Ich will ja den Mainthread möglichst entlasten und nicht noch zusätzlich mit Statusanzeigen aus den Workerthreads belasten. Zumal das die chronologische Reihenfolge ja wieder durcheinander bringen würde.

Konnte das leider noch nicht testen und mich weiter reinlesen, werde das aber sobald ich Zeit habe natürlich nachholen!

[mschnell]

Ich will ja den Mainthread möglichst entlasten und nicht noch zusätzlich mit Statusanzeigen aus den Workerthreads belasten.

Das geht leider nicht. Da bei Lazarus (genau wie bei Delphi) jegliche GUI-Aktionen nur im Main-Thread zulässig sind, kannst Du "Statusanzeigen" nur dort machen.

-Michael

[Michl]

Das geht leider nicht. Da bei Lazarus (genau wie bei Delphi) jegliche GUI-Aktionen nur im Main-Thread zulässig sind, kannst Du "Statusanzeigen" nur dort machen.

Naja, das sind ja keine GUI-Aktionen (der StatusText, ein String, wird per SimpleIPC an ein externes Programm geschickt, was ihn weiterverarbeitet). Ich muss es eigentlich nur hinbekommen, chronologisch Strings in einen Thread zu bekommen, egal was der "Sender" (Mainthread oder WorkerThread) ist. Das schaffe ich z.Zt. nur dadurch, dass ich jeglichen Thread warten lasse, wenn ein String in den StatusThread kopiert wird.

[Socke]

Das geht leider nicht. Da bei Lazarus (genau wie bei Delphi) jegliche GUI-Aktionen nur im Main-Thread zulässig sind, kannst Du "Statusanzeigen" nur dort machen.

Naja, das sind ja keine GUI-Aktionen (der StatusText, ein String, wird per SimpleIPC an ein externes Programm geschickt, was ihn weiterverarbeitet). Ich muss es eigentlich nur hinbekommen, chronologisch Strings in einen Thread zu bekommen, egal was der "Sender" (Mainthread oder WorkerThread) ist. Das schaffe ich z.Zt. nur dadurch, dass ich jeglichen Thread warten lasse, wenn ein String in den StatusThread kopiert wird.

Die zeitliche Reihenfolge ist eine grundsätzliches Problem bei der Nebenläufigkeit (von Threads oder Prozessen). Du kannst im Status-Thread gar nicht bestimmen, ob die Statusmeldungen verschiedener Threads in der zeitlich korrekten Reihenfolge übergeben wurden.

[mschnell]

der StatusText, ein String, wird per SimpleIPC an ein externes Programm geschick

ich kenne "SimpleIPC" nicht, aber übertragungen per TCP/IP oder Serienschnittstelle kann man prima in einem Worker-Thread machen. Dann brauchst Du da kein QueuAsyncCall und u.U. auch auch keinen separaten Therad für das Schicken. Wenn "SimpleIPC" nur im Mainthread läuft ist es für diese Aufgabe wohl ungeeignet.

-Michael

[mschnell]

Du kannst im Status-Thread gar nicht bestimmen, ob die Statusmeldungen verschiedener Threads in der zeitlich korrekten Reihenfolge übergeben wurden.

Genau. Also die Mitteilungen in ein Memory-Fifo (z.B. TThreadList mit Pointern auf die Texte) schreiben und dann z.B. in einem separaten Thread (kann der Mainthread sein, muss aber nicht) der Reihe nach ausgeben.

-Michael

[Michl]

D.h. würde ich einen Puffer-Thread für den MainThread verwenden wollen, würde ich für alle Threads jeweils einen erstellen

Das habe ich probiert. Das Erstellen dauert zu lange. Es werden innerhalb kürzester Zeit so viele Threads erstellt, dass es zu einem Speicherüberlauf kommt von der Performance ganz zu schweigen.

Die zeitliche Reihenfolge ist eine grundsätzliches Problem bei der Nebenläufigkeit (von Threads oder Prozessen). Du kannst im Status-Thread gar nicht bestimmen, ob die Statusmeldungen verschiedener Threads in der zeitlich korrekten Reihenfolge übergeben wurden.

Das ist prinzipiell richtig und ich verstehe auch, dass parallel laufende Threads sich schwer kontrollieren lassen. Ich kann aber einen einfachen Test machen:

Code: Alles auswählen

procedure TStatus.StatLn(const aMethName, aStr: String);    
const
  Cnt: Integer = 0;
begin
  EnterCriticalsection(StatusThread.ACriticalSection);
  StatusThread.AktText:=IntToStr(Cnt)+'['+IntToStr(FLastView)+']['+aMethName+'] '+aStr;
  LeaveCriticalsection(StatusThread.ACriticalSection);
  RtlEventSetEvent(StatusThread.StartWrite);
  RtlEventWaitFor(StatusThread.EndWrite);
  inc(Cnt);
end;  

So wird schön der Zähler "Cnt" hochgezählt. Allerdings wartet jeder Thread darauf, dass er, nachdem er einen Status geschrieben hat, vom StatusThread abgenommen wurden ist. Ich konnte damit zu keiner Zeit feststellen, dass ein von einem anderen Thread gestarteter/resumter Thread sich in der Reihenfolge vorgedrängelt hat.

Schmeiße ich "RtlEventWaitFor(StatusThread.EndWrite);" raus und nutze Criticalsections werden Statusausgaben übersprungen (der eine Thread, der die Strings rausschickt schafft das nicht, alle abzuarbeiten). Das ist für den späteren Parser tödlich.

ich kenne "SimpleIPC" nicht, aber übertragungen per TCP/IP oder Serienschnittstelle kann man prima in einem Worker-Thread machen. Dann brauchst Du da kein QueuAsyncCall und u.U. auch auch keinen separaten Therad für das Schicken. Wenn "SimpleIPC" nur im Mainthread läuft ist es für diese Aufgabe wohl ungeeignet.

SimpleIPC läuft nicht im MainThread sondern im StatusThread! Die Übergabe der Strings per SimpleIPC ist bei mir sogar über Faktor 10 mal schneller, als ein einfaches WriteLn in der Console!

aber übertragungen per TCP/IP oder Serienschnittstelle kann man prima in einem Worker-Thread machen.

Genau das würde mich interessieren, wie man das realisiert, dass alle Meldungen gesendet werden, ohne dass der Mainthread geblockt wird, wenn dieser mal einen Status sendet.

Also die Mitteilungen in ein Memory-Fifo (z.B. TThreadList mit Pointern auf die Texte) schreiben und dann z.B. in einem separaten Thread (kann der Mainthread sein, muss aber nicht) der Reihe nach ausgeben.

Das werde ich jetzt mal probieren. Danke für den Tip.

[Michl]

Mitteilungen in ein Memory-Fifo

Perfekt, es funktioniert!

Und zwar so gut, dass ich innerhalb 1 Sekunde 1.471.493 Statusmeldungen setzen konnte. Zuvor waren es mit dem Blockieren des Mainthreads 52.606! Allerdings füllt sich die Liste im Speicher viel mehr mit Strings, als versendet werden können (bei einem Thread, der die Information verarbeitet und 100 Threads, die senden, ist das ja auch kein Wunder). Im Test bricht das Programm nach ein paar Sekunden zusammen. Füge ich in die Threads ein "Sleep(1);" ein, läuft das Programm durch. Später werden die Threads hauptsächlich arbeiten und weniger Fehler- oder Statusmeldungen senden

Mal überlegen, ob ich den Listenauf- und Listenabbau noch anders gestalten kann, so funktioniert das erstmal:

Code: Alles auswählen

type
 
  TListString = Record
    Str:  String;
    Next: Pointer
  end;
  PListString = ^TListString;
...
var
  FirstListString, LastListString: PListString;
 
...
 
procedure TStatus.StatLn(const aMethName, aStr: String);    //Sendet einen Status
var
  aListString: ^TListString;
begin
  New(aListString);
  aListString^.Str:=IntToStr(Cnt)+'['+IntToStr(FLastView)+']['+aMethName+'] '+aStr;
  aListString^.Next:=Nil;
 
  if FirstListString = Nil then
  begin
    EnterCriticalsection(StatusThread.ACriticalSection);
    FirstListString:=aListString;
    LeaveCriticalsection(StatusThread.ACriticalSection);
  end
  else
    if FirstListString^.Next = Nil then
    begin
      EnterCriticalsection(StatusThread.ACriticalSection);
      LastListString:=aListString;
      FirstListString^.Next:=LastListString;
      LeaveCriticalsection(StatusThread.ACriticalSection);
    end
    else
    begin
      EnterCriticalsection(StatusThread.ACriticalSection);
      LastListString^.Next:=aListString;
      LastListString:=aListString;
      LeaveCriticalsection(StatusThread.ACriticalSection);
    end;
 
  RtlEventSetEvent(StatusThread.StartWrite);
  inc(Cnt);
end;
 
...
 
procedure TStatusThread.Schreib;
var
  aListString: ^TListString;
begin
  while FirstListString <> Nil do begin
    aListString:=FirstListString;
    try
      FClient.Connect;
      FClient.SendStringMessage(aListString^.Str);
    except
      on e: Exception do begin
        Status.MaxView:=0;
        Exit;
      end;
    end;
 
    if aListString^.Next = Nil then
    begin
      EnterCriticalsection(ACriticalSection);
      FirstListString:=Nil;
      LeaveCriticalsection(ACriticalSection);
    end
    else
    begin
      EnterCriticalsection(ACriticalSection);
      FirstListString:=FirstListString^.Next;
      LeaveCriticalsection(ACriticalSection);
    end;
    Dispose(aListString);
  end;
end; 
 
procedure TStatusThread.Execute;
begin
  while not Terminated do begin
    RtlEventWaitFor(StartWrite);
    Schreib;
  end;
end;

Nochmal ein fettes Danke an mschnell und socke, daß Ihr mir auf die Sprünge geholfen habt!!!

[Michl]

Für die, die es evtl. interessiert?!

Ich habe jetzt die Liste auf 100.000 Einträge begrenzt (ab dann bremse ich wieder jeden Thread aus). Somit kommt es zu keinem Speicherüberlauf mehr. Im Normalfall wird auch nie ein Thread aufgrund einer Statussetzung warten müssen.

Der Test mit

Code: Alles auswählen

procedure TMyThread.Execute;
begin
  while not terminated do begin
    StatLn('Test',
      'Thread['+IntToStr(ThreadID)+'] sendet Zahl['+IntToStr(FZahl)+']');
    inc(FZahl);
  end;
end; 

ergibt ein dann ein LogFile, worin man wunderbar sieht, dass keine Meldung vergessen wird und schön chronologisch gespeichert wird:

Code: Alles auswählen

...
[0][Test] Thread[5160] sendet Zahl[53]
[0][Test] Thread[5160] sendet Zahl[54]
[0][Test] Thread[5160] sendet Zahl[55]
[0][Test] Thread[5160] sendet Zahl[56]
[0][Test] Thread[5160] sendet Zahl[57]
[0][Test] Thread[5160] sendet Zahl[58]
[0][Test] Thread[5160] sendet Zahl[59]
[0][Test] Thread[3368] sendet Zahl[0]     <- hier läuft der zweite Thread los
[0][Test] Thread[5160] sendet Zahl[60]
[0][Test] Thread[5160] sendet Zahl[61]
[0][Test] Thread[3368] sendet Zahl[1]
[0][Test] Thread[5160] sendet Zahl[62]
[0][Test] Thread[3368] sendet Zahl[2]
[0][Test] Thread[5160] sendet Zahl[63]
[0][Test] Thread[5160] sendet Zahl[64]
[0][Test] Thread[3368] sendet Zahl[3]
[0][Test] Thread[5160] sendet Zahl[65]
[0][Test] Thread[3368] sendet Zahl[4]
[0][Test] Thread[5160] sendet Zahl[66]
...