Verständnisfrage zur Verwendung von TMessageManager im Thread

[TiGü]

Um ein bisschen mit den System.Messaging.TMessageManager zu spielen, habe ich mir eine kleine Beispielapplikation in XE7 geschrieben.
Es ist natürlich ein akademisches und an den Haaren herbeigezogendes Beispiel.

Das VCL-Formular versendet im Application.OnIdle eine selbstdefinierte Nachricht namens TIdleMessage.
Diese Nachricht wird in einen externen Thread empfangen und die Uhrzeit des Empfangs mittels System.SysUtils.Now in einem Container (TQueue<TDateTime>) gespeichert.

Im TThread.Execute wird nach 25 gesammelten TDateTimes, die Werte zurück an das Formular geschickt.
Hier sende ich per TThread.Queue, damit nichts blockiert, eine weitere Nachricht (System.Messaging.TMessage<TDateTime>).

Folgendes Verständnisproblem:
Ich beobachte im Empfangshandler (OnNewDateTimeMessage) des Formulars, das ich mehrfach die gleiche Instanz und damit den gleichen Wert von TDateTime erhalte (siehe Memo-Ausgabe).
Warum ist das so?
Liegt das am TMessageManager selbst oder gehe ich falsch mit TThread.Queue um?

Delphi-Quellcode:

unit Messagner.View;

interface

uses
  System.SysUtils, System.Classes, System.Types,
  System.Messaging, System.SyncObjs, System.Generics.Collections,
  Vcl.Controls, Vcl.Forms, Vcl.StdCtrls;

type
  TIdleMessage = class(System.Messaging.TMessage)
  end;

  TDateTimeMessage = class(System.Messaging.TMessage<TDateTime>)
  end;

  TMessageThread = class(TThread)
  private
    FLock : TCriticalSection;
    FQueue : TQueue<TDateTime>;

    procedure GetIdleMessage(const Sender : TObject; const M : TMessage);
    procedure DoSendMessage(const ADateTime : TDateTime);
    procedure DoInternalExecute;
  protected
    procedure Execute; override;
  public
    procedure BeforeDestruction; override;
    constructor Create;
  end;

  TForm1 = class(TForm)
    mmoLog : TMemo;
    procedure FormCreate(Sender : TObject);
    procedure FormDestroy(Sender : TObject);
  private
    FMessageThread : TMessageThread;
    FIdleMessage : TIdleMessage;
    FMessage : TDateTimeMessage;
    procedure ThreadTerminated(Sender : TObject);
    procedure OnNewDateTimeMessage(const Sender : TObject; const M : TMessage);
    procedure LogToMemo(const Text : string);
  public
    procedure DoIdle(Sender : TObject; var Done : Boolean);
  end;

var
  Form1 : TForm1;

implementation

{$R *.dfm}


procedure TForm1.LogToMemo(const Text : string);
begin
  mmoLog.Lines.Add(Text);
end;

procedure TForm1.OnNewDateTimeMessage(const Sender : TObject; const M : TMessage);
var
  LMessage : TDateTimeMessage;
begin
  LMessage := M as TDateTimeMessage;
  if FMessage <> LMessage then
  begin
    FMessage := LMessage;
    LogToMemo(sLineBreak + '- - - > ' + FormatDateTime('hh:mm:ss:zzz', FMessage.Value) + sLineBreak);
  end
  else
  begin
    LogToMemo('An der Stelle erhalte ich mehrmals die gleiche Instanz der gesendeten Nachricht.' + sLineBreak + 'Warum ist das so?')
  end;
end;

procedure TForm1.DoIdle(Sender : TObject; var Done : Boolean);
begin
  TMessageManager.DefaultManager.SendMessage(Self, FIdleMessage, False);
end;

procedure TForm1.FormCreate(Sender : TObject);
begin
  FIdleMessage := TIdleMessage.Create;
  Vcl.Forms.Application.OnIdle := DoIdle;
  FMessageThread := TMessageThread.Create;
  FMessageThread.OnTerminate := ThreadTerminated;
  TMessageManager.DefaultManager.SubscribeToMessage(TIdleMessage, FMessageThread.GetIdleMessage);
  TMessageManager.DefaultManager.SubscribeToMessage(TDateTimeMessage, OnNewDateTimeMessage);
end;

procedure TForm1.ThreadTerminated(Sender : TObject);
var
  LException : Exception;
