TThread mit Queue und Events

[AJ_Oldendorf]

Guten Morgen,
basierend auf diesem Thread (https://www.delphipraxis.net/217196-...angeblich.html), wurde mir ziemlich oft gesagt, warum machst du das so und so und nicht mit Queues und Events im Thread.
Jetzt mache ich einfach ein separaten Beitrag dazu hier auf und hoffe, jemand kann genau diese Aussage in Form von Quelltext mal zeigen, wie es funktioniert bzw. funktionen sollte.

Delphi-Quellcode:

TDataQueue = class
  private
    FQueue: TQueue<TDataRec>;
    FLock: TCriticalSection;
  public
    constructor Create;
    destructor Destroy; override;
    procedure Enqueue(const Data: TDataRec);
    function Dequeue: TDataRec;
  end;

constructor TDataQueue.Create;
begin
  FQueue := TQueue<TDataRec>.Create;
  FLock := TCriticalSection.Create;
end;

destructor TDataQueue.Destroy;
begin
  FQueue.Free;
  FLock.Free;
  inherited;
end;

procedure TDataQueue.Enqueue(const Data: TDataRec);
begin
  FLock.Acquire;
  try
    FQueue.Enqueue(Data);
  finally
    FLock.Release;
  end;
end;

function TDataQueue.Dequeue: TDataRec;
begin
  FLock.Acquire;
  try
    if FQueue.Count > 0 then
      Result := FQueue.Dequeue
    else
    begin
      Result.Clear;
    end;
  finally
    FLock.Release;
  end;
end;


TMyThread = class(TThread)
private
  FDataQueue: TDataQueue;
protected
  procedure Execute; override;
public
  constructor Create(ADataQueue: TDataQueue);
end;

constructor TMyThread.Create(ADataQueue: TDataQueue);
begin
  FDataQueue := ADataQueue;

  Inherited Create(False);
end;

procedure TMyThread.Execute;
var
  RecData : TDataRec;
begin
  while not Terminated do
  begin
    if Terminated then
      Exit;

    if Assigned(FDataQueue) then
    begin
      if FDataQueue.Count > 0 then
      begin
        RecData := FDataQueue.Dequeue;
        //Hier etwas mit den Daten/der Nachricht machen
      end
      else
      begin
        Sleep(1);
      end;
    end;
  end;
end;

Aufgabenstellung:

Ein Thread soll über die Queue Daten erhalten und damit irgendwas machen.
Gleichzeitig will ich den Threads mit Events von "außen" bestimmte Dinge mitteilen (z.B. die eingelesenen Nachrichten anders zu verarbeiten bzw gar nicht zu verarbeiten oder das sich der gesamte Thread selber terminieren soll). Wie ich es aktuell gemacht habe, ist in dem oben verlinkten Beitrag ausführlich beschrieben. Hier in diesem Beitrag soll es nur um die Alternative gehen. Danke

[Kas Ob.]

I see problems and useless CPU usage in your suggested code

1) No need for "if Terminated then" right after the "while not Terminated" it is redundant, if you want to check for termination to exit long job/process then ... well i will add few lines down here.
2) Sleep(1) is good, but not efficient, you can't combine Sleep(1) with high performance, and here i talking about if 1ms against 16ms might make difference or needed, so this loop should wait for signal instead of looping over what might be 1ms in best case scenario.
3) The code have clear and visible race condition, unless of course one consumer thread will be used, see this line "if FDataQueue.Count > 0 then", what will happen if two thread at this exact moment reached this point and had the Count = 1 then one will Dequeue, while the other might end up with what ? this should be protected from such case, there is few different approaches here.


Now to extra talk about (1), threads are one thing and jobs/tasks/process are different thing, mixing them is bad practice and bad design, think about this, if you are going after checking/using TThread.Terminated as signal to exit thread is thing, it is right and correct, but your tasks/jobs are and can be modified on their own and they agnostic to the Thread executing them, so mixing the tangling these two is somehow wrong, specially in this very design, when you are after multi consumer with multi producer, jobs/task must have their own exit/terminated/stop sginal, right where/with their parameters and data and their result if there is any, i hope that is clear.

