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