begin
  TMessageManager.DefaultManager.Unsubscribe(TIdleMessage, FMessageThread.GetIdleMessage);
  if Sender is TThread then
  begin
    if TThread(Sender).FatalException is Exception then
    begin
      LException := Exception(TThread(Sender).FatalException);
      LogToMemo(LException.ToString + ' ' + LException.Message);
    end;
  end;
end;

procedure TForm1.FormDestroy(Sender : TObject);
begin
  FMessageThread.Free;
  FIdleMessage.Free;
end;

{ TMessageThread }

procedure TMessageThread.BeforeDestruction;
begin
  FQueue.Free;
  FLock.Free;
  inherited;
end;

constructor TMessageThread.Create;
begin
  inherited Create;
  NameThreadForDebugging('Message-Thread');
  FLock := TCriticalSection.Create;
  FQueue := TQueue<TDateTime>.Create;
end;

procedure TMessageThread.DoInternalExecute;
var
  LDateTime : TDateTime;
begin
  while not Terminated do
  begin
    FLock.Enter;
    try
      if FQueue.Count >= 25 then
      begin
        while FQueue.Count > 0 do
        begin
          if not Terminated then
          begin
            LDateTime := FQueue.Dequeue;

            Queue(
              procedure
              begin
                DoSendMessage(LDateTime);
              end);
          end;
        end;
      end;
    finally
      FLock.Leave;
    end;
  end;
end;

procedure TMessageThread.DoSendMessage(const ADateTime : TDateTime);
var
  LMessage : TDateTimeMessage;
begin
  LMessage := TDateTimeMessage.Create(ADateTime);
  TMessageManager.DefaultManager.SendMessage(Self, LMessage, True);
end;

procedure TMessageThread.Execute;
begin
  inherited;
  DoInternalExecute;
end;

procedure TMessageThread.GetIdleMessage(const Sender : TObject; const M : TMessage);
var
  LDateTime : TDateTime;
begin
  FLock.Enter;
  try
    LDateTime := System.SysUtils.Now;

    if FQueue.Count <> 0 then
    begin
      if FQueue.Peek <> LDateTime then
      begin
        FQueue.Enqueue(LDateTime);
      end;
    end
    else
    begin
      FQueue.Enqueue(LDateTime);
    end;
  finally
    FLock.Leave;
  end;
end;

end.

[Mavarik]

OK ich sehe auf Anhieb keinen Fehler, aber:

- Warum einen Thread für den Empfang?
- Dein Thread Läuft wie ein wild gewordener Affe immer im Kreis und Locked und Leaved die CS

Mavarik

Ungetestet... Auf die schnelle zusammengestrichen... Sollte so reichen...

Delphi-Quellcode:

unit Messagner.View;

interface

uses
   System.SysUtils, System.Classes, System.Types,
   System.Messaging, System.SyncObjs, System.Generics.Collections,
   Vcl.Controls, Vcl.Forms, Vcl.StdCtrls;

type
   TIdleMessage = class(System.Messaging.TMessage)
   end;

   TDateTimeMessage = class(System.Messaging.TMessage<TDateTime>)
   end;

   TForm1 = class(TForm)
     mmoLog : TMemo;
     procedure FormCreate(Sender : TObject);
     procedure FormDestroy(Sender : TObject);
   private
     FIdleID,
     FDateTimeID : Integer;
     FQueue : TQueue<TDateTime>;

     procedure OnNewDateTimeMessage(const Sender : TObject; const M : TMessage);
     procedure GetIdleMessage(const Sender : TObject; const M : TMessage);
     procedure DoSendMessage(const ADateTime : TDateTime);
     procedure LogToMemo(const Text : string);
   public
     procedure DoIdle(Sender : TObject; var Done : Boolean);
   end;

var
   Form1 : TForm1;

implementation

{$R *.dfm}


procedure TForm1.LogToMemo(const Text : string);
begin
   mmoLog.Lines.Add(Text);
end;

procedure TForm1.OnNewDateTimeMessage(const Sender : TObject; const M : TMessage);
var
   LMessage : TDateTimeMessage;
begin
   LMessage := M as TDateTimeMessage;
   LogToMemo(sLineBreak + '- - - > ' + FormatDateTime('hh:mm:ss:zzz', LMessage.Value) + sLineBreak);
end;

procedure TForm1.DoIdle(Sender : TObject; var Done : Boolean);
begin
   TMessageManager.DefaultManager.SendMessage(Self, TIdleMessage.Create, False);