Switch to Events, TEvent in Delphi is more than enough to be used, as you can manually signal all waiting thread or use the auto reset to wake only one and only one waiting thread on the event, meaning out of many consumer waiting to can signal one, and this will not take 1ms or 16ms, it is faster.

Hope that help !

[Kas Ob.]

Forgot to shout one thing !

If you are going to use the wait approach right, then make it wait with infinite timeout ! no 250ms or any thing like that, any use of limited time for wait is bad design and shows weak logic with hope to not fail instead will work as intended.

[AJ_Oldendorf]

Ok, das "if Terminated" ist überflüssig. Liegt daran, dass der QT zusammen kopiert wurde.
Bitte ignorieren.

Ok, aus Sleep(1), könnte man Sleep(16) machen, da der Aufruf sowieso nie schneller als 16ms (normalerweise) ist.

Das der Aufruf "if FDataQueue.Count > 0 then" von zwei Threads gleichzeitig bearbeitet wird, habe ich nie gesagt.
Die eine Liste ist genau für den einen Thread. Jeder hat sozusagen seine eigene Liste und nur der eine Thread selber macht das Dequeue.

Es wurden wieder einige Dinge gesagt aber auf die konkrete Fragestelle bei meinem Codebeispiel wurde nicht eingegangen (1x Dataqueue, 1x Thread, 2x Events genau an dem Codebeispiel darstellen).

[Edelfix]

Ein Beispiel (ungeprüft):

Hauptformular (Form1)

Delphi-Quellcode:

unit Unit1;

interface

uses
  Winapi.Windows, Winapi.Messages, System.SysUtils, System.Variants, System.Classes,
  Vcl.Graphics, Vcl.Controls, Vcl.Forms, Vcl.Dialogs, Vcl.StdCtrls,
  WorkerThreadUnit;

type
  TForm1 = class(TForm)
    btnStartThread: TButton;
    btnSendData: TButton;
    btnModeNormal: TButton;
    btnModeIgnore: TButton;
    btnModeAlternative: TButton;
    btnStopThread: TButton;
    MemoLog: TMemo;
    procedure FormCreate(Sender: TObject);
    procedure FormDestroy(Sender: TObject);
    procedure btnStartThreadClick(Sender: TObject);
    procedure btnSendDataClick(Sender: TObject);
    procedure btnModeNormalClick(Sender: TObject);
    procedure btnModeIgnoreClick(Sender: TObject);
    procedure btnModeAlternativeClick(Sender: TObject);
    procedure btnStopThreadClick(Sender: TObject);
  private
    FWorker: TMyWorkerThread;
    FDataCounter: Integer;
    procedure LogMessage(const Msg: string);
  public
  end;

var
  Form1: TForm1;

implementation

{$R *.dfm}

procedure TForm1.FormCreate(Sender: TObject);
begin
  FWorker := nil;
  FDataCounter := 0;
end;

procedure TForm1.FormDestroy(Sender: TObject);
begin
  if Assigned(FWorker) then
  begin
    FWorker.StopThread;
    FWorker.Terminate;
    FWorker.WaitFor;
    FreeAndNil(FWorker);
  end;
end;

procedure TForm1.btnStartThreadClick(Sender: TObject);
begin
  if not Assigned(FWorker) then
  begin
    FWorker := TMyWorkerThread.Create(
      procedure(const Msg: string)
      begin
        LogMessage(Msg);
      end);
    LogMessage('Thread gestartet.');
  end;
end;

procedure TForm1.btnSendDataClick(Sender: TObject);
begin
  if Assigned(FWorker) then
  begin
    Inc(FDataCounter);
    FWorker.EnqueueData('Nachricht #' + FDataCounter.ToString);
  end;
end;

procedure TForm1.btnModeNormalClick(Sender: TObject);
begin
  if Assigned(FWorker) then
  begin
    FWorker.SetProcessingMode(0);
    LogMessage('Modus: Normal');
  end;
end;

procedure TForm1.btnModeIgnoreClick(Sender: TObject);
begin
  if Assigned(FWorker) then
  begin
    FWorker.SetProcessingMode(1);
    LogMessage('Modus: Ignorieren');
  end;
end;

procedure TForm1.btnModeAlternativeClick(Sender: TObject);
begin
  if Assigned(FWorker) then
  begin
    FWorker.SetProcessingMode(2);
    LogMessage('Modus: Alternative Verarbeitung');
  end;
end;

procedure TForm1.btnStopThreadClick(Sender: TObject);
begin
  if Assigned(FWorker) then
  begin
    FWorker.StopThread;
    FWorker.Terminate;
    FWorker.WaitFor;
    FreeAndNil(FWorker);
    LogMessage('Thread gestoppt.');
  end;
end;

procedure TForm1.LogMessage(const Msg: string);
begin
  TThread.Queue(nil, procedure begin
    MemoLog.Lines.Add(Msg);
  end);
end;

end.

Worker-Thread (WorkerThreadUnit)

Delphi-Quellcode:

unit WorkerThreadUnit;

interface

uses
  System.Classes, System.SysUtils, System.SyncObjs, System.Generics.Collections;

type
  TMyWorkerThread = class(TThread)
  private
    FQueue: TThreadedQueue<string>;
    FTerminateEvent: TEvent;
    FProcessingMode: Integer;
    FModeLock: TCriticalSection;
    FLogProc: TProc<string>;
  protected
    procedure Execute; override;
  public
    constructor Create(LogProc: TProc<string>);
    destructor Destroy; override;

    procedure EnqueueData(const AData: string);
    procedure SetProcessingMode(AMode: Integer);
    procedure StopThread;
  end;

implementation

constructor TMyWorkerThread.Create(LogProc: TProc<string>);
begin
  inherited Create(False);
  FQueue := TThreadedQueue<string>.Create(100);
  FTerminateEvent := TEvent.Create(nil, True, False, '');
  FModeLock := TCriticalSection.Create;
  FLogProc := LogProc;
end;

destructor TMyWorkerThread.Destroy;
begin
  FQueue.Free;
  FTerminateEvent.Free;
  FModeLock.Free;
  inherited;
end;

procedure TMyWorkerThread.EnqueueData(const AData: string);
begin
  FQueue.Enqueue(AData);
end;

procedure TMyWorkerThread.SetProcessingMode(AMode: Integer);
begin
  FModeLock.Acquire;
  try
    FProcessingMode := AMode;
  finally
    FModeLock.Release;
  end;
end;

procedure TMyWorkerThread.StopThread;
begin
  FTerminateEvent.SetEvent;
end;

procedure TMyWorkerThread.Execute;
var
  Data: string;
  Mode: Integer;
begin
  while not Terminated do
  begin
    if FTerminateEvent.WaitFor(0) = wrSignaled then
      Break;

    if FQueue.Dequeue(Data, 100) = TWaitResult.wrSignaled then
    begin
      FModeLock.Acquire;
      try
        Mode := FProcessingMode;
      finally
        FModeLock.Release;
      end;

      case Mode of
        0: FLogProc('Verarbeite: ' + Data);
        1: FLogProc('Ignoriert: ' + Data);
        2: FLogProc('Alternative Verarbeitung: ' + Data);
      end;
    end
    else
    begin
      Sleep(10);
    end;
  end;

  FLogProc('Thread beendet.');
end;

end.

Oder habe ich es falsch verstanden?

[Mavarik]

Oje...

Rule 1 - ein Thread hat kein Sleep!

Auch der Zugriff auf Count muss gelocked werden.

Sodo Code.

Thread:

Delphi-Quellcode:

While not Terminated do
  begin
    if Queue.Count = 0 then
      E_Event.WaitFor(INFINITE);

    if not terminated then
      begin
        // DeQueue  
      end;  
  end;

Queue:

Delphi-Quellcode:

Procedure Enqueue(aData : TData);
begin
  TMonitor.Enter;
  try
    fQueue.Add(aData);
  finally
    TMonitor.Leave;
  end;

  E_Event.SetEvent;
end;

Den E_Event des Threads wird and die Queue übergeben.

Nicht vergessen:

Delphi-Quellcode:

procedure TMyThread.Terminate;
begin
  inherited Terminate;

  E_Event.SetEvent;
end;

Hier nochmal den Link zu meiner CodeRage 2020 Session zum Thema Threads & Queues!

Mavarik :coder:

[AJ_Oldendorf]

Danke, beide Beispiele helfen um es besser zu verstehen und es evtl. zu implementieren.
@Mavarik: Dein Video werde ich mir mal angucken, kannte ich ehrlich gesagt nicht :pale:

[Rollo62]

Zitat:

Zitat von Mavarik (Beitrag 1548967)

Oje...
Rule 1 - ein Thread hat kein Sleep!

Hmmm, meinst Du damit, er SOLLTE besser kein Sleep einsetzen? .... oder er kann gar kein Sleep ausführen? :gruebel:

https://docwiki.embarcadero.com/Libr....TThread.Sleep

https://docwiki.embarcadero.com/RADS...ervice_Threads

Delphi-Quellcode:

//Add the following code to the .cpp file for the TSparkyThread Execute method (the thread function):

void __fastcall TSparkyThread::Execute() {
    while (!Terminated) {
        Beep();
        Sleep(500);
    }
}

[jaenicke]

Ein Thread funktioniert problemlos mit Sleep. Wenn es rein um eine Verzögerung geht, sehe ich auch keinen Grund, das mit Events oder ähnlichem umzusetzen. Das Ergebnis ist das gleiche, ob ich auf ein Event eine bestimmte Zeit bis zum Timeout warte oder dies mit Sleep direkt mache.

Ich persönlich finde den zusätzlichen Code für z.B. ein Event nicht gerade gut für die Lesbarkeit, denn das Event hat ja selbst dann gar keine Funktion.

Anders sieht es aus, wenn man nicht die Verzögerung an sich möchte (z.B. zur Reduktion der CPU-Last), sondern eigentlich auf etwas wartet, das man per Event signalisieren kann. Das ist aber nicht immer möglich. Dann macht das Event natürlich deutlich mehr Sinn, weil man dann gezielt warten und weitermachen kann.

[AJ_Oldendorf]

Danke Sebastian, würde ich auch so sehen wie du.
Das Codebeispiel oben ist ja von dir, dass will ich jetzt mal noch mit 2 unterschiedlichen Events ausstatten und dann damit mal testen

[Rollo62]

Zitat:

Zitat von jaenicke (Beitrag 1549027)

... sehe ich auch keinen Grund, das mit Events oder ähnlichem umzusetzen. Das Ergebnis ist das gleiche, ...

Vielleicht wird darauf abgezielt, dass Windows Sleep nicht unterbrechbar ist und generell die Events schon?
Deshalb könnte es mit Events insgesamt reaktiver werden und auch sicherer gegen Deadlocks, weil Events sich von aussen killen lassen.

Die Timer-Präzision sollte nicht den Unterschied machen, das ist meiner Meinung nach gleich.
Nur bei Cross-Plattform gibt es da generell Vorteile, aber dann auch bei Sleep und WaitFor gleichermaßen.

[Mavarik]

Zitat:

Zitat von Rollo62 (Beitrag 1549016)

Hmmm, meinst Du damit, er SOLLTE besser kein Sleep einsetzen? .... oder er kann gar kein Sleep ausführen? :gruebel:

Natürlich geht ein Sleep, aber ich nutze doch einen Thread um etwas parallel oder schneller zu machen.
Daher macht für mich ein Sleep in einem Thread nie einen Sinn.

Zitat:

Zitat von jaenicke (Beitrag 1549027)

Wenn es rein um eine Verzögerung geht, sehe ich auch keinen Grund, das mit Events oder ähnlichem umzusetzen. Das Ergebnis ist das gleiche, ob ich auf ein Event eine bestimmte Zeit bis zum Timeout warte oder dies mit Sleep direkt mache.

Ob das das gleiche ist möchte ich (ungetestet) bezweifeln. Aber in diesem Beispiel nutzt man ja einen TEvent um keinerlei CPU Zyklen zu verschwenden. Immer warten und pollen macht halt überhaupt keinen Sinn, wenn ich auch auf einen Event in Best Case: ~1–10 µs (Mikrosekunden) reagieren kann.

Zitat:

Zitat von jaenicke (Beitrag 1549027)

Ich persönlich finde den zusätzlichen Code für z.B. ein Event nicht gerade gut für die Lesbarkeit, denn das Event hat ja selbst dann gar keine Funktion.

Doch genau das was er soll... Warten bis etwas neues in der Queue ist. (Oder Terminate)

Mavarik :coder:

[jaenicke]

Zitat:

Zitat von Mavarik (Beitrag 1549058)

Doch genau das was er soll... Warten bis etwas neues in der Queue ist. (Oder Terminate)

Ich habe ja geschrieben, dass das für den Fall passt.

Aber wer soll das Event auslösen, wenn ich z.B. einen Server pollen möchte, ob er neue Daten hat? Aber eben mit etwas Wartezeit dazwischen.

Es gibt auch andere Konstellationen als dass ein Thread zur Beschleunigung durch Parallelisierung dienen soll oder auf konkret abfragbare Events wartet.

[Kas Ob.]

Zitat:

Zitat von jaenicke (Beitrag 1549059)

Zitat:

Zitat von Mavarik (Beitrag 1549058)

Doch genau das was er soll... Warten bis etwas neues in der Queue ist. (Oder Terminate)

Aber wer soll das Event auslösen, wenn ich z.B. einen Server pollen möchte, ob er neue Daten hat? Aber eben mit etwas Wartezeit dazwischen.

I don't recall a single Windows IO operation that doesn't support both blocking and non-blocking, synchronous or asynchronous, they all fall in one way or other under overlapped, and they all have the possibility to attach an event to automatically triggered, as for the same example like a server then either
1) block on "recv" and skip the event altogether or
2) block on "select" to get the state of the socket and have the exact moment of/for receiving data, or
3) use WSAEventSelect and WSAAsyncSelect , these are powerful to combine with other complex or non-complex operation, events here become central notification for one or more threads.
4) IOCP and don't even wait manually, get the operation and process the data in place within the thread knowing there is other threads allowed to process other client...
5) ..
6) ..
there is many approaches and that only for handling sockets on server or client side, same goes for files ReadFile and ReadFileEx, they have differences and can combined in many different ways, if you prefer events then overlapped with supplied event is the way to go, work like charm with com and legacy communication ports, allowing you to utilize the maximum throughput being writing files to disk or sending data to hardware wire or a driver ...

The overlapped design and operation are powerful and efficient, the core of overlapped mechanism is low level kernel triggered events.

Zitat:

Zitat von jaenicke (Beitrag 1549059)

Es gibt auch andere Konstellationen als dass ein Thread zur Beschleunigung durch Parallelisierung dienen soll oder auf konkret abfragbare Events wartet.

Of course there is, there is always such needed design where time interval is needed at the exact point, in other words, time is essential for non performed operation or non triggered event, may be a timer, may be we want to control the time out for a socket, short one and dont want to depend on socket provider timeout..etc ,
other than that i see polling as waste of time and CPU cycle as as Mavarik, for fast and lazy or limited usage then yes, poll then execute then sleep for short time, but with servers with many clients and many threads, this is a waste.