end;

procedure TForm1.FormCreate(Sender : TObject);
begin
   Vcl.Forms.Application.OnIdle := DoIdle;
   FQueue := TQueue<TDateTime>.Create;
   FIdleID := TMessageManager.DefaultManager.SubscribeToMessage(TIdleMessage, GetIdleMessage);
   FDateTimeID := TMessageManager.DefaultManager.SubscribeToMessage(TDateTimeMessage, OnNewDateTimeMessage);
end;

procedure TForm1.FormDestroy(Sender : TObject);
begin
  TMessageManager.DefaultManager.Unsubscribe(TIdleMessage,FIdleID,true);
  TMessageManager.DefaultManager.Unsubscribe(TDateTimeMessage,FDateTimeID,true);
end;

procedure TForm1.DoSendMessage(const ADateTime : TDateTime);
var
   LMessage : TDateTimeMessage;
begin
   LMessage := TDateTimeMessage.Create(ADateTime);
   TMessageManager.DefaultManager.SendMessage(Self, LMessage, True);
end;

procedure TForm1.GetIdleMessage(const Sender : TObject; const M : TMessage);
var
   LDateTime : TDateTime;
begin
   LDateTime := System.SysUtils.Now;

   if FQueue.Count <> 0 then
   begin
     if FQueue.Peek <> LDateTime then
     begin
       FQueue.Enqueue(LDateTime);
     end;
   end
   else
   begin
     FQueue.Enqueue(LDateTime);
   end;
end;

end.

[Sir Rufo]

Um den Fehler mit den gleichen DateTime Werten zu lösen, sollte man sich die Frage stellen, was macht

Delphi-Quellcode:

TQueue.Peek

und danach sollte man sich die Frage stellen, was für ein Wert steht denn in so einem

Delphi-Quellcode:

TDateTime

.

Und wenn der aktuelle Wert ungleich dem Peek-Wert ist, trägst du das in die Queue ein ;)

Das ist mit dem Dauerlauf des Thread ein schönes Feuerwerk :D

BTW: Prüf doch mal auf MemLeaks, da hast du noch ein Feuerwerk (oh, doch nicht)

[TiGü]

Zitat:

Zitat von Sir Rufo (Beitrag 1310095)

Um den Fehler mit den gleichen DateTime Werten zu lösen, sollte man sich die Frage stellen, was macht

Delphi-Quellcode:

TQueue.Peek

"Guckt" nach dem obersten Element im Stapel, ohne es zu entfernen, im Gegensatz zu Dequeue.

Zitat:

Zitat von Sir Rufo (Beitrag 1310095)

und danach sollte man sich die Frage stellen, was für ein Wert steht denn in so einem

Delphi-Quellcode:

TDateTime

.

So an und für sich ein Double, daher ist die Prüfung auf Ungleichheit doch okay, oder?

Zitat:

Zitat von Sir Rufo (Beitrag 1310095)

Und wenn der aktuelle Wert ungleich dem Peek-Wert ist, trägst du das in die Queue ein ;)

So war das gedacht...habe ich hier schon einen Denkfehler?
Manchmal sehe ich den Wald vor lauter Bäumen nicht.

Zitat:

Zitat von Sir Rufo (Beitrag 1310095)

Das ist mit dem Dauerlauf des Thread ein schönes Feuerwerk :D

Na ja, das wäre ja nur ein Nebenaspekt in diesem Beispiel.
Ein liebevoll eingestreutes try-finally mit Sleep(100) oder meinetwegen auch ein Event kann dem Abhilfe schaffen, löst aber nicht das grundsätzliche Problem.

In

Delphi-Quellcode:

procedure TMessageThread.DoSendMessage(const ADateTime : TDateTime);

wird jedes Mal eine neue Instanz erzeugt, nur im Empfänger

Delphi-Quellcode:

procedure TForm1.OnNewDateTimeMessage(const Sender : TObject; const M : TMessage);

kommt es häufig vor, dass die gleiche Instanz empfangen wird.
Aber warum?

Zitat:

Zitat von Mavarik

Warum einen Thread für den Empfang?

Rein akademisches Interesse, sonst hätte ich ja auch im OnIdle direkt ins Memo schreiben können.
Mir ging es um die Threadübergreifende Kommunikation mithilfe des RTL-eigenen MessageManagers.

[Sir Rufo]

Du räumst ja die Queue, wenn dort mehr als 25 Einträge enthalten sind, ok.

Gehen wir also mal davon aus, dass der erste Wert in der Queue 29.07.2015 12:00:00.000 ist.
Jetzt kommen über die IdleMessage 24 exakt die gleichen Werte an und zwar jeweils 29.07.2015 12:00:00.001.

Was passiert? Richtig, alle Werte landen in der Queue, denn alle sind ungleich dem ersten Wert in der Queue.

War es das, was du haben wolltest?

[Rollo62]

Warum machst du denn das ?

Code:

  while not Terminated do
  begin
    FLock.Enter;

    try
      TueEtwas;

    finally
      FLock.Leave;
    end;
  end;

Im Thread Execute muss man doch keine Locks setzen, nur wenn etwas synchronisiert werden müsste.
Aber die Ganze Zeit ???

[TiGü]

Zitat:

Zitat von Sir Rufo (Beitrag 1310118)

War es das, was du haben wolltest?

Du meinst, dass mich der Blitz beim Scheißen treffen soll, weil ich Queue und Stack verwechselt habe? :shock:
Klar, ich will ja nach dem zuletzt hingefügten Element gucken und nicht was als allererstes hinzugefügt wurde.

Aber selbst wenn ich mein Programm (dieses Mal im Anhang) entsprechend umändere, wird DoSendMessage immer mit den gleichen TDateTime-Wert aufgerufen, obwohl laut Debugger die FItems vom Stack immer unterschiedlich sind (in der Nachkommastelle...).

Wende ich TThread.Queue falsch an? Oder muss ich die Methode DoSendMessage, die ich innerhalb der anonymen Methode aufrufe, umändern?

[TiGü]

Zitat:

Zitat von Rollo62 (Beitrag 1310129)

Warum machst du denn das ?

Code:

  while not Terminated do
  begin
    FLock.Enter;

    try
      TueEtwas;

    finally
      FLock.Leave;
    end;
  end;

Im Thread Execute muss man doch keine Locks setzen, nur wenn etwas synchronisiert werden müsste.
Aber die Ganze Zeit ???

Ich ging/gehe davon aus, dass während der Verarbeitung (im Execute) von außen neue Daten reingeschaufelt werden können (durch GetIdleMessage).
Ist mein Ansatz falsch?
In beiden Methoden wird der Datencontainer angefasst, Items hingefügt oder entfernt.

[TiGü]

Ich bin jetzt dazu übergegangen, mir ein Array zu übergeben.
Der Aufruf der anonymen Methode von TThread.Queue wird wahrscheinlich immer den letzten Wert für die lokale TDateTime-Variable genommen haben.
Daher die gleichen Werte in der Message.
Die gleichen Instanz-Pointer der Messages sind eher Zufall und liegen einfach nur auf der gleichen Speicheradresse.

Also war mein Problem wahrscheinlich eher ein Missverständis, wie und wann TThread.Queue die übergebenden Methoden aufruft und wie zu diesen Zeitpunkt die Werte außerhalb der anonymen Methode belegt sind.

Delphi-Quellcode:

procedure TMessageThread.DoInternalExecute;
var
  DateTimes : TArray<TDateTime>;
begin
  while not Terminated do
  begin
    try
      FLock.Enter;
      try
        if FStack.Count >= 25 then
        begin
          if not Terminated then
          begin
            DateTimes := FStack.ToArray;

            Queue(
              procedure
              begin
                DoSendMessage(DateTimes);
              end);

            FStack.Clear;
          end;
        end;
      finally
        FLock.Leave;
      end;
    finally
      Sleep(100);
    end;
  end;
end;

procedure TMessageThread.DoSendMessage(const DateTimes : TArray<TDateTime>);
var
  LMessage : TDateTimeMessage;
  LDateTime : TDateTime;
begin
  for LDateTime in DateTimes do
  begin
    LMessage := TDateTimeMessage.Create(LDateTime);
    TMessageManager.DefaultManager.SendMessage(Self, LMessage, True);
  end;
end;

[Mavarik]

So geht der Q-Trick

Delphi-Quellcode:

  Procedure QWas(ADateTime : TDateTime);
  var  
    LDateTime : TDateTime;
  begin
    LDateTime := ADateTeim;
    TThread.Queue(NIL,Procedure
     begin
       MachWasMit(LDateTime);
     end;
  end;

...
  QWas(DateTime);